spew_bsky_posts/
spew-bsky-posts.rs

1use async_trait::async_trait;
2use rocketman::{
3    connection::JetstreamConnection,
4    handler::{self, Ingestors},
5    ingestion::LexiconIngestor,
6    options::JetstreamOptions,
7    types::event::{Commit, Event},
8};
9use serde_json::Value;
10use std::{sync::Arc, sync::Mutex};
11
12#[tokio::main]
13async fn main() {
14    // set up logging
15    tracing_subscriber::fmt()
16        .with_max_level(tracing::Level::INFO)
17        .init();
18    // init the builder
19    let opts = JetstreamOptions::builder()
20        // your EXACT nsids
21        .wanted_collections(vec!["app.bsky.feed.post".to_string()])
22        .build();
23    // create the jetstream connector
24    let jetstream = JetstreamConnection::new(opts);
25
26    // create your ingestors
27    let mut ingestors = Ingestors::new();
28
29    // register commit ingestor for posts
30    ingestors.commits.insert(
31        // your EXACT nsid
32        "app.bsky.feed.post".to_string(),
33        Box::new(PostIngestor),
34    );
35
36    // optionally register identity/account ingestors
37    // ingestors.identity = Some(Box::new(MyIdentityIngestor));
38    // ingestors.account = Some(Box::new(MyAccountIngestor));
39
40    // tracks the last message we've processed
41    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
42
43    // get channels
44    let msg_rx = jetstream.get_msg_rx();
45    let reconnect_tx = jetstream.get_reconnect_tx();
46
47    // spawn a task to process messages from the queue.
48    // this is a simple implementation, you can use a more complex one based on needs.
49    let c_cursor = cursor.clone();
50    tokio::spawn(async move {
51        while let Ok(message) = msg_rx.recv().await {
52            if let Err(e) =
53                handler::handle_message(message, &ingestors, reconnect_tx.clone(), c_cursor.clone())
54                    .await
55            {
56                eprintln!("Error processing message: {}", e);
57            };
58        }
59    });
60
61    // connect to jetstream
62    // retries internally, but may fail if there is an extreme error.
63    if let Err(e) = jetstream.connect(cursor.clone()).await {
64        eprintln!("Failed to connect to Jetstream: {}", e);
65        std::process::exit(1);
66    }
67}
68
69pub struct PostIngestor;
70
71/// A cool ingestor implementation. Will just print the message. Does not do verification.
72#[async_trait]
73impl LexiconIngestor for PostIngestor {
74    async fn ingest(&self, message: Event<Value>) -> anyhow::Result<()> {
75        if let Some(Commit {
76            record: Some(record),
77            ..
78        }) = message.commit
79        {
80            if let Some(Value::String(text)) = record.get("text") {
81                println!("{text:?}");
82            }
83        }
84        Ok(())
85    }
86}
spew_bsky_posts/spew-bsky-posts.rs

spew_bsky_posts/
spew-bsky-posts.rs
