microcosm.blue
Server tools to backfill, tail, mirror, and verify PLC logs
add example for op polling
bad-example.com
+35
examples/poll.rs
+35
examples/poll.rs
···
1
+
use allegedly::{ExportPage, poll_upstream};
2
+
3
+
#[tokio::main]
4
+
async fn main() {
5
+
// set to `None` to replay from the beginning of the PLC history
6
+
let after = Some(chrono::Utc::now());
7
+
8
+
// the PLC server to poll for new ops
9
+
let upstream = "https://plc.wtf/export".parse().unwrap();
10
+
11
+
// self-rate-limit (plc.directory's limit interval is 600ms)
12
+
let throttle = std::time::Duration::from_millis(300);
13
+
14
+
// pages are sent out of the poller via a tokio mpsc channel
15
+
let (tx, mut rx) = tokio::sync::mpsc::channel(1);
16
+
17
+
// spawn a tokio task to run the poller
18
+
tokio::task::spawn(poll_upstream(after, upstream, throttle, tx));
19
+
20
+
// receive pages of plc ops from the poller
21
+
while let Some(ExportPage { ops }) = rx.recv().await {
22
+
println!("received {} plc ops", ops.len());
23
+
24
+
for op in ops {
25
+
// in this example we're alerting when changes are found for one
26
+
// specific identity
27
+
if op.did == "did:plc:hdhoaan3xa3jiuq4fg4mefid" {
28
+
println!(
29
+
"Update found for {}! cid={}\n -> operation: {}",
30
+
op.did, op.cid, op.operation.get()
31
+
);
32
+
}
33
+
}
34
+
}
35
+
}
+54
-2
src/poll.rs
+54
-2
src/poll.rs
···
51
51
}
52
52
}
53
53
54
-
/// PLC
54
+
/// State for removing duplicates ops between PLC export page boundaries
55
55
#[derive(Debug, PartialEq)]
56
56
pub struct PageBoundaryState {
57
+
/// The previous page's last timestamp
58
+
///
59
+
/// Duplicate ops from /export only occur for the same exact timestamp
57
60
pub last_at: Dt,
61
+
/// The previous page's ops at its last timestamp
58
62
keys_at: Vec<OpKey>, // expected to ~always be length one
59
63
}
60
64
61
-
/// track keys at final createdAt to deduplicate the start of the next page
62
65
impl PageBoundaryState {
66
+
/// Initialize the boundary state with a PLC page
63
67
pub fn new(page: &ExportPage) -> Option<Self> {
64
68
// grab the very last op
65
69
let (last_at, last_key) = page.ops.last().map(|op| (op.created_at, op.into()))?;
···
75
79
76
80
Some(me)
77
81
}
82
+
/// Apply the deduplication and update state
83
+
///
84
+
/// The beginning of the page will be modified to remove duplicates from the
85
+
/// previous page.
86
+
///
87
+
/// The end of the page is inspected to update the deduplicator state for
88
+
/// the next page.
78
89
fn apply_to_next(&mut self, page: &mut ExportPage) {
79
90
// walk ops forward, kicking previously-seen ops until created_at advances
80
91
let to_remove: Vec<usize> = page
···
124
135
}
125
136
}
126
137
138
+
/// Get one PLC export page
139
+
///
140
+
/// Extracts the final op so it can be used to fetch the following page
127
141
pub async fn get_page(url: Url) -> Result<(ExportPage, Option<LastOp>), GetPageError> {
128
142
log::trace!("Getting page: {url}");
129
143
···
152
166
Ok((ExportPage { ops }, last_op))
153
167
}
154
168
169
+
/// Poll an upstream PLC server for new ops
170
+
///
171
+
/// Pages of operations are written to the `dest` channel.
172
+
///
173
+
/// ```no_run
174
+
/// # #[tokio::main]
175
+
/// # async fn main() {
176
+
/// use allegedly::{ExportPage, Op, poll_upstream};
177
+
///
178
+
/// // set to `None` to replay from the beginning of the PLC history
179
+
/// let after = Some(chrono::Utc::now());
180
+
///
181
+
/// // the PLC server to poll for new ops
182
+
/// let upstream = "https://plc.wtf/export".parse().unwrap();
183
+
///
184
+
/// // self-rate-limit (plc.directory's limit interval is 600ms)
185
+
/// let throttle = std::time::Duration::from_millis(300);
186
+
///
187
+
/// // pages are sent out of the poller via a tokio mpsc channel
188
+
/// let (tx, mut rx) = tokio::sync::mpsc::channel(1);
189
+
///
190
+
/// // spawn a tokio task to run the poller
191
+
/// tokio::task::spawn(poll_upstream(after, upstream, throttle, tx));
192
+
///
193
+
/// // receive pages of plc ops from the poller
194
+
/// while let Some(ExportPage { ops }) = rx.recv().await {
195
+
/// println!("received {} plc ops", ops.len());
196
+
///
197
+
/// for Op { did, cid, operation, .. } in ops {
198
+
/// // in this example we're alerting when changes are found for one
199
+
/// // specific identity
200
+
/// if did == "did:plc:hdhoaan3xa3jiuq4fg4mefid" {
201
+
/// println!("Update found for {did}! cid={cid}\n -> operation: {}", operation.get());
202
+
/// }
203
+
/// }
204
+
/// }
205
+
/// # }
206
+
/// ```
155
207
pub async fn poll_upstream(
156
208
after: Option<Dt>,
157
209
base: Url,