Class: Rage::Cable::Protocols::Base
- Inherits:
-
Object
- Object
- Rage::Cable::Protocols::Base
- Defined in:
- lib/rage/cable/protocols/base.rb
Overview
A protocol defines the structure, rules and semantics for exchanging data between the client and the server. A protocol class should inherit from Base and implement the following methods:
on_openon_messageserializeprotocol_id
The optional methods are:
protocol_definitionsupports_rpc?on_shutdownon_closeon_remote_disconnect
Direct Known Subclasses
Class Method Summary collapse
-
.broadcast(name, data) ⇒ Object
Broadcast data to all clients connected to a stream.
- .init ⇒ Object
-
.on_remote_disconnect(connection, options) ⇒ Object
Process the disconnect initiated by the server.
- .protocol_definition ⇒ Object
-
.subscribe(connection, name, params) ⇒ Object
Subscribe to a stream.
-
.supports_rpc? ⇒ Boolean
Whether the protocol allows remote procedure calls.
-
.unsubscribe(connection, name, params) ⇒ Object
Unsubscribe from a stream.
Class Method Details
.broadcast(name, data) ⇒ Object
Broadcast data to all clients connected to a stream.
79 80 81 82 83 |
# File 'lib/rage/cable/protocols/base.rb', line 79 def broadcast(name, data) @subscription_identifiers[name].each do |params| ::Iodine.publish("cable:#{protocol_id}:#{name}:#{stream_id(params)}", serialize(params, data)) end end |
.init ⇒ Object
27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 |
# File 'lib/rage/cable/protocols/base.rb', line 27 def init # Hash<String(stream name) => Set<Hash>(subscription params)> @subscription_identifiers = Hash.new { |hash, key| hash[key] = Set.new } @sync_channel = "cable:#{protocol_id}:synchronize" Iodine.on_state(:pre_start) do # this is a fallback to synchronize subscription identifiers across different worker processes; # we expect connections to be distributed among all workers, so this code will almost never be called; # we also synchronize subscriptions with the master process so that the forks that are spun up instead # of the crashed ones also had access to the identifiers; Iodine.subscribe(@sync_channel) do |_, subscription_msg| stream_name, params = Rage::ParamsParser.json_parse(subscription_msg) @subscription_identifiers[stream_name] << params end end Iodine.on_state(:on_finish) do Iodine.unsubscribe(@sync_channel) end end |
.on_remote_disconnect(connection, options) ⇒ Object
Process the disconnect initiated by the server.
96 97 |
# File 'lib/rage/cable/protocols/base.rb', line 96 def on_remote_disconnect(connection, ) end |
.protocol_definition ⇒ Object
48 49 50 |
# File 'lib/rage/cable/protocols/base.rb', line 48 def protocol_definition HANDSHAKE_HEADERS end |
.subscribe(connection, name, params) ⇒ Object
Subscribe to a stream.
57 58 59 60 61 62 63 64 |
# File 'lib/rage/cable/protocols/base.rb', line 57 def subscribe(connection, name, params) connection.subscribe("cable:#{protocol_id}:#{name}:#{stream_id(params)}") unless @subscription_identifiers[name].include?(params) @subscription_identifiers[name] << params ::Iodine.publish(@sync_channel, [name, params].to_json) end end |
.supports_rpc? ⇒ Boolean
Whether the protocol allows remote procedure calls.
88 89 90 |
# File 'lib/rage/cable/protocols/base.rb', line 88 def supports_rpc? true end |
.unsubscribe(connection, name, params) ⇒ Object
Unsubscribe from a stream.
71 72 73 |
# File 'lib/rage/cable/protocols/base.rb', line 71 def unsubscribe(connection, name, params) connection.unsubscribe("cable:#{protocol_id}:#{name}:#{stream_id(params)}") end |