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 



131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
 
# File 'lib/cable_room/channel_base.rb', line 131

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 



170
171
172
173
174
 
# File 'lib/cable_room/channel_base.rb', line 170

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

#check_room_watchdog ⇒ Object 



85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/cable_room/channel_base.rb', line 85

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 



104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/cable_room/channel_base.rb', line 104

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 



78
79
80
81
82
83
 
# File 'lib/cable_room/channel_base.rb', line 78

def ping_watchdog
  return if state == :dead

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

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



147
148
149
150
151
152
153
 
# File 'lib/cable_room/channel_base.rb', line 147

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.



157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/cable_room/channel_base.rb', line 157

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 



51
52
53
 
# File 'lib/cable_room/channel_base.rb', line 51

def state
  @current_state
end

#stream_from ⇒ Object

Raises:

  • (ArgumentError) 


46
47
48
49
 
# File 'lib/cable_room/channel_base.rb', line 46

def stream_from(...)
  raise ArgumentError, "Block required" unless block_given?
  super
end

#terminate! ⇒ Object 



122
123
124
125
126
127
128
129
 
# File 'lib/cable_room/channel_base.rb', line 122

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 



176
177
178
 
# File 'lib/cable_room/channel_base.rb', line 176

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