InMemory — Audited Consumer

Use this when you want to add audit logging — trace IDs, outcomes, handler duration — to an existing handler without changing its logic. The Audited decorator wraps any MessageHandler transparently; this example shows the wiring with no external services required.

Prerequisites

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

Cargo features: inmemory,audit

Run

cargo run --example inmemory_audited_consumer --features inmemory,audit

Expected output

Each message produces two lines — one from the primary handler and one from the audit handler:

handled event 0
audit trace_id=<uuid> outcome=Ack duration_ms=0
handled event 1
audit trace_id=<uuid> outcome=Ack duration_ms=0
handled event 2
audit trace_id=<uuid> outcome=Ack duration_ms=0

Source

//! In-memory consumer wrapped in the `Audited` decorator. The audit handler
//! prints each record to stdout; swap in `ShoveAuditHandler` to publish to a
//! dedicated audit topic.
 
use shove::error::Result;
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::{
    AuditHandler, AuditRecord, Broker, ConsumerGroupConfig, InMemory, JsonCodec, MessageHandler,
    MessageHandlerExt, MessageMetadata, Outcome, Topic, TopologyBuilder,
};
 
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Event {
    id: u32,
}
 
struct EventTopic;
impl Topic for EventTopic {
    type Message = Event;
    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("audited-demo").dlq().build())
    }
}
 
#[derive(Clone)]
struct Inner {
    count: Arc<AtomicUsize>,
}
impl MessageHandler<EventTopic> for Inner {
    type Context = ();
    async fn handle(&self, msg: Event, _: MessageMetadata, _: &()) -> Outcome {
        self.count.fetch_add(1, Ordering::Relaxed);
        println!("handled event {}", msg.id);
        Outcome::Ack
    }
}
 
#[derive(Clone, Default)]
struct StdoutAudit;
impl AuditHandler<EventTopic> for StdoutAudit {
    async fn audit(&self, record: &AuditRecord<Event>) -> Result<()> {
        println!(
            "audit trace_id={} outcome={:?} duration_ms={}",
            record.trace_id, record.outcome, record.duration_ms
        );
        Ok(())
    }
}
 
#[tokio::main]
async fn main() {
    tracing_subscriber::fmt::init();
 
    let broker = Broker::<InMemory>::new(InMemoryConfig::default())
        .await
        .expect("connect InMemory");
    broker
        .topology()
        .declare::<EventTopic>()
        .await
        .expect("declare");
 
    let count = Arc::new(AtomicUsize::new(0));
 
    let mut group = broker.consumer_group();
    let c = count.clone();
    group
        .register::<EventTopic, _>(
            ConsumerGroupConfig::new(
                InMemoryConsumerGroupConfig::new(1..=1).with_prefetch_count(1),
            ),
            move || Inner { count: c.clone() }.audited(StdoutAudit),
        )
        .await
        .expect("register");
 
    let publisher = broker.publisher().await.expect("publisher");
    for id in 0..3 {
        publisher
            .publish::<EventTopic>(&Event { id })
            .await
            .expect("publish");
    }
 
    // Stop when all three events have been processed (or after a deadline).
    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) < 3 && 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

Primary handler

Inner is a straightforward MessageHandler\<EventTopic\> that increments a shared Arc<AtomicUsize> counter and prints the event id. It has no knowledge of auditing — it simply processes the message and returns Outcome::Ack.

Audit handler

StdoutAudit implements AuditHandler\<EventTopic\>. Its single method audit receives an AuditRecord\<Event\> that carries the trace_id (a UUID assigned per message by the framework), the outcome returned by the inner handler, and duration_ms (wall-clock time the inner handler spent). In this example the record is printed to stdout; in production you would typically publish it to a dedicated audit topic using ShoveAuditHandler.

Wiring with `.audited()`

The factory passed to group.register calls Inner { count: c.clone() }.audited(StdoutAudit). The .audited() method comes from the MessageHandlerExt trait and wraps Inner in the framework's Audited\<Inner, StdoutAudit\> combinator. The consumer group sees only a single MessageHandler — the auditing happens transparently inside.

Consumer group configuration

InMemoryConsumerGroupConfig::new(1..=1).with_prefetch_count(1) runs a single worker with no pipelining. With prefetch_count(1) only one message at a time is dispatched to the handler, which makes the output strictly ordered and easier to read in this demonstration.

What to try next

Replace StdoutAudit with a struct that publishes to a second topic via broker.publisher() — this is the pattern ShoveAuditHandler follows on real deployments.
Set with_prefetch_count(4) and new(1..=2) — verify that audit records are still emitted for every message even under concurrent processing.
Return Outcome::Nack from Inner::handle — confirm that AuditRecord::outcome reflects the nack and the message is requeued.
See Guides: Audit for a full description of the AuditRecord fields.