Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
set our jetstream websocket user agent
not sure if jaz ever looks at these but be nice to them in case we show up
+10
-4
+1
link_aggregator/readme.md
+1
link_aggregator/readme.md
+9
-4
link_aggregator/src/consumer/jetstream.rs
+9
-4
link_aggregator/src/consumer/jetstream.rs
···
5
5
use std::thread;
6
6
use std::time;
7
7
use tinyjson::JsonValue;
8
-
use tungstenite::{Error as TError, Message};
8
+
use tungstenite::{client::IntoClientRequest, Error as TError, Message};
9
9
use zstd::dict::DecoderDictionary;
10
10
11
11
const JETSTREAM_ZSTD_DICTIONARY: &[u8] = include_bytes!("../../zstd/dictionary");
···
72
72
let mut latest_cursor = cursor;
73
73
'outer: loop {
74
74
let url = WS_URLS[connect_retries % WS_URLS.len()];
75
-
let stream = format!(
75
+
let stream_url = format!(
76
76
"{url}?compress=true{}",
77
77
latest_cursor
78
78
.map(|c| {
···
81
81
})
82
82
.unwrap_or("".into())
83
83
);
84
+
let mut req = (&stream_url).into_client_request()?;
85
+
let ua = format!("ucosm/link aggregator v{}", env!("CARGO_PKG_VERSION"));
86
+
req.headers_mut().insert("user-agent", ua.parse()?);
87
+
84
88
counter!("jetstream_connect", "url" => url, "is_retry" => (connect_retries > 0).to_string()).increment(1);
85
-
println!("jetstream connecting, attempt #{connect_retries}: {stream}...");
86
-
let mut socket = match tungstenite::connect(&stream) {
89
+
println!("jetstream connecting, attempt #{connect_retries}, {stream_url:?} with user-agent: {ua:?}");
90
+
let mut socket = match tungstenite::connect(req) {
87
91
Ok((socket, _)) => {
88
92
println!("jetstream connected.");
89
93
connect_retries = 0;
···
92
96
Err(e) => {
93
97
connect_retries += 1;
94
98
if connect_retries >= 7 {
99
+
eprintln!("jetstream: no more connect retries, breaking out.");
95
100
break;
96
101
}
97
102
let backoff = time::Duration::from_secs(connect_retries.try_into().unwrap());