vielle.dev
this repo has no description
move from Arc<Mutex<VecDeque<T>>> to broadcast for channel system
this creates a spmc architecture so we can create com.atproto.sync.subscribeRepos and implement jetstream etc
+15
-5
src/ingest/handler.rs
+15
-5
src/ingest/handler.rs
···
1
-
use std::{collections::VecDeque, sync::Arc};
2
3
use futures_util::future;
4
use ipld_core::ipld::Ipld;
···
9
use jacquard_repo::{BlockStore, MemoryBlockStore};
10
use sqlx::{Pool, Postgres, query};
11
use thiserror::Error;
12
-
use tokio::{sync::Mutex, task::JoinHandle};
13
14
use crate::{backfill::backfill, utils::ipld_json::ipld_to_json_value};
15
···
157
}
158
159
pub fn ingest(
160
-
queue: Arc<Mutex<VecDeque<SubscribeReposMessage<'static>>>>,
161
conn: Arc<Pool<Postgres>>,
162
) -> JoinHandle<()> {
163
tokio::spawn(async move {
164
loop {
165
-
let Some(next) = queue.lock().await.pop_front() else {
166
-
continue;
167
};
168
169
match next {
···
1
+
use std::sync::Arc;
2
3
use futures_util::future;
4
use ipld_core::ipld::Ipld;
···
9
use jacquard_repo::{BlockStore, MemoryBlockStore};
10
use sqlx::{Pool, Postgres, query};
11
use thiserror::Error;
12
+
use tokio::{sync::broadcast, task::JoinHandle};
13
14
use crate::{backfill::backfill, utils::ipld_json::ipld_to_json_value};
15
···
157
}
158
159
pub fn ingest(
160
+
mut reciever: broadcast::Receiver<SubscribeReposMessage<'static>>,
161
conn: Arc<Pool<Postgres>>,
162
) -> JoinHandle<()> {
163
tokio::spawn(async move {
164
loop {
165
+
let next = match reciever.recv().await {
166
+
Ok(val) => val,
167
+
Err(err) => match err {
168
+
broadcast::error::RecvError::Closed => {
169
+
eprintln!("Ingestion failed. Quitting");
170
+
break;
171
+
}
172
+
broadcast::error::RecvError::Lagged(skipped) => {
173
+
eprintln!("Warning: lagging behind. Skipping {skipped} messages");
174
+
continue;
175
+
}
176
+
},
177
};
178
179
match next {
+8
-18
src/ingest/queue.rs
+8
-18
src/ingest/queue.rs
···
1
use std::thread;
2
use std::time::Duration;
3
-
use std::{collections::VecDeque, sync::Arc};
4
5
use futures_util::stream::StreamExt;
6
use jacquard::api::com_atproto::sync::subscribe_repos::{SubscribeRepos, SubscribeReposMessage};
7
use jacquard::url::Url;
8
use jacquard::{common::xrpc::TungsteniteSubscriptionClient, xrpc::SubscriptionClient};
9
-
use tokio::{
10
-
sync::Mutex,
11
-
task::{self, JoinHandle},
12
-
};
13
14
use crate::config;
15
16
-
pub async fn queue() -> (
17
-
Arc<Mutex<VecDeque<SubscribeReposMessage<'static>>>>,
18
-
JoinHandle<()>,
19
-
) {
20
-
let queue = Arc::new(Mutex::new(VecDeque::new()));
21
22
// USER_SUBSCRIBE_URL is formatted as a domain
23
let uri = Url::parse(&format!("wss://{}/", config::USER_SUBSCRIBE_URL))
···
29
.expect("Could not subscribe to new events")
30
.into_stream();
31
32
-
let queue_clone = queue.clone();
33
-
let handle = task::spawn(async move {
34
-
let queue = queue_clone;
35
-
36
loop {
37
if let Some(msg) = messages.next().await {
38
let msg = match msg {
···
126
SubscribeReposMessage::Unknown(data) => SubscribeReposMessage::Unknown(data),
127
};
128
129
-
queue.lock().await.push_back(ev);
130
}
131
}
132
-
});
133
-
134
-
(queue, handle)
135
}
···
1
use std::thread;
2
use std::time::Duration;
3
4
use futures_util::stream::StreamExt;
5
use jacquard::api::com_atproto::sync::subscribe_repos::{SubscribeRepos, SubscribeReposMessage};
6
use jacquard::url::Url;
7
use jacquard::{common::xrpc::TungsteniteSubscriptionClient, xrpc::SubscriptionClient};
8
+
use tokio::sync::broadcast;
9
+
use tokio::task::{self, JoinHandle};
10
11
use crate::config;
12
13
+
pub async fn queue<'a>(tx: broadcast::Sender<SubscribeReposMessage<'static>>) -> JoinHandle<()> {
14
+
// let queue = Arc::new(Mutex::new(VecDeque::new()));
15
16
// USER_SUBSCRIBE_URL is formatted as a domain
17
let uri = Url::parse(&format!("wss://{}/", config::USER_SUBSCRIBE_URL))
···
23
.expect("Could not subscribe to new events")
24
.into_stream();
25
26
+
// let queue_clone = queue.clone();
27
+
task::spawn(async move {
28
loop {
29
if let Some(msg) = messages.next().await {
30
let msg = match msg {
···
118
SubscribeReposMessage::Unknown(data) => SubscribeReposMessage::Unknown(data),
119
};
120
121
+
let _ = tx.send(ev);
122
}
123
}
124
+
})
125
}
+6
-3
src/main.rs
+6
-3
src/main.rs
···
1
use crate::backfill::backfill;
2
3
mod backfill;
···
35
let conn = db::conn().await;
36
println!("Database connected and initialized");
37
38
-
let (queue, queue_handle) = ingest::queue().await;
39
40
println!("Starting backfill");
41
let timer = std::time::Instant::now();
···
55
});
56
57
println!("Handling new events. Ctrl+C to quit.");
58
-
let ingest_handle = ingest::ingest(queue, conn.clone());
59
60
let _ = rx.await;
61
-
queue_handle.abort();
62
ingest_handle.abort();
63
64
Ok(())
···
1
+
use tokio::sync::broadcast;
2
+
3
use crate::backfill::backfill;
4
5
mod backfill;
···
37
let conn = db::conn().await;
38
println!("Database connected and initialized");
39
40
+
let (tx, ingest_reciever) = broadcast::channel(16);
41
+
let broadcast_handle = ingest::queue(tx).await;
42
43
println!("Starting backfill");
44
let timer = std::time::Instant::now();
···
58
});
59
60
println!("Handling new events. Ctrl+C to quit.");
61
+
let ingest_handle = ingest::ingest(ingest_reciever, conn.clone());
62
63
let _ = rx.await;
64
+
broadcast_handle.abort();
65
ingest_handle.abort();
66
67
Ok(())