Class: Terminalwire::V2::Server::Rack::ReactorBridge

Inherits:
Object
  • Object
show all
Defined in:
lib/terminalwire/v2/server/rack.rb

Overview

One WebSocket connection over async-websocket. inbound: connection.read (reactor fiber) -> transport.deliver; outbound: the CLI thread pushes to a Thread::Queue that a writer fiber drains -> connection.

The cross-thread hand-off is a plain Thread::Queue, NOT a self-pipe. A fiber blocked in outbox.pop yields the reactor, and a push from the CLI thread wakes it through the fiber scheduler's own cross-thread wakeup (Async::Scheduler#unblock -> selector.wakeup). We verified this empirically against async 2.39: the reactor keeps running while the writer is parked, and a push from a real OS thread resumes it. An earlier version hand-rolled an IO.pipe to wake the reactor — that just reimplemented selector.wakeup, so it is gone. The CLI runs on a real Thread (not a fiber) on purpose: a user's CLI makes arbitrary blocking calls, which would stall the whole reactor if run as a fiber — Sam's own guidance is to offload blocking work to a thread.

Constant Summary collapse

JOIN_TIMEOUT = 
2

Instance Method Summary collapse

Constructor Details

#initialize(connection, handler, request: {}) ⇒ ReactorBridge

Returns a new instance of ReactorBridge.



332
333
334
335
336
 
# File 'lib/terminalwire/v2/server/rack.rb', line 332

def initialize(connection, handler, request: {})
  @connection = connection
  @handler = handler
  @request = request
end

Instance Method Details

#run ⇒ Object

:nocov: reactor-fiber bridge — exercised live by the conformance suite, not units.



339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
 
# File 'lib/terminalwire/v2/server/rack.rb', line 339

def run
  # Falcon already runs us in a reactor; Sync reuses it (and would create
  # one if absent), giving the connection's fiber I/O a scheduler.
  Sync do |task|
    outbox = ::Queue.new # Thread::Queue: thread-safe + fiber-scheduler aware
    transport = Transport::Queue.new(sink: ->(bytes) { outbox << bytes })

    cli = Thread.new { @handler.call(transport: transport, request: @request) }

    writer = task.async do
      # pop blocks the fiber until the CLI thread pushes (cross-thread wakeup
      # via the scheduler); nil means the outbox was closed in teardown.
      while (bytes = outbox.pop)
        @connection.send_binary(bytes)
        @connection.flush if outbox.empty? # batch: flush once the burst drains
      end
    rescue EOFError, IOError, Errno::EPIPE
      # connection died mid-write
    end

    begin
      while (message = @connection.read)
        transport.deliver(message.buffer.b)
      end
    rescue EOFError, IOError, Errno::ECONNRESET, Errno::EPIPE
      # client disconnected
    end
  ensure
    transport&.close      # unblock the handler's pending reads/requests
    cli&.join(JOIN_TIMEOUT) # let it emit the exit frame (writer drains it meanwhile)
    outbox&.close         # -> writer's pop returns nil -> writer fiber ends
    writer&.wait
  end
end