Class: Terminalwire::V2::Server::Runtime

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

Overview

Drives a Server::Connection over a transport. A background read pump continuously drains incoming frames and routes them: responses go to the caller blocked in #request, and unsolicited control frames (resize, and later interrupt) update state / fire callbacks. This is what lets the server always know the client's terminal size, not just while it happens to be inside a request.

Threading: the pump runs on its own thread; the CLI runs on the caller's thread and blocks in #request on a per-stream queue the pump fulfills.

Constant Summary collapse

MAX_FRAME =

Largest payload in a single output data frame; actual frame size is min(this, available flow credit).

32 * 1024

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(transport:, server_min: Protocol::MIN_VERSION, server_max: Protocol::MAX_VERSION, server_capabilities: Protocol::CAPABILITIES) ⇒ Runtime

Returns a new instance of Runtime.



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 21

def initialize(transport:,
               server_min: Protocol::MIN_VERSION,
               server_max: Protocol::MAX_VERSION,
               server_capabilities: Protocol::CAPABILITIES)
  @transport = transport
  @connection = Connection.new(
    server_min: server_min, server_max: server_max,
    server_capabilities: server_capabilities
  )
  @terminal = Terminal.new
  @flow = FlowController.new
  @client_window = Protocol::DEFAULT_WINDOW
  @waiters = {}
  @raw_inputs = {}
  @lock = Mutex.new
  @ready = Queue.new
  @signaled = false
  @on_resize = nil
  @interrupted = false
end

Instance Attribute Details

#connection ⇒ Object (readonly)

Returns the value of attribute connection.



19
20
21
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 19

def connection
  @connection
end

#entitlement ⇒ Object (readonly)

Returns the value of attribute entitlement.



19
20
21
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 19

def entitlement
  @entitlement
end

#program ⇒ Object (readonly)

Returns the value of attribute program.



19
20
21
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 19

def program
  @program
end

#terminal ⇒ Object (readonly)

Returns the value of attribute terminal.



19
20
21
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 19

def terminal
  @terminal
end

Instance Method Details

#close ⇒ Object

Stop the pump and release the transport.



152
153
154
155
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 152

def close
  @transport.close
  @pump&.join(2)
end

#close_output(sid) ⇒ Object 



94
95
96
97
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 94

def close_output(sid)
  emit(@connection.close_stream(sid))
  @flow.close(sid)
end

#close_raw_input(sid) ⇒ Object 



125
126
127
128
129
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 125

def close_raw_input(sid)
  emit(@connection.close_stream(sid))
  queue = @lock.synchronize { @raw_inputs.delete(sid) }
  queue&.push(:closed) # unblock a pending read_raw
end

#emit(frame) ⇒ Object

Fire-and-forget: write a single control frame (welcome, exit, request, open/close). NOT flow-controlled — these are small control-plane frames.



61
62
63
64
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 61

def emit(frame)
  @transport.write(Codec.encode(frame))
  nil
end

#handshake ⇒ Object

Start the read pump and block until the handshake reaches ready (or fails). The calling thread is the CLI thread; an interrupt signal is raised into it.



50
51
52
53
54
55
56
57
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 50

def handshake
  @cli_thread = Thread.current
  @pump = Thread.new { pump }
  result = @ready.pop
  raise result if result.is_a?(Exception)

  self
end

#on_resize(&block) ⇒ Object

Register a callback fired (on the pump thread) whenever the client's window resizes. The Terminal is already updated before it runs.



44
45
46
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 44

def on_resize(&block)
  @on_resize = block
end

#open_output(stream) ⇒ Object

Open an output stream (:stdout/:stderr) and start its flow window at the client's advertised offer. Returns the stream id.



68
69
70
71
72
73
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 68

def open_output(stream)
  sid, frame = @connection.open_stream(stream)
  @flow.open(sid, @client_window)
  emit(frame)
  sid
end

#open_raw_input(mode: Protocol::Mode::RAW) ⇒ Object

Open a raw input stream: the client puts its terminal in mode (raw or cbreak) and streams keystrokes as data frames until we close it, restoring the prior mode on close. Returns the stream id.



102
103
104
105
106
107
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 102

def open_raw_input(mode: Protocol::Mode::RAW)
  sid, frame = @connection.open_stream(Protocol::Stream::STDIN_RAW, mode: mode)
  @lock.synchronize { @raw_inputs[sid] = Queue.new }
  emit(frame)
  sid
end

#read_raw(sid) ⇒ Object

Read the next keystroke chunk from a raw input stream; blocks until input arrives, returns nil when the stream is closed or the connection dies.

Raises:



111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 111

def read_raw(sid)
  queue = @lock.synchronize { @raw_inputs[sid] }
  return nil unless queue

  value = queue.pop
  # An interrupt and the connection's :closed both unblock this pop and can
  # race; the interrupt is the user's intent, so it wins (-> exit 130). This
  # makes the outcome deterministic regardless of which arrives first (the
  # async/Falcon bridge could let :closed land before Thread#raise lands).
  raise Interrupted if interrupted?

  value == :closed ? nil : value
end

#request(resource, method, params = {}) ⇒ Object

Synchronous resource call: register a waiter, write the request, and block until the pump delivers the correlated response (or the connection dies).

Raises:



133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 133

def request(resource, method, params = {})
  sid, frame = @connection.call(resource, method, params)
  waiter = Queue.new
  @lock.synchronize { @waiters[sid] = waiter }
  emit(frame)

  answer = waiter.pop
  # Interrupt wins over a racing connection-closed failure (see read_raw).
  raise Interrupted if interrupted?
  raise answer if answer.is_a?(Exception)

  unless answer[:ok]
    error = answer[:error] || {}
    raise ResponseError.new(error["code"] || "internal", error["message"] || "request failed")
  end
  answer[:value]
end

#write_data(sid, bytes) ⇒ Object

Write output to a stream, flow-controlled: each frame is sized to the currently available credit (blocking when the window is empty), so the server can never outrun the client. Raises if the connection dies.



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/terminalwire/v2/server/runtime.rb', line 78

def write_data(sid, bytes)
  bytes = bytes.b
  total = bytes.bytesize
  if total.zero?
    emit(Frames.data(sid: sid, bytes: "".b))
    return
  end

  offset = 0
  while offset < total
    take = @flow.reserve(sid, [total - offset, MAX_FRAME].min)
    emit(Frames.data(sid: sid, bytes: bytes.byteslice(offset, take)))
    offset += take
  end
end