InMemory — Sequenced

Use this when you want to verify per-key ordering before wiring a real broker. Messages for three accounts are published in interleaved order; each account's entries arrive in strict sequence, proving that ledger-style FIFO delivery works correctly. No external infrastructure required.

Prerequisites

No external services required. The in-memory backend runs entirely in-process.

Cargo feature: inmemory

Run

cargo run --example inmemory_sequenced --features inmemory

Expected output

Output order within a single account is always ascending (seq=0 before seq=1), but entries for different accounts may interleave:

applied entry account=alice seq=0 amount=+100
applied entry account=bob seq=0 amount=+100
applied entry account=carol seq=0 amount=+100
applied entry account=alice seq=1 amount=+100
applied entry account=bob seq=1 amount=+100
...

Source

//! Sequenced (per-key FIFO) delivery with `SequenceFailure::FailAll`.
//!
//! Note: per-key FIFO consumption (`run_fifo`) isn't yet surfaced on the
//! generic `Broker<B>` / `ConsumerSupervisor<B>` / `ConsumerGroup<B>`
//! wrappers — this example therefore keeps using the backend-specific
//! `InMemoryConsumer::run_fifo` directly.
 
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
 
use serde::{Deserialize, Serialize};
use tokio_util::sync::CancellationToken;
 
use shove::inmemory::{
    InMemoryBroker, InMemoryConsumer, InMemoryPublisher, InMemoryTopologyDeclarer,
};
use shove::{
    ConsumerOptions, InMemory, JsonCodec, MessageHandler, MessageMetadata, Outcome,
    SequenceFailure, SequencedTopic, Topic, TopologyBuilder,
};
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct LedgerEntry {
    account: String,
    seq: u64,
    amount: i64,
}
 
struct LedgerTopic;
impl Topic for LedgerTopic {
    type Message = LedgerEntry;
    type Codec = JsonCodec;
    fn topology() -> &'static shove::QueueTopology {
        static T: std::sync::OnceLock<shove::QueueTopology> = std::sync::OnceLock::new();
        T.get_or_init(|| {
            TopologyBuilder::new("ledger")
                .sequenced(SequenceFailure::FailAll)
                .routing_shards(4)
                .hold_queue(Duration::from_millis(50))
                .dlq()
                .build()
        })
    }
    const SEQUENCE_KEY_FN: Option<fn(&Self::Message) -> String> = Some(Self::sequence_key);
}
impl SequencedTopic for LedgerTopic {
    fn sequence_key(msg: &LedgerEntry) -> String {
        msg.account.clone()
    }
}
 
#[derive(Clone)]
struct Handler {
    acked: Arc<AtomicUsize>,
}
impl MessageHandler<LedgerTopic> for Handler {
    type Context = ();
    async fn handle(&self, msg: LedgerEntry, _: MessageMetadata, _: &()) -> Outcome {
        println!(
            "applied entry account={} seq={} amount={:+}",
            msg.account, msg.seq, msg.amount
        );
        self.acked.fetch_add(1, Ordering::Relaxed);
        Outcome::Ack
    }
}
 
#[tokio::main]
async fn main() {
    tracing_subscriber::fmt::init();
 
    let broker = InMemoryBroker::new();
    let declarer = InMemoryTopologyDeclarer::new(broker.clone());
    declarer.declare(LedgerTopic::topology()).await.unwrap();
 
    let acked = Arc::new(AtomicUsize::new(0));
    let handler = Handler {
        acked: acked.clone(),
    };
 
    let shutdown = CancellationToken::new();
    let consumer = InMemoryConsumer::new(broker.clone());
    let shutdown_for_task = shutdown.clone();
    let consume_handle = tokio::spawn(async move {
        let opts = ConsumerOptions::<InMemory>::new()
            .with_shutdown(shutdown_for_task)
            .with_prefetch_count(1);
        consumer.run_fifo::<LedgerTopic, _>(handler, (), opts).await
    });
 
    let publisher = InMemoryPublisher::new(broker.clone());
    for seq in 0..5 {
        for account in ["alice", "bob", "carol"] {
            publisher
                .publish::<LedgerTopic>(&LedgerEntry {
                    account: account.into(),
                    seq,
                    amount: 100,
                })
                .await
                .unwrap();
        }
    }
 
    while acked.load(Ordering::Relaxed) < 15 {
        tokio::time::sleep(Duration::from_millis(10)).await;
    }
 
    shutdown.cancel();
    let _ = consume_handle.await;
}

Walkthrough

Topic definition with sequence key

LedgerTopic implements both Topic and SequencedTopic. The topology is built with .sequenced(SequenceFailure::FailAll) — if any message for an account fails, all subsequent messages for that account are also failed immediately rather than left queued. .routing_shards(4) tells the broker to spread keys across four in-memory shard channels, which improves parallelism when many distinct keys are in flight. .hold_queue(Duration::from_millis(50)) configures a 50 ms retry delay before a failed message is redelivered.

Sequence key function

SEQUENCE_KEY_FN is set to Some(Self::sequence_key), which extracts msg.account as a String. The broker uses this function to route each LedgerEntry to the correct FIFO lane — messages with the same account name always land in the same shard and are processed one at a time.

Backend-specific consumer path

Because per-key FIFO (run_fifo) is not yet exposed on the generic Broker\<B\> / ConsumerGroup\<B\> API, this example drops down to the concrete InMemoryConsumer type directly. ConsumerOptions::<InMemory>::new().with_prefetch_count(1) keeps exactly one message in-flight per consumer, which is the correct setting for strict FIFO — allowing more would let the consumer pick up a later message for the same key before finishing the earlier one.

Shutdown after all messages land

The main thread busy-loops on acked.load(Ordering::Relaxed) < 15 (five sequence numbers × three accounts) before calling shutdown.cancel(). This pattern is simple for a bounded benchmark; in production you would instead react to an external signal (see Guides: Shutdown).

What to try next

Change SequenceFailure::FailAll to SequenceFailure::FailOne — observe that only the individual failing message is retried; subsequent messages for the same account are not held back.
Remove .routing_shards(4) — all keys collapse into a single shard; per-key parallelism disappears and throughput drops proportionally.
Add a second hold queue stage with .hold_queue(Duration::from_secs(1)) — failed entries wait longer before redelivery.
Read Guides: Sequenced for a deeper explanation of shard routing and failure modes.