Menu

Guide

Message Broker Rust SDK examples.

Tested Rust SDK workflows for Message Broker stream publish, replay, cursor commit, shared queue processing, retry, and dead-letter handling with private beta route boundaries.

Current availability

Private Beta. Public beta is scheduled for .

Confirm First

Example Status

Rust Examples

Private Beta

These examples are tested as complete native API Cargo programs. They use in-memory state or canned HTTP responses.

Endpoint Scope

Confirm First

Treat these snippets as local native API and scripted HTTP transport examples. Check the reference and compatibility pages before applying the same flow to a live remote endpoint.

Example Snapshot

Example status
Tested
Rust examples
4
Validation
Cargo examples
Source scope
Public documentation boundary

Source Files

Example source bundle
WorkflowStatusCargo targetSource
Publish, replay, and cursor committestedpublish_replaypublish_replay.rs
Shared queue ack, nack, and redeliverytestedshared_queue_processingshared_queue_processing.rs
HTTP queue retry, dead-letter, and DLQ replaytestedhttp_queue_workerhttp_queue_worker.rs
Failure-policy retry exhaustiontestedretry_dead_letterretry_dead_letter.rs

Example Boundaries

Tested Examples

The complete Rust files compile and run as native API Cargo examples. Each uses synthetic in-memory data or scripted responses.

Endpoint Scope

Use the examples for the documented local native API and scripted HTTP transport behavior. Live remote auth and network behavior stay tied to the reference and compatibility pages.

Remote Boundary

In-process examples use InMemoryBroker. The HTTP example uses a scripted transport, so live remote auth and network behavior stay tied to reference and compatibility docs.

Before you begin

This source workflow requires an existing Message Broker source checkout containing the stornamics-message-broker-native package, its Cargo.lock, and a Rust toolchain. It was checked with Rust 1.96.0 and Cargo 1.96.0. For the local service workflow, use the local quickstart.

Use an isolated copy of that source so saving these examples does not overwrite your edits. Work in its Service.MessageBroker directory, which contains Cargo.toml and crates/native-api. The native package has no external dependencies and the commands below run offline.

Each program creates its own synthetic in-memory state and exits. The LocalDisk value in these programs does not exercise disk persistence. The HTTP program uses a scripted transport; it neither opens a connection nor validates the outgoing request. No broker service, token, or second terminal is needed.

  1. Save the four complete files below under their shown crates/native-api/examples/ paths in the isolated source copy.
  2. From the Service.MessageBroker directory, check and run the four targets in order.

Bash

Check and run the Rust examples

Run from the Message Broker source workspace. Each cargo run executes one independent program.

cargo check --locked --offline -p stornamics-message-broker-native --examples
cargo run --locked --offline --quiet -p stornamics-message-broker-native --example publish_replay
cargo run --locked --offline --quiet -p stornamics-message-broker-native --example shared_queue_processing
cargo run --locked --offline --quiet -p stornamics-message-broker-native --example http_queue_worker
cargo run --locked --offline --quiet -p stornamics-message-broker-native --example retry_dead_letter

Each program exits successfully and prints three workflow lines; compare them with Expected Output. If Cargo cannot find the package or target, check the working directory and underscore-form filenames. If compilation fails, confirm that the source package and toolchain match this workflow before adapting the example.

In-memory records disappear when the program exits. The source copy and Cargo build files remain on disk; retain or remove that isolated copy according to your normal local workspace cleanup.

Rust SDK Snippets

Every block is a complete program with its own imports, initialization, and helpers. Save the file under the shown name in your isolated source copy before running its matching Cargo target.

Publish and replay

Rust

Publish and replay

publish_replay.rs

Complete Rust program. Creates synthetic in-memory records, replays them, and commits a cursor. Save as crates/native-api/examples/publish_replay.rs in your isolated source copy.

use std::error::Error;
use stornamics_message_broker_native::{
    ClientConfig, DurabilityMode, InMemoryBroker, NativeClient, PublishMessage, RetentionPolicy,
    StartPosition, Timestamp,
};

fn main() -> Result<(), Box<dyn Error>> {
    let broker = InMemoryBroker::with_start_time(Timestamp::from_unix_millis(10_000));
    let mut client = NativeClient::new(broker, ClientConfig::new("telemetry-producer"))?;

    client.create_stream(
        "telemetry",
        DurabilityMode::LocalDisk,
        RetentionPolicy::default(),
    )?;

    let first = client.publish(
        PublishMessage::new(
            "telemetry",
            Timestamp::from_unix_millis(1_700_000_000_000),
            b"reading-1".to_vec(),
        )
        .with_message_id("reading-1")
        .with_key(b"machine-17".to_vec()),
    )?;

    let batch = client.batch_publish(vec![
        PublishMessage::new(
            "telemetry",
            Timestamp::from_unix_millis(1_700_000_000_001),
            b"reading-2".to_vec(),
        )
        .with_message_id("reading-2"),
        PublishMessage::new(
            "telemetry",
            Timestamp::from_unix_millis(1_700_000_000_002),
            b"reading-3".to_vec(),
        )
        .with_message_id("reading-3"),
    ])?;

    let replay = client.subscribe_replay("telemetry", "analytics", StartPosition::Earliest)?;
    let fetched = client.fetch(&replay, 10)?;

    client.commit_cursor("telemetry", "analytics", "processed", fetched.next_offset)?;

    let mut offsets = vec![first.offset];
    offsets.extend(batch.responses.iter().map(|response| response.offset));

    println!("workflow=publish_replay published_offsets={offsets:?}");
    println!(
        "workflow=publish_replay replayed={} next_offset={}",
        fetched.messages.len(),
        fetched.next_offset
    );
    println!(
        "workflow=publish_replay committed_cursor=processed offset={}",
        fetched.next_offset
    );

    Ok(())
}

Shared queue

Rust

Shared queue

shared_queue_processing.rs

Complete Rust program. Initializes its own in-memory broker and jobs, then acknowledges, rejects, and redelivers work. Save as crates/native-api/examples/shared_queue_processing.rs.

use std::error::Error;
use stornamics_message_broker_native::{
    ClientConfig, DurabilityMode, InMemoryBroker, NativeClient, PublishMessage, RetentionPolicy,
    StartPosition, Timestamp,
};

fn main() -> Result<(), Box<dyn Error>> {
    let broker = InMemoryBroker::with_start_time(Timestamp::from_unix_millis(20_000));
    let mut client = NativeClient::new(broker, ClientConfig::new("worker-producer"))?;

    client.create_stream(
        "jobs",
        DurabilityMode::LocalDisk,
        RetentionPolicy::default(),
    )?;
    client.batch_publish(vec![
        PublishMessage::new(
            "jobs",
            Timestamp::from_unix_millis(1_700_000_100_000),
            b"job-1".to_vec(),
        )
        .with_message_id("job-1"),
        PublishMessage::new(
            "jobs",
            Timestamp::from_unix_millis(1_700_000_100_001),
            b"job-2".to_vec(),
        )
        .with_message_id("job-2"),
    ])?;

    let workers =
        client.subscribe_queue("jobs", "workers", StartPosition::Earliest, Some(30_000))?;

    let first = client.fetch(&workers, 1)?;
    let first_delivery = &first.messages[0];
    let acked = client.ack(&workers, first_delivery)?;

    let second = client.fetch(&workers, 1)?;
    let second_delivery = &second.messages[0];
    let nacked = client.nack(&workers, second_delivery, "worker shutting down", true)?;

    let redelivery = client.fetch(&workers, 1)?;
    let redelivered = &redelivery.messages[0];
    client.ack(&workers, redelivered)?;

    println!(
        "workflow=shared_queue acked_offset={} delivery_id={}",
        acked.offset, acked.delivery_id
    );
    println!(
        "workflow=shared_queue nacked_offset={} will_redeliver={}",
        nacked.offset, nacked.will_redeliver
    );
    println!(
        "workflow=shared_queue redelivered_offset={} attempt={}",
        redelivered.envelope.offset, redelivered.attempt
    );

    Ok(())
}

Scripted HTTP queue

Rust

Scripted HTTP queue

http_queue_worker.rs

Complete Rust program, including ScriptedTransport and response helpers. The transport returns canned responses and discards requests; it does not contact the shown loopback endpoint. Save as crates/native-api/examples/http_queue_worker.rs.

use std::collections::VecDeque;
use std::error::Error;
use stornamics_message_broker_native::{
    ClientConfig, HttpNativeClient, SdkHttpRequest, SdkHttpResponse, SdkHttpTransport,
    SdkTransportError,
};

fn main() -> Result<(), Box<dyn Error>> {
    let transport = ScriptedTransport::new(vec![
        ok(&queue_claim_json("del-1", 1)),
        ok("{\"api_version\":\"v1\",\"stream\":\"jobs\",\"subscription_id\":\"sub-1\",\"delivery_id\":\"del-1\",\"offset\":0,\"attempt\":1,\"available_at_ms\":1700000000250}\n"),
        ok(&queue_claim_json("del-2", 2)),
        ok("{\"api_version\":\"v1\",\"stream\":\"jobs\",\"subscription_id\":\"sub-1\",\"delivery_id\":\"del-2\",\"source_offset\":0,\"dead_letter_stream\":\"jobs.dlq\",\"dead_letter_offset\":0,\"attempt\":2}\n"),
        ok("{\"api_version\":\"v1\",\"stream\":\"jobs.dlq\",\"records\":[{\"offset\":0,\"message_id\":\"dead-letter:jobs:0:del-2\",\"key_hex\":null,\"event_time_ms\":1700000000000,\"ingest_time_ms\":1700000000500,\"headers\":[{\"name\":\"x-stornamics-dead-letter-reason\",\"value_hex\":\"7374696c6c20756e617661696c61626c65\"}],\"payload_hex\":\"6a6f622d31\",\"payload_bytes\":5,\"durability_mode\":\"local-append\"}],\"next_offset\":1,\"end_of_stream\":true,\"diagnostics\":{\"returned_records\":1}}\n"),
    ]);
    let mut client = HttpNativeClient::with_transport(
        "http://127.0.0.1:9101",
        ClientConfig::new("worker-a"),
        transport,
    )?;

    let first_claim = client.claim_queue("jobs", "workers", 0, 1, Some(30_000))?;
    let first_delivery = first_claim.messages[0].clone();
    let retry =
        client.retry_delivery(&first_claim, &first_delivery, "dependency unavailable", 250)?;
    println!(
        "workflow=http_queue_worker retry_scheduled delivery_id={} available_at={}",
        retry.delivery_id, retry.available_at.unix_millis
    );

    let retry_claim = client.claim_queue("jobs", "workers", retry.offset, 1, Some(30_000))?;
    let retry_delivery = retry_claim.messages[0].clone();
    let dead_letter = client.dead_letter(
        &retry_claim,
        &retry_delivery,
        "jobs.dlq",
        "still unavailable",
    )?;
    println!(
        "workflow=http_queue_worker dead_lettered source_offset={} dead_letter_stream={} dead_letter_offset={}",
        dead_letter.source_offset, dead_letter.dead_letter_stream, dead_letter.dead_letter_offset
    );

    let dlq = client.list_dead_letter_records("jobs.dlq", 0, 10)?;
    println!(
        "workflow=http_queue_worker dlq_message_id={} payload={}",
        dlq.records[0].message_id,
        String::from_utf8_lossy(&dlq.records[0].payload)
    );

    Ok(())
}

struct ScriptedTransport {
    responses: VecDeque<Result<SdkHttpResponse, SdkTransportError>>,
}

impl ScriptedTransport {
    fn new(responses: Vec<Result<SdkHttpResponse, SdkTransportError>>) -> Self {
        Self {
            responses: VecDeque::from(responses),
        }
    }
}

impl SdkHttpTransport for ScriptedTransport {
    fn send(&mut self, _request: SdkHttpRequest) -> Result<SdkHttpResponse, SdkTransportError> {
        self.responses
            .pop_front()
            .expect("example scripted response should exist")
    }
}

fn ok(body: &str) -> Result<SdkHttpResponse, SdkTransportError> {
    Ok(SdkHttpResponse {
        status: 200,
        body: body.as_bytes().to_vec(),
    })
}

fn queue_claim_json(delivery_id: &str, attempt: u32) -> String {
    format!(
        "{{\"api_version\":\"v1\",\"stream\":\"jobs\",\"consumer_group\":\"workers\",\"subscription_id\":\"sub-1\",\"deliveries\":[{{\"delivery_id\":\"{delivery_id}\",\"lease_id\":\"lease-1\",\"attempt\":{attempt},\"leased_at_ms\":1700000000000,\"lease_expires_at_ms\":1700000030000,\"record\":{{\"offset\":0,\"message_id\":\"job-1\",\"key_hex\":null,\"event_time_ms\":1700000000000,\"ingest_time_ms\":1700000000000,\"headers\":[],\"payload_hex\":\"6a6f622d31\",\"payload_bytes\":5,\"durability_mode\":\"local-append\"}}}}],\"next_offset\":1,\"delivery_count\":1}}\n"
    )
}

Retry and dead letter

Rust

Retry and dead letter

retry_dead_letter.rs

Complete Rust program, including broker setup and the header helper. Advances synthetic time to inspect retry exhaustion. Save as crates/native-api/examples/retry_dead_letter.rs.

use std::error::Error;
use stornamics_message_broker_native::{
    ClientConfig, DurabilityMode, FailurePolicyAction, Header, InMemoryBroker, NativeClient,
    PublishMessage, RetentionPolicy, StartPosition, StreamFailurePolicy, Timestamp,
};

fn main() -> Result<(), Box<dyn Error>> {
    let broker = InMemoryBroker::with_start_time(Timestamp::from_unix_millis(30_000));
    let mut client = NativeClient::new(broker, ClientConfig::new("worker-producer"))?;

    client.create_stream(
        "jobs",
        DurabilityMode::LocalDisk,
        RetentionPolicy::default(),
    )?;
    client.create_stream(
        "jobs.dlq",
        DurabilityMode::LocalDisk,
        RetentionPolicy::default(),
    )?;
    client.configure_failure_policy(
        "jobs",
        StreamFailurePolicy::fixed(2, 100, Some("jobs.dlq".to_string())),
    )?;

    client.publish(
        PublishMessage::new(
            "jobs",
            Timestamp::from_unix_millis(1_700_000_200_000),
            b"job-1".to_vec(),
        )
        .with_message_id("job-1"),
    )?;

    let workers =
        client.subscribe_queue("jobs", "workers", StartPosition::Earliest, Some(30_000))?;

    let first = client.fetch(&workers, 1)?;
    let first_nack = client.nack(&workers, &first.messages[0], "dependency unavailable", true)?;
    let available_at = first_nack
        .available_at
        .expect("scheduled retry should include availability");

    client.broker_mut().advance_time_millis(100);
    let retry = client.fetch(&workers, 1)?;
    let exhausted = client.nack(&workers, &retry.messages[0], "still unavailable", true)?;

    let dlq_replay = client.subscribe_replay("jobs.dlq", "dlq-reader", StartPosition::Earliest)?;
    let dlq = client.fetch(&dlq_replay, 10)?;
    let dead_letter = &dlq.messages[0];

    println!(
        "workflow=retry_dead_letter first_action={:?} available_at={}",
        first_nack.policy_action, available_at.unix_millis
    );
    println!(
        "workflow=retry_dead_letter exhausted_action={:?} dead_letter_stream={} dead_letter_offset={}",
        exhausted.policy_action,
        exhausted.dead_letter_stream.as_deref().unwrap_or("<none>"),
        exhausted.dead_letter_offset.unwrap_or_default()
    );
    println!(
        "workflow=retry_dead_letter dlq_message_id={} original_offset={} attempt={} reason={}",
        dead_letter.envelope.message_id,
        header_value(
            &dead_letter.envelope.headers,
            "x-stornamics-original-offset"
        )
        .unwrap_or("<missing>"),
        header_value(
            &dead_letter.envelope.headers,
            "x-stornamics-delivery-attempt"
        )
        .unwrap_or("<missing>"),
        header_value(
            &dead_letter.envelope.headers,
            "x-stornamics-dead-letter-reason"
        )
        .unwrap_or("<missing>")
    );

    assert_eq!(
        first_nack.policy_action,
        FailurePolicyAction::ScheduledRetry
    );
    assert_eq!(exhausted.policy_action, FailurePolicyAction::DeadLettered);

    Ok(())
}

fn header_value<'a>(headers: &'a [Header], name: &str) -> Option<&'a str> {
    headers
        .iter()
        .find(|header| header.name == name)
        .and_then(|header| std::str::from_utf8(&header.value).ok())
}

Coverage Map

Behavior, SDK surface, and review mapping
BehaviorSDK surfaceSource scopeMaturity
Create stream with local-disk durabilityNativeClient::create_streamdocumented source scope and publish_replay.rsControlled beta
Publish one record and a batchPublishMessage, NativeClient::publish, NativeClient::batch_publishdocumented source scope and publish_replay.rsControlled beta
Replay from earliest and commit cursorsubscribe_replay, fetch, commit_cursordocumented source scope and publish_replay.rsControlled beta
Fetch queue work and ack successful deliverysubscribe_queue, fetch, ackdocumented source scope and shared_queue_processing.rsControlled beta
Nack and redeliver worknack and queue fetchdocumented source scope and shared_queue_processing.rsControlled beta
Retry and dead-letter through HTTP clientHttpNativeClient::retry_delivery and dead_letterdocumented source scope and http_queue_worker.rsVerification required for live remote endpoint posture
Apply failure policy and inspect DLQconfigure_failure_policy and subscribe_replaydocumented source scope and retry_dead_letter.rsControlled beta with provisional wording

Expected Output

Each group is the standard output of one independent program, in the same order as the run commands above. The values come from synthetic inputs and clocks.

Text

Expected Output

Deterministic output from the four complete Rust programs using in-memory state and scripted responses.

workflow=publish_replay published_offsets=[0, 1, 2]
workflow=publish_replay replayed=3 next_offset=3
workflow=publish_replay committed_cursor=processed offset=3

workflow=shared_queue acked_offset=0 delivery_id=del-1
workflow=shared_queue nacked_offset=1 will_redeliver=true
workflow=shared_queue redelivered_offset=1 attempt=2

workflow=http_queue_worker retry_scheduled delivery_id=del-1 available_at=1700000000250
workflow=http_queue_worker dead_lettered source_offset=0 dead_letter_stream=jobs.dlq dead_letter_offset=0
workflow=http_queue_worker dlq_message_id=dead-letter:jobs:0:del-2 payload=job-1

workflow=retry_dead_letter first_action=ScheduledRetry available_at=30104
workflow=retry_dead_letter exhausted_action=DeadLettered dead_letter_stream=jobs.dlq dead_letter_offset=0
workflow=retry_dead_letter dlq_message_id=job-1.dead-letter.del-2 original_offset=0 attempt=2 reason=still unavailable

Validation And status notes

Validation source scope
CommandWhat it provesValidation status
cargo check --locked --offline -p stornamics-message-broker-native --examplesAll native API Rust example targets compile.passed on 2026-09-14
cargo run --locked --offline --quiet -p stornamics-message-broker-native --example publish_replayStream publish, batch publish, replay count, next offset, and cursor commit output.passed on 2026-09-14
cargo run --locked --offline --quiet -p stornamics-message-broker-native --example shared_queue_processingQueue ack, nack, redelivery, and delivery attempt output.passed on 2026-09-14
cargo run --locked --offline --quiet -p stornamics-message-broker-native --example http_queue_workerHTTP client retry scheduling, dead-letter response parsing, and DLQ record listing with scripted transport.passed on 2026-09-14
cargo run --locked --offline --quiet -p stornamics-message-broker-native --example retry_dead_letterFailure policy scheduled retry, retry exhaustion, dead-letter stream output, and DLQ replay metadata.passed on 2026-09-14

Route Readiness Notes

Stream, replay, cursor, queue, retry, and dead-letter examples stay scoped to Message Broker private beta behavior. Use the compatibility matrix before strengthening SDK, auth, replication, or Kafka compatibility claims.

Safe Next Steps

  • Available

    Run the local quickstart

    Use the curl workflow for a copy/paste first success before adapting Rust SDK code.

  • Available

    Read CLI and SDK reference

    Use lifecycle flags, diagnostics, remote boundaries, and SDK entry point names when extending examples.

  • Available

    Check route and auth scope

    Confirm route maturity, auth modes, queue behavior, and Kafka boundaries before adding more SDK flows.

  • Available

    Review guide patterns

    Use the manifest check alongside product Cargo checks before changing status or evidence wording.