Graceful Shutdown

Overview

A graceful shutdown stops a KubeMQ client without dropping in-flight messages or leaking server-side subscription state. Killing a process outright, or dropping the client mid-callback, can truncate a handler or leave the server thinking a consumer is still there. In a container platform that sends SIGTERM before force-killing a pod, an orderly shutdown sequence turns a rolling deploy into a clean handoff instead of a burst of errors.

The pattern has a fixed order: stop new work by calling sub.unsubscribe().await, which cancels the subscription's internal CancellationToken and waits for its task to finish, then call client.close().await, which cancels remaining child tasks and releases the gRPC channel.

Gotchas: calling client.close() before the subscription's task has finished can race the channel teardown — unsubscribe() is async precisely so you can await that first. Dropping the client without ever calling close() skips the graceful drain, relying instead on whatever cleanup happens on Drop.

Prerequisites

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

Code

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

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

    let channel = "rust-error-handling.graceful-shutdown";

    let sub = client
        .subscribe_to_events(
            channel,
            "",
            |event| {
                Box::pin(async move {
                    println!(
                        "Received: body={}",
                        String::from_utf8_lossy(&event.body)
                    );
                })
            },
            None,
        )
        .await?;

    tokio::time::sleep(Duration::from_millis(500)).await;
    println!("Subscription active, sending events...");

    for i in 0..3 {
        let event = EventBuilder::new()
            .channel(channel)
            .body(format!("shutdown-msg-{}", i).into_bytes())
            .build();
        client.send_event(event).await?;
        tokio::time::sleep(Duration::from_millis(200)).await;
    }

    tokio::time::sleep(Duration::from_secs(1)).await;

    println!("Cancelling subscription...");
    sub.unsubscribe().await;
    println!("Subscription cancelled");

    client.close().await?;
    println!("Client closed — shutdown complete");

    Ok(())
}

How It Works

• The shutdown sequence is: (1) unsubscribe active subscriptions, (2) close the client.
• unsubscribe() cancels the subscription via CancellationToken and waits for the task to finish.
• close() cancels all remaining child tasks and releases the gRPC channel.
• 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
• Reconnection
• Close