KubemqClient

The primary entry point for all messaging operations. Thread-safe — cloning is cheap (Arc-based).

main.rs

use kubemq::prelude::*;

let client = KubemqClient::builder()
    .host("localhost")
    .port(50000)
    .client_id("my-service")
    .build()
    .await?;

Methods

Method	Returns	Description
`builder()`	`ClientConfigBuilder`	Creates a new configuration builder
`ping()`	`Result<ServerInfo>`	Health-check that returns server metadata
`close()`	`Result<()>`	Graceful shutdown; cancels subscriptions and releases the gRPC channel
`state()`	`ConnectionState`	Returns current connection state (synchronous)
`config()`	`&ClientConfig`	Returns a reference to the client configuration

ClientConfigBuilder

Builder for creating a KubemqClient. Configuration precedence: builder method > environment variable > compiled default.

Builder Methods

Method	Default	Description
`.host(h)`	`"localhost"`	Broker hostname. Falls back to `KUBEMQ_ADDRESS` env var
`.port(p)`	`50000`	Broker port number
`.client_id(id)`	UUID v4	Client identifier sent with every request
`.auth_token(token)`	None	Authentication token for gRPC metadata
`.tls_config(config)`	None	TLS/mTLS configuration via `TlsConfig`
`.connection_timeout(d)`	10s	Timeout for establishing the initial connection
`.check_connection(bool)`	`false`	Ping the broker during `build()` to verify connectivity
`.drain_timeout(d)`	5s	Time allowed for in-flight tasks during `close()`
`.keepalive_time(d)`	10s	HTTP/2 keepalive interval (must be >= 5s)
`.keepalive_timeout(d)`	5s	HTTP/2 keepalive ping timeout
`.max_receive_message_size(n)`	4 MB	Maximum inbound message size in bytes
`.max_send_message_size(n)`	100 MB	Maximum outbound message size in bytes
`.retry_policy(policy)`	3 retries, 100ms–10s	`RetryPolicy` for automatic reconnection
`.rpc_timeout(d)`	60s	Timeout for RPC operations (Commands/Queries)
`.on_connected(cb)`	None	Async callback on connection establishment
`.on_closed(cb)`	None	Async callback on connection close
`.credential_provider(p)`	None	Dynamic credential provider for per-request tokens

ConnectionState

main.rs

match client.state() {
    ConnectionState::Idle => println!("Not yet connected"),
    ConnectionState::Ready => println!("Connected"),
    ConnectionState::Closed => println!("Closed"),
}

TlsConfig

Configuration for TLS and mutual TLS (mTLS) connections.

Field	Type	Description
`ca_cert_file`	`Option<String>`	Path to CA certificate file
`ca_cert_pem`	`Option<Vec<u8>>`	PEM-encoded CA certificate bytes
`cert_file`	`Option<String>`	Path to client certificate file (mTLS)
`key_file`	`Option<String>`	Path to client private key file (mTLS)
`cert_pem`	`Option<Vec<u8>>`	PEM-encoded client certificate (mTLS)
`key_pem`	`Option<Vec<u8>>`	PEM-encoded client private key (mTLS)
`server_name`	`Option<String>`	Override server name for TLS verification

RetryPolicy

Retry policy with exponential backoff and configurable jitter.

Field	Default	Description
`max_retries`	3	Maximum retry attempts (0 = unlimited)
`initial_backoff`	100ms	Backoff duration for the first retry
`max_backoff`	10s	Upper bound for backoff duration
`multiplier`	2.0	Exponential multiplier per attempt
`jitter_mode`	`Full`	`None`, `Full`, or `Equal` jitter

main.rs

use kubemq::RetryPolicy;
use std::time::Duration;

let policy = RetryPolicy {
    max_retries: 5,
    initial_backoff: Duration::from_millis(200),
    max_backoff: Duration::from_secs(30),
    multiplier: 2.0,
    ..Default::default()
};

ServerInfo

Returned by client.ping().

Field	Type	Description
`host`	`String`	Server hostname
`version`	`String`	Server version
`server_start_time`	`i64`	Server start time (Unix timestamp)
`server_up_time_seconds`	`i64`	Server uptime in seconds

Client

On this page