Rust

[dependencies]
tonic       = "0.12"
prost       = "0.13"
tokio       = { version = "1", features = ["rt-multi-thread", "macros"] }
tokio-stream = "0.1"

[build-dependencies]
tonic-build = "0.12"

fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic_build::configure()
        .build_server(false)
        .compile_protos(
            &[
                "proto/tinybroker/v1/service.proto",
                "proto/tinybroker/v1/client.proto",
                "proto/tinybroker/v1/server.proto",
                "proto/tinybroker/v1/shared.proto",
            ],
            &["proto"],
        )?;
    Ok(())
}

pub mod tinybroker {
    pub mod v1 {
        tonic::include_proto!("tinybroker.v1");
    }
}

use tinybroker::v1::broker_client::BrokerClient;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut client = BrokerClient::connect("http://tinybroker:50051").await?;
    // ...
    Ok(())
}

use tinybroker::v1::PublishRequest;

client.publish(PublishRequest {
    topic:   "events.user.signup".into(),
    payload: b"{\"user_id\":\"abc123\"}".to_vec(),
    ..Default::default()
}).await?;

use tinybroker::v1::{
    SubscribeCommand, OpenSubscription, DeliveryMode,
    subscribe_command, subscription_event,
};
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;

let (tx, rx) = mpsc::channel(32);

// Open the subscription
tx.send(SubscribeCommand {
    command: Some(subscribe_command::Command::Open(OpenSubscription {
        topic_patterns: vec!["events.user.*".into(), "events.order.*".into()],
        mode: DeliveryMode::Multi as i32,
        ..Default::default()
    })),
}).await?;

let mut stream = client
    .subscribe(ReceiverStream::new(rx))
    .await?
    .into_inner();

let mut subscription_id = String::new();

while let Some(event) = stream.message().await? {
    match event.event {
        Some(subscription_event::Event::SubscriptionId(id)) => {
            subscription_id = id;
        }
        Some(subscription_event::Event::Message(msg)) => {
            println!("topic={} payload={}", msg.topic, String::from_utf8_lossy(&msg.payload));
        }
        Some(subscription_event::Event::PatternsUpdated(p)) => {
            println!("patterns now: {:?}", p.topic_patterns);
        }
        None => {}
    }
}

use tinybroker::v1::AckRequest;

tx.send(SubscribeCommand {
    command: Some(subscribe_command::Command::Open(OpenSubscription {
        subscription_id: "job-workers".into(),
        topic_patterns:  vec!["jobs.*".into()],
        mode:            DeliveryMode::Balanced as i32,
        consumer_group:  "job-workers".into(),
    })),
}).await?;

while let Some(event) = stream.message().await? {
    if let Some(subscription_event::Event::Message(msg)) = event.event {
        process_job(&msg.payload);
        client.ack(AckRequest {
            subscription_id: "job-workers".into(),
            system_id:       msg.system_id,
        }).await?;
    }
}

use tinybroker::v1::{AddTopicPatterns, RemoveTopicPatterns};
use subscribe_command::Command;

// Add a pattern on the live subscription
tx.send(SubscribeCommand {
    command: Some(Command::AddPatterns(AddTopicPatterns {
        topic_patterns: vec!["alerts.*".into()],
    })),
}).await?;

// Remove a pattern
tx.send(SubscribeCommand {
    command: Some(Command::RemovePatterns(RemoveTopicPatterns {
        topic_patterns: vec!["events.user.*".into()],
    })),
}).await?;

async fn subscribe_with_reconnect(addr: &str) {
    loop {
        match run_subscription(addr).await {
            Ok(_) => {}
            Err(e) => {
                eprintln!("subscription error: {e} — reconnecting in 2s");
                tokio::time::sleep(std::time::Duration::from_secs(2)).await;
            }
        }
    }
}