hermes-rs

Hermes is a lightweight, async event broker for Rust built on gRPC. It supports fire-and-forget fanout, queue groups for load-balanced consumption, and durable delivery with ack/nack and automatic redelivery.

Features

• Typed events — define events as Rust structs with #[derive(Event)], serialized automatically
• Fanout — every subscriber receives every message
• Queue groups — competing consumers, each message delivered to exactly one subscriber per group
• Event groups — subscribe to multiple event types as a single merged stream
• Durable delivery — at-least-once with persistence, ack/nack, redelivery, and dead-lettering
• Batch publishing — high-throughput buffered publisher
• Wildcard subscriptions — * (single segment) and > (rest) pattern matching
• gRPC transport — built on tonic, HTTP/2 multiplexed

Quick start

1. Start the broker

use tokio::net::TcpListener;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("0.0.0.0:4222").await?;
    hermes_server::run(listener).await?;
    Ok(())
}

Or run directly:

cargo run -p hermes-broker-server

2. Publish and subscribe

use futures::StreamExt;
use hermes_client::HermesClient;
use hermes_core::Event;
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Serialize, Deserialize, Event)]
#[event(subject = "chat.message")]
struct ChatMessage {
    user: String,
    text: String,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = HermesClient::connect("http://127.0.0.1:4222").await?;

    // Subscribe FIRST — fanout delivers only to active subscribers
    let sub = client.clone();
    tokio::spawn(async move {
        let mut stream = sub.subscribe::<ChatMessage>(&[]).await.unwrap();
        while let Some(Ok(msg)) = stream.next().await {
            println!("[{}]: {}", msg.user, msg.text);
        }
    });

    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    // Publish
    client.publish(&ChatMessage {
        user: "alice".into(),
        text: "Hello, world!".into(),
    }).await?;

    Ok(())
}

3. Queue groups (load balancing)

# use futures::StreamExt;
# use hermes_client::HermesClient;
# use hermes_core::Event;
# use serde::{Deserialize, Serialize};
# #[derive(Debug, Clone, Serialize, Deserialize, Event)]
# struct Task { id: u32 }
# async fn example(client: &HermesClient) {
// Multiple subscribers in the same group — each message goes to exactly one
let mut stream = client.subscribe::<Task>(&["workers"]).await.unwrap();
# }

4. Durable delivery (at-least-once)

# use hermes_client::HermesClient;
# use hermes_core::Event;
# use serde::{Deserialize, Serialize};
# #[derive(Debug, Clone, Serialize, Deserialize, Event)]
# struct Payment { id: String }
# async fn example(client: &HermesClient) -> Result<(), Box<dyn std::error::Error>> {
// Publish with persistence
client.publish_durable(&Payment { id: "PAY-1".into() }).await?;

// Subscribe with ack/nack — un-acked messages are automatically redelivered
let mut sub = client
    .subscribe_durable::<Payment>("payment-processor", &[], 10, 30)
    .await?;

while let Some(msg) = sub.next().await {
    let msg = msg?;
    println!("payment: {:?}", msg.event);
    msg.ack().await?;
}
# Ok(())
# }

Crates

Server configuration

Configure via environment variables or ServerConfig struct:

HERMES_STORE_PATH=./hermes.redb \   # enables durable mode
HERMES_ACK_TIMEOUT=30 \
HERMES_MAX_DELIVERY_ATTEMPTS=5 \
cargo run -p hermes-broker-server

Without HERMES_STORE_PATH, the server runs in fire-and-forget mode only.

See the server README for all configuration options.

License

Licensed under either of:

• Apache License, Version 2.0 (LICENSE-APACHE)
• MIT license (LICENSE-MIT)

at your option.