Reconnection

Overview

Production connections drop for reasons that have nothing to do with your application code: the broker restarts during a rolling upgrade, a load balancer fails over, a network blip severs the connection. Without built-in reconnection, every client in your fleet needs its own hand-rolled retry-connect loop, and it's easy to get the backoff wrong — too aggressive and you hammer a recovering broker, too slow and you leave the application dark longer than necessary. Automatic reconnection moves that logic into the client itself, so the connection self-heals without any code the application has to write or maintain.

It works by setting a RetryPolicy on KubemqClient::builder() — max_retries, initial_backoff, max_backoff, and multiplier shape the backoff curve. The on_connected and on_closed callbacks fire on connection lifecycle transitions for diagnostics, and client.state() returns the current ConnectionState synchronously. Gotchas: callbacks are async closures run by the client, so long-running work inside one can delay processing of further state transitions; send_event calls issued during the outage window still fail immediately — the policy governs the connection, not individual sends, so the retry loop in the sample handles that at the application level instead; and max_retries is a hard cap, so a broker outage longer than the total backoff window leaves the connection closed rather than retrying indefinitely.

Prerequisites

• KubeMQ server running on localhost:50000
• Rust SDK installed (cargo add kubemq)

Code

use kubemq::prelude::*;
use kubemq::{EventBuilder, RetryPolicy};
use std::time::Duration;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .retry_policy(RetryPolicy {
            max_retries: 3,
            initial_backoff: Duration::from_millis(100),
            max_backoff: Duration::from_secs(10),
            multiplier: 2.0,
            ..Default::default()
        })
        .on_connected(|| async {
            println!("[STATE] Connected");
        })
        .on_closed(|| async {
            println!("[STATE] Closed");
        })
        .build()
        .await?;

    println!("Client state: {:?}", client.state());

    let channel = "rust-error-handling.reconnection";
    for i in 0..30 {
        let event = EventBuilder::new()
            .channel(channel)
            .body(format!("heartbeat-{}", i).into_bytes())
            .build();

        match client.send_event(event).await {
            Ok(()) => println!("Sent heartbeat {}", i),
            Err(e) => println!("Send failed (retrying): {}", e),
        }

        tokio::time::sleep(Duration::from_secs(2)).await;
    }

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

How It Works

• RetryPolicy configures exponential backoff with jitter for automatic reconnection.
• on_connected and on_closed callbacks provide visibility into connection state changes.
• The SDK automatically reconnects subscriptions using the configured retry policy.
• client.state() returns the current ConnectionState (synchronous, no .await).
• Review timeouts, channel names, and client IDs before running against shared environments.
• Run the program while the server from the prerequisites is available.

Related

• Rust SDK Reference
• Connection Error
• Graceful Shutdown
• Custom Timeouts