Class: Libuv::Loop

Inherits:
Object show all
Extended by:
ClassMethods
Includes:
Assertions, Resource
Defined in:
lib/libuv/loop.rb,
lib/libuv/coroutines.rb

Defined Under Namespace

Modules: ClassMethods

Constant Summary collapse

LOOPS = 
ThreadSafe::Cache.new
CRITICAL = 
Mutex.new
@@use_fibers = 
true

Constants included from Assertions

Assertions::MSG_NO_PROC

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from ClassMethods

create, current, default, new

Methods included from Resource

#check_result, #check_result!, #resolve, #to_ptr

Methods included from Assertions

#assert_block, #assert_boolean, #assert_type

Constructor Details

#initialize(pointer) ⇒ Loop

Initialize a loop using an FFI::Pointer to a libuv loop



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/libuv/loop.rb', line 49

def initialize(pointer) # :notnew:
    @pointer = pointer
    @loop = self

    # Create an async call for scheduling work from other threads
    @run_queue = Queue.new
    @process_queue = @loop.async method(:process_queue_cb)
    @process_queue.unref

    # Create a next tick timer
    @next_tick = @loop.timer method(:next_tick_cb)
    @next_tick.unref

    # Create an async call for ending the loop
    @stop_loop = @loop.async method(:stop_cb)
    @stop_loop.unref

    # Libuv can prevent the application shutting down once the main thread has ended
    # The addition of a prepare function prevents this from happening.
    @loop_prep = Libuv::Prepare.new(@loop, method(:noop))
    @loop_prep.unref
    @loop_prep.start
end

Instance Attribute Details

#reactor_thread ⇒ Thread (readonly)

Exposed to allow joining on the thread, when run in a multithreaded environment. Performing other actions on the thread has undefined semantics (read: a dangerous endevor).

Returns:

  • (Thread) 


437
438
439
 
# File 'lib/libuv/loop.rb', line 437

def reactor_thread
  @reactor_thread
end

Instance Method Details

#all(*promises) ⇒ ::Libuv::Q::Promise

Combines multiple promises into a single promise that is resolved when all of the input promises are resolved. (thread safe)

Parameters:

  • *promises (::Libuv::Q::Promise) —

    a number of promises that will be combined into a single promise

Returns:

  • (::Libuv::Q::Promise) —

    Returns a single promise that will be resolved with an array of values, each value corresponding to the promise at the same index in the promises array. If any of the promises is resolved with a rejection, this resulting promise will be resolved with the same rejection.



180
181
182
 
# File 'lib/libuv/loop.rb', line 180

def all(*promises)
    Q.all(@loop, *promises)
end

#any(*promises) ⇒ ::Libuv::Q::Promise

Combines multiple promises into a single promise that is resolved when any of the input promises are resolved.

Parameters:

  • *promises (::Libuv::Q::Promise) —

    a number of promises that will be combined into a single promise

Returns:



190
191
192
 
# File 'lib/libuv/loop.rb', line 190

def any(*promises)
    Q.any(@loop, *promises)
end

#async(callback = nil, &block) ⇒ ::Libuv::Async

Get a new Async handle

Returns:



305
306
307
308
309
310
 
# File 'lib/libuv/loop.rb', line 305

def async(callback = nil, &block)
    callback ||= block
    handle = Async.new(@loop)
    handle.progress callback if callback
    handle
end

#check(callback = nil, &blk) ⇒ ::Libuv::Check

Get a new Check handle

Returns:



290
291
292
 
# File 'lib/libuv/loop.rb', line 290

def check(callback = nil, &blk)
    Check.new(@loop, callback || blk)
end

#defer ⇒ ::Libuv::Q::Deferred

Creates a deferred result object for where the result of an operation may only be returned at some point in the future or is being processed on a different thread (thread safe)

Returns:



168
169
170
 
# File 'lib/libuv/loop.rb', line 168

def defer
    Q.defer(@loop)
end

#file(path, flags = 0, mode = 0) ⇒ ::Libuv::File

Opens a file and returns an object that can be used to manipulate it

Parameters:

  • path (String) —

    the path to the file or folder for watching

  • flags (Integer) (defaults to: 0) —

    see ruby File::Constants

  • mode (Integer) (defaults to: 0)

Returns:



362
363
364
365
366
367
 
# File 'lib/libuv/loop.rb', line 362

def file(path, flags = 0, mode = 0)
    assert_type(String, path, "path must be a String")
    assert_type(Integer, flags, "flags must be an Integer")
    assert_type(Integer, mode, "mode must be an Integer")
    File.new(@loop, path, flags, mode)
end

#filesystem ⇒ ::Libuv::Filesystem

Returns an object for manipulating the filesystem

Returns:



372
373
374
 
# File 'lib/libuv/loop.rb', line 372

def filesystem
    Filesystem.new(@loop)
end

#finally(*promises) ⇒ ::Libuv::Q::Promise

Combines multiple promises into a single promise that is resolved when all of the input promises are resolved or rejected.

Parameters:

  • *promises (::Libuv::Q::Promise) —

    a number of promises that will be combined into a single promise

Returns:

  • (::Libuv::Q::Promise) —

    Returns a single promise that will be resolved with an array of values, each [result, wasResolved] value pair corresponding to a at the same index in the promises array.



201
202
203
 
# File 'lib/libuv/loop.rb', line 201

def finally(*promises)
    Q.finally(@loop, *promises)
end

#fs_event(path) ⇒ ::Libuv::FSEvent

Get a new FSEvent instance

Parameters:

  • path (String) —

    the path to the file or folder for watching

Returns:

Raises:

  • (ArgumentError) —

    if path is not a string



351
352
353
354
 
# File 'lib/libuv/loop.rb', line 351

def fs_event(path)
    assert_type(String, path)
    FSEvent.new(@loop, path)
end

#handle ⇒ Object 



120
 
# File 'lib/libuv/loop.rb', line 120

def handle; @pointer; end

#idle(callback = nil, &block) ⇒ ::Libuv::Idle

Get a new Idle handle

Parameters:

  • callback (Proc) (defaults to: nil) —

    the callback to be called on idle trigger

Returns:



298
299
300
 
# File 'lib/libuv/loop.rb', line 298

def idle(callback = nil, &block)
    Idle.new(@loop, callback || block)
end

#inspect ⇒ Object

Overwrite as errors in jRuby can literally hang VM when inspecting as many many classes will reference this class



115
116
117
 
# File 'lib/libuv/loop.rb', line 115

def inspect
    "#<#{self.class}:0x#{self.__id__.to_s(16)} NT=#{@run_queue.length}>"
end

#log(level, id, *args) ⇒ Object

Notifies the loop there was an event that should be logged

Parameters:

  • level (Symbol) —

    the error level (info, warn, error etc)

  • id (Object) —

    some kind of identifying information

  • *args (*args) —

    any additional information



418
419
420
 
# File 'lib/libuv/loop.rb', line 418

def log(level, id, *args)
    @loop_notify.notify(level, id, *args)
end

#lookup(hostname, hint = :IPv4, port = 9, &block) ⇒ ::Libuv::Dns

Lookup a hostname

Parameters:

  • hostname (String) —

    the domain name to lookup

  • port (Integer, String) (defaults to: 9) —

    the service being connected too

  • callback (Proc) —

    the callback to be called on success

Returns:



340
341
342
343
344
 
# File 'lib/libuv/loop.rb', line 340

def lookup(hostname, hint = :IPv4, port = 9, &block)
    dns = Dns.new(@loop, hostname, port, hint)    # Work is a promise object
    dns.then block if block_given?
    dns
end

#lookup_error(err) ⇒ ::Libuv::Error

Lookup an error code and return is as an error object

Parameters:

  • err (Integer) —

    The error code to look up.

Returns:



224
225
226
227
228
229
230
231
232
233
234
235
236
237
 
# File 'lib/libuv/loop.rb', line 224

def lookup_error(err)
    name = ::Libuv::Ext.err_name(err)

    if name
        msg  = ::Libuv::Ext.strerror(err)
        ::Libuv::Error.const_get(name.to_sym).new(msg)
    else
        # We want a back-trace in this case
        raise "error lookup failed for code #{err}"
    end
rescue Exception => e
    @loop.log :warn, :error_lookup_failed, e
    e
end

#next_tick(callback = nil, &block) ⇒ Object

Queue some work to be processed in the next iteration of the event loop (thread safe)

Parameters:

  • callback (Proc) (defaults to: nil) —

    the callback to be called on the reactor thread

Raises:

  • (ArgumentError) —

    if block is not given



396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
 
# File 'lib/libuv/loop.rb', line 396

def next_tick(callback = nil, &block)
    callback ||= block
    assert_block(callback)

    @run_queue << callback
    if reactor_thread?
        # Create a next tick timer
        if not @next_tick_scheduled
            @next_tick.start(0)
            @next_tick_scheduled = true
            @next_tick.ref
        end
    else
        @process_queue.call
    end
end

#notifier ⇒ ::Libuv::Q::Promise

Provides a promise notifier for receiving un-handled exceptions

Returns:



160
161
162
 
# File 'lib/libuv/loop.rb', line 160

def notifier
    @loop_notify.promise
end

#now ⇒ Fixnum

Get current time in milliseconds

Returns:

  • (Fixnum) 


216
217
218
 
# File 'lib/libuv/loop.rb', line 216

def now
    ::Libuv::Ext.now(@pointer)
end

#pipe(ipc = false) ⇒ ::Libuv::Pipe

Get a new Pipe instance

Parameters:

  • ipc (true, false) (defaults to: false) —

    indicate if a handle will be used for ipc, useful for sharing tcp socket between processes

Returns:



268
269
270
 
# File 'lib/libuv/loop.rb', line 268

def pipe(ipc = false)
    Pipe.new(@loop, ipc)
end

#prepare(callback = nil, &blk) ⇒ ::Libuv::Prepare

Get a new Prepare handle

Returns:



283
284
285
 
# File 'lib/libuv/loop.rb', line 283

def prepare(callback = nil, &blk)
    Prepare.new(@loop, callback || blk)
end

#reactor_running? ⇒ Boolean

Tells you whether the Libuv reactor loop is currently running.

Returns:

  • (Boolean) 


442
443
444
 
# File 'lib/libuv/loop.rb', line 442

def reactor_running?
    !@reactor_thread.nil?
end

#reactor_thread? ⇒ Boolean

True if the calling thread is the same thread as the reactor.

Returns:

  • (Boolean) 


430
431
432
 
# File 'lib/libuv/loop.rb', line 430

def reactor_thread?
    @reactor_thread == Thread.current
end

#run(run_type = :UV_RUN_DEFAULT) {|promise| ... } ⇒ Object

Run the actual event loop. This method will block until the loop is stopped.

Parameters:

  • run_type (:UV_RUN_DEFAULT, :UV_RUN_ONCE, :UV_RUN_NOWAIT) (defaults to: :UV_RUN_DEFAULT)

Yield Parameters:

  • promise (::Libuv::Q::Promise) —

    Yields a promise that can be used for logging unhandled exceptions on the loop.



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
 
# File 'lib/libuv/loop.rb', line 127

def run(run_type = :UV_RUN_DEFAULT)
    if @reactor_thread.nil?
        @loop_notify = @loop.defer

        begin
            @reactor_thread = Thread.current
            LOOPS[@reactor_thread] = @loop
            if block_given?
                if @@use_fibers
                    Fiber.new { yield @loop_notify.promise }.resume
                else
                    yield @loop_notify.promise
                end
            end
            ::Libuv::Ext.run(@pointer, run_type)  # This is blocking
        ensure
            @reactor_thread = nil
            @run_queue.clear
        end
    elsif block_given?
        if @@use_fibers
            schedule { Fiber.new { yield @loop_notify.promise }.resume }
        else
            schedule { yield @loop_notify.promise }
        end
    end
    @loop
end

#schedule(callback = nil, &block) ⇒ Object

Schedule some work to be processed on the event loop as soon as possible (thread safe)

Parameters:

  • callback (Proc) (defaults to: nil) —

    the callback to be called on the reactor thread

Raises:

  • (ArgumentError) —

    if block is not given



380
381
382
383
384
385
386
387
388
389
390
 
# File 'lib/libuv/loop.rb', line 380

def schedule(callback = nil, &block)
    callback ||= block
    assert_block(callback)

    if reactor_thread?
        callback.call
    else
        @run_queue << callback
        @process_queue.call
    end
end

#signal(signum = nil, callback = nil, &block) ⇒ ::Libuv::Signal

Get a new signal handler

Returns:



315
316
317
318
319
320
321
 
# File 'lib/libuv/loop.rb', line 315

def signal(signum = nil, callback = nil, &block)
    callback ||= block
    handle = Signal.new(@loop)
    handle.progress callback if callback
    handle.start(signum) if signum
    handle
end

#stop ⇒ Object

Closes handles opened by the loop class and completes the current loop iteration (thread safe)



423
424
425
 
# File 'lib/libuv/loop.rb', line 423

def stop
    @stop_loop.call
end

#tcp ⇒ ::Libuv::TCP

Get a new TCP instance

Returns:



242
243
244
 
# File 'lib/libuv/loop.rb', line 242

def tcp
    TCP.new(@loop)
end

#timer(callback = nil, &blk) ⇒ ::Libuv::Timer

Get a new timer instance

Parameters:

  • callback (Proc) (defaults to: nil) —

    the callback to be called on timer trigger

Returns:



276
277
278
 
# File 'lib/libuv/loop.rb', line 276

def timer(callback = nil, &blk)
    Timer.new(@loop, callback || blk)
end

#tty(fileno, readable = false) ⇒ ::Libuv::TTY

Get a new TTY instance

Parameters:

  • fileno (Integer) —

    Integer file descriptor of a tty device

  • readable (true, false) (defaults to: false) —

    Boolean indicating if TTY is readable

Returns:



258
259
260
261
262
 
# File 'lib/libuv/loop.rb', line 258

def tty(fileno, readable = false)
    assert_type(Integer, fileno, "io#fileno must return an integer file descriptor, #{fileno.inspect} given")

    TTY.new(@loop, fileno, readable)
end

#udp ⇒ ::Libuv::UDP

Get a new UDP instance

Returns:



249
250
251
 
# File 'lib/libuv/loop.rb', line 249

def udp
    UDP.new(@loop)
end

#update_time ⇒ Object

forces loop time update, useful for getting more granular times

Returns:

  • nil



209
210
211
 
# File 'lib/libuv/loop.rb', line 209

def update_time
    ::Libuv::Ext.update_time(@pointer)
end

#work(callback = nil, &block) ⇒ ::Libuv::Work

Queue some work for processing in the libuv thread pool

Parameters:

  • callback (Proc) (defaults to: nil) —

    the callback to be called in the thread pool

Returns:

Raises:

  • (ArgumentError) —

    if block is not given



328
329
330
331
332
 
# File 'lib/libuv/loop.rb', line 328

def work(callback = nil, &block)
    callback ||= block
    assert_block(callback)
    Work.new(@loop, callback)    # Work is a promise object
end