Struct JetstreamConnection

Source

pub struct JetstreamConnection {
    pub opts: JetstreamOptions,
    /* private fields */
}

Fields§

§opts: JetstreamOptions

Implementations§

Source §

impl JetstreamConnection

Source

pub fn new(opts: JetstreamOptions) -> Self

Examples found in repository ?

examples/comprehensive-example.rs (line 28)

13async fn main() {
14    // set up logging
15    tracing_subscriber::fmt()
16        .with_max_level(tracing::Level::INFO)
17        .init();
18
19    // init the builder with wanted collections
20    let opts = JetstreamOptions::builder()
21        .wanted_collections(vec![
22            "app.bsky.feed.post".to_string(),
23            "xyz.statusphere.status".to_string(),
24        ])
25        .build();
26
27    // create the jetstream connector
28    let jetstream = JetstreamConnection::new(opts);
29
30    // create your ingestors
31    let mut ingestors = Ingestors::new();
32
33    // register commit ingestors
34    ingestors.commits.insert(
35        "xyz.statusphere.status".to_string(),
36        Box::new(StatusphereIngestor),
37    );
38
39    // register identity ingestor
40    ingestors.identity = Some(Box::new(IdentityIngestor));
41
42    // register account ingestor
43    ingestors.account = Some(Box::new(AccountIngestor));
44
45    // tracks the last message we've processed
46    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
47
48    // get channels
49    let msg_rx = jetstream.get_msg_rx();
50    let reconnect_tx = jetstream.get_reconnect_tx();
51
52    // spawn a task to process messages from the queue.
53    let c_cursor = cursor.clone();
54    tokio::spawn(async move {
55        while let Ok(message) = msg_rx.recv().await {
56            if let Err(e) =
57                handler::handle_message(message, &ingestors, reconnect_tx.clone(), c_cursor.clone())
58                    .await
59            {
60                eprintln!("Error processing message: {}", e);
61            };
62        }
63    });
64
65    // connect to jetstream
66    if let Err(e) = jetstream.connect(cursor.clone()).await {
67        eprintln!("Failed to connect to Jetstream: {}", e);
68        std::process::exit(1);
69    }
70}

More examples

Hide additional examples

examples/spew-bsky-posts.rs (line 24)

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}

examples/bsky-posts.rs (line 54)

42pub async fn start_idle_test() -> Result<()> {
43    info!("Testing reconnection fix with idle connection...");
44
45    let opts = JetstreamOptions::builder()
46        .wanted_collections(vec![
47            // This is a custom collection that will get zero traffic
48            "app.bsky.feed.post".to_string(),
49        ])
50        .timeout_time_sec(20) // Shorter timeout for faster testing
51        .bound(65536)
52        .build();
53
54    let jetstream = JetstreamConnection::new(opts);
55
56    let mut ingestors: HashMap<String, Box<dyn LexiconIngestor + Send + Sync>> = HashMap::new();
57    ingestors.insert(
58        "fm.teal.alpha.feed.play".to_string(),
59        Box::new(ProfileIngestor),
60    );
61
62    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
63
64    let msg_rx = jetstream.get_msg_rx();
65    let reconnect_tx = jetstream.get_reconnect_tx();
66
67    let monitor_rx = msg_rx.clone();
68    tokio::spawn(async move {
69        let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
70
71        loop {
72            interval.tick().await;
73            let queue_len = monitor_rx.len();
74            if queue_len > 0 {
75                warn!("Queue has {} messages pending", queue_len);
76            }
77        }
78    });
79
80    // Spawn task to process messages (there should be none)
81    let c_cursor = cursor.clone();
82    tokio::spawn(async move {
83        info!("Message processing task started");
84        let mut message_count = 0u64;
85        let mut last_log = std::time::Instant::now();
86
87        let ing = Ingestors {
88            commits: ingestors,
89            identity: None,
90            account: None,
91        };
92
93        while let Ok(message) = msg_rx.recv().await {
94            message_count += 1;
95
96            // Log every 10 messages to see if we're actually processing
97            if message_count % 10 == 0 || last_log.elapsed() > std::time::Duration::from_secs(5) {
98                info!(
99                    "Processing message #{} (queue len: {}/{})",
100                    message_count,
101                    msg_rx.len(),
102                    msg_rx.capacity().unwrap_or(0)
103                );
104                last_log = std::time::Instant::now();
105            }
106
107            match handler::handle_message(message, &ing, reconnect_tx.clone(), c_cursor.clone())
108                .await
109            {
110                Ok(_) => {}
111                Err(e) => {
112                    error!("Error processing message #{}: {}", message_count, e);
113                }
114            }
115        }
116
117        error!("Message processing task ended unexpectedly!");
118    });
119
120    // Add a monitoring task to track connection health
121    let start_time = std::time::Instant::now();
122    tokio::spawn(async move {
123        let mut interval = tokio::time::interval(std::time::Duration::from_secs(15));
124        let mut tick_count = 0;
125
126        loop {
127            interval.tick().await;
128            tick_count += 1;
129            let elapsed = start_time.elapsed().as_secs();
130            info!(
131                "Health check #{} - uptime: {}s (expecting reconnects every ~20s)",
132                tick_count, elapsed
133            );
134        }
135    });
136
137    info!("Connecting to jetstream (testing reconnection fix)...");
138    info!("Watch for 'Stream closed by server' or 'No messages received' followed by reconnection");
139
140    // This should connect, sit idle, get disconnected, then reconnect automatically
141    // The fix should prevent infinite loops when reconnecting
142    jetstream
143        .connect(cursor.clone())
144        .await
145        .map_err(|e| anyhow::anyhow!("error running ingest: {}", e))
146}

examples/reproduce-stall.rs (line 54)

42pub async fn start_idle_test() -> Result<()> {
43    info!("Testing reconnection fix with idle connection...");
44
45    let opts = JetstreamOptions::builder()
46        .wanted_collections(vec![
47            // This is a custom collection that will get zero traffic
48            "moe.hayden.blogi.actor.profile".to_string(),
49        ])
50        .timeout_time_sec(20) // Shorter timeout for faster testing
51        .bound(65536)
52        .build();
53
54    let jetstream = JetstreamConnection::new(opts);
55
56    let mut ingestors: HashMap<String, Box<dyn LexiconIngestor + Send + Sync>> = HashMap::new();
57    ingestors.insert(
58        "fm.teal.alpha.feed.play".to_string(),
59        Box::new(ProfileIngestor),
60    );
61
62    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
63
64    let msg_rx = jetstream.get_msg_rx();
65    let reconnect_tx = jetstream.get_reconnect_tx();
66
67    let monitor_rx = msg_rx.clone();
68    tokio::spawn(async move {
69        let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
70
71        loop {
72            interval.tick().await;
73            let queue_len = monitor_rx.len();
74            if queue_len > 0 {
75                warn!("Queue has {} messages pending", queue_len);
76            }
77        }
78    });
79
80    // Spawn task to process messages (there should be none)
81    let c_cursor = cursor.clone();
82    tokio::spawn(async move {
83        info!("Message processing task started");
84        let mut message_count = 0u64;
85        let mut last_log = std::time::Instant::now();
86
87        let ing = Ingestors {
88            commits: ingestors,
89            identity: None,
90            account: None,
91        };
92
93        while let Ok(message) = msg_rx.recv().await {
94            message_count += 1;
95
96            // Log every 10 messages to see if we're actually processing
97            if message_count % 10 == 0 || last_log.elapsed() > std::time::Duration::from_secs(5) {
98                info!(
99                    "Processing message #{} (queue len: {}/{})",
100                    message_count,
101                    msg_rx.len(),
102                    msg_rx.capacity().unwrap_or(0)
103                );
104                last_log = std::time::Instant::now();
105            }
106
107            match handler::handle_message(message, &ing, reconnect_tx.clone(), c_cursor.clone())
108                .await
109            {
110                Ok(_) => {}
111                Err(e) => {
112                    error!("Error processing message #{}: {}", message_count, e);
113                }
114            }
115        }
116
117        error!("Message processing task ended unexpectedly!");
118    });
119
120    // Add a monitoring task to track connection health
121    let start_time = std::time::Instant::now();
122    tokio::spawn(async move {
123        let mut interval = tokio::time::interval(std::time::Duration::from_secs(15));
124        let mut tick_count = 0;
125
126        loop {
127            interval.tick().await;
128            tick_count += 1;
129            let elapsed = start_time.elapsed().as_secs();
130            info!(
131                "Health check #{} - uptime: {}s (expecting reconnects every ~20s)",
132                tick_count, elapsed
133            );
134        }
135    });
136
137    info!("Connecting to jetstream (testing reconnection fix)...");
138    info!("Watch for 'Stream closed by server' or 'No messages received' followed by reconnection");
139
140    // This should connect, sit idle, get disconnected, then reconnect automatically
141    // The fix should prevent infinite loops when reconnecting
142    jetstream
143        .connect(cursor.clone())
144        .await
145        .map_err(|e| anyhow::anyhow!("error running ingest: {}", e))
146}

Source

pub fn get_reconnect_tx(&self) -> Sender<()>

Examples found in repository ?

examples/comprehensive-example.rs (line 50)

13async fn main() {
14    // set up logging
15    tracing_subscriber::fmt()
16        .with_max_level(tracing::Level::INFO)
17        .init();
18
19    // init the builder with wanted collections
20    let opts = JetstreamOptions::builder()
21        .wanted_collections(vec![
22            "app.bsky.feed.post".to_string(),
23            "xyz.statusphere.status".to_string(),
24        ])
25        .build();
26
27    // create the jetstream connector
28    let jetstream = JetstreamConnection::new(opts);
29
30    // create your ingestors
31    let mut ingestors = Ingestors::new();
32
33    // register commit ingestors
34    ingestors.commits.insert(
35        "xyz.statusphere.status".to_string(),
36        Box::new(StatusphereIngestor),
37    );
38
39    // register identity ingestor
40    ingestors.identity = Some(Box::new(IdentityIngestor));
41
42    // register account ingestor
43    ingestors.account = Some(Box::new(AccountIngestor));
44
45    // tracks the last message we've processed
46    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
47
48    // get channels
49    let msg_rx = jetstream.get_msg_rx();
50    let reconnect_tx = jetstream.get_reconnect_tx();
51
52    // spawn a task to process messages from the queue.
53    let c_cursor = cursor.clone();
54    tokio::spawn(async move {
55        while let Ok(message) = msg_rx.recv().await {
56            if let Err(e) =
57                handler::handle_message(message, &ingestors, reconnect_tx.clone(), c_cursor.clone())
58                    .await
59            {
60                eprintln!("Error processing message: {}", e);
61            };
62        }
63    });
64
65    // connect to jetstream
66    if let Err(e) = jetstream.connect(cursor.clone()).await {
67        eprintln!("Failed to connect to Jetstream: {}", e);
68        std::process::exit(1);
69    }
70}

More examples

Hide additional examples

examples/spew-bsky-posts.rs (line 45)

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}

examples/bsky-posts.rs (line 65)

42pub async fn start_idle_test() -> Result<()> {
43    info!("Testing reconnection fix with idle connection...");
44
45    let opts = JetstreamOptions::builder()
46        .wanted_collections(vec![
47            // This is a custom collection that will get zero traffic
48            "app.bsky.feed.post".to_string(),
49        ])
50        .timeout_time_sec(20) // Shorter timeout for faster testing
51        .bound(65536)
52        .build();
53
54    let jetstream = JetstreamConnection::new(opts);
55
56    let mut ingestors: HashMap<String, Box<dyn LexiconIngestor + Send + Sync>> = HashMap::new();
57    ingestors.insert(
58        "fm.teal.alpha.feed.play".to_string(),
59        Box::new(ProfileIngestor),
60    );
61
62    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
63
64    let msg_rx = jetstream.get_msg_rx();
65    let reconnect_tx = jetstream.get_reconnect_tx();
66
67    let monitor_rx = msg_rx.clone();
68    tokio::spawn(async move {
69        let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
70
71        loop {
72            interval.tick().await;
73            let queue_len = monitor_rx.len();
74            if queue_len > 0 {
75                warn!("Queue has {} messages pending", queue_len);
76            }
77        }
78    });
79
80    // Spawn task to process messages (there should be none)
81    let c_cursor = cursor.clone();
82    tokio::spawn(async move {
83        info!("Message processing task started");
84        let mut message_count = 0u64;
85        let mut last_log = std::time::Instant::now();
86
87        let ing = Ingestors {
88            commits: ingestors,
89            identity: None,
90            account: None,
91        };
92
93        while let Ok(message) = msg_rx.recv().await {
94            message_count += 1;
95
96            // Log every 10 messages to see if we're actually processing
97            if message_count % 10 == 0 || last_log.elapsed() > std::time::Duration::from_secs(5) {
98                info!(
99                    "Processing message #{} (queue len: {}/{})",
100                    message_count,
101                    msg_rx.len(),
102                    msg_rx.capacity().unwrap_or(0)
103                );
104                last_log = std::time::Instant::now();
105            }
106
107            match handler::handle_message(message, &ing, reconnect_tx.clone(), c_cursor.clone())
108                .await
109            {
110                Ok(_) => {}
111                Err(e) => {
112                    error!("Error processing message #{}: {}", message_count, e);
113                }
114            }
115        }
116
117        error!("Message processing task ended unexpectedly!");
118    });
119
120    // Add a monitoring task to track connection health
121    let start_time = std::time::Instant::now();
122    tokio::spawn(async move {
123        let mut interval = tokio::time::interval(std::time::Duration::from_secs(15));
124        let mut tick_count = 0;
125
126        loop {
127            interval.tick().await;
128            tick_count += 1;
129            let elapsed = start_time.elapsed().as_secs();
130            info!(
131                "Health check #{} - uptime: {}s (expecting reconnects every ~20s)",
132                tick_count, elapsed
133            );
134        }
135    });
136
137    info!("Connecting to jetstream (testing reconnection fix)...");
138    info!("Watch for 'Stream closed by server' or 'No messages received' followed by reconnection");
139
140    // This should connect, sit idle, get disconnected, then reconnect automatically
141    // The fix should prevent infinite loops when reconnecting
142    jetstream
143        .connect(cursor.clone())
144        .await
145        .map_err(|e| anyhow::anyhow!("error running ingest: {}", e))
146}

examples/reproduce-stall.rs (line 65)

42pub async fn start_idle_test() -> Result<()> {
43    info!("Testing reconnection fix with idle connection...");
44
45    let opts = JetstreamOptions::builder()
46        .wanted_collections(vec![
47            // This is a custom collection that will get zero traffic
48            "moe.hayden.blogi.actor.profile".to_string(),
49        ])
50        .timeout_time_sec(20) // Shorter timeout for faster testing
51        .bound(65536)
52        .build();
53
54    let jetstream = JetstreamConnection::new(opts);
55
56    let mut ingestors: HashMap<String, Box<dyn LexiconIngestor + Send + Sync>> = HashMap::new();
57    ingestors.insert(
58        "fm.teal.alpha.feed.play".to_string(),
59        Box::new(ProfileIngestor),
60    );
61
62    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
63
64    let msg_rx = jetstream.get_msg_rx();
65    let reconnect_tx = jetstream.get_reconnect_tx();
66
67    let monitor_rx = msg_rx.clone();
68    tokio::spawn(async move {
69        let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
70
71        loop {
72            interval.tick().await;
73            let queue_len = monitor_rx.len();
74            if queue_len > 0 {
75                warn!("Queue has {} messages pending", queue_len);
76            }
77        }
78    });
79
80    // Spawn task to process messages (there should be none)
81    let c_cursor = cursor.clone();
82    tokio::spawn(async move {
83        info!("Message processing task started");
84        let mut message_count = 0u64;
85        let mut last_log = std::time::Instant::now();
86
87        let ing = Ingestors {
88            commits: ingestors,
89            identity: None,
90            account: None,
91        };
92
93        while let Ok(message) = msg_rx.recv().await {
94            message_count += 1;
95
96            // Log every 10 messages to see if we're actually processing
97            if message_count % 10 == 0 || last_log.elapsed() > std::time::Duration::from_secs(5) {
98                info!(
99                    "Processing message #{} (queue len: {}/{})",
100                    message_count,
101                    msg_rx.len(),
102                    msg_rx.capacity().unwrap_or(0)
103                );
104                last_log = std::time::Instant::now();
105            }
106
107            match handler::handle_message(message, &ing, reconnect_tx.clone(), c_cursor.clone())
108                .await
109            {
110                Ok(_) => {}
111                Err(e) => {
112                    error!("Error processing message #{}: {}", message_count, e);
113                }
114            }
115        }
116
117        error!("Message processing task ended unexpectedly!");
118    });
119
120    // Add a monitoring task to track connection health
121    let start_time = std::time::Instant::now();
122    tokio::spawn(async move {
123        let mut interval = tokio::time::interval(std::time::Duration::from_secs(15));
124        let mut tick_count = 0;
125
126        loop {
127            interval.tick().await;
128            tick_count += 1;
129            let elapsed = start_time.elapsed().as_secs();
130            info!(
131                "Health check #{} - uptime: {}s (expecting reconnects every ~20s)",
132                tick_count, elapsed
133            );
134        }
135    });
136
137    info!("Connecting to jetstream (testing reconnection fix)...");
138    info!("Watch for 'Stream closed by server' or 'No messages received' followed by reconnection");
139
140    // This should connect, sit idle, get disconnected, then reconnect automatically
141    // The fix should prevent infinite loops when reconnecting
142    jetstream
143        .connect(cursor.clone())
144        .await
145        .map_err(|e| anyhow::anyhow!("error running ingest: {}", e))
146}

Source

pub fn get_msg_rx(&self) -> Receiver<Message>

Examples found in repository ?

examples/comprehensive-example.rs (line 49)

13async fn main() {
14    // set up logging
15    tracing_subscriber::fmt()
16        .with_max_level(tracing::Level::INFO)
17        .init();
18
19    // init the builder with wanted collections
20    let opts = JetstreamOptions::builder()
21        .wanted_collections(vec![
22            "app.bsky.feed.post".to_string(),
23            "xyz.statusphere.status".to_string(),
24        ])
25        .build();
26
27    // create the jetstream connector
28    let jetstream = JetstreamConnection::new(opts);
29
30    // create your ingestors
31    let mut ingestors = Ingestors::new();
32
33    // register commit ingestors
34    ingestors.commits.insert(
35        "xyz.statusphere.status".to_string(),
36        Box::new(StatusphereIngestor),
37    );
38
39    // register identity ingestor
40    ingestors.identity = Some(Box::new(IdentityIngestor));
41
42    // register account ingestor
43    ingestors.account = Some(Box::new(AccountIngestor));
44
45    // tracks the last message we've processed
46    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
47
48    // get channels
49    let msg_rx = jetstream.get_msg_rx();
50    let reconnect_tx = jetstream.get_reconnect_tx();
51
52    // spawn a task to process messages from the queue.
53    let c_cursor = cursor.clone();
54    tokio::spawn(async move {
55        while let Ok(message) = msg_rx.recv().await {
56            if let Err(e) =
57                handler::handle_message(message, &ingestors, reconnect_tx.clone(), c_cursor.clone())
58                    .await
59            {
60                eprintln!("Error processing message: {}", e);
61            };
62        }
63    });
64
65    // connect to jetstream
66    if let Err(e) = jetstream.connect(cursor.clone()).await {
67        eprintln!("Failed to connect to Jetstream: {}", e);
68        std::process::exit(1);
69    }
70}

More examples

Hide additional examples

examples/spew-bsky-posts.rs (line 44)

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}

examples/bsky-posts.rs (line 64)

42pub async fn start_idle_test() -> Result<()> {
43    info!("Testing reconnection fix with idle connection...");
44
45    let opts = JetstreamOptions::builder()
46        .wanted_collections(vec![
47            // This is a custom collection that will get zero traffic
48            "app.bsky.feed.post".to_string(),
49        ])
50        .timeout_time_sec(20) // Shorter timeout for faster testing
51        .bound(65536)
52        .build();
53
54    let jetstream = JetstreamConnection::new(opts);
55
56    let mut ingestors: HashMap<String, Box<dyn LexiconIngestor + Send + Sync>> = HashMap::new();
57    ingestors.insert(
58        "fm.teal.alpha.feed.play".to_string(),
59        Box::new(ProfileIngestor),
60    );
61
62    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
63
64    let msg_rx = jetstream.get_msg_rx();
65    let reconnect_tx = jetstream.get_reconnect_tx();
66
67    let monitor_rx = msg_rx.clone();
68    tokio::spawn(async move {
69        let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
70
71        loop {
72            interval.tick().await;
73            let queue_len = monitor_rx.len();
74            if queue_len > 0 {
75                warn!("Queue has {} messages pending", queue_len);
76            }
77        }
78    });
79
80    // Spawn task to process messages (there should be none)
81    let c_cursor = cursor.clone();
82    tokio::spawn(async move {
83        info!("Message processing task started");
84        let mut message_count = 0u64;
85        let mut last_log = std::time::Instant::now();
86
87        let ing = Ingestors {
88            commits: ingestors,
89            identity: None,
90            account: None,
91        };
92
93        while let Ok(message) = msg_rx.recv().await {
94            message_count += 1;
95
96            // Log every 10 messages to see if we're actually processing
97            if message_count % 10 == 0 || last_log.elapsed() > std::time::Duration::from_secs(5) {
98                info!(
99                    "Processing message #{} (queue len: {}/{})",
100                    message_count,
101                    msg_rx.len(),
102                    msg_rx.capacity().unwrap_or(0)
103                );
104                last_log = std::time::Instant::now();
105            }
106
107            match handler::handle_message(message, &ing, reconnect_tx.clone(), c_cursor.clone())
108                .await
109            {
110                Ok(_) => {}
111                Err(e) => {
112                    error!("Error processing message #{}: {}", message_count, e);
113                }
114            }
115        }
116
117        error!("Message processing task ended unexpectedly!");
118    });
119
120    // Add a monitoring task to track connection health
121    let start_time = std::time::Instant::now();
122    tokio::spawn(async move {
123        let mut interval = tokio::time::interval(std::time::Duration::from_secs(15));
124        let mut tick_count = 0;
125
126        loop {
127            interval.tick().await;
128            tick_count += 1;
129            let elapsed = start_time.elapsed().as_secs();
130            info!(
131                "Health check #{} - uptime: {}s (expecting reconnects every ~20s)",
132                tick_count, elapsed
133            );
134        }
135    });
136
137    info!("Connecting to jetstream (testing reconnection fix)...");
138    info!("Watch for 'Stream closed by server' or 'No messages received' followed by reconnection");
139
140    // This should connect, sit idle, get disconnected, then reconnect automatically
141    // The fix should prevent infinite loops when reconnecting
142    jetstream
143        .connect(cursor.clone())
144        .await
145        .map_err(|e| anyhow::anyhow!("error running ingest: {}", e))
146}

examples/reproduce-stall.rs (line 64)

42pub async fn start_idle_test() -> Result<()> {
43    info!("Testing reconnection fix with idle connection...");
44
45    let opts = JetstreamOptions::builder()
46        .wanted_collections(vec![
47            // This is a custom collection that will get zero traffic
48            "moe.hayden.blogi.actor.profile".to_string(),
49        ])
50        .timeout_time_sec(20) // Shorter timeout for faster testing
51        .bound(65536)
52        .build();
53
54    let jetstream = JetstreamConnection::new(opts);
55
56    let mut ingestors: HashMap<String, Box<dyn LexiconIngestor + Send + Sync>> = HashMap::new();
57    ingestors.insert(
58        "fm.teal.alpha.feed.play".to_string(),
59        Box::new(ProfileIngestor),
60    );
61
62    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
63
64    let msg_rx = jetstream.get_msg_rx();
65    let reconnect_tx = jetstream.get_reconnect_tx();
66
67    let monitor_rx = msg_rx.clone();
68    tokio::spawn(async move {
69        let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
70
71        loop {
72            interval.tick().await;
73            let queue_len = monitor_rx.len();
74            if queue_len > 0 {
75                warn!("Queue has {} messages pending", queue_len);
76            }
77        }
78    });
79
80    // Spawn task to process messages (there should be none)
81    let c_cursor = cursor.clone();
82    tokio::spawn(async move {
83        info!("Message processing task started");
84        let mut message_count = 0u64;
85        let mut last_log = std::time::Instant::now();
86
87        let ing = Ingestors {
88            commits: ingestors,
89            identity: None,
90            account: None,
91        };
92
93        while let Ok(message) = msg_rx.recv().await {
94            message_count += 1;
95
96            // Log every 10 messages to see if we're actually processing
97            if message_count % 10 == 0 || last_log.elapsed() > std::time::Duration::from_secs(5) {
98                info!(
99                    "Processing message #{} (queue len: {}/{})",
100                    message_count,
101                    msg_rx.len(),
102                    msg_rx.capacity().unwrap_or(0)
103                );
104                last_log = std::time::Instant::now();
105            }
106
107            match handler::handle_message(message, &ing, reconnect_tx.clone(), c_cursor.clone())
108                .await
109            {
110                Ok(_) => {}
111                Err(e) => {
112                    error!("Error processing message #{}: {}", message_count, e);
113                }
114            }
115        }
116
117        error!("Message processing task ended unexpectedly!");
118    });
119
120    // Add a monitoring task to track connection health
121    let start_time = std::time::Instant::now();
122    tokio::spawn(async move {
123        let mut interval = tokio::time::interval(std::time::Duration::from_secs(15));
124        let mut tick_count = 0;
125
126        loop {
127            interval.tick().await;
128            tick_count += 1;
129            let elapsed = start_time.elapsed().as_secs();
130            info!(
131                "Health check #{} - uptime: {}s (expecting reconnects every ~20s)",
132                tick_count, elapsed
133            );
134        }
135    });
136
137    info!("Connecting to jetstream (testing reconnection fix)...");
138    info!("Watch for 'Stream closed by server' or 'No messages received' followed by reconnection");
139
140    // This should connect, sit idle, get disconnected, then reconnect automatically
141    // The fix should prevent infinite loops when reconnecting
142    jetstream
143        .connect(cursor.clone())
144        .await
145        .map_err(|e| anyhow::anyhow!("error running ingest: {}", e))
146}

Source

pub async fn connect( &self, cursor: Arc<Mutex<Option<u64>>>, ) -> Result<(), Box<dyn Error>>

Examples found in repository ?

examples/comprehensive-example.rs (line 66)

13async fn main() {
14    // set up logging
15    tracing_subscriber::fmt()
16        .with_max_level(tracing::Level::INFO)
17        .init();
18
19    // init the builder with wanted collections
20    let opts = JetstreamOptions::builder()
21        .wanted_collections(vec![
22            "app.bsky.feed.post".to_string(),
23            "xyz.statusphere.status".to_string(),
24        ])
25        .build();
26
27    // create the jetstream connector
28    let jetstream = JetstreamConnection::new(opts);
29
30    // create your ingestors
31    let mut ingestors = Ingestors::new();
32
33    // register commit ingestors
34    ingestors.commits.insert(
35        "xyz.statusphere.status".to_string(),
36        Box::new(StatusphereIngestor),
37    );
38
39    // register identity ingestor
40    ingestors.identity = Some(Box::new(IdentityIngestor));
41
42    // register account ingestor
43    ingestors.account = Some(Box::new(AccountIngestor));
44
45    // tracks the last message we've processed
46    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
47
48    // get channels
49    let msg_rx = jetstream.get_msg_rx();
50    let reconnect_tx = jetstream.get_reconnect_tx();
51
52    // spawn a task to process messages from the queue.
53    let c_cursor = cursor.clone();
54    tokio::spawn(async move {
55        while let Ok(message) = msg_rx.recv().await {
56            if let Err(e) =
57                handler::handle_message(message, &ingestors, reconnect_tx.clone(), c_cursor.clone())
58                    .await
59            {
60                eprintln!("Error processing message: {}", e);
61            };
62        }
63    });
64
65    // connect to jetstream
66    if let Err(e) = jetstream.connect(cursor.clone()).await {
67        eprintln!("Failed to connect to Jetstream: {}", e);
68        std::process::exit(1);
69    }
70}

More examples

Hide additional examples

examples/spew-bsky-posts.rs (line 63)

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}

examples/bsky-posts.rs (line 143)

42pub async fn start_idle_test() -> Result<()> {
43    info!("Testing reconnection fix with idle connection...");
44
45    let opts = JetstreamOptions::builder()
46        .wanted_collections(vec![
47            // This is a custom collection that will get zero traffic
48            "app.bsky.feed.post".to_string(),
49        ])
50        .timeout_time_sec(20) // Shorter timeout for faster testing
51        .bound(65536)
52        .build();
53
54    let jetstream = JetstreamConnection::new(opts);
55
56    let mut ingestors: HashMap<String, Box<dyn LexiconIngestor + Send + Sync>> = HashMap::new();
57    ingestors.insert(
58        "fm.teal.alpha.feed.play".to_string(),
59        Box::new(ProfileIngestor),
60    );
61
62    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
63
64    let msg_rx = jetstream.get_msg_rx();
65    let reconnect_tx = jetstream.get_reconnect_tx();
66
67    let monitor_rx = msg_rx.clone();
68    tokio::spawn(async move {
69        let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
70
71        loop {
72            interval.tick().await;
73            let queue_len = monitor_rx.len();
74            if queue_len > 0 {
75                warn!("Queue has {} messages pending", queue_len);
76            }
77        }
78    });
79
80    // Spawn task to process messages (there should be none)
81    let c_cursor = cursor.clone();
82    tokio::spawn(async move {
83        info!("Message processing task started");
84        let mut message_count = 0u64;
85        let mut last_log = std::time::Instant::now();
86
87        let ing = Ingestors {
88            commits: ingestors,
89            identity: None,
90            account: None,
91        };
92
93        while let Ok(message) = msg_rx.recv().await {
94            message_count += 1;
95
96            // Log every 10 messages to see if we're actually processing
97            if message_count % 10 == 0 || last_log.elapsed() > std::time::Duration::from_secs(5) {
98                info!(
99                    "Processing message #{} (queue len: {}/{})",
100                    message_count,
101                    msg_rx.len(),
102                    msg_rx.capacity().unwrap_or(0)
103                );
104                last_log = std::time::Instant::now();
105            }
106
107            match handler::handle_message(message, &ing, reconnect_tx.clone(), c_cursor.clone())
108                .await
109            {
110                Ok(_) => {}
111                Err(e) => {
112                    error!("Error processing message #{}: {}", message_count, e);
113                }
114            }
115        }
116
117        error!("Message processing task ended unexpectedly!");
118    });
119
120    // Add a monitoring task to track connection health
121    let start_time = std::time::Instant::now();
122    tokio::spawn(async move {
123        let mut interval = tokio::time::interval(std::time::Duration::from_secs(15));
124        let mut tick_count = 0;
125
126        loop {
127            interval.tick().await;
128            tick_count += 1;
129            let elapsed = start_time.elapsed().as_secs();
130            info!(
131                "Health check #{} - uptime: {}s (expecting reconnects every ~20s)",
132                tick_count, elapsed
133            );
134        }
135    });
136
137    info!("Connecting to jetstream (testing reconnection fix)...");
138    info!("Watch for 'Stream closed by server' or 'No messages received' followed by reconnection");
139
140    // This should connect, sit idle, get disconnected, then reconnect automatically
141    // The fix should prevent infinite loops when reconnecting
142    jetstream
143        .connect(cursor.clone())
144        .await
145        .map_err(|e| anyhow::anyhow!("error running ingest: {}", e))
146}

examples/reproduce-stall.rs (line 143)

42pub async fn start_idle_test() -> Result<()> {
43    info!("Testing reconnection fix with idle connection...");
44
45    let opts = JetstreamOptions::builder()
46        .wanted_collections(vec![
47            // This is a custom collection that will get zero traffic
48            "moe.hayden.blogi.actor.profile".to_string(),
49        ])
50        .timeout_time_sec(20) // Shorter timeout for faster testing
51        .bound(65536)
52        .build();
53
54    let jetstream = JetstreamConnection::new(opts);
55
56    let mut ingestors: HashMap<String, Box<dyn LexiconIngestor + Send + Sync>> = HashMap::new();
57    ingestors.insert(
58        "fm.teal.alpha.feed.play".to_string(),
59        Box::new(ProfileIngestor),
60    );
61
62    let cursor: Arc<Mutex<Option<u64>>> = Arc::new(Mutex::new(None));
63
64    let msg_rx = jetstream.get_msg_rx();
65    let reconnect_tx = jetstream.get_reconnect_tx();
66
67    let monitor_rx = msg_rx.clone();
68    tokio::spawn(async move {
69        let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
70
71        loop {
72            interval.tick().await;
73            let queue_len = monitor_rx.len();
74            if queue_len > 0 {
75                warn!("Queue has {} messages pending", queue_len);
76            }
77        }
78    });
79
80    // Spawn task to process messages (there should be none)
81    let c_cursor = cursor.clone();
82    tokio::spawn(async move {
83        info!("Message processing task started");
84        let mut message_count = 0u64;
85        let mut last_log = std::time::Instant::now();
86
87        let ing = Ingestors {
88            commits: ingestors,
89            identity: None,
90            account: None,
91        };
92
93        while let Ok(message) = msg_rx.recv().await {
94            message_count += 1;
95
96            // Log every 10 messages to see if we're actually processing
97            if message_count % 10 == 0 || last_log.elapsed() > std::time::Duration::from_secs(5) {
98                info!(
99                    "Processing message #{} (queue len: {}/{})",
100                    message_count,
101                    msg_rx.len(),
102                    msg_rx.capacity().unwrap_or(0)
103                );
104                last_log = std::time::Instant::now();
105            }
106
107            match handler::handle_message(message, &ing, reconnect_tx.clone(), c_cursor.clone())
108                .await
109            {
110                Ok(_) => {}
111                Err(e) => {
112                    error!("Error processing message #{}: {}", message_count, e);
113                }
114            }
115        }
116
117        error!("Message processing task ended unexpectedly!");
118    });
119
120    // Add a monitoring task to track connection health
121    let start_time = std::time::Instant::now();
122    tokio::spawn(async move {
123        let mut interval = tokio::time::interval(std::time::Duration::from_secs(15));
124        let mut tick_count = 0;
125
126        loop {
127            interval.tick().await;
128            tick_count += 1;
129            let elapsed = start_time.elapsed().as_secs();
130            info!(
131                "Health check #{} - uptime: {}s (expecting reconnects every ~20s)",
132                tick_count, elapsed
133            );
134        }
135    });
136
137    info!("Connecting to jetstream (testing reconnection fix)...");
138    info!("Watch for 'Stream closed by server' or 'No messages received' followed by reconnection");
139
140    // This should connect, sit idle, get disconnected, then reconnect automatically
141    // The fix should prevent infinite loops when reconnecting
142    jetstream
143        .connect(cursor.clone())
144        .await
145        .map_err(|e| anyhow::anyhow!("error running ingest: {}", e))
146}