1
2
3
4
5
6
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
|
# frozen_string_literal: true
module ActionCable
module Channel
module PeriodicTimers
extend ActiveSupport::Concern
included do
class_attribute :periodic_timers, instance_reader: false, default: []
after_subscribe :start_periodic_timers
after_unsubscribe :stop_periodic_timers
end
module ClassMethods
# Periodically performs a task on the channel, like updating an online
# user counter, polling a backend for new status messages, sending
# regular "heartbeat" messages, or doing some internal work and giving
# progress updates.
#
# Pass a method name or lambda argument or provide a block to call.
# Specify the calling period in seconds using the <tt>every:</tt>
# keyword argument.
#
# periodically :transmit_progress, every: 5.seconds
#
# periodically every: 3.minutes do
# transmit action: :update_count, count: current_count
# end
#
def periodically(callback_or_method_name = nil, every:, &block)
callback =
if block_given?
raise ArgumentError, "Pass a block or provide a callback arg, not both" if callback_or_method_name
block
else
case callback_or_method_name
when Proc
callback_or_method_name
when Symbol
-> { __send__ callback_or_method_name }
else
raise ArgumentError, "Expected a Symbol method name or a Proc, got #{callback_or_method_name.inspect}"
end
end
unless every.kind_of?(Numeric) && every > 0
raise ArgumentError, "Expected every: to be a positive number of seconds, got #{every.inspect}"
end
self.periodic_timers += [[ callback, every: every ]]
end
end
private
def active_periodic_timers
@active_periodic_timers ||= []
end
def start_periodic_timers
self.class.periodic_timers.each do |callback, options|
active_periodic_timers << start_periodic_timer(callback, every: options.fetch(:every))
end
end
def start_periodic_timer(callback, every:)
connection.server.event_loop.timer every do
connection.worker_pool.async_exec self, connection: connection, &callback
end
end
def stop_periodic_timers
active_periodic_timers.each { |timer| timer.shutdown }
active_periodic_timers.clear
end
end
end
end
|
