Crate arminius_lib_message_bus 

Message bus abstractions: Publisher, Subscriber, Event, and ReceivedEvent.

arminius-lib-message-bus

Transport-agnostic message bus abstractions for arminius services.

This crate defines the core traits for event-driven and request-reply communication between services — Event, Publisher, Subscriber, Command, Requester, Responder — plus the derive macros that generate their boilerplate. The traits carry no transport dependency: the NATS implementation lives in arminius-lib-nats, and in-memory mocks for unit testing live behind this crate’s mock feature.

Two communication styles are supported:

• Events — fire-and-forget, published to a versioned subject and consumed by any number of subscribers (Publisher / Subscriber).
• Commands — request-reply, sent to a subject and answered with a typed reply (Requester / Responder).

Feature flags

Feature	Default	Description
mock	off	Enables the [mock] module: in-memory MockBus, MockPublisher, MockSubscriber, MockReceivedEvent, MockReceivedCommand, and ReplyChecker for unit testing without a NATS server. Enable it as a dev-dependency feature so the mocks do not ship in release builds.

Events

Use #[derive(Event)] with a #[subject(...)] attribute to attach a versioned NATS subject to a serializable type:

use serde::{Deserialize, Serialize};
use arminius_lib_message_bus::{Event, Publisher, Subscriber, ReceivedEvent};
use futures::StreamExt as _;

#[derive(Clone, Serialize, Deserialize, Event)]
#[subject(version = "1")]
struct RoomCreated {
    room_id: String,
}

// RoomCreated::SUBJECT == "RoomCreated.v1"

// Publishing — accept `&impl Publisher` to stay transport-agnostic and testable.
async fn announce(publisher: &impl Publisher, room_id: String) {
    publisher.publish(RoomCreated { room_id }).await.unwrap();
}

// Subscribing — every event must be acknowledged once handled.
async fn consume(subscriber: &impl Subscriber) {
    let stream = subscriber.subscribe::<RoomCreated>().await.unwrap();
    futures::pin_mut!(stream);
    while let Some(Ok(event)) = stream.next().await {
        println!("room created: {}", event.room_id);
        event.ack().await.unwrap();
    }
}

The subject defaults to <StructName>.v<version>. Override the name with #[subject(name = "custom", version = "1")], or supply a full literal with #[subject("custom.v1")].

Per-instance (dynamic) subjects

A subject template may contain {field} placeholders naming fields of the struct. Each becomes a NATS * wildcard in Event::SUBJECT (the subscribe/stream pattern) and is interpolated into Event::subject() (the concrete publish subject):

use serde::{Deserialize, Serialize};
use arminius_lib_message_bus::{Event, Publisher, Subscriber, ReceivedEvent};

#[derive(Clone, Serialize, Deserialize, Event)]
#[subject(name = "rooms.{room_id}.events", version = "1")]
struct RoomCreated {
    room_id: u64,
}

// RoomCreated::SUBJECT                  == "rooms.*.events.v1"   (subscribe to all)
// RoomCreated { room_id: 7 }.subject()  == "rooms.7.events.v1"   (publish target)

NOTE: subscribe() only takes an event type, so it always consumes the full subject pattern — a {field} placeholder cannot be pinned to a single value. Consumers therefore see every instance’s events and select what they need in code.

Schema migration: compatible subjects

When introducing a new version, list older types in compatible so subscribers also consume their subjects and upcast them via From:

use serde::{Deserialize, Serialize};
use arminius_lib_message_bus::Event;

pub mod v1 {
    use serde::{Deserialize, Serialize};
    use arminius_lib_message_bus::Event;

    #[derive(Clone, Serialize, Deserialize, Event)]
    #[subject(version = "1")]
    pub struct OldRoomCreated {
        pub room_id: String,
    }    
}
#[derive(Clone, Serialize, Deserialize, Event)]
#[subject(version = "2", compatible = [v1::OldRoomCreated])]
struct RoomCreated {
    room_id: String,
    display_name: String, // new field
}

impl From<v1::OldRoomCreated> for RoomCreated {
    fn from(old: v1::OldRoomCreated) -> Self {
        Self { room_id: old.room_id, display_name: String::new() }
    }
}

The subscriber receives both RoomCreated.v1 and RoomCreated.v2 messages and converts v1 payloads automatically.

Commands

All commands sharing a namespace must resolve to the same filter so they live on a single JetStream stream (each command then attaches its own consumer filtered to SUBJECT). The filter is everything before the first variable segment, with .> appended. A variable segment is either a * wildcard or — for a fully-static subject like "rooms.create" — its trailing command segment. So "rooms.*.add_event" and "rooms.create" both yield "rooms.>" and share one stream. A subject whose first segment is already variable yields ">".

use serde::{Deserialize, Serialize};
use arminius_lib_message_bus::{Command, Requester, Responder, ReceivedCommand};
use futures::StreamExt as _;

#[derive(Debug, Serialize, Deserialize)]
pub struct CreateRoomError;

// The whole command is serialised, so it must derive `Serialize`/`Deserialize`.
// Fields can be inlined directly — no separate request struct required.
#[derive(Serialize, Deserialize, Command)]
#[subject("rooms.{room_id}.create")]
#[command(reply = Result<String, CreateRoomError>)]
pub struct CreateRoomCommand {
    pub room_id: String,    // also encoded into the subject
    pub name: String,       // part of the serialised payload
    pub topics: Vec<String>,
}

// CreateRoomCommand::SUBJECT        == "rooms.*.create"

// Requesting side — send a command and await the typed reply.
async fn create(requester: impl Requester, room_id: String) {
    let cmd = CreateRoomCommand { room_id, name: "lobby".into(), topics: vec![] };
    let reply: Result<String, CreateRoomError> = requester.request(cmd).await.unwrap();
}

// Responding side — answer each command with `reply`, which acknowledges too.
async fn serve(responder: impl Responder) {
    let stream = responder.respond::<CreateRoomCommand>().await.unwrap();
    futures::pin_mut!(stream);
    while let Some(Ok(cmd)) = stream.next().await {
        let result = handle(&*cmd);
        cmd.reply(result).await.unwrap();
    }
}

fn handle(command: &CreateRoomCommand) -> Result<String, CreateRoomError> {
    todo!("your logic goes here")
}

Delivery guarantees

By default a command is redelivered until it is acknowledged, which assumes handling it twice is harmless: a handler that dies before it answers leaves the acknowledgement window to expire, and the command is retried. A handler’s own failures belong in the reply type, where the caller can act on them; redelivery is for a handler that never finished at all.

Where handling twice is not harmless — the handler has an effect that cannot be repeated, and losing the answer is the cheaper failure — say so on the command:

#[derive(Serialize, Deserialize, Command)]
#[subject("keys.claim")]
#[command(reply = Result<Keys, ClaimError>, delivery = at_most_once)]
pub struct ClaimKeysCommand { /* … */ }

Such a handler acknowledges before it starts work, so nothing past that point can hand the command out again:

while let Some(Ok(cmd)) = stream.next().await {
    // From here on a crash costs the answer rather than claiming a second key.
    cmd.ack().await.unwrap();
    let result = claim_keys(&*cmd).await;
    cmd.reply(result).await.unwrap();
}

The consumer is capped at a single delivery, so an expiring acknowledgement window hands the command to nobody rather than to a second handler. The guarantee lives on the command type rather than on a responder because request-reply gives a command one responder: every instance of the answering service derives the same value, so they cannot disagree about it.

Testing with the mock feature

The mock module provides in-memory trait implementations so service code that takes &impl Publisher, &impl Subscriber, or impl ReceivedCommand can be tested without a running NATS server.

MockBus — publish and subscribe in one test

MockBus implements both Publisher and Subscriber. Published events are delivered to active subscriptions on a matching subject and recorded for assertions via published_as:

#[cfg(feature = "mock")]
async fn example() {
    use serde::{Deserialize, Serialize};
    use arminius_lib_message_bus::{Event, Publisher};

    #[derive(Clone, Serialize, Deserialize, Event)]
    #[subject(version = "1")]
    struct RoomCreated { room_id: String }

    async fn announce(publisher: &impl Publisher, room_id: String) {
        publisher.publish(RoomCreated { room_id }).await.unwrap();
    }

    let bus = arminius_lib_message_bus::mock::MockBus::new();

    // Code under test only sees `&impl Publisher`.
    announce(&bus, "!abc".into()).await;

    let published = bus.published_as::<RoomCreated>();
    assert_eq!(published[0].room_id, "!abc");
}

It also delivers live to subscribers (clones share the same backing state):

#[cfg(feature = "mock")]
async fn example() {
    use serde::{Deserialize, Serialize};
    use arminius_lib_message_bus::{Event, Publisher, Subscriber, ReceivedEvent};
    use futures::StreamExt as _;
    
    #[derive(Clone, Serialize, Deserialize, Event)]
    #[subject(version = "1")]
    struct RoomCreated { room_id: String }

    let bus = arminius_lib_message_bus::mock::MockBus::new();
    let stream = bus.subscribe::<RoomCreated>().await.unwrap();
    futures::pin_mut!(stream);

    bus.publish(RoomCreated { room_id: "!abc".into() }).await.unwrap();

    let event = stream.next().await.unwrap().unwrap();
    assert_eq!(event.room_id, "!abc");
    event.ack().await.unwrap();
}

MockSubscriber — drive a consumer over a fixed list

create_stream wraps a Vec of events into a stream and hands back parallel “checkers” so a test can assert each event was acknowledged:

#[cfg(feature = "mock")]
async fn example() {
    use serde::{Deserialize, Serialize};
    use arminius_lib_message_bus::{Event, ReceivedEvent};
    use futures::StreamExt as _;
    
    #[derive(Clone, Serialize, Deserialize, Event)]
    #[subject(version = "1")]
    struct RoomCreated { room_id: String }
    
    let (mut stream, checkers) = arminius_lib_message_bus::mock::MockSubscriber::new()
        .create_stream(vec![RoomCreated { room_id: "!a".into() },
                            RoomCreated { room_id: "!b".into() }]);

    while let Some(Ok(event)) = stream.next().await {
        event.ack().await.unwrap();
    }

    assert!(checkers.iter().all(|c| c.was_acked()));
}

MockPublisher — capture published events

MockPublisher<T> records events of exactly type T. To capture several event types in one publisher, wrap them in an enum that also implements Event:

#[cfg(feature = "mock")]
async fn example() {
    use serde::{Deserialize, Serialize};
    use arminius_lib_message_bus::{Event, Publisher};

    #[derive(Clone, Serialize, Deserialize, Event)]
    #[subject(version = "1")]
    struct RoomCreated { room_id: String }

    let publisher: arminius_lib_message_bus::mock::MockPublisher<RoomCreated> = arminius_lib_message_bus::mock::MockPublisher::new();
    publisher.publish(RoomCreated { room_id: "!abc".into() }).await.unwrap();
    assert_eq!(publisher.published_messages().len(), 1);
}

MockReceivedCommand — assert on command replies

MockReceivedCommand::new returns the mock command alongside a ReplyChecker. Run the handler under test, then read back the captured reply:

#[cfg(feature = "mock")]
async fn example() {
    use serde::{Deserialize, Serialize};
    use arminius_lib_message_bus::{Command, ReceivedCommand};
    
    #[derive(Debug, PartialEq, Serialize, Deserialize)]
    pub struct CreateRoomError;
    
    #[derive(Debug, Default, Serialize, Deserialize)]
    pub struct CreateRoomRequest { fields: Vec<String> }
    
    #[derive(Default, Command, Serialize, Deserialize)]
    #[subject("rooms.{room_id}.create")]
    #[command(reply = Result<String, CreateRoomError>)]
    pub struct CreateRoomCommand {
        pub room_id: String,
        pub request: CreateRoomRequest,
    }

    async fn handle_create(cmd: &impl ReceivedCommand<CreateRoomCommand>) {
        let room_id = cmd.room_id.clone();
        cmd.reply(Ok(room_id)).await.unwrap();
    }

    let room_id = String::from("!abc");
    let req = CreateRoomRequest::default();

    let cmd = CreateRoomCommand { room_id: room_id.clone(), request: req };
    let (received, checker) = arminius_lib_message_bus::mock::MockReceivedCommand::new(cmd);

    // Code under test replies via `reply`.
    handle_create(&received).await;

    let reply: Result<String, CreateRoomError> = checker.into_reply().await.unwrap();
    assert_eq!(reply, Ok(room_id));
}

into_reply returns None if the handler dropped the command without replying.

Run the crate’s tests with cargo nextest run (process isolation). The mock tests above need the mock feature enabled.

Work in progress

Structs with same name in compatible chain do not work

This does currently not compile:

use arminius_lib_nats::{Event, Publisher, EventSubscriber as _};
use serde::{Deserialize, Serialize};
pub mod v1 {
    use arminius_lib_nats::{Event, Publisher, EventSubscriber as _};
    use serde::{Deserialize, Serialize};

    #[derive(Clone, Serialize, Deserialize, Event)]
    #[subject(version = "1")]
    pub struct RoomCreated {
        pub room_id: String,
    }
}

#[derive(Clone, Serialize, Deserialize, Event)]
#[subject(version = "2", compatible = [v1::RoomCreated])]
pub struct RoomCreated {
    pub room_id: String,
    pub display_name: String, // new field
}

impl From<v1::RoomCreated> for RoomCreated {
    fn from(old: v1::RoomCreated) -> Self {
        Self { room_id: old.room_id, display_name: String::new() }
    }
}

Modules

command 🔒

Request-reply command messaging: traits for sending commands and awaiting typed replies, and for responding to received commands.

event 🔒

Fire-and-forget event messaging: traits for publishing events and subscribing to them as acknowledgeable streams.

mock

Mock implementation that combines both publish/subscribe and request/response patterns for unit testing without a NATS server.

subject 🔒

Utilities for building NATS subjects out of arbitrary tokens.

Structs

EscapingSubject

Struct that allows building a NATS subject out of tokens, escaping any reserved characters that would otherwise split up or wildcard the tokens.

Enums

CommandError

Errors during command payload serialization or subject parsing.

Delivery

How often a command may reach its handler.

Error

Errors that can occur during message bus operations.

TokenEscapeError

Errors that occur when unescaping a NATS token

Traits

Command

Typed command endpoint for request-reply over a message bus.

Event

Generic event type that can be sent between services.

Publisher

Publishes events to the message bus.

ReceivedCommand

A received command that must be acknowledged with a typed reply.

ReceivedEvent

An event received from the message bus that can be acknowledged.

Requester

Sends commands and awaits typed replies. Mirrors Publisher.

Responder

Subscribes to incoming commands. Mirrors Subscriber.

Subscriber

Subscribes to events from the message bus.

Functions

encode_subject_segment

Encodes a string for use as a single NATS subject segment by escaping the characters NATS reserves (see EscapingSubject).

subject_matches

Returns true if the concrete subject matches the NATS pattern.

Derive Macros

Command

Derive macro that implements arminius_lib_message_bus::Command for a struct.

Event

Derive macro that implements arminius_lib_message_bus::Event.