Class: Upkeep::Delivery::ActionCableAdapter

Inherits:
Object
  • Object
show all
Defined in:
lib/upkeep/delivery/action_cable_adapter.rb

Constant Summary collapse

STREAM_PREFIX = 
"upkeep:subscriber"
SUBSCRIPTION_STREAM_PREFIX = 
"upkeep:subscription"

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(server: default_server) ⇒ ActionCableAdapter

Returns a new instance of ActionCableAdapter.



20
21
22
 
# File 'lib/upkeep/delivery/action_cable_adapter.rb', line 20

def initialize(server: default_server)
  @server = server
end

Class Method Details

.stream_name_for(subscriber_id) ⇒ Object 



12
13
14
 
# File 'lib/upkeep/delivery/action_cable_adapter.rb', line 12

def self.stream_name_for(subscriber_id)
  "#{STREAM_PREFIX}:#{Digest::SHA256.hexdigest(subscriber_id.to_s)[0, 32]}"
end

.subscription_stream_name_for(subscription_id) ⇒ Object 



16
17
18
 
# File 'lib/upkeep/delivery/action_cable_adapter.rb', line 16

def self.subscription_stream_name_for(subscription_id)
  "#{SUBSCRIPTION_STREAM_PREFIX}:#{Digest::SHA256.hexdigest(subscription_id.to_s)[0, 32]}"
end

Instance Method Details

#deliver(envelope) ⇒ Object 



24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/upkeep/delivery/action_cable_adapter.rb', line 24

def deliver(envelope)
  stream_name = envelope.stream_name || self.class.stream_name_for(envelope.subscriber_id)
  payload = {
    subscriber_id: envelope.subscriber_id,
    stream_name: stream_name,
    envelope_digest: Transport.envelope_digest(envelope),
    bytesize: envelope.body.bytesize
  }

  ActiveSupport::Notifications.instrument("deliver.upkeep", payload) do
    server.broadcast(stream_name, envelope.body)
  end
end