Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
fmt-1: nightly fmt on jetstream
since its rustfmt config is nightly-only :/
(need to fix the make command and gha to use nightly for that one package still)
author
phil
date
(Mar 6, 2025, 8:14 PM -0500)
commit
1519dbc8
1519dbc8dc39d0e1c93c1d59f99d3dba4ab4e9a4
parent
4f765cd5
4f765cd5ad08660bd859a0184cd66f7a58858ba1
+66
-26
+10
jetstream/README.md
+10
jetstream/README.md
···
57
57
This listens for posts that *I personally make*. You can substitute your own DID and make a few test posts yourself if
58
58
you'd
59
59
like of course!
60
+
61
+
62
+
## Running `rustfmt` (requires nightly)
63
+
64
+
```bash
65
+
# get nightly set up
66
+
rustup toolchain install nightly --allow-downgrade -c rustfmt
67
+
# run the nightly version of fmt
68
+
cargo +nightly fmt
69
+
```
+11
-8
jetstream/examples/arbitrary_record.rs
+11
-8
jetstream/examples/arbitrary_record.rs
···
1
-
//! An example of how to listen for create/delete events on a specific DID and potentialy unknown NSID
1
+
//! An example of how to listen for create/delete events on a specific DID and potentialy unknown
2
+
//! NSID
2
3
3
4
use atrium_api::types::string;
4
5
use clap::Parser;
5
6
use jetstream::{
6
-
events::{commit::CommitEvent, JetstreamEvent::Commit},
7
-
DefaultJetstreamEndpoints, JetstreamCompression, JetstreamConfig, JetstreamConnector,
7
+
events::{
8
+
commit::CommitEvent,
9
+
JetstreamEvent::Commit,
10
+
},
11
+
DefaultJetstreamEndpoints,
12
+
JetstreamCompression,
13
+
JetstreamConfig,
14
+
JetstreamConnector,
8
15
};
9
16
10
17
#[derive(Parser, Debug)]
···
34
41
let jetstream: JetstreamConnector<serde_json::Value> = JetstreamConnector::new(config)?;
35
42
let receiver = jetstream.connect().await?;
36
43
37
-
println!(
38
-
"Listening for '{}' events on DIDs: {:?}",
39
-
&*args.nsid,
40
-
dids,
41
-
);
44
+
println!("Listening for '{}' events on DIDs: {:?}", &*args.nsid, dids);
42
45
43
46
while let Ok(event) = receiver.recv_async().await {
44
47
if let Commit(CommitEvent::Create { commit, .. }) = event {
+13
-8
jetstream/examples/basic.rs
+13
-8
jetstream/examples/basic.rs
···
1
1
//! A very basic example of how to listen for create/delete events on a specific DID and NSID.
2
2
3
-
use atrium_api::{record::KnownRecord::AppBskyFeedPost, types::string};
3
+
use atrium_api::{
4
+
record::KnownRecord::AppBskyFeedPost,
5
+
types::string,
6
+
};
4
7
use clap::Parser;
5
8
use jetstream::{
6
-
events::{commit::CommitEvent, JetstreamEvent::Commit},
7
-
DefaultJetstreamEndpoints, JetstreamCompression, JetstreamConfig, JetstreamConnector,
9
+
events::{
10
+
commit::CommitEvent,
11
+
JetstreamEvent::Commit,
12
+
},
13
+
DefaultJetstreamEndpoints,
14
+
JetstreamCompression,
15
+
JetstreamConfig,
16
+
JetstreamConnector,
8
17
};
9
18
10
19
#[derive(Parser, Debug)]
···
34
43
let jetstream = JetstreamConnector::new(config)?;
35
44
let receiver = jetstream.connect().await?;
36
45
37
-
println!(
38
-
"Listening for '{}' events on DIDs: {:?}",
39
-
&*args.nsid,
40
-
dids,
41
-
);
46
+
println!("Listening for '{}' events on DIDs: {:?}", &*args.nsid, dids);
42
47
43
48
while let Ok(event) = receiver.recv_async().await {
44
49
if let Commit(commit) = event {
+32
-10
jetstream/src/lib.rs
+32
-10
jetstream/src/lib.rs
···
3
3
pub mod exports;
4
4
5
5
use std::{
6
-
io::{Cursor, Read},
6
+
io::{
7
+
Cursor,
8
+
Read,
9
+
},
7
10
marker::PhantomData,
8
11
sync::Arc,
9
-
time::{Instant, Duration},
12
+
time::{
13
+
Duration,
14
+
Instant,
15
+
},
10
16
};
17
+
11
18
use atrium_api::record::KnownRecord;
12
19
use chrono::Utc;
13
-
use futures_util::{stream::StreamExt, SinkExt};
20
+
use futures_util::{
21
+
stream::StreamExt,
22
+
SinkExt,
23
+
};
14
24
use serde::de::DeserializeOwned;
15
-
use tokio::{net::TcpStream, sync::Mutex};
16
-
use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
25
+
use tokio::{
26
+
net::TcpStream,
27
+
sync::Mutex,
28
+
};
29
+
use tokio_tungstenite::{
30
+
connect_async,
31
+
tungstenite::Message,
32
+
MaybeTlsStream,
33
+
WebSocketStream,
34
+
};
17
35
use tokio_util::sync::CancellationToken;
18
36
use url::Url;
19
37
use zstd::dict::DecoderDictionary;
20
38
21
39
use crate::{
22
-
error::{ConfigValidationError, ConnectionError, JetstreamEventError},
40
+
error::{
41
+
ConfigValidationError,
42
+
ConnectionError,
43
+
JetstreamEventError,
44
+
},
23
45
events::JetstreamEvent,
24
46
};
25
47
···
318
340
.map_err(JetstreamEventError::ReceivedMalformedJSON)?;
319
341
320
342
if send_channel.send(event).is_err() {
321
-
// We can assume that all receivers have been dropped, so we can close the
322
-
// connection and exit the task.
343
+
// We can assume that all receivers have been dropped, so we can close
344
+
// the connection and exit the task.
323
345
log::info!(
324
346
"All receivers for the Jetstream connection have been dropped, closing connection."
325
347
);
···
343
365
.map_err(JetstreamEventError::ReceivedMalformedJSON)?;
344
366
345
367
if send_channel.send(event).is_err() {
346
-
// We can assume that all receivers have been dropped, so we can close the
347
-
// connection and exit the task.
368
+
// We can assume that all receivers have been dropped, so we can close
369
+
// the connection and exit the task.
348
370
log::info!(
349
371
"All receivers for the Jetstream connection have been dropped, closing connection..."
350
372
);