KubeMQ Rust SDK

The official Rust client library for KubeMQ message broker. Supports Events, Events Store, Commands, Queries, and Queues messaging patterns.

Table of Contents

• Installation
• Quick Start
• Features
• Configuration
  • Environment Variables
• Examples
• Error Handling
• Reconnection
• TLS & Authentication
  • Server-side TLS
  • Mutual TLS (mTLS)
  • Auth Token
• Minimum Supported Rust Version
• Contributing
• License

Installation

cargo add kubemq

Or add to your Cargo.toml:

[dependencies]
kubemq = "1.0"

Quick Start

use kubemq::prelude::*;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .build()
        .await?;

    // Send a fire-and-forget event
    let event = Event::builder()
        .channel("my-channel")
        .body(b"Hello KubeMQ!".to_vec())
        .build();
    client.send_event(&event).await?;

    client.close().await?;
    Ok(())
}

Features

• Events (Pub/Sub) -- Fire-and-forget messaging with wildcard subscriptions and consumer groups
• Events Store (Persistent Pub/Sub) -- Persistent events with 6 replay start positions
• Queues (Stream + Simple) -- Reliable message queuing with ack/nack, dead-letter, delay, and expiration
• Commands (RPC) -- Request/response with timeout
• Queries (RPC with Cache) -- Request/response with server-side caching
• Channel Management -- Create, delete, and list channels for all pattern types
• Auto-Reconnection -- Automatic reconnection with configurable backoff
• TLS & mTLS -- Secure connections via rustls
• OpenTelemetry -- Optional tracing and metrics instrumentation (feature-gated)

Configuration

Environment Variables

All configuration can be set via environment variables with precedence: Builder method > Environment variable > Default.

Examples

See the examples/ directory for 34 standalone examples covering all patterns:

• Connection: ping, TLS, mTLS, auth token
• Events: pub/sub, streaming, wildcards, consumer groups
• Events Store: pub/sub, all 6 start types, streaming, consumer groups
• Commands: send/handle, consumer groups
• Queries: send/handle, caching, consumer groups
• Queues Simple: send/receive, batch, peek, ack-all, delay, expiration, dead-letter
• Queues Stream: upstream, downstream, auto-ack, ack-range, nack, requeue, poll
• Management: create/list/delete channels
• Error handling and reconnection

Run any example with:

cargo run --example connection_ping
cargo run --example events_pubsub
cargo run --example queues_stream_downstream

Error Handling

All operations return kubemq::Result<T> with structured error types:

match client.send_event(&event).await {
    Ok(()) => println!("Sent"),
    Err(e) => {
        println!("Error: {}", e);
        println!("Code: {:?}", e.code());
        println!("Retryable: {}", e.is_retryable());
        println!("Suggestion: {}", e.suggestion());
    }
}

Error codes: Transient, Timeout, Throttling, Authentication, Authorization, Validation, NotFound, Fatal, Cancellation, Backpressure.

Reconnection

Auto-reconnect is enabled by default with unlimited attempts and exponential backoff (1s-30s). Events and queues are buffered during reconnection (up to 1000 messages). Commands and queries fail immediately during reconnection.

let client = KubemqClient::builder()
    .host("localhost")
    .port(50000)
    .on_disconnected(|| println!("Disconnected"))
    .on_reconnecting(|| println!("Reconnecting..."))
    .on_reconnected(|| println!("Reconnected!"))
    .build()
    .await?;

TLS & Authentication

Server-side TLS

let tls = TlsConfig {
    ca_cert_file: Some("/path/to/ca.pem".to_string()),
    ..Default::default()
};
let client = KubemqClient::builder()
    .tls_config(tls)
    .build()
    .await?;

Mutual TLS (mTLS)

let tls = TlsConfig {
    ca_cert_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/client.pem".to_string()),
    key_file: Some("/path/to/client-key.pem".to_string()),
    ..Default::default()
};

Auth Token

let client = KubemqClient::builder()
    .auth_token("your-token")
    .build()
    .await?;

Minimum Supported Rust Version

This crate requires Rust 1.75 or later.

Contributing

See CONTRIBUTING.md for development setup, build/test commands, and PR process.

License

Apache-2.0. See LICENSE for details.