mackuba.eu
A Ruby gem for streaming data from the Bluesky/ATProto firehose
rolled back the heartbeat timer for now
This reverts commit 5d485ae61eccc16a138c509c3a4d643b7586e6b0.
# Conflicts:
# lib/skyfall/stream.rb
+2
-22
lib/skyfall/stream.rb
+2
-22
lib/skyfall/stream.rb
···
14
14
:subscribe_labels => SUBSCRIBE_LABELS
15
15
}
16
16
17
-
EVENTS = %w(message raw_message connecting connect disconnect reconnect error timeout)
17
+
EVENTS = %w(message raw_message connecting connect disconnect reconnect error)
18
18
19
19
MAX_RECONNECT_INTERVAL = 300
20
20
21
-
attr_accessor :heartbeat_timeout, :heartbeat_interval, :cursor, :auto_reconnect, :last_update
21
+
attr_accessor :heartbeat_timeout, :heartbeat_interval, :cursor, :auto_reconnect
22
22
23
23
def initialize(server, endpoint, cursor = nil)
24
24
@endpoint = check_endpoint(endpoint)
···
27
27
@handlers = {}
28
28
@auto_reconnect = true
29
29
@connection_attempts = 0
30
-
@heartbeat_interval = 10
31
-
@heartbeat_timeout = 300
32
-
@last_update = nil
33
30
34
31
@handlers[:error] = proc { |e| puts "ERROR: #{e}" }
35
32
end
···
54
51
55
52
@ws.on(:open) do |e|
56
53
@handlers[:connect]&.call
57
-
@last_update = Time.now
58
-
start_heartbeat_timer
59
54
end
60
55
61
56
@ws.on(:message) do |msg|
62
57
@reconnecting = false
63
58
@connection_attempts = 0
64
-
@last_update = Time.now
65
59
66
60
data = msg.data.pack('C*')
67
61
@handlers[:raw_message]&.call(data)
···
115
109
end
116
110
117
111
alias close disconnect
118
-
119
-
def start_heartbeat_timer
120
-
return if @timer || @heartbeat_interval.to_f <= 0 || @heartbeat_timeout.to_f <= 0
121
-
122
-
@timer = EM::PeriodicTimer.new(@heartbeat_interval) do
123
-
next if @ws.nil? || @heartbeat_timeout.to_f <= 0
124
-
time_passed = Time.now - @last_update
125
-
126
-
if time_passed > @heartbeat_timeout
127
-
@handlers[:timeout]&.call
128
-
reconnect
129
-
end
130
-
end
131
-
end
132
112
133
113
EVENTS.each do |event|
134
114
define_method "on_#{event}" do |&block|