Class: Terminalwire::V2::Transport::Queue

Inherits:
Object
  • Object
show all
Defined in:
lib/terminalwire/v2/transport/queue.rb

Overview

A queue-backed transport for callback-driven servers (ActionCable, async websocket Rack endpoints, etc.). The endpoint pushes each received frame with #deliver; the blocking Runtime consumes them via #read. Outgoing frames are handed to the sink callable. This bridges an event-loop/callback world to the synchronous server runtime.

Constant Summary collapse

CLOSED = 
Object.new

Instance Method Summary collapse

Constructor Details

#initialize(sink:) ⇒ Queue

Returns a new instance of Queue.



15
16
17
18
19
20
 
# File 'lib/terminalwire/v2/transport/queue.rb', line 15

def initialize(sink:)
  @sink = sink
  @inbox = ::Queue.new
  @mutex = Mutex.new
  @closed = false
end

Instance Method Details

#close ⇒ Object 



36
37
38
39
40
41
42
43
 
# File 'lib/terminalwire/v2/transport/queue.rb', line 36

def close
  @mutex.synchronize do
    next if @closed

    @closed = true
    @inbox << CLOSED
  end
end

#deliver(bytes) ⇒ Object

Called by the websocket endpoint when a frame arrives from the client.



23
24
25
 
# File 'lib/terminalwire/v2/transport/queue.rb', line 23

def deliver(bytes)
  @mutex.synchronize { @inbox << bytes unless @closed }
end

#read ⇒ Object 



27
28
29
30
 
# File 'lib/terminalwire/v2/transport/queue.rb', line 27

def read
  value = @inbox.pop
  value.equal?(CLOSED) ? nil : value
end

#write(bytes) ⇒ Object 



32
33
34
 
# File 'lib/terminalwire/v2/transport/queue.rb', line 32

def write(bytes)
  @sink.call(bytes)
end