Reconnection

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 TCP stream. 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 configuring a ReconnectPolicy on the client — an initial delay, a backoff multiplier, a cap on the maximum delay, and either a bounded or unlimited number of attempts. As the connection moves through its lifecycle, the client fires state callbacks (WithOnConnected, WithOnDisconnected, WithOnReconnecting, WithOnReconnected, WithOnClosed) so the application can log or alert on each transition, and client.State() exposes the current state on demand. Once reconnected, active subscriptions are transparently re-registered — no manual re-subscribe logic needed. Gotchas: state callbacks fire synchronously on the connection's internal goroutine, so blocking work inside one stalls reconnection itself; in-flight publishes or requests issued during the outage window still fail immediately — the policy governs the connection, not individual calls, so you still need your own retry for those; and MaxAttempts: 0 (unlimited) will retry forever against a broker that's gone for good, so pair it with alerting on the reconnecting state rather than assuming it will eventually succeed.

Prerequisites

• KubeMQ server running on localhost:50000
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// 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: go-error-handling.reconnection
// Client ID: go-error-handling-reconnection-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/kubemq-io/kubemq-go/v2"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// Configure reconnection policy and state callbacks.
	client, err := kubemq.NewClient(ctx,
		kubemq.WithAddress("localhost", 50000),
		kubemq.WithClientId("go-error-handling-reconnection-client"),
		// Custom reconnect policy with exponential backoff.
		kubemq.WithReconnectPolicy(kubemq.ReconnectPolicy{
			InitialDelay: 1 * time.Second,
			MaxDelay:     30 * time.Second,
			Multiplier:   2.0,
			MaxAttempts:  0, // 0 = unlimited
		}),
		// State callbacks to monitor connection lifecycle.
		kubemq.WithOnConnected(func() {
			fmt.Println("[State] Connected to KubeMQ server")
		}),
		kubemq.WithOnDisconnected(func() {
			fmt.Println("[State] Disconnected from KubeMQ server")
		}),
		kubemq.WithOnReconnecting(func() {
			fmt.Println("[State] Reconnecting to KubeMQ server...")
		}),
		kubemq.WithOnReconnected(func() {
			fmt.Println("[State] Reconnected to KubeMQ server")
		}),
		kubemq.WithOnClosed(func() {
			fmt.Println("[State] Connection closed")
		}),
	)
	if err != nil {
		log.Fatalf("Failed to create client: %v", err)
	}
	defer client.Close()

	// Check connection state.
	state := client.State()
	fmt.Printf("Current state: %v\n", state)

	// Verify connectivity.
	info, err := client.Ping(ctx)
	if err != nil {
		log.Printf("Ping failed: %v", err)
		return
	}
	fmt.Printf("Connected: host=%s version=%s\n", info.Host, info.Version)
	fmt.Println("Client is configured with automatic reconnection")
}

How It Works

1. kubemq.WithReconnectPolicy(kubemq.ReconnectPolicy{...}) configures exponential backoff starting at 1 second, doubling up to 30 seconds, with MaxAttempts: 0 meaning unlimited retries.
2. The state callbacks (WithOnConnected, WithOnDisconnected, WithOnReconnecting, WithOnReconnected, WithOnClosed) are fired synchronously on each transition — do not block inside them.
3. client.State() returns the current ConnectionState atomically; possible values are StateConnecting, StateReady, StateReconnecting, and StateClosed.
4. Active subscriptions are transparently re-registered on the broker after a successful reconnection — no manual re-subscribe logic is needed.

Related

• Go SDK Reference
• Connection Error
• Graceful Shutdown