Module: Ruflet::ConnectionProtocol

Defined in:
lib/ruflet/server/connection_protocol.rb

Overview

Transport-agnostic implementation of the Ruflet wire protocol: one connection loop shared by every server that speaks to Flutter clients.

The standalone TCP server (Ruflet::Server) and host-server adapters such as ruflet_rails' Rack-hijack endpoint include this module and provide only their transport plus the integration hooks below — the protocol itself is never reimplemented.

Includers must initialize:

@app_block       — proc invoked with the Page on first registration
@sessions        — Hash mapping connection key => Page
@sessions_mutex  — Mutex guarding @sessions

Instance Method Summary collapse

Instance Method Details

#attach_sender(page, ws) ⇒ Object 



255
256
257
 
# File 'lib/ruflet/server/connection_protocol.rb', line 255

def attach_sender(page, ws)
  page.instance_variable_set(:@sender, sender_for(ws))
end

#before_dispatch_event(ws, event) ⇒ Object

Called before a control event is dispatched to the Page.



40
 
# File 'lib/ruflet/server/connection_protocol.rb', line 40

def before_dispatch_event(ws, event); end

#close_connection(ws) ⇒ Object 



72
73
74
75
76
77
78
 
# File 'lib/ruflet/server/connection_protocol.rb', line 72

def close_connection(ws)
  return unless ws

  remove_session(ws)
  connection_closed(ws)
  ws.close
end

#connection_closed(ws) ⇒ Object

Called when a connection leaves the protocol loop.



25
 
# File 'lib/ruflet/server/connection_protocol.rb', line 25

def connection_closed(ws); end

#connection_opened(ws) ⇒ Object

Called when a connection enters the protocol loop.



22
 
# File 'lib/ruflet/server/connection_protocol.rb', line 22

def connection_opened(ws); end

#decode_incoming(raw) ⇒ Object 



109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/ruflet/server/connection_protocol.rb', line 109

def decode_incoming(raw)
  parsed = normalize_incoming(Ruflet::WireCodec.unpack(raw.to_s.b))

  if parsed.is_a?(Array) && parsed.length >= 2
    return [parsed[0], parsed[1]]
  end

  if parsed.is_a?(Hash)
    action = parsed["action"] || parsed[:action]
    payload = parsed["payload"] || parsed[:payload]
    return [action, payload] unless action.nil?

    if (parsed.key?("target") || parsed.key?(:target)) && (parsed.key?("name") || parsed.key?(:name))
      return [Protocol::ACTIONS[:control_event], parsed]
    end
  end

  raise "Unsupported payload format"
end

#disconnect_error?(error) ⇒ Boolean

Returns:

  • (Boolean) 


259
260
261
262
263
264
265
266
267
268
269
270
 
# File 'lib/ruflet/server/connection_protocol.rb', line 259

def disconnect_error?(error)
  return true if error.is_a?(IOError)
  return true if error.is_a?(Errno::EPIPE)
  return true if error.is_a?(Errno::ECONNRESET)
  return true if error.is_a?(Errno::ECONNABORTED)
  return true if error.is_a?(Errno::ENOTCONN)
  return true if error.is_a?(Errno::ESHUTDOWN)
  return true if error.is_a?(Errno::EBADF)
  return true if error.is_a?(Errno::EINVAL)

  false
end

#fetch_page(ws) ⇒ Object 



212
213
214
215
216
217
 
# File 'lib/ruflet/server/connection_protocol.rb', line 212

def fetch_page(ws)
  page = @sessions_mutex.synchronize { @sessions[ws.session_key] }
  raise "Session not found" unless page

  page
end

#handle_message(ws, raw) ⇒ Object

Protocol core.



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/ruflet/server/connection_protocol.rb', line 84

def handle_message(ws, raw)
  action, payload = decode_incoming(raw)
  payload ||= {}

  warn "incoming action=#{action.inspect}" if ENV["RUFLET_DEBUG"] == "1"

  case action
  when Protocol::ACTIONS[:register_client], Protocol::ACTIONS[:register_web_client]
    on_register_client(ws, payload)
  when Protocol::ACTIONS[:control_event], Protocol::ACTIONS[:page_event_from_web]
    on_control_event(ws, payload)
  when Protocol::ACTIONS[:update_control], Protocol::ACTIONS[:update_control_props]
    on_update_control(ws, payload)
  when Protocol::ACTIONS[:invoke_control_method]
    on_invoke_control_method(ws, payload)
  when Protocol::ACTIONS[:python_output]
    # Flet clients can emit this legacy diagnostic action. It carries no
    # server-side state, but it is a valid wire message and must not be
    # treated as a protocol error or tear down host-adapter connections.
    nil
  else
    raise "Unknown action: #{action.inspect}"
  end
end

#handle_upgraded_socket(io) ⇒ Object

For hosts that already performed the HTTP upgrade (Rack hijack, the embedded runtime, tests with socket pairs).



53
54
55
 
# File 'lib/ruflet/server/connection_protocol.rb', line 53

def handle_upgraded_socket(io)
  run_connection(Ruflet::WebSocketConnection.new(io))
end

#log_connection_error(error) ⇒ Object 



42
43
44
45
 
# File 'lib/ruflet/server/connection_protocol.rb', line 42

def log_connection_error(error)
  warn "server error: #{error.class}: #{error.message}"
  warn error.backtrace.join("\n") if error.backtrace
end

#normalize_event_data(value) ⇒ Object 



227
228
229
230
231
232
233
234
235
236
 
# File 'lib/ruflet/server/connection_protocol.rb', line 227

def normalize_event_data(value)
  case value
  when Hash
    value.each_with_object({}) { |(k, v), out| out[k.to_sym] = normalize_event_data(v) }
  when Array
    value.map { |entry| normalize_event_data(entry) }
  else
    value
  end
end

#normalize_incoming(value) ⇒ Object 



129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/ruflet/server/connection_protocol.rb', line 129

def normalize_incoming(value)
  case value
  when String
    value.dup.force_encoding("UTF-8")
  when Integer, Float, TrueClass, FalseClass, NilClass
    value
  when Symbol
    value.to_s
  when Array
    value.map { |v| normalize_incoming(v) }
  when Hash
    value.each_with_object({}) do |(k, v), out|
      out[k.to_s] = normalize_incoming(v)
    end
  else
    value.to_s
  end
end

#on_control_event(ws, payload) ⇒ Object 



183
184
185
186
187
188
189
190
191
192
193
194
195
 
# File 'lib/ruflet/server/connection_protocol.rb', line 183

def on_control_event(ws, payload)
  event = Protocol.normalize_control_event_payload(payload)
  page = fetch_page(ws)
  return if event["target"].nil? || event["name"].to_s.empty?

  attach_sender(page, ws)
  before_dispatch_event(ws, event)
  page.dispatch_event(
    target: event["target"],
    name: event["name"],
    data: normalize_event_data(event["data"])
  )
end

#on_invoke_control_method(ws, payload) ⇒ Object 



206
207
208
209
210
 
# File 'lib/ruflet/server/connection_protocol.rb', line 206

def on_invoke_control_method(ws, payload)
  page = fetch_page(ws)
  attach_sender(page, ws)
  page.handle_invoke_method_result(Protocol.normalize_invoke_method_result_payload(payload))
end

#on_register_client(ws, payload) ⇒ Object 



148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
 
# File 'lib/ruflet/server/connection_protocol.rb', line 148

def on_register_client(ws, payload)
  normalized = Protocol.normalize_register_payload(payload)
  session_id = normalized["session_id"].to_s.empty? ? pseudo_uuid : normalized["session_id"]

  page = resume_session(session_id)
  first_registration = page.nil?

  if page
    attach_sender(page, ws)
    page.prepare_for_reconnect!
  else
    page = Page.new(
      session_id: session_id,
      client_details: normalized,
      sender: sender_for(ws)
    )
    page.title = "Ruflet App"
  end

  @sessions_mutex.synchronize { @sessions[ws.session_key] = page }
  session_stored(page, ws)

  initial_response = [
    Protocol::ACTIONS[:register_client],
    Protocol.register_response(session_id: session_id)
  ]
  ws.send_binary(Ruflet::WireCodec.pack(initial_response))

  @app_block.call(page) if first_registration
  page.update
rescue StandardError => e
  send_message(ws, Protocol::ACTIONS[:session_crashed], { "message" => e.message.to_s })
  raise
end

#on_update_control(ws, payload) ⇒ Object 



197
198
199
200
201
202
203
204
 
# File 'lib/ruflet/server/connection_protocol.rb', line 197

def on_update_control(ws, payload)
  update = Protocol.normalize_update_control_payload(payload)
  page = fetch_page(ws)
  return if update["id"].nil?

  attach_sender(page, ws)
  page.apply_client_update(update["id"], update["props"] || {})
end

#pseudo_uuid ⇒ Object 



272
273
274
275
276
277
278
279
280
281
282
 
# File 'lib/ruflet/server/connection_protocol.rb', line 272

def pseudo_uuid
  now = Process.clock_gettime(Process::CLOCK_REALTIME, :nanosecond)
  rnd = rand(0..0xffff_ffff)
  "%08x-%04x-%04x-%04x-%012x" % [
    rnd,
    now & 0xffff,
    (now >> 16) & 0xffff,
    (now >> 32) & 0xffff,
    (now >> 48) & 0xffff_ffff_ffff
  ]
end

#remove_session(ws) ⇒ Object 



219
220
221
222
223
224
225
 
# File 'lib/ruflet/server/connection_protocol.rb', line 219

def remove_session(ws)
  return unless ws

  page = @sessions_mutex.synchronize { @sessions.delete(ws.session_key) }
  session_removed(page, ws) if page
  page
end

#resume_session(_session_id) ⇒ Object

Return an existing Page to resume for this session id, or nil to create a fresh one (hosts with a session registry override this).



29
30
31
 
# File 'lib/ruflet/server/connection_protocol.rb', line 29

def resume_session(_session_id)
  nil
end

#run_connection(ws) ⇒ Object 



57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/ruflet/server/connection_protocol.rb', line 57

def run_connection(ws)
  connection_opened(ws)

  while (raw = ws.read_message)
    handle_message(ws, raw)
  end
rescue StandardError => e
  return if disconnect_error?(e)

  log_connection_error(e)
  send_message(ws, Protocol::ACTIONS[:session_crashed], { "message" => e.message.to_s.dup.force_encoding("UTF-8") })
ensure
  close_connection(ws)
end

#send_message(ws, action, payload) ⇒ Object 



238
239
240
241
242
243
244
245
246
247
 
# File 'lib/ruflet/server/connection_protocol.rb', line 238

def send_message(ws, action, payload)
  return if ws.nil? || ws.closed?

  ws.send_binary(Ruflet::WireCodec.pack([action, payload]))
rescue StandardError => e
  log_connection_error(e) unless disconnect_error?(e)
  remove_session(ws)
  connection_closed(ws)
  nil
end

#sender_for(ws) ⇒ Object 



249
250
251
252
253
 
# File 'lib/ruflet/server/connection_protocol.rb', line 249

def sender_for(ws)
  lambda do |action, msg_payload|
    send_message(ws, action, msg_payload)
  end
end

#session_removed(page, ws) ⇒ Object

Called after a Page is removed for a connection.



37
 
# File 'lib/ruflet/server/connection_protocol.rb', line 37

def session_removed(page, ws); end

#session_stored(page, ws) ⇒ Object

Called after a Page is stored for a connection.



34
 
# File 'lib/ruflet/server/connection_protocol.rb', line 34

def session_stored(page, ws); end