Class: Rage::Cable::Protocols::ActioncableV1Json

Inherits:
Base
  • Object
show all
Defined in:
lib/rage/cable/protocols/actioncable_v1_json.rb

Overview

This is an implementation of the Action Cable protocol. Clients are expected to use @rails/actioncable to connect to the server.

Examples:

Server side

class TodoItemsChannel
  def subscribed
    stream_from "todo-items"
  end

  def add_item(data)
    puts "Adding Todo item: #{data}"
  end

  def remove_item(data)
    puts "Removing Todo item: #{data}"
  end
end

Client side

import { createConsumer } from '@rails/actioncable'

const cable = createConsumer('ws://localhost:3000/cable')

const channel = cable.subscriptions.create('TodoItemsChannel', {
  connected: () => console.log('connected')
})

channel.perform('add_item', { item: 'New Item' })
channel.perform('remove_item', { item_id: 123 })

See Also:

Defined Under Namespace

Modules: COMMAND, MESSAGES, REASON, TYPE

Constant Summary collapse

HANDSHAKE_HEADERS = 
{ "Sec-WebSocket-Protocol" => "actioncable-v1-json" }

Class Method Summary collapse

Methods inherited from Base

broadcast, subscribe, supports_rpc?, unsubscribe

Class Method Details

.init ⇒ Object

This method serves as a constructor to prepare the object or set up recurring tasks (e.g. heartbeats).



73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/rage/cable/protocols/actioncable_v1_json.rb', line 73

def self.init
  super

  Iodine.on_state(:on_start) do
    ping_counter = Time.now.to_i

    Iodine.run_every(3000) do
      ping_counter += 1
      Iodine.publish("cable:ping", { type: TYPE::PING, message: ping_counter }.to_json, Iodine::PubSub::PROCESS)
    end
  end
end

.on_close(connection, env, router) ⇒ Object

Note:

This method is optional.

The method should process client disconnections and call Router#process_disconnection.

Parameters:



156
157
158
 
# File 'lib/rage/cable/protocols/actioncable_v1_json.rb', line 156

def self.on_close(connection, env, router)
  router.process_disconnection(connection, env)
end

.on_message(connection, env, raw_data, router) ⇒ Object

The method processes messages from existing connections. It should parse the message, call either Router#process_subscription or Router#process_message, and handle its return value.

Parameters:



111
112
113
114
115
116
117
118
119
120
121
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
 
# File 'lib/rage/cable/protocols/actioncable_v1_json.rb', line 111

def self.on_message(connection, env, raw_data, router)
  parsed_data = Rage::ParamsParser.json_parse(raw_data)

  command, identifier = parsed_data[:command], parsed_data[:identifier]
  params = Rage::ParamsParser.json_parse(identifier)

  # process subscription messages
  if command == COMMAND::SUBSCRIBE
    status = router.process_subscription(connection, env, identifier, params[:channel], params)
    if status == :subscribed
      connection.write({ identifier: identifier, type: TYPE::CONFIRM }.to_json)
    elsif status == :rejected
      connection.write({ identifier: identifier, type: TYPE::REJECT }.to_json)
    elsif status == :invalid
      connection.write(MESSAGES::INVALID)
    end

    return
  elsif command == COMMAND::UNSUBSCRIBE
    router.process_unsubscription(connection, env, identifier)
    return
  end

  # process data messages;
  # plain `JSON` is used here to conform with the ActionCable API that passes `data` as a Hash with string keys;
  data = JSON.parse(parsed_data[:data])

  message_status = if command == COMMAND::MESSAGE && data.has_key?("action")
    router.process_message(connection, env, identifier, data["action"].to_sym, data)

  elsif command == COMMAND::MESSAGE
    router.process_message(connection, env, identifier, :receive, data)
  end

  unless message_status == :processed
    connection.write(MESSAGES::INVALID)
  end
end

.on_open(connection, env, router) ⇒ Object

The method is called any time a new WebSocket connection is established. It is expected to call Router#process_connection and handle its return value.

Parameters:



92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/rage/cable/protocols/actioncable_v1_json.rb', line 92

def self.on_open(connection, env, router)
  accepted = router.process_connection(connection, env)

  if accepted
    connection.subscribe("cable:ping")
    connection.write(MESSAGES::WELCOME)
  else
    connection.write(MESSAGES::UNAUTHORIZED)
    connection.close
  end
end

.on_remote_disconnect(connection, options) ⇒ Object

Process the disconnect initiated by the server.

Parameters:



178
179
180
181
182
183
 
# File 'lib/rage/cable/protocols/actioncable_v1_json.rb', line 178

def self.on_remote_disconnect(connection, options)
  reconnect = options.nil? || options.fetch(:reconnect, true)
  message = reconnect ? MESSAGES::DISCONNECT : MESSAGES::DISCONNECT_NO_RECONNECT

  connection.write(message)
end

.protocol_definition ⇒ Object

The method defines the headers to send to the client after the handshake process.



68
69
70
 
# File 'lib/rage/cable/protocols/actioncable_v1_json.rb', line 68

def self.protocol_definition
  HANDSHAKE_HEADERS
end

.protocol_id ⇒ String

A unique identifier of the protocol used in multi-server setups by PubSub.

Returns:

  • (String) 


170
171
172
 
# File 'lib/rage/cable/protocols/actioncable_v1_json.rb', line 170

def self.protocol_id
  "acv1"
end

.serialize(params, data) ⇒ Object

Serialize a Ruby object into the format the client would understand.

Parameters:

  • params (Hash) —

    parameters associated with the client

  • data (Object) —

    the object to serialize



164
165
166
 
# File 'lib/rage/cable/protocols/actioncable_v1_json.rb', line 164

def self.serialize(params, data)
  { identifier: params.to_json, message: data }.to_json
end