A Ruby gem for streaming data from the Bluesky/ATProto firehose
mackuba.eu
exponential backoff on reconnection errors
+19
-2
lib/skyfall/stream.rb
+19
-2
lib/skyfall/stream.rb
···
12
12
:subscribe_repos => SUBSCRIBE_REPOS
13
13
}
14
14
15
+
MAX_RECONNECT_INTERVAL = 300
16
+
15
17
attr_accessor :heartbeat_timeout, :heartbeat_interval, :cursor, :auto_reconnect
16
18
17
19
def initialize(server, endpoint, cursor = nil)
···
24
26
@heartbeat_timeout = 30
25
27
@last_update = nil
26
28
@auto_reconnect = true
29
+
@connection_attempts = 0
27
30
end
28
31
29
32
def connect
···
46
49
end
47
50
48
51
@ws.on(:message) do |msg|
52
+
@connection_attempts = 0
53
+
49
54
data = msg.data.pack('C*')
50
55
@handlers[:raw_message]&.call(data)
51
56
···
64
69
65
70
@ws.on(:close) do |e|
66
71
@ws = nil
72
+
67
73
if @auto_reconnect && @engines_on
68
-
@handlers[:reconnect]&.call(e)
69
-
connect
74
+
EM.add_timer(reconnect_delay) do
75
+
@connection_attempts += 1
76
+
@handlers[:reconnect]&.call(e)
77
+
connect
78
+
end
70
79
else
71
80
@engines_on = false
72
81
@handlers[:disconnect]&.call(e)
···
115
124
116
125
117
126
private
127
+
128
+
def reconnect_delay
129
+
if @connection_attempts == 0
130
+
0
131
+
else
132
+
[2 ** (@connection_attempts - 1), MAX_RECONNECT_INTERVAL].min
133
+
end
134
+
end
118
135
119
136
def build_websocket_url
120
137
url = "wss://#{@server}/xrpc/#{@endpoint}"