Class: Rage::Cable::Protocols::Base

Inherits:
Object
  • Object
show all
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_open
  • on_message
  • serialize
  • protocol_id

The optional methods are:

  • protocol_definition
  • supports_rpc?
  • on_shutdown
  • on_close
  • on_remote_disconnect

Direct Known Subclasses

ActioncableV1Json, RawWebSocketJson

Class Method Summary collapse

Class Method Details

.broadcast(name, data) ⇒ Object

Broadcast data to all clients connected to a stream.

Parameters:

  • name (String) —

    the stream name

  • data (Object) —

    the data to send



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.

Parameters:



96
97
 
# File 'lib/rage/cable/protocols/base.rb', line 96

def on_remote_disconnect(connection, options)
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.

Parameters:

  • connection (Rage::Cable::WebSocketConnection) —

    the connection object

  • name (String) —

    the stream name

  • params (Hash) —

    parameters associated with the client



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.

Returns:

  • (Boolean) 


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.

Parameters:

  • connection (Rage::Cable::WebSocketConnection) —

    the connection object

  • name (String) —

    the stream name

  • params (Hash) —

    parameters associated with the client



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