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 ReconnectPolicy on ClientOptions — initial_delay, max_delay, backoff_multiplier, max_attempts, and jitter shape the backoff curve. State callbacks (set_on_connected, set_on_disconnected, set_on_reconnecting, set_on_reconnected, set_on_closed) fire on each transition so the application can log or alert without polling, and client->State() exposes the current state on demand via ConnectionStateToString(). Gotchas: state callbacks run on the client's internal thread, so blocking work inside one stalls reconnection itself; in-flight calls issued during the outage window still fail immediately — the policy governs the connection, not individual RPCs; and max_attempts = 0 (unlimited) will retry forever against a broker that's gone for good, so pair it with alerting on the reconnecting callback rather than assuming it will eventually succeed.

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: error_handling/reconnection
//
// Demonstrates configuring automatic reconnection with state callbacks.
// The client registers callbacks for connection state transitions
// (connected, disconnected, reconnecting, reconnected, closed).
//
// Channel: cpp-error-handling.reconnection
// Client ID: cpp-error-handling-reconnection-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).

#include <kubemq/kubemq.h>

#include <chrono>
#include <iostream>

int main() {
    std::cout << "[1] Configuring reconnection policy and state callbacks" << std::endl;

    kubemq::ClientOptions options;
    options.set_address("localhost", 50000);
    options.set_client_id("cpp-error-handling-reconnection-client");

    // Custom reconnect policy with exponential backoff.
    kubemq::ReconnectPolicy policy;
    policy.initial_delay = std::chrono::milliseconds(1000);
    policy.max_delay = std::chrono::milliseconds(30000);
    policy.backoff_multiplier = 2.0;
    policy.max_attempts = 0;  // 0 = unlimited
    policy.jitter = kubemq::JitterMode::kFull;
    options.set_reconnect_policy(policy);

    // State callbacks to monitor connection lifecycle.
    options.set_on_connected(
        []() { std::cout << "[State] Connected to KubeMQ server" << std::endl; });
    options.set_on_disconnected(
        []() { std::cout << "[State] Disconnected from KubeMQ server" << std::endl; });
    options.set_on_reconnecting(
        []() { std::cout << "[State] Reconnecting to KubeMQ server..." << std::endl; });
    options.set_on_reconnected(
        []() { std::cout << "[State] Reconnected to KubeMQ server" << std::endl; });
    options.set_on_closed([]() { std::cout << "[State] Connection closed" << std::endl; });

    std::cout << "[2] Connecting to localhost:50000" << std::endl;
    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;

    // Check connection state.
    auto state = client->State();
    std::cout << "[3] Current state: " << kubemq::ConnectionStateToString(state) << std::endl;

    // Verify connectivity.
    auto ping_result = client->Ping();
    if (!ping_result.ok()) {
        std::cerr << "[ERROR] Ping failed: " << ping_result.status().message() << 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 << "[4] Connected: host=" << ping_result->host << " version=" << ping_result->version
              << std::endl;
    std::cout << "[5] Client is configured with automatic reconnection" << std::endl;

    // Close the client explicitly.
    // Note: The Client destructor also calls Close(), but explicit
    // cleanup is shown here for clarity and to match Go's defer pattern.
    auto close_status = client->Close();
    if (!close_status.ok()) {
        std::cerr << "[ERROR] Close failed: " << close_status.message() << std::endl;
        return 1;
    }
    std::cout << "[6] Client closed" << std::endl;

    return 0;
}

How It Works

Configures a ReconnectPolicy with initial delay, max delay, backoff multiplier, and jitter.
Registers state callbacks for connected, disconnected, reconnecting, reconnected, and closed events.
Uses ConnectionStateToString() to display the current connection state.
The client automatically reconnects when the connection is lost.

Reconnection

Overview

Prerequisites

Code

How It Works

On this page

Reconnection

Overview

Prerequisites

Code

How It Works

Related

On this page