Overview

A live Events subscription holds a client-side stream and its background thread 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->Cancel() on the returned handle stops delivery cleanly and frees those resources on both sides.

SubscribeToEvents returns a Subscription handle rather than blocking, so the callback keeps firing in the background until you cancel it. sub->Cancel() stops receiving events, and sub->IsDone() gives you a synchronous check to confirm the stream has fully terminated — useful before the object goes out of scope. The Subscription destructor also calls Cancel() via RAII, so explicit cancellation is really about controlling when teardown happens rather than whether it happens at all.

Gotchas: Cancel() 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 afterward. 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
C++ SDK installed (vcpkg or CMake FetchContent)
C++17 compiler (GCC 9+, Clang 9+, MSVC 2019+)

Code

main.cc

// Example: events/cancel_subscription
//
// Demonstrates how to cancel (unsubscribe from) an event subscription.
// After cancellation, no more events are delivered to the handler.
//
// Channel: cpp-events.cancel-subscription
// Client ID: cpp-events-cancel-subscription-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).

#include <kubemq/kubemq.h>

#include <chrono>
#include <iostream>
#include <thread>

int main() {
    std::cout << "[1] Connecting to localhost:50000" << std::endl;

    kubemq::ClientOptions options;
    options.set_address("localhost", 50000);
    options.set_client_id("cpp-events-cancel-subscription-client");

    auto client_result = kubemq::Client::Create(options);
    if (!client_result.ok()) {
        std::cerr << "[ERROR] Failed to create client: " << client_result.status().message()
                  << std::endl;
        return 1;
    }
    auto& client = *client_result;

    std::string channel = "cpp-events.cancel-subscription";

    // Create a subscription.
    std::cout << "[2] Subscribing to channel " << channel << std::endl;
    auto sub_result = client->SubscribeToEvents(
        channel, "",
        [](const kubemq::EventReceive& event) {
            std::cout << "[4] Received: body=" << event.body << std::endl;
        },
        [](const kubemq::Status& err) {
            std::cerr << "[ERROR] Subscription error: " << err.message() << std::endl;
        });
    if (!sub_result.ok()) {
        std::cerr << "[ERROR] SubscribeToEvents: " << sub_result.status().message() << std::endl;
        return 1;
    }
    auto& sub = *sub_result;

    // Allow time for subscription to register on server.
    std::this_thread::sleep_for(std::chrono::seconds(1));

    // Send an event and wait for it.
    auto ev_result = kubemq::Event::Builder()
                         .SetChannel(channel)
                         .SetBody("before cancel")
                         .SetMetadata("test")
                         .Build();
    if (!ev_result.ok()) {
        std::cerr << "[ERROR] Build event: " << ev_result.status().message() << std::endl;
        return 1;
    }
    auto send_status = client->SendEvent(*ev_result);
    if (!send_status.ok()) {
        std::cerr << "[ERROR] SendEvent: " << send_status.message() << std::endl;
        return 1;
    }
    std::cout << "[3] Event sent" << std::endl;

    std::this_thread::sleep_for(std::chrono::seconds(2));
    std::cout << "[5] Event received before cancellation" << std::endl;

    // Cancel the subscription. After this, no more events are delivered.
    // Note: Subscription destructor would also handle cleanup (RAII)
    sub->Cancel();
    std::cout << "[6] Subscription cancelled" << std::endl;

    // Check that the subscription is done.
    if (sub->IsDone()) {
        std::cout << "[7] Subscription confirmed done" << std::endl;
    }

    auto close_status = client->Close();
    if (!close_status.ok()) {
        std::cerr << "[ERROR] Close failed: " << close_status.message() << std::endl;
        return 1;
    }
    std::cout << "[8] Client closed" << std::endl;

    return 0;
}

How It Works

Creates a subscription and receives an event before cancellation.
Calls sub->Cancel() to stop receiving events.
Verifies with sub->IsDone() that the subscription is fully terminated.
The Subscription destructor also calls Cancel() (RAII), but explicit cancel gives control over timing.

Cancel Subscription

Overview

Prerequisites

Code

How It Works

On this page

Cancel Subscription

Overview

Prerequisites

Code

How It Works

Related

On this page