A decentralized music tracking and discovery platform built on AT Protocol 🎵
rocksky.app
spotify
atproto
lastfm
musicbrainz
scrobbling
listenbrainz
feat: implement reconnection logic for ScrobbleSubscriber in feed and jetstream modules
+23
-4
+11
-3
crates/feed/src/feed.rs
+11
-3
crates/feed/src/feed.rs
···
143
143
);
144
144
let subscriber = ScrobbleSubscriber::new(&url);
145
145
146
-
match subscriber.run(ddb).await {
147
-
Ok(_) => tracing::info!("Firehose listener exited normally"),
148
-
Err(e) => tracing::error!(error = %e, "Firehose listener exited with error"),
146
+
// loop, reconnecting on failure
147
+
loop {
148
+
match subscriber.run(ddb.clone()).await {
149
+
Ok(_) => tracing::info!("Connected to jetstream server"),
150
+
Err(e) => {
151
+
tracing::error!(error = %e, "Failed to connect to jetstream server, retrying in 1 second...");
152
+
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
153
+
continue;
154
+
}
155
+
}
156
+
break;
149
157
}
150
158
151
159
Ok::<(), Error>(())
+12
-1
crates/jetstream/src/lib.rs
+12
-1
crates/jetstream/src/lib.rs
···
31
31
);
32
32
let subscriber = ScrobbleSubscriber::new(&url);
33
33
34
-
subscriber.run(state).await?;
34
+
// loop, reconnecting on failure
35
+
loop {
36
+
match subscriber.run(state.clone()).await {
37
+
Ok(_) => tracing::info!("Connected to jetstream server"),
38
+
Err(e) => {
39
+
tracing::error!(error = %e, "Failed to connect to jetstream server, retrying in 1 second...");
40
+
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
41
+
continue;
42
+
}
43
+
}
44
+
break;
45
+
}
35
46
36
47
Ok(())
37
48
}