Struct Subscription 

pub struct Subscription<S> { /* private fields */ }

User-facing subscription handle.

Holds a shared reference to a Store<S> backend (one Arc clone) and exposes subscribe / subscribe_all. Cheap to construct; no Arc ever appears in user code.

Example

use core::pin::pin;
use futures::StreamExt;
use mnesis_store::{Step, StepStreamExt, Store, Subscription};

let store = Store::new(FjallStore::builder("path").open()?);
// Items are `Step<PersistedEnvelope>`: tell catch-up from live directly.
let cursor = Subscription::new(&store).subscribe(&account_id, None)?;
let mut cursor = pin!(cursor);
while let Some(item) = cursor.next().await {
    match item? {
        Step::CaughtUp => { /* backlog drained — now live */ }
        Step::Event(env) => { /* handle the raw envelope */ }
    }
}
// …or `.events()` to drop the phase, `.decoded(codec)` to decode+keep it.

Implementations

impl<S> Subscription<S>

pub fn new(store: &Store<S>) -> Self

Construct from a Store<S> handle. One Arc::clone per call.

impl<S: RawEventStore + WakeSource> Subscription<S>

pub fn subscribe<I: Id>( &self, id: &I, from: Option<Version>, ) -> Result<impl Stream<Item = Result<Step<PersistedEnvelope>, <S as RawEventStore>::Error>> + Send + use<S, I>, <S as WakeSource>::Error>

where <S as RawEventStore>::Stream: Unpin,

Open a per-stream catch-up + live-tail subscription.

from: None starts from version 1; from: Some(v) starts from the event strictly after version v. Items are Step<PersistedEnvelope>: the replay events, then exactly one Step::CaughtUp at the backlog→live boundary, then live events — the phase marker is intrinsic to a subscription (a finite read has none). Checkpoint by version() on each event.

Compose from here: .events() drops the phase for a consumer that only wants events (then the full DecodedStreamExt applies); .decoded(codec) keeps the phase and hands you typed Step<Decoded<E>>. The returned stream never returns None — it waits for new events when caught up — and is !Unpin, so pin! it before polling.

Errors

<S as WakeSource>::Error if wake-registration fails. Read errors are surfaced as Err items in the stream.

pub fn subscribe_all( &self, from: Option<<S as RawEventStore>::AllPosition>, ) -> Result<impl Stream<Item = Result<Step<(<S as RawEventStore>::AllPosition, StreamKey, PersistedEnvelope)>, <S as RawEventStore>::Error>> + Send + use<S>, <S as WakeSource>::Error>

where <S as RawEventStore>::AllStream: Unpin,

Open an all-streams ($all) catch-up + live-tail subscription in AllPosition order.

from: None starts from the first event ever appended; from: Some(p) starts from the event strictly after position p. Items are Step<(AllPosition, StreamKey, PersistedEnvelope)>: the replay events, then exactly one Step::CaughtUp, then live events. Each event carries three parts beside the box: the position to checkpoint (the consumer hands it back here or to read_all to resume; the checkpoint type is adapter-defined and must be serializable), the stream key to route on without decoding the payload, and the envelope for content.

Compose with .events() / .decoded(codec) exactly as subscribe. The returned stream never returns None and is !Unpin, so pin! it before polling.

Errors

<S as WakeSource>::Error if wake-registration fails. Read errors are surfaced as Err items in the stream.

Trait Implementations

impl<S> Debug for Subscription<S>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.