ruststream 0.6.1

Async messaging framework for Rust: broker-agnostic traits, router, codecs, and a conformance harness for broker authors.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

//! Repositioning live subscriptions with the `Seekable` capability: a `Seek` handler
//! parameter injects the subscription's own seeker (skipping forward past a poison region),
//! and a `StartAt` source opens a subscription at a chosen log position (replaying history
//! into a fresh subscription, or starting from the latest).
//!
//! ```text
//! cargo run --example seek --features macros,memory,json
//! ```

use std::error::Error;
use std::time::Duration;

use ruststream::memory::{MemoryBroker, MemoryPosition, MemorySeeker, MemorySource};
use ruststream::runtime::{AppInfo, HandlerResult, RustStream, Seek};
use ruststream::{OutgoingMessage, Publisher, Seeker, subscriber};
use serde::{Deserialize, Serialize};
use tokio::time::sleep;

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

// --8<-- [start:start_at]
/// The audit trail: its subscription opens at the start of the log, so entries published
/// before the service started are replayed into it.
#[subscriber("audit", start_at(MemoryPosition::start()))]
async fn record(entry: &Job) -> HandlerResult {
    println!("audit: entry {}", entry.id);
    HandlerResult::Ack
}
// --8<-- [end:start_at]

// --8<-- [start:handler]
/// Skips forward when the producer marks a poison region: everything queued before the
/// resume point is dropped without touching the subscription itself.
#[subscriber(MemorySource::new("jobs"))]
async fn work(job: &Job, Seek(seeker): Seek<MemorySeeker>) -> HandlerResult {
    if job.id == 999 {
        // The poison marker carries the resume point: skip to the fourth log entry.
        if seeker.seek(MemoryPosition::sequence(3)).await.is_err() {
            return HandlerResult::retry();
        }
        return HandlerResult::Ack;
    }
    println!("jobs: processed {}", job.id);
    HandlerResult::Ack
}
// --8<-- [end:handler]

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let broker = MemoryBroker::new();
    let ingress = broker.publisher();

    // Published before the app even exists; only the chosen start position below makes these
    // visible to the audit subscription.
    for id in 1..=2u64 {
        let payload = serde_json::to_vec(&Job { id })?;
        ingress
            .publish(OutgoingMessage::new("audit", payload.as_slice()))
            .await?;
    }

    // --8<-- [start:mount]
    // Both mount plainly: the runtime mints the seek handler's seeker off its subscription
    // right after it opens, and the start_at clause seeks the audit one before its first
    // delivery.
    let app = RustStream::new(AppInfo::new("seek-demo", "0.1.0")).with_broker(broker, |b| {
        b.include(work);
        b.include(record);
    });
    // --8<-- [end:mount]
    let running = app.start().await?;

    // The jobs stream hits a poison marker at the second position; the handler's own seek
    // jumps it to the fourth, so id 3 is never processed.
    for id in [1, 999, 3, 4] {
        let payload = serde_json::to_vec(&Job { id })?;
        ingress
            .publish(OutgoingMessage::new("jobs", payload.as_slice()))
            .await?;
    }
    // A demo-only pause so the dispatch loops drain; a real service reacts to its own signals.
    sleep(Duration::from_millis(100)).await;

    running.shutdown().await?;
    println!("ok: replayed the audit history and skipped the poisoned region");
    Ok(())
}