very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
rust fjall at-protocol atproto indexer
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273use std::sync::atomic::{AtomicBool, Ordering};use std::time::{Duration, Instant};
use parking_lot::{Condvar, Mutex};
/// tells hydrant's own threads to wind down. they sleep on it instead of/// `thread::sleep`, so a shutdown doesn't wait out a whole interval.#[derive(Default)]pub struct Stop { // the ingest shards check this per message, so it stays a plain load stopped: AtomicBool, lock: Mutex<()>, changed: Condvar,}
impl Stop { pub fn trigger(&self) { // set under the lock, or a waiter that just checked could miss the notify let _lock = self.lock.lock(); self.stopped.store(true, Ordering::Release); self.changed.notify_all(); }
pub fn is_set(&self) -> bool { self.stopped.load(Ordering::Acquire) }
/// sleeps for `timeout`, or less if stopped meanwhile. true means stop. pub fn wait(&self, timeout: Duration) -> bool { let deadline = Instant::now() + timeout; let mut lock = self.lock.lock(); while !self.is_set() { if self.changed.wait_until(&mut lock, deadline).timed_out() { break; } } self.is_set() }}
#[cfg(test)]mod tests { use super::*; use std::sync::Arc;
#[test] fn wait_times_out_without_a_trigger() { let stop = Stop::default(); let started = Instant::now(); assert!(!stop.wait(Duration::from_millis(20))); assert!(started.elapsed() >= Duration::from_millis(20)); }
#[test] fn trigger_wakes_a_waiting_thread_early() { let stop = Arc::new(Stop::default()); let waiter = std::thread::spawn({ let stop = stop.clone(); move || { let started = Instant::now(); (stop.wait(Duration::from_secs(60)), started.elapsed()) } }); std::thread::sleep(Duration::from_millis(20)); stop.trigger(); let (stopped, waited) = waiter.join().unwrap(); assert!(stopped); assert!(waited < Duration::from_secs(5)); // and stays set for anyone who checks later assert!(stop.wait(Duration::from_secs(60))); }}