Class: Plushie::Bridge

Inherits:
Object
  • Object
show all
Defined in:
lib/plushie/bridge.rb,
sig/plushie/bridge.rbs

Overview

Renderer lifecycle manager.

Wraps a Connection with restart logic. When the renderer crashes, the Bridge reconnects with exponential backoff and notifies the Runtime via the event queue. The Runtime owns the resync flow: it re-sends settings, renders a fresh snapshot, and re-syncs subscriptions after a successful restart.

The Bridge pushes decoded events to the Runtime's event queue:

  • [:renderer_event, msg] for normal protocol messages
  • [:renderer_exited, reason] when the connection drops
  • [:renderer_restarted] after a successful reconnect

Constant Summary collapse

SDK_LOG_LEVELS =

Returns:

  • (Hash[Symbol, Integer]) 
{
  off: Logger::UNKNOWN,
  error: Logger::ERROR,
  warning: Logger::WARN,
  warn: Logger::WARN,
  info: Logger::INFO,
  debug: Logger::DEBUG,
  trace: Logger::DEBUG
}.freeze
DEFAULT_LOG_LEVEL =

Returns:

  • (Object) 
Object.new.freeze
BACKOFF_BASE_MS =

Exponential backoff parameters. Shared with the other host SDKs (Elixir, Rust, Gleam, Python, TypeScript) so renderer restart behavior is consistent across implementations.

Returns:

  • (Integer) 
100
BACKOFF_MAX_MS =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.

Maximum backoff delay in milliseconds.

Returns:

  • (Integer) 
5000
MAX_RETRIES =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.

Maximum retry attempts before giving up.

Returns:

  • (Integer) 
5
DEFAULT_HEARTBEAT_INTERVAL =

Default watchdog interval in seconds. Set to nil to disable.

30

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(event_queue:, format: :msgpack, binary: nil, transport: :spawn, log_level: DEFAULT_LOG_LEVEL, token: nil, heartbeat_interval: DEFAULT_HEARTBEAT_INTERVAL) ⇒ Bridge

Returns a new instance of Bridge.

Parameters:

  • heartbeat_interval (Numeric, nil) (defaults to: DEFAULT_HEARTBEAT_INTERVAL) —

    max seconds between renderer messages before triggering a restart. nil disables the watchdog.



61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/plushie/bridge.rb', line 61

def initialize(event_queue:, format: :msgpack, binary: nil,
  transport: :spawn, log_level: DEFAULT_LOG_LEVEL, token: nil,
  heartbeat_interval: DEFAULT_HEARTBEAT_INTERVAL)
  @event_queue = event_queue
  @format = format
  @binary = binary
  @transport = transport
  @log_level = renderer_log_level(log_level)
  @token = token
  @connection = nil
  @retry_count = 0
  @settings = {}
  @heartbeat_interval = heartbeat_interval
  @heartbeat_timer = nil
  @forwarder_thread = nil
  @conn_queue = nil
  @logger = Logger.new($stderr, level: sdk_log_level(log_level), progname: "plushie")
end

Instance Attribute Details

#format ⇒ :msgpack, :json (readonly)

Returns wire format.

Returns:

  • (:msgpack, :json) —

    wire format



44
45
46
 
# File 'lib/plushie/bridge.rb', line 44

def format
  @format
end

#hello ⇒ Hash? (readonly)

Returns hello response from the current connection.

Returns:

  • (Hash, nil) —

    hello response from the current connection



47
48
49
 
# File 'lib/plushie/bridge.rb', line 47

def hello
  @hello
end

Instance Method Details

#attempt_restart

This method returns an undefined value.



204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
 
# File 'lib/plushie/bridge.rb', line 204

def attempt_restart
  return if @retry_count >= MAX_RETRIES

  @retry_count += 1
  delay_ms = [BACKOFF_BASE_MS * (2**(@retry_count - 1)), BACKOFF_MAX_MS].min
  @logger.warn("plushie: renderer exited, retry #{@retry_count}/#{MAX_RETRIES} in #{delay_ms}ms")

  sleep(delay_ms / 1000.0)
  @connection&.close

  return unless connect!

  # Notify the runtime that the renderer is back. The runtime owns
  # the resync flow: re-send settings, render, sync subscriptions.
  BoundedQueue.push(@event_queue, [:renderer_restarted])
rescue => e
  @logger.error("plushie: restart failed: #{e.class}: #{e.message}")
  BoundedQueue.push(@event_queue, [:renderer_exited, e]) if @retry_count >= MAX_RETRIES
end

#cancel_heartbeat_timer

This method returns an undefined value.



259
260
261
262
 
# File 'lib/plushie/bridge.rb', line 259

def cancel_heartbeat_timer
  stop_thread(@heartbeat_timer, timeout: 1)
  @heartbeat_timer = nil
end

#connect!

This method returns an undefined value.



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
 
# File 'lib/plushie/bridge.rb', line 127

def connect!
  queue = BoundedQueue.new(BoundedQueue::CONNECTION_CAPACITY)
  settings = @token ? @settings.merge(token: @token) : @settings

  @connection = case @transport
  when :spawn
    Connection.spawn(
      format: @format, binary: @binary,
      mode: nil, log_level: @log_level,
      settings: settings, queue: queue
    )
  when :stdio
    Connection.attach(
      stdin: $stdout, stdout: $stdin,
      format: @format, settings: settings, queue: queue
    )
  when Array
    kind, adapter = @transport
    raise ArgumentError, "unsupported transport tuple: #{@transport.inspect}" unless kind == :iostream
    Connection.iostream(
      adapter: adapter, format: @format,
      settings: settings, queue: queue
    )
  else
    raise ArgumentError, "unsupported transport: #{@transport.inspect}"
  end

  @hello = @connection.hello
  check_renderer_version(@hello)
  @retry_count = 0

  # Forward messages from connection queue to event queue
  start_forwarder(queue)
  true
rescue => e
  handle_connect_failure(e)
  false
end

#handle_connect_failure(error)

This method returns an undefined value.

Parameters:

  • error (Exception) 


272
273
274
275
 
# File 'lib/plushie/bridge.rb', line 272

def handle_connect_failure(error)
  @logger.error("plushie: connection failed: #{error.class}: #{error.message}")
  BoundedQueue.push(@event_queue, [:renderer_exited, error])
end

#renderer_log_level(level) ⇒ Symbol

Parameters:

  • level (Object)

Returns:

  • (Symbol) 


172
173
174
175
176
 
# File 'lib/plushie/bridge.rb', line 172

def renderer_log_level(level)
  return :error if level.equal?(DEFAULT_LOG_LEVEL)

  level
end

#sdk_log_level(level) ⇒ Integer

Parameters:

  • level (Object)

Returns:

  • (Integer) 


166
167
168
169
170
 
# File 'lib/plushie/bridge.rb', line 166

def sdk_log_level(level)
  return Logger::WARN if level.equal?(DEFAULT_LOG_LEVEL)

  SDK_LOG_LEVELS.fetch(level, Logger::ERROR)
end

#send_encoded(data)

This method returns an undefined value.

Send pre-encoded wire bytes to the renderer. Thread-safe.

Parameters:

  • data (String) —

    encoded message bytes



91
92
93
 
# File 'lib/plushie/bridge.rb', line 91

def send_encoded(data)
  @connection&.send_encoded(data)
end

#send_register_effect_stub(kind, response)

This method returns an undefined value.

Register an effect stub with the renderer.

Parameters:

  • kind (String) —

    effect kind

  • response (Object) —

    canned response



99
100
101
102
103
 
# File 'lib/plushie/bridge.rb', line 99

def send_register_effect_stub(kind, response)
  @connection&.send_encoded(
    Protocol::Encode.encode_register_effect_stub(kind, response, @format)
  )
end

#send_unregister_effect_stub(kind)

This method returns an undefined value.

Remove a previously registered effect stub.

Parameters:

  • kind (String) —

    effect kind



108
109
110
111
112
 
# File 'lib/plushie/bridge.rb', line 108

def send_unregister_effect_stub(kind)
  @connection&.send_encoded(
    Protocol::Encode.encode_unregister_effect_stub(kind, @format)
  )
end

#start(settings: {})

This method returns an undefined value.

Start the connection and perform handshake.

Parameters:

  • settings (Hash) (defaults to: {}) —

    application settings to send

  • settings: (Hash[Symbol, untyped]) (defaults to: {}) 


83
84
85
86
 
# File 'lib/plushie/bridge.rb', line 83

def start(settings: {})
  @settings = settings
  connect!
end

#start_forwarder(conn_queue) ⇒ Thread

Parameters:

  • conn_queue (Thread::Queue)

Returns:

  • (Thread) 


178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
 
# File 'lib/plushie/bridge.rb', line 178

def start_forwarder(conn_queue)
  @conn_queue = conn_queue
  reset_heartbeat_timer
  @forwarder_thread = Thread.new do
    while (msg = conn_queue.pop)
      begin
        case msg
        in {type: :connection_closed} | {type: :connection_error}
          cancel_heartbeat_timer
          BoundedQueue.push(@event_queue, [:renderer_exited, msg])
          attempt_restart
          break
        else
          reset_heartbeat_timer
          BoundedQueue.push(@event_queue, [:renderer_event, msg])
        end
      rescue => e
        @logger.warn("plushie: bridge forwarder message error: #{e.class}: #{e.message}")
      end
    end
  rescue => e
    cancel_heartbeat_timer
    BoundedQueue.push(@event_queue, [:renderer_exited, e])
  end.tap { |t| t.name = "plushie-bridge-forwarder" }
end

#stop

This method returns an undefined value.

Stop the connection and clean up.



115
116
117
118
119
120
121
122
123
 
# File 'lib/plushie/bridge.rb', line 115

def stop
  cancel_heartbeat_timer
  stop_thread(@forwarder_thread, timeout: 1)
  @forwarder_thread = nil
  @conn_queue&.close
  @connection&.close
  @connection = nil
  @retry_count = 0
end

#stop_thread(thread, timeout:)

This method returns an undefined value.

Parameters:

  • thread (Thread, nil)
  • timeout: (Numeric) 


264
265
266
267
268
269
270
 
# File 'lib/plushie/bridge.rb', line 264

def stop_thread(thread, timeout:)
  return unless thread
  return if thread == Thread.current

  thread.kill
  thread.join(timeout)
end