Module: YrbLite::ActionCable::Sync

Defined in:
lib/yrb_lite/action_cable/sync.rb

Overview

y-websocket protocol over ActionCable.

Include this module in an ActionCable channel to sync Y.js documents (and awareness/presence) with browser clients. Messages are the standard y-protocols binary messages, base64-encoded in a JSON envelope:

{ "update" => "<base64 bytes>", "id" => 42 } # client -> server
{ "update" => "...", "origin" => "<id>" }    # server -> subscribers
{ "ack" => 42 }                              # server -> sender

Example:

class DocumentChannel < ApplicationCable::Channel
  include YrbLite::ActionCable::Sync

  on_load { |key| Document.find_by(key: key)&.content }
  # on_change blocks run in the channel instance's context, so instance
  # methods (current_user, params, ...) are available without plumbing:
  on_change { |key, update| Document.record!(key, update, by: current_user) }

  def subscribed
    sync_for params[:id]
  end

  def receive(data)
    sync_receive(data)
  end

  def unsubscribed
    sync_unsubscribed
  end
end

The concern is store-backed and fail-closed: every document update is validated against ‘on_load`, recorded through `on_change`, then broadcast. No authoritative document state is kept in ActionCable process memory.

Defined Under Namespace

Modules: ClassMethods

Constant Summary collapse

MSG_KIND_SYNC_STEP1 =

Frame kinds we act on, from Awareness#message_kind. The other codes it can return – 0 (drop: malformed/truncated/multi-message/unknown) and 4 (awareness query) – fall through to a no-op in the dispatch below.

1
MSG_KIND_UPDATE = 
2
MSG_KIND_AWARENESS = 
3
DEFAULT_MAX_FRAME_BYTES =

Default incoming-frame size cap (decoded bytes). Generous enough for a large initial SyncStep2, small enough to bound a single message’s allocation/parse cost. Override per channel with ‘max_frame_bytes`.

8 * 1024 * 1024

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.included(base) ⇒ Object 



56
57
58
 
# File 'lib/yrb_lite/action_cable/sync.rb', line 56

def self.included(base)
  base.extend(ClassMethods)
end

.process_id ⇒ Object

A stable id for this server process, stamped on every broadcast so other processes know to apply it to their replica and this process knows to skip its own. Survives for the life of the process.



293
294
295
 
# File 'lib/yrb_lite/action_cable/sync.rb', line 293

def process_id
  @process_id ||= SecureRandom.hex(8)
end

Instance Method Details

#sync_for(key) ⇒ Object

Call from ‘subscribed`. Streams broadcasts for this document and transmits the server’s opening handshake (SyncStep1 from the store).



101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/yrb_lite/action_cable/sync.rb', line 101

def sync_for(key)
  @sync_key = key.to_s
  @sync_origin = SecureRandom.hex(8)
  sync_require_store_recorder!

  # The document stream is never whisper-enabled; under AnyCable we also
  # subscribe an awareness stream with `whisper: true`, scoping the client-to-
  # client path to ephemeral presence rather than the durable document stream.
  stream_from sync_stream_name
  stream_from sync_awareness_stream_name, whisper: true if respond_to?(:whispers_to)
  sync_transmit(sync_load_doc.sync_step1)
end

#sync_receive(data, key = nil) ⇒ Object

Call from ‘receive`. Applies the client’s message, replies directly when the protocol calls for it, and relays document/awareness changes to the other subscribers.

Reliable delivery: document updates carry an “id”, and the server replies ‘{ “ack” => id }` once the update has been durably recorded. A causally-gapped update is not acked – it gets a resync instead – so the client retransmits until the update lands.



122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/yrb_lite/action_cable/sync.rb', line 122

def sync_receive(data, key = nil)
  # Pass `key` (params[:id]) when your transport doesn't keep the channel
  # instance alive across actions. Under AnyCable each RPC command gets a
  # fresh channel, so instance variables set in `subscribed` are gone here.
  @sync_key = key.to_s if key

  encoded = data.is_a?(Hash) ? data["update"] : nil
  return unless encoded.is_a?(String)

  # Optional client-supplied id for reliable delivery (see sync_send_ack).
  id = data.is_a?(Hash) ? data["id"] : nil

  # Frame-size cap: drop oversized frames before decoding (the encoded form
  # is ~4/3 the decoded size) and again after, so a client can't force large
  # base64 decodes / native parses / merges. A dropped frame is never acked.
  cap = self.class.max_frame_bytes
  return if cap && encoded.bytesize > (cap * 4 / 3) + 4

  begin
    bytes = Base64.strict_decode64(encoded)
  rescue ArgumentError
    return # not valid base64; ignore the frame and keep the connection
  end

  return if cap && bytes.bytesize > cap

  sync_send_ack(id, sync_handle_frame(encoded, bytes))
end

#sync_unsubscribed(key = nil) ⇒ Object

The ‘unsubscribed` hook target. Nothing to clean up: the server keeps no per-connection document or presence state.



153
154
155
 
# File 'lib/yrb_lite/action_cable/sync.rb', line 153

def sync_unsubscribed(key = nil)
  @sync_key = key.to_s if key
end