Streams and Watchers
Streams and watchers in Locus — live file reads, change notifications, and per-path event subscriptions.
Purpose
File and directory model with metadata, mounts, streams, watchers, journals, permission modes, and governance.
This page follows the real source shape for Locus and explains the workflow a developer is likely to use first.
Developer workflow
Start from the smallest constructor or builder, perform one meaningful operation, inspect the returned state, then add the relevant policy, storage, or network integration. The examples below should be expanded whenever the crate API changes.
Watchers receive events for every path change in the drive. Filter at the watcher boundary rather than in your handler — discarding events early keeps memory bounded under bursty writes.
Primary types to know
AllowAllLocusPolicy— models/locus/src/policy.rsCreateOptions— models/locus/src/ops.rsFileEntry— models/locus/src/entry.rsFileMetadata— models/locus/src/entry.rsJournalRecord— models/locus/src/journal.rsJournalStore— models/locus/src/journal.rsLocus— models/locus/src/locus.rsLocusConfig— models/locus/src/config.rsMountOptions— models/locus/src/config.rsMountPoint— models/locus/src/mount.rsMountTable— models/locus/src/mount.rsOpenOptions— models/locus/src/ops.rs
Example shape
use bytes::Bytes;
use locus::{ReadStream, WatchEvent, Watcher, WriteStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// ReadStream is an AsyncRead over an in-memory Bytes payload.
let mut reader = ReadStream::new(Bytes::from_static(b"hello locus stream"));
let mut buf = Vec::new();
reader.read_to_end(&mut buf).await?;
// WriteStream is an AsyncWrite with an in-memory buffer up to `max_size`,
// optionally draining via a flush callback (commit to blob storage, etc).
let mut writer = WriteStream::new(64 * 1024);
writer.write_all(b"locus write").await?;
writer.shutdown().await?;
let committed: Bytes = writer.into_bytes();
// Watchers fan out path-scoped events to broadcast subscribers; the engine
// wires them in via Locus::watch(). Here we drive one manually to inspect
// the WatchEvent variants a subscriber will receive.
let watcher = Watcher::new();
let mut rx = watcher.subscribe();
watcher.emit_created("/docs/notes.md".into(), false);
watcher.emit_modified("/docs/notes.md".into(), false);
let evt = rx.recv().await?;
assert!(matches!(evt, WatchEvent::Created { .. }));
println!(
"read={} bytes committed={} bytes watcher_id={:?}",
buf.len(),
committed.len(),
watcher.id(),
);
Ok(())
}