Custom Timeouts

Overview

Every client operation has an implicit deadline — how long to wait for the initial connection, how long before a dead socket is detected, how long reconnection retries keep running, and whether your application even finds out when the connection state changes. The defaults are reasonable for a healthy local network, but they're wrong for high-latency links, connections that pass through load balancers or NAT gateways, or servers that occasionally run slow under load. Tuning timeouts explicitly is how you trade fast-fail behavior against tolerance for transient slowness.

The retry_policy() builder method configures exponential backoff — max_retries, initial_backoff, max_backoff, and multiplier — for reconnection attempts, while on_connected / on_closed register async callbacks so your application can react to connection-state transitions instead of polling for them. Keepalive is a separate pair of builder methods: keepalive_time(Duration) sets the HTTP/2 keepalive interval (how often an idle connection pings the server) and keepalive_timeout(Duration) sets how long to wait for a ping response before the connection is considered dead — together they control how quickly a silently-broken connection (e.g. behind a NAT gateway or load balancer that drops idle sockets) is detected instead of hanging until the next real request fails. Gotchas: a retry_policy with a high max_retries and no cap on total elapsed time will keep retrying against a server that's down for good, silently masking an outage instead of surfacing it; ..Default::default() silently fills in defaults for fields you didn't set (including jitter_mode: Full) — review what those defaults actually are before relying on them; on_connected / on_closed fire on every reconnect, not just the first connect, so idempotent handling matters if the callback has side effects; and keepalive_time must be at least 5 seconds — the SDK rejects a shorter interval at build time.

Prerequisites

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

Code

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

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

    let info = client.ping().await?;
    println!("Connected with custom timeouts. Server: {}", info.version);

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

To tune keepalive specifically, chain keepalive_time() and keepalive_timeout() on the same builder:

use kubemq::prelude::*;
use std::time::Duration;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .client_id("rust-keepalive-example")
        .keepalive_time(Duration::from_secs(30))
        .keepalive_timeout(Duration::from_secs(10))
        .build()
        .await?;

    let info = client.ping().await?;
    println!("Connected with keepalive tuned. Server: {}", info.version);

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

How It Works

• RetryPolicy configures exponential backoff with jitter for subscription reconnection.
• on_connected and on_closed are async callbacks invoked on connection lifecycle events.
• The ..Default::default() syntax fills remaining fields with defaults (jitter_mode: Full).
• keepalive_time() / keepalive_timeout() detect a dead connection (defaults 10s / 5s); retry_policy governs reconnection after that detection.
• Review timeouts, channel names, and client IDs before running against shared environments.
• Run the program while the server from the prerequisites is available.

Related

• Getting Started
• Rust SDK Reference
• Reconnection