Cancel Subscription

Overview

A live Events subscription holds a background task and its underlying stream open indefinitely, so a long-running service needs an explicit way to tear one down without closing the whole client connection — for example when a feature flag disables a channel, a worker is draining before shutdown, or a subscription needs to be re-created with different options. Calling sub.unsubscribe().await on the returned handle stops delivery cleanly and frees those resources on both sides.

subscribe_to_events returns a subscription handle backed by a CancellationToken, so the callback keeps firing in the background until you cancel it. sub.unsubscribe().await cancels that token and awaits cleanup, so by the time the call returns you know the underlying task has actually stopped rather than merely been signaled to stop.

Gotchas: unsubscribing only affects this one handle — other subscribers on the same channel keep receiving events. Events already in flight when you call it may still reach the callback briefly beforehand. And because Events are fire-and-forget, anything published after cancellation is simply dropped for this subscriber — there's no queue to catch up from later.

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-events.cancel-subscription";

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

    tokio::time::sleep(Duration::from_millis(500)).await;

    let event = EventBuilder::new()
        .channel(channel)
        .body(b"before-cancel".to_vec())
        .build();
    client.send_event(event).await?;

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

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

    tokio::time::sleep(Duration::from_millis(500)).await;

    let event = EventBuilder::new()
        .channel(channel)
        .body(b"after-cancel".to_vec())
        .build();
    client.send_event(event).await?;
    println!("Event sent after cancel (should not be received)");

    tokio::time::sleep(Duration::from_secs(1)).await;
    client.close().await?;
    Ok(())
}

How It Works

• sub.unsubscribe().await cancels the subscription via CancellationToken and waits for cleanup.
• Events published after cancellation are not delivered to the cancelled subscriber.
• Review timeouts, channel names, and client IDs before running against shared environments.
• Run the program while the server from the prerequisites is available.

Related

• Events overview
• Rust SDK Reference
• Basic Pub/Sub