Class: ActionCable::Connection::StreamEventLoop
- Inherits:
-
Object
- Object
- ActionCable::Connection::StreamEventLoop
- Defined in:
- lib/action_cable/connection/stream_event_loop.rb
Instance Method Summary collapse
- #attach(io, stream) ⇒ Object
- #detach(io, stream) ⇒ Object
-
#initialize ⇒ StreamEventLoop
constructor
A new instance of StreamEventLoop.
- #post(task = nil, &block) ⇒ Object
- #stop ⇒ Object
- #timer(interval, &block) ⇒ Object
Constructor Details
#initialize ⇒ StreamEventLoop
Returns a new instance of StreamEventLoop.
7 8 9 10 11 12 13 14 |
# File 'lib/action_cable/connection/stream_event_loop.rb', line 7 def initialize @nio = @thread = nil @map = {} @stopping = false @todo = Queue.new @spawn_mutex = Mutex.new end |
Instance Method Details
#attach(io, stream) ⇒ Object
26 27 28 29 30 31 32 |
# File 'lib/action_cable/connection/stream_event_loop.rb', line 26 def attach(io, stream) @todo << lambda do @map[io] = stream @nio.register(io, :r) end wakeup end |
#detach(io, stream) ⇒ Object
34 35 36 37 38 39 40 |
# File 'lib/action_cable/connection/stream_event_loop.rb', line 34 def detach(io, stream) @todo << lambda do @nio.deregister io @map.delete io end wakeup end |
#post(task = nil, &block) ⇒ Object
20 21 22 23 24 |
# File 'lib/action_cable/connection/stream_event_loop.rb', line 20 def post(task = nil, &block) task ||= block Concurrent.global_io_executor << task end |
#stop ⇒ Object
42 43 44 45 |
# File 'lib/action_cable/connection/stream_event_loop.rb', line 42 def stop @stopping = true wakeup if @nio end |
#timer(interval, &block) ⇒ Object
16 17 18 |
# File 'lib/action_cable/connection/stream_event_loop.rb', line 16 def timer(interval, &block) Concurrent::TimerTask.new(execution_interval: interval, &block).tap(&:execute) end |