pamoja_sync/lib.rs
1//! Offline-first store-and-forward buffering for pamoja.
2//!
3//! Devices on intermittent links append records while offline and drain them in
4//! order when a link returns, so a node that is disconnected for hours or days
5//! loses nothing. This crate provides durable first-in first-out implementations
6//! of the core [`Store`](pamoja_core::Store) trait:
7//!
8//! - [`MemoryStore`] - a fast in-memory queue, optionally capacity-bounded so a
9//! full queue becomes an explicit backpressure signal.
10//! - [`FileStore`] - a crash-safe on-disk queue that survives a restart or a
11//! power cut, optionally bounded so a long outage cannot fill the disk.
12//!
13//! Both buffer raw bytes, so an application pairs a [`Store`](pamoja_core::Store)
14//! with a [`Codec`](https://docs.rs/pamoja-codec) to persist encoded payloads and
15//! a [`Transport`](pamoja_core::Transport) to forward them when a link appears.
16//! [`drain_to`] is the forward half: it publishes buffered records onto a
17//! transport in order, removing each only after it is sent, so a failed link
18//! loses nothing.
19//!
20//! # Examples
21//!
22//! A rain gauge out of range for a night keeps its readings in an outbox, and hands them
23//! on in order when the link returns. Here the link is an in-process loopback, so the
24//! gateway on the other end can check what arrived.
25//!
26//! ```
27//! use pamoja_core::{Receive, Store, Transport};
28//! use pamoja_loopback::{LoopbackBroker, LoopbackTransport};
29//! use pamoja_sync::{drain_to, MemoryStore};
30//!
31//! # tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
32//! let mut outbox = MemoryStore::new();
33//! outbox.append(b"0.4").await?;
34//! outbox.append(b"1.2").await?;
35//!
36//! let broker = LoopbackBroker::new();
37//! let mut gateway = LoopbackTransport::new(broker.clone());
38//! let mut link = LoopbackTransport::new(broker);
39//! gateway.connect().await?;
40//! gateway.subscribe("gauges/3/rain").await?;
41//! link.connect().await?;
42//!
43//! let forwarded = drain_to(&mut outbox, &mut link, "gauges/3/rain").await?;
44//! assert_eq!(forwarded, 2);
45//! assert_eq!(outbox.len().await?, 0, "each record goes once it is sent");
46//! assert_eq!(gateway.recv().await?.expect("the first").payload, b"0.4");
47//! assert_eq!(gateway.recv().await?.expect("the second").payload, b"1.2");
48//! # Ok::<(), pamoja_core::Error>(())
49//! # }).unwrap();
50//! ```
51
52mod file;
53mod forward;
54mod memory;
55
56pub use file::FileStore;
57pub use forward::drain_to;
58pub use memory::MemoryStore;