Getting Started

You can run a real shove publisher and consumer in 60 seconds against the in-process backend — no Docker, no AWS credentials, no config. This page is a copy-paste walkthrough.

Add the dependency

cargo add shove --features inmemory
cargo add tokio --features full
cargo add serde --features derive
cargo add tracing-subscriber
cargo add tokio-util --features rt

Define a topic

A topic binds a Rust message type to a queue topology. The define_topic! macro is the fast path; this example shows the manual Topic impl for clarity.

#[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())
    }
}

Reference: Topics & Topology.

Write a handler

#[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
    }
}

The handler returns an Outcome. Ack removes the message from the queue.

Wire it up

    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());

This connects to the in-process broker, declares the topology, registers a one-worker consumer group, publishes five messages, and exits cleanly when all five are processed.

Run it

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

What's next

Core concepts — what Topic, Outcome, and Broker<B> mean.
Pick a real backend — RabbitMQ, SNS+SQS, NATS, Kafka, or Redis/Valkey. Same handler, swap one marker.
Sequenced topics — strict per-key ordering when message order matters.
Audit logging — wrap any handler with structured per-delivery records.