Are you an LLM? Read llms.txt for a summary of the docs, or llms-full.txt for the full context.
Skip to content

In-process

The fastest way to get started with shove and the right choice for tests. No Docker, no AWS, no broker process — the broker lives in process RAM. Suitable for unit tests, integration tests, and single-process applications where you want the full pub/sub API without external infrastructure.

What you need

Nothing. No services to start, no credentials to configure.

Install

cargo add shove --features inmemory

Connect

    let broker = Broker::<InMemory>::new(InMemoryConfig::default())
        .await
        .expect("connect InMemory");

InMemoryConfig::default() is the only config needed. The broker lives in process RAM and is dropped when the variable goes out of scope.

Declare topology

    broker
        .topology()
        .declare::<PingTopic>()
        .await
        .expect("declare");

topology().declare::<T>() registers the topic with the in-process broker. Idempotent — safe to call multiple times.

Publish

    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");
    }

publisher().await? returns a Publisher<InMemory>. Messages are delivered to consumers in the same process synchronously via in-memory channels.

Consume

    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;

consumer_group() creates an in-process consumer group. InMemoryConsumerGroupConfig::new(min..=max) sets the concurrency bounds. The run_until_timeout call blocks until the shutdown signal fires or the drain timeout elapses.

Sequenced delivery

Per-key FIFO shards in process memory. The same T::sequence_key() semantics as durable backends — messages for the same key are processed in order, and different keys are independent and may run concurrently. No persistence: ordering is enforced in-memory only and is lost on shutdown.

See Sequenced Topics for the full ordering model.

Consumer groups + autoscaling

In-process queue-depth tracking drives the autoscaler. The autoscale signal works the same shape as on durable backends — queue depth within the in-process channels drives scale-up and scale-down.

Configure via InMemoryConsumerGroupConfig:

use shove::inmemory::InMemoryConsumerGroupConfig;
use shove::{Broker, ConsumerGroupConfig, InMemory};
use std::time::Duration;

let mut group = broker.consumer_group();
group
    .register::<PingTopic, _>(
        ConsumerGroupConfig::new(
            InMemoryConsumerGroupConfig::new(1..=4)
                .with_prefetch_count(8)
                .with_max_retries(5)
                .with_handler_timeout(Duration::from_secs(10))
                .with_max_message_size(64 * 1024)
                .with_max_pending_per_key(50),
        ),
        || MyHandler,
    )
    .await?;

All builder methods mirror the durable backends, so test code written against the in-process backend is structurally identical to production code on RabbitMQ/Kafka/Redis. with_default_handler_timeout(Duration) on consumer_group() sets a registry-wide default — useful when most test topics share a deadline.

Gotchas

  • No durability. Messages live in process RAM and are dropped on shutdown. Do not use the in-process backend for production business-critical flows.
  • No cross-process delivery. Only consumers registered in the same process see messages. This is by design — the in-process backend is not a substitute for a real broker.
  • Intended for tests and single-process apps. It is the right default for unit tests, integration tests that don't need a broker, and single-binary applications where all producers and consumers run in the same process.

Examples

See also