InMemory — Basic

Use this when you want to see the full publish/consume cycle with no external services: declare a topic, publish five messages, handle them, and exit cleanly. The simplest end-to-end proof that your topic definition and handler are wired correctly.

Prerequisites

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

Cargo feature: inmemory

Run

cargo run --example inmemory_basic --features inmemory

Expected output

received #0: hello 0
received #1: hello 1
received #2: hello 2
received #3: hello 3
received #4: hello 4

Source

//! Basic in-memory publish/consume round-trip.
 
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::{InMemoryConfig, InMemoryConsumerGroupConfig};
use shove::{
    Broker, ConsumerGroupConfig, InMemory, JsonCodec, MessageHandler, MessageMetadata, Outcome,
    Topic, TopologyBuilder,
};
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Ping {
    id: u32,
    note: String,
}
 
struct PingTopic;
impl Topic for PingTopic {
    type Message = Ping;
    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("ping").dlq().build())
    }
}
 
#[derive(Clone)]
struct PingHandler {
    count: Arc<AtomicUsize>,
}
impl MessageHandler<PingTopic> for PingHandler {
    type Context = ();
    async fn handle(&self, msg: Ping, _: MessageMetadata, _: &()) -> Outcome {
        println!("received #{}: {}", msg.id, msg.note);
        self.count.fetch_add(1, Ordering::Relaxed);
        Outcome::Ack
    }
}
 
#[tokio::main]
async fn main() {
    tracing_subscriber::fmt::init();
 
    let broker = Broker::<InMemory>::new(InMemoryConfig::default())
        .await
        .expect("connect InMemory");
    broker
        .topology()
        .declare::<PingTopic>()
        .await
        .expect("declare");
 
    let count = Arc::new(AtomicUsize::new(0));
 
    let mut group = broker.consumer_group();
    let c = count.clone();
    group
        .register::<PingTopic, _>(
            ConsumerGroupConfig::new(
                InMemoryConsumerGroupConfig::new(1..=1).with_prefetch_count(4),
            ),
            move || PingHandler { count: c.clone() },
        )
        .await
        .expect("register");
 
    let publisher = broker.publisher().await.expect("publisher");
    for i in 0..5 {
        publisher
            .publish::<PingTopic>(&Ping {
                id: i,
                note: format!("hello {i}"),
            })
            .await
            .expect("publish");
    }
 
    // Stop when all five messages have been processed (or after 5 s).
    let stop = CancellationToken::new();
    let waiter_stop = stop.clone();
    let waiter_count = count.clone();
    tokio::spawn(async move {
        let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
        while waiter_count.load(Ordering::Relaxed) < 5 && tokio::time::Instant::now() < deadline {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        waiter_stop.cancel();
    });
 
    let signal_stop = stop.clone();
    let outcome = group
        .run_until_timeout(
            async move { signal_stop.cancelled().await },
            Duration::from_secs(5),
        )
        .await;
 
    std::process::exit(outcome.exit_code());
}

Walkthrough

Topic definition

PingTopic is the type-level identifier for the queue. It implements the Topic trait, which associates the topic with its message type (Ping) and its QueueTopology. The topology is built once via OnceLock and describes the queue name ("ping") along with a dead-letter queue (.dlq()). Every shove backend consults this topology to provision the necessary queues on startup.

Connecting and declaring

Broker::<InMemory>::new(InMemoryConfig::default()) creates the in-process broker. Calling broker.topology().declare::<PingTopic>() then materialises the "ping" queue (and its DLQ) inside the broker's in-memory store. Declaration is idempotent — safe to call more than once.

Handler and consumer group

PingHandler wraps an Arc<AtomicUsize> counter so the main thread can observe how many messages have been processed. The handler always returns Outcome::Ack, which tells the broker to remove the message from the queue. broker.consumer_group() creates a ConsumerGroup\<InMemory\>, and group.register binds the handler factory to PingTopic with a worker range of 1..=1 and a prefetch of 4. The factory closure is called once per spawned worker.

Publishing

Five Ping messages are published in a loop via publisher.publish::<PingTopic>. The in-memory backend delivers them immediately into the in-process queue — no serialisation round-trip to a remote broker.

Graceful shutdown

A background task polls count every 10 ms and cancels a CancellationToken once all five messages are processed (or after a 5-second safety deadline). The consumer group's run_until_timeout call drives the event loop until that token fires, then allows in-flight handlers to drain before returning. std::process::exit(outcome.exit_code()) propagates a non-zero code if the drain timed out.

What to try next

Change InMemoryConsumerGroupConfig::new(1..=1) to new(1..=4) — observe how the consumer group scales workers automatically as the queue grows.
Change Outcome::Ack to Outcome::Nack in PingHandler::handle — watch messages loop back into the queue and eventually land in the DLQ.
Swap the InMemory marker for another backend (e.g. RabbitMq) and supply the matching config — the handler code remains unchanged.
See Concepts: Outcomes for the full list of outcome variants and their retry semantics.