Class: Terminalwire::V2::Server::Rack::ThreadBridge

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

Overview

The streaming body for a threaded server. #call(stream) gets the raw blocking socket after the 101; it runs the CLI on its own thread and pumps the socket on another, then returns so the web server can reuse the worker.

Instance Method Summary collapse

Constructor Details

#initialize(handler, request: {}) ⇒ ThreadBridge

Returns a new instance of ThreadBridge.



274
275
276
277
 
# File 'lib/terminalwire/v2/server/rack.rb', line 274

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

Instance Method Details

#call(stream) ⇒ Object

:nocov: blocking-socket threading — exercised live by the conformance suite, not units.



280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
 
# File 'lib/terminalwire/v2/server/rack.rb', line 280

def call(stream)
  write_lock = Mutex.new
  parser = Parser.new
  transport = Transport::Queue.new(
    # Serialize writes: the client forbids concurrent writes, and the
    # runtime emits from more than one thread.
    sink: ->(bytes) { write_lock.synchronize { stream.write(Frame.binary(bytes)) } }
  )

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

  Thread.new do
    loop do
      parser.push(stream.readpartial(4096)) do |opcode, payload|
        case opcode
        when 0x1, 0x2 then transport.deliver(payload)        # protocol frame
        when 0x9 then write_lock.synchronize { stream.write(Frame.pong(payload)) }
        when 0x8 then raise EOFError                          # client close
        end
      end
    end
  rescue EOFError, IOError, Errno::ECONNRESET, Errno::EPIPE
    # client went away — normal end
  ensure
    transport.close
    cli.join(2)
    write_lock.synchronize { stream.write(Frame::CLOSE) rescue nil }
    stream.close rescue nil
  end
end