Enum FilePosition 

pub enum FilePosition {
    Beginning,
    End,
    Sequence(u64),
    Timestamp(u64),
}

A position in a stream file’s retained log: where start_at(..) opens a subscription, and where Seeker::seek moves a running one.

Captured positions (Positioned::position) carry the pinned semantics the framework defines: seeking to one redelivers exactly that message (the transport’s sequence rewind is inclusive). The other forms keep the transport’s own semantics: Beginning replays everything retained, End skips to the tip, and Timestamp resumes at the earliest message strictly later than the instant (milliseconds since the Unix epoch).

Seek to a position the stream has reached, or to the sequence it writes next, which is the tip: a service that handled every message resumes there. A sequence further ahead, and an instant later than every message it holds, are both refused - the transport reads forward looking for the message and runs out of file - and the subscription ends with that refusal rather than staying where it was. A captured position, the beginning and the tip are always safe; a number a service worked out for itself is what needs the care.

The FilePosition page shows it in a service.

Examples

A handler reads its delivery’s place in the file with Ctx<Position> and moves its own subscription with Ctx<SeekHandle>:

use ruststream_sea_file::file::prelude::*;
use serde::Deserialize;

#[derive(Debug, Deserialize)]
struct Job {
    id: u64,
    poisoned: bool,
}

#[subscriber(FileStream::new("jobs"), start_at(FilePosition::beginning()))]
async fn work(
    job: &Job,
    Ctx(at): Ctx<Position>,
    Ctx(seeker): Ctx<SeekHandle>,
) -> HandlerOutcome {
    if job.poisoned && seeker.seek(FilePosition::end()).await.is_err() {
        return HandlerOutcome::retry();
    }
    tracing::info!(job.id, position = ?at, "worked");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("jobs", "0.1.0"))
        .with_broker(FileBroker::new("/tmp/jobs.ss"), |b| {
            b.include(work);
        })
}

Variants

Beginning

Everything retained.

End

The tip of the stream.

Sequence(u64)

A message sequence (inclusive). A stream key is numbered from one, and the numbering carries on when a later connection appends to the same file.

Timestamp(u64)

Milliseconds since the Unix epoch (exclusive).

Implementations

impl FilePosition

pub const fn beginning() -> Self

Everything retained: FilePosition::Beginning.

pub const fn end() -> Self

The tip of the stream: FilePosition::End.

pub const fn sequence(sequence: u64) -> Self

A message sequence, redelivered inclusively: FilePosition::Sequence.

pub const fn timestamp(millis: u64) -> Self

Milliseconds since the Unix epoch, resuming strictly later: FilePosition::Timestamp.

Trait Implementations

impl Clone for FilePosition

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Copy for FilePosition

impl Debug for FilePosition

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

Formats the value using the given formatter.

impl Eq for FilePosition

impl PartialEq for FilePosition

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl StructuralPartialEq for FilePosition