topmesys

An embeddable topic-based message broker.

What it does

topmesys is an in-process publish/subscribe event bus for tokio applications. It lets loosely coupled parts of an application exchange messages through hierarchical topics instead of calling each other directly.

• Topic-based routing — messages carry a routing key like orders.eu.created; consumers subscribe with patterns and only receive matching messages.
• Flexible patterns — subscription topics support literal segments (orders), wildcards (orders.*.created), selections (orders.[eu,us].paid) and tail matching (orders.* matches orders.eu, orders.eu.created, ...).
• Multi-topic consumers — a consumer declares several subscriptions, each identified by a value of its own Topic type (typically an enum), and handles every message knowing which subscription it arrived on.
• Independent subscriptions — every subscription has its own bounded inbox and worker with configurable concurrency; a concurrency of 1 handles messages strictly in order.
• Retries and dead letters — failed deliveries are retried with fixed, linear or exponential backoff; permanent failures, exhausted retries and overflowing inboxes end up in a DeadLetterSink.
• Transport handles — messages bridged in from NATS, Kafka and the like can carry their transport's handle. Consumers can access it, and the broker settles it exactly once with the outcome of every subscription, so the bridge can ack, nak or terminate the message upstream.
• Typestate lifecycle — invalid states are unrepresentable: an EventTopic must be turned into a routing key or a subscription pattern before use, and messages can only be sent to an EventBroker<Running>.
• Batched, cancel-safe submission — the EventEmitter trait submits single messages or batches through channel permits.
• Graceful shutdown — stopping the broker drains all buffered messages and waits for in-flight deliveries, including pending retries; stopping on Ctrl-C can be opted into with with_ctrl_c_handling().

Matching messages are delivered to every matching subscription; messages with no matching subscription are dropped silently. When a subscription's inbox is full, routing waits for it by default (Overflow::Block), which means a subscription busy retrying holds up all others once its inbox fills. Subscriptions depending on unreliable systems can use Overflow::DeadLetter instead.

Quick start

use std::time::Duration;

use tokio::sync::mpsc;
use topmesys::{
    Delivery, EventBroker, EventConsumer, EventEmitter, EventMessage, EventSubmission,
    HandlerError, RetryPolicy, Subscription, Subscriptions,
};

// Identifies the subscriptions of the consumer below.
#[derive(Debug)]
enum OrderTopic {
    Created,
    Cancelled,
}

// A consumer subscribes to topic patterns and handles matching events per subscription.
#[derive(Debug)]
struct Invoicing;

#[async_trait::async_trait]
impl EventConsumer for Invoicing {
    type Topic = OrderTopic;

    fn subscriptions(&self) -> Subscriptions<OrderTopic> {
        Subscriptions::new()
            .on(OrderTopic::Created, "orders.*.created")
            .on(
                OrderTopic::Cancelled,
                Subscription::new("orders.*.cancelled")
                    .with_retry_policy(RetryPolicy::exponential(3, Duration::from_millis(100))),
            )
    }

    async fn handle_event(
        &self,
        topic: &OrderTopic,
        delivery: &Delivery,
    ) -> Result<(), HandlerError> {
        let order = std::str::from_utf8(delivery.message().content())?;
        match topic {
            OrderTopic::Created => println!("invoicing {order}"),
            OrderTopic::Cancelled => println!("voiding the invoice of {order}"),
        }
        Ok(())
    }
}

// An emitter wraps a sender obtained from the running broker.
#[derive(Debug)]
struct OrderService {
    sender: mpsc::Sender<EventMessage>,
}

#[async_trait::async_trait]
impl EventEmitter for OrderService {
    fn get_sender(&self) -> &mpsc::Sender<EventMessage> {
        &self.sender
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let broker = EventBroker::default();
    broker.add_topic_consumer(Invoicing)?;

    let broker = broker.run()?;
    let orders = OrderService {
        sender: broker.get_sender(),
    };

    let event = EventMessage::new("orders.eu.created", r#"{"order":"A-1001"}"#)?;
    orders.submit_event(EventSubmission::Single(event)).await?;

    broker.stop().await;
    Ok(())
}

Errors returned from handle_event are retried according to the subscription's RetryPolicy; return HandlerError::permanent(error) for errors retrying won't fix. Messages a subscription gives up on are handed to the DeadLetterSink set with EventBroker::with_dead_letter_sink or Subscription::with_dead_letter_sink, and logged and dropped without one.

Bridging other messaging systems

A message received from another messaging system can carry the transport's handle. The broker settles it once, after every subscription the message was routed to finished, e.g. for NATS JetStream:

#[derive(Debug)]
struct JetStreamHandle(async_nats::jetstream::Message);

#[async_trait::async_trait]
impl TransportHandle for JetStreamHandle {
    async fn settle(&self, settlement: &Settlement) -> anyhow::Result<()> {
        let kind = if settlement.all_handled() || settlement.is_unrouted() {
            AckKind::Ack
        } else if settlement.any(DeliveryOutcome::Aborted) {
            AckKind::Nak(None)
        } else {
            AckKind::Term
        };
        Ok(self.0.ack_with(kind).await?)
    }

    // Keeps JetStream from redelivering while a subscription waits to retry.
    async fn on_retry(&self, _attempt: u32, _delay: Duration) -> anyhow::Result<()> {
        Ok(self.0.ack_with(AckKind::Progress).await?)
    }
}

let message = EventMessage::new(js_message.subject.as_str(), js_message.payload.clone())?
    .with_transport(JetStreamHandle(js_message));

Consumers and dead letter sinks get the handle back with transport::<JetStreamHandle>().

Complete, runnable scenarios live in examples/:

cargo run --example sensor_hub        # wildcard and selection patterns
cargo run --example order_pipeline    # fan-out to multiple services
cargo run --example transport_bridge  # retries, dead letters and transport handles

Benchmarks for topic parsing, subscription matching and end-to-end dispatch:

cargo bench --bench broker

Development

ℹ️ Make sure just, git-cliff and cargo-llvm-cov are installed:

cargo install cargo-llvm-cov just git-cliff

Common operations related to development and release can be found in the justfile. For an overview of available recipes, run:

just

License

Licensed under either of

• Apache License, Version 2.0 (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0)
• MIT license (LICENSE-MIT or http://opensource.org/licenses/MIT)

at your option.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in the work by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions.