Class: WSDirector::Protocols::ActionCable

Inherits:
Base
  • Object
show all
Defined in:
lib/wsdirector/protocols/action_cable.rb

Overview

ActionCable protocol

Constant Summary collapse

PING_IGNORE = 
/['"]type['"]:\s*['"]ping['"]/

Constants included from Utils

Utils::MULTIPLIER_FORMAT

Instance Attribute Summary

Attributes included from Utils

#scale

Instance Method Summary collapse

Methods inherited from Base

#debug, #handle_step, #initialize, #send, #sleep, #to_proc, #wait_all

Methods included from Utils

#parse_multiplier

Constructor Details

This class inherits a constructor from WSDirector::Protocols::Base

Instance Method Details

#init_client(**options) ⇒ Object

Add ping ignore and make sure that we receive Welcome message



10
11
12
13
14
15
16
17
18
 
# File 'lib/wsdirector/protocols/action_cable.rb', line 10

def init_client(**options)
  options[:ignore] ||= [PING_IGNORE]
  options[:subprotocol] ||= "actioncable-v1-json"

  super

  receive("data>" => {"type" => "welcome"})
  log(:done) { "Welcomed" }
end

#perform(step) ⇒ Object

Raises:



46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/wsdirector/protocols/action_cable.rb', line 46

def perform(step)
  identifier = extract_identifier(step)
  action = step.delete("action")

  raise Error, "Action is missing" unless action

  data = step.fetch("data", {}).merge(action:).to_json

  client.send({command: "message", data:, identifier:}.to_json)

  log(nil) { "Performed #{action} on #{identifier}" }
end

#receive(step) ⇒ Object 



59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/wsdirector/protocols/action_cable.rb', line 59

def receive(step)
  return super unless step.key?("channel")

  identifier = extract_identifier(step)

  key = step.key?("data") ? "data" : "data>"

  message = step.fetch(key, {})

  # Move all protocol-level fields to data
  step[key] = {"identifier" => identifier, "message" => message}.merge(step.slice("offset", "stream_id", "epoch"))

  super
end

#receive_all(step) ⇒ Object 



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/wsdirector/protocols/action_cable.rb', line 74

def receive_all(step)
  messages = step["messages"]

  return super if messages.nil? || messages.empty?

  messages.each do |msg|
    next unless msg.key?("channel")
    identifier = extract_identifier(msg)

    key = msg.key?("data") ? "data" : "data>"

    msg[key] = {"identifier" => identifier, "message" => msg[key]}.merge(step.slice("offset", "stream_id", "epoch"))
  end

  super
end

#subscribe(step) ⇒ Object 



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/wsdirector/protocols/action_cable.rb', line 20

def subscribe(step)
  identifier = extract_identifier(step)

  log { "Subsribing to #{identifier}" }

  client.send({command: "subscribe", identifier:}.to_json)

  begin
    receive(
      "data" => {"type" => "confirm_subscription", "identifier" => identifier}
    )
    log(:done) { "Subsribed to #{identifier}" }
  rescue UnmatchedExpectationError => e
    raise unless /reject_subscription/.match?(e.message)
    raise UnmatchedExpectationError, "Subscription rejected to #{identifier}"
  end
end

#unsubscribe(step) ⇒ Object 



38
39
40
41
42
43
44
 
# File 'lib/wsdirector/protocols/action_cable.rb', line 38

def unsubscribe(step)
  identifier = extract_identifier(step)

  client.send({command: "unsubscribe", identifier:}.to_json)

  log(nil) { "Unsubscribed from #{identifier}" }
end