Class: CableRoom::ChannelBase

Inherits:
ActionCable::Channel::Base
  • Object
show all
Defined in:
lib/cable_room/channel_base.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(lock_info, room_class, key, config) ⇒ ChannelBase

Returns a new instance of ChannelBase.



7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/cable_room/channel_base.rb', line 7

def initialize(lock_info, room_class, key, config)
  # We don't really have a "connection" in the ActionCable sense, so
  #   stuff in something that half looks like one
  super(DummyConnection.new(self), "Room[]", {})

  # Used mainly for logs and being able to follow a specific Room instance
  @uuid = SecureRandom.hex(6)

  @mutex = Monitor.new
  @lock_info = lock_info
  @current_state = :initializing

  @tenant = Apartment::Tenant.current if defined?(Apartment)

  @server = ChannelTracker.instance
  @logger = ActionCable::Connection::TaggedLoggerProxy.new(
    @server.logger,
    tags: ["#{room_class.name} #{@uuid}"]
  )

  logger.info "Initializing new #{room_class.name}"
  logger.info "  UUID: #{@uuid}"
  logger.info "  Key: #{room_class.room_port_key(key)}"

  @watchdog_interval = config[:watchdog_interval]
  @lock_duration = config[:lock_duration]

  @processing_work = false
  @work_queue = []

  @room = room_class.new(self, key)
  @server.track_room_channel self
  ping_watchdog
end

Instance Attribute Details

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



4
5
6
 
# File 'lib/cable_room/channel_base.rb', line 4

def logger
  @logger
end

#room ⇒ Object (readonly)

Returns the value of attribute room.



3
4
5
 
# File 'lib/cable_room/channel_base.rb', line 3

def room
  @room
end

#server ⇒ Object (readonly)

Returns the value of attribute server.



4
5
6
 
# File 'lib/cable_room/channel_base.rb', line 4

def server
  @server
end

#tenant ⇒ Object (readonly)

Returns the value of attribute tenant.



3
4
5
 
# File 'lib/cable_room/channel_base.rb', line 3

def tenant
  @tenant
end

Class Method Details

.channel_name ⇒ Object 



42
43
44
 
# File 'lib/cable_room/channel_base.rb', line 42

def self.channel_name
  module_parent.name
end

Instance Method Details

#_post_wrapped_work(async: false, silent: false, &blk) ⇒ Object 



146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
 
# File 'lib/cable_room/channel_base.rb', line 146

def _post_wrapped_work(async: false, silent: false, &blk)
  if async
    # Async stuff is mostly untracked - we just post it to the worker pool and forget about it
    worker_pool.executor.post(&blk)
  else
    @mutex.synchronize do
      if @current_state == :dead || @current_state == :shutting_down
        raise "Attempt to post work to dead or shutting down room" unless silent
        return
      end
      @work_queue << blk
    end
    schedule_work
  end
end

#beat ⇒ Object 



185
186
187
188
189
 
# File 'lib/cable_room/channel_base.rb', line 185

def beat
  post_work(async: true) do
    check_room_watchdog
  end
end

#check_room_watchdog ⇒ Object 



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/cable_room/channel_base.rb', line 100

def check_room_watchdog
  @mutex.synchronize do
    return if state == :dead || state == :shutting_down
  end

  relock = CableRoom.lock_manager.lock(@lock_info[:resource], @lock_duration.in_milliseconds, extend: @lock_info)
  unless relock
    logger.warn "Lost lock, shutting down"
    unsubscribe_from_channel
    return
  end

  unless @last_watchdog_ping_at && @last_watchdog_ping_at > @watchdog_interval.ago
    logger.warn "Watchdog timeout for room #{@room.class.name}[#{@room.key}], shutting down"
    initiate_shutdown("Watchdog timeout")
    return
  end
end

#initiate_shutdown(reason) ⇒ Object 



119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/cable_room/channel_base.rb', line 119

def initiate_shutdown(reason)
  @mutex.synchronize do
    return if @current_state == :dead || @current_state == :shutting_down

    logger.info "Initiating shutdown: #{reason}"

    # Stop streams immediately to prevent further messages from being added
    stop_all_streams

    # Append the final unsubscribe to the work queue so we can process remaining messages first
    post_work(async: false) do
      unsubscribe_from_channel
    end

    @current_state = :shutting_down
  end
end

#ping_watchdog ⇒ Object 



93
94
95
96
97
98
 
# File 'lib/cable_room/channel_base.rb', line 93

def ping_watchdog
  return if state == :dead

  logger.debug "Ping watchdog"
  @last_watchdog_ping_at = Time.current
end

#post_work(**kwargs, &blk) ⇒ Object 



162
163
164
165
166
167
168
 
# File 'lib/cable_room/channel_base.rb', line 162

def post_work(**kwargs, &blk)
  _post_wrapped_work(**kwargs) do
    worker_pool.invoke(self, :instance_exec, connection: self, &blk)
  rescue => e
    report_work_error(e)
  end
end

#report_work_error(error) ⇒ Object

Work errors are swallowed so one bad message can't take the Room down with it. Log the backtrace and hand the error to the application so the failure is still discoverable.



172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/cable_room/channel_base.rb', line 172

def report_work_error(error)
  logger.error "Error during work execution: #{error.class.name}: #{error.message}"
  Array(error.backtrace).first(20).each { |line| logger.error "  #{line}" }

  CableRoom.report_error(
    error,
    room: room,
    room_class: room&.class,
    room_key: room&.key,
    channel: self
  )
end

#state ⇒ Object 



66
67
68
 
# File 'lib/cable_room/channel_base.rb', line 66

def state
  @current_state
end

#stream_from(broadcasting, callback = nil, coder: nil, &block) ⇒ Object

def stream_from(...) # TODO Try making sync raise ArgumentError, "Block required" unless block_given? super end

Raises:

  • (ArgumentError) 


51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/cable_room/channel_base.rb', line 51

def stream_from(broadcasting, callback = nil, coder: nil, &block)
  raise ArgumentError, "Block required" unless block.present?

  broadcasting = String(broadcasting)

  # Build a stream handler by wrapping the user-provided callback with a decoder
  # or defaulting to a JSON-decoding retransmitter.
  handler = worker_pool_stream_handler(broadcasting, callback || block, coder: coder)
  streams[broadcasting] = handler

  pubsub.subscribe(broadcasting, handler, lambda do
    logger.info "#{self.class.name} is streaming from #{broadcasting}"
  end)
end

#terminate! ⇒ Object 



137
138
139
140
141
142
143
144
 
# File 'lib/cable_room/channel_base.rb', line 137

def terminate!
  @mutex.synchronize do
    stop_all_streams
    @current_state = :dead
    CableRoom.lock_manager.unlock(@lock_info) if @lock_info
    server.untrack_room_channel self
  end
end

#transmit(*args) ⇒ Object 



191
192
193
 
# File 'lib/cable_room/channel_base.rb', line 191

def transmit(*args)
  logger.info("Channel.transmit called, ignoring: #{args.inspect}")
end