Error Handling Patterns

package main

import (
    "context"
    "log"
    "time"

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

func connectWithRetry(ctx context.Context) *kubemq.Client {
    var client *kubemq.Client
    var err error
    backoff := time.Second

    for {
        client, err = kubemq.NewClient(ctx,
            kubemq.WithAddress("localhost", 50000),
            kubemq.WithClientId("resilient-client"),
        )
        if err == nil {
            if _, pingErr := client.Ping(ctx); pingErr == nil {
                log.Println("Connected to KubeMQ")
                return client
            }
            client.Close()
        }

        log.Printf("Connection failed: %v — retrying in %v", err, backoff)
        time.Sleep(backoff)
        if backoff < 30*time.Second {
            backoff *= 2
        }
    }
}

import time
from kubemq.pubsub import Client as PubSubClient

def connect_with_retry(address: str) -> PubSubClient:
    backoff = 1.0
    while True:
        try:
            client = PubSubClient(
                address=address,
                client_id="resilient-client",
            )
            client.ping()
            print("Connected to KubeMQ")
            return client
        except Exception as e:
            print(f"Connection failed: {e} — retrying in {backoff}s")
            time.sleep(backoff)
            backoff = min(backoff * 2, 30)

const { KubeMQClient } = require("kubemq-js");

async function connectWithRetry(address) {
  let backoff = 1000;
  while (true) {
    try {
      const client = new KubeMQClient({
        address,
        clientId: "resilient-client",
      });
      await client.ping();
      console.log("Connected to KubeMQ");
      return client;
    } catch (err) {
      console.error(`Connection failed: ${err.message} — retrying in ${backoff}ms`);
      await new Promise((r) => setTimeout(r, backoff));
      backoff = Math.min(backoff * 2, 30000);
    }
  }
}

PubSubClient connectWithRetry(String address) throws InterruptedException {
    long backoff = 1000;
    while (true) {
        try {
            PubSubClient client = PubSubClient.builder()
                .address(address)
                .clientId("resilient-client")
                .build();
            client.ping();
            System.out.println("Connected to KubeMQ");
            return client;
        } catch (Exception e) {
            System.err.printf("Connection failed: %s — retrying in %dms%n",
                e.getMessage(), backoff);
            Thread.sleep(backoff);
            backoff = Math.min(backoff * 2, 30000);
        }
    }
}

async Task<KubeMQClient> ConnectWithRetryAsync(string address)
{
    var backoff = TimeSpan.FromSeconds(1);
    while (true)
    {
        try
        {
            var client = new KubeMQClient(new KubeMQClientOptions
            {
                Address = address,
                ClientId = "resilient-client",
            });
            await client.ConnectAsync();
            await client.PingAsync();
            Console.WriteLine("Connected to KubeMQ");
            return client;
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Connection failed: {ex.Message} — retrying in {backoff}");
            await Task.Delay(backoff);
            if (backoff < TimeSpan.FromSeconds(30))
                backoff *= 2;
        }
    }
}

import io.kubemq.sdk.pubsub.PubSubClient
import kotlinx.coroutines.delay

suspend fun connectWithRetry(address: String): PubSubClient {
    var backoff = 1000L
    while (true) {
        try {
            val client = PubSubClient(address)
            client.ping()
            println("Connected to KubeMQ")
            return client
        } catch (e: Exception) {
            println("Connection failed: ${e.message} — retrying in ${backoff}ms")
            delay(backoff)
            backoff = minOf(backoff * 2, 30000)
        }
    }
}

#include <kubemq/client.h>
#include <iostream>
#include <thread>
#include <chrono>

kubemq::PubSubClient connectWithRetry(const std::string& address) {
    int backoff = 1000;
    while (true) {
        try {
            kubemq::PubSubClient client(address);
            client.ping();
            std::cout << "Connected to KubeMQ" << std::endl;
            return client;
        } catch (const std::exception& e) {
            std::cerr << "Connection failed: " << e.what()
                      << " — retrying in " << backoff << "ms" << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(backoff));
            backoff = std::min(backoff * 2, 30000);
        }
    }
}

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

async fn connect_with_retry(host: &str, port: u16) -> KubemqClient {
    let mut backoff = Duration::from_secs(1);
    loop {
        // The builder applies its own retry policy on transient failures;
        // check_connection verifies the connection before returning.
        let result = KubemqClient::builder()
            .host(host)
            .port(port)
            .client_id("resilient-client")
            .check_connection(true)
            .retry_policy(RetryPolicy {
                max_retries: 3,
                initial_backoff: Duration::from_millis(100),
                max_backoff: Duration::from_secs(10),
                multiplier: 2.0,
                ..Default::default()
            })
            .build()
            .await;

        match result {
            Ok(client) => {
                println!("Connected to KubeMQ");
                return client;
            }
            Err(e) => {
                println!("Connection failed: {e} — retrying in {backoff:?}");
                tokio::time::sleep(backoff).await;
                backoff = std::cmp::min(backoff * 2, Duration::from_secs(30));
            }
        }
    }
}

require 'kubemq'

def connect_with_retry(address)
  backoff = 1.0
  loop do
    begin
      client = KubeMQ::PubSubClient.new(
        address: address,
        client_id: 'resilient-client'
      )
      client.ping
      puts 'Connected to KubeMQ'
      return client
    rescue KubeMQ::Error, StandardError => e
      puts "Connection failed: #{e.message} — retrying in #{backoff}s"
      sleep backoff
      backoff = [backoff * 2, 30].min
    end
  end
end

# KubeMQ.Client.start_link/1 returns {:ok, client} or {:error, reason}.
# Retry with exponential backoff until the connection succeeds.
defmodule Resilient do
  def connect_with_retry(address, backoff \\ 1_000) do
    case KubeMQ.Client.start_link(address: address, client_id: "resilient-client") do
      {:ok, client} ->
        IO.puts("Connected to KubeMQ")
        client

      {:error, reason} ->
        IO.puts("Connection failed: #{inspect(reason)} — retrying in #{backoff}ms")
        Process.sleep(backoff)
        connect_with_retry(address, min(backoff * 2, 30_000))
    end
  end
end

func sendWithRetry(ctx context.Context, client *kubemq.Client, event *kubemq.Event, maxRetries int) error {
    backoff := 100 * time.Millisecond
    for attempt := 0; attempt <= maxRetries; attempt++ {
        err := client.SendEvent(ctx, event)
        if err == nil {
            return nil
        }
        if attempt == maxRetries {
            return fmt.Errorf("failed after %d retries: %w", maxRetries, err)
        }
        log.Printf("Attempt %d failed: %v — retrying in %v", attempt+1, err, backoff)
        time.Sleep(backoff)
        backoff *= 2
    }
    return nil
}

import time

def send_with_retry(client, event, max_retries=3):
    backoff = 0.1
    for attempt in range(max_retries + 1):
        try:
            client.send_event(event)
            return
        except Exception as e:
            if attempt == max_retries:
                raise RuntimeError(f"Failed after {max_retries} retries") from e
            print(f"Attempt {attempt + 1} failed: {e} — retrying in {backoff}s")
            time.sleep(backoff)
            backoff *= 2

async function sendWithRetry(client, event, maxRetries = 3) {
  let backoff = 100;
  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      await client.sendEvent(event);
      return;
    } catch (err) {
      if (attempt === maxRetries) {
        throw new Error(`Failed after ${maxRetries} retries: ${err.message}`);
      }
      console.warn(`Attempt ${attempt + 1} failed: ${err.message} — retrying in ${backoff}ms`);
      await new Promise((r) => setTimeout(r, backoff));
      backoff *= 2;
    }
  }
}

void sendWithRetry(PubSubClient client, EventMessage event, int maxRetries)
        throws Exception {
    long backoff = 100;
    for (int attempt = 0; attempt <= maxRetries; attempt++) {
        try {
            client.sendEventsMessage(event);
            return;
        } catch (Exception e) {
            if (attempt == maxRetries) {
                throw new RuntimeException("Failed after " + maxRetries + " retries", e);
            }
            System.err.printf("Attempt %d failed: %s — retrying in %dms%n",
                attempt + 1, e.getMessage(), backoff);
            Thread.sleep(backoff);
            backoff *= 2;
        }
    }
}

async Task SendWithRetryAsync(KubeMQClient client, EventMessage message, int maxRetries = 3)
{
    var backoff = TimeSpan.FromMilliseconds(100);
    for (int attempt = 0; attempt <= maxRetries; attempt++)
    {
        try
        {
            await client.SendEventAsync(message);
            return;
        }
        catch (Exception ex)
        {
            if (attempt == maxRetries)
                throw new InvalidOperationException($"Failed after {maxRetries} retries", ex);
            Console.WriteLine($"Attempt {attempt + 1} failed: {ex.Message} — retrying in {backoff}");
            await Task.Delay(backoff);
            backoff *= 2;
        }
    }
}

suspend fun sendWithRetry(client: PubSubClient, event: EventMessage, maxRetries: Int = 3) {
    var backoff = 100L
    for (attempt in 0..maxRetries) {
        try {
            client.sendEvent(event)
            return
        } catch (e: Exception) {
            if (attempt == maxRetries) {
                throw RuntimeException("Failed after $maxRetries retries", e)
            }
            println("Attempt ${attempt + 1} failed: ${e.message} — retrying in ${backoff}ms")
            delay(backoff)
            backoff *= 2
        }
    }
}

void sendWithRetry(kubemq::PubSubClient& client, kubemq::EventMessage& event, int maxRetries = 3) {
    int backoff = 100;
    for (int attempt = 0; attempt <= maxRetries; ++attempt) {
        try {
            client.sendEvent(event);
            return;
        } catch (const std::exception& e) {
            if (attempt == maxRetries) {
                throw std::runtime_error(
                    "Failed after " + std::to_string(maxRetries) + " retries: " + e.what());
            }
            std::cerr << "Attempt " << attempt + 1 << " failed: " << e.what()
                      << " — retrying in " << backoff << "ms" << std::endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(backoff));
            backoff *= 2;
        }
    }
}

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

async fn send_with_retry(
    client: &KubemqClient,
    event: Event,
    max_retries: u32,
) -> kubemq::Result<()> {
    let mut backoff = Duration::from_millis(100);
    for attempt in 0..=max_retries {
        match client.send_event(event.clone()).await {
            Ok(()) => return Ok(()),
            Err(e) => {
                if attempt == max_retries {
                    return Err(e);
                }
                println!("Attempt {} failed: {e} — retrying in {backoff:?}", attempt + 1);
                tokio::time::sleep(backoff).await;
                backoff *= 2;
            }
        }
    }
    Ok(())
}

def send_with_retry(client, message, max_retries = 3)
  backoff = 0.1
  (0..max_retries).each do |attempt|
    begin
      client.send_event(message)
      return
    rescue KubeMQ::Error, StandardError => e
      raise "Failed after #{max_retries} retries: #{e.message}" if attempt == max_retries

      puts "Attempt #{attempt + 1} failed: #{e.message} — retrying in #{backoff}s"
      sleep backoff
      backoff *= 2
    end
  end
end

# send_event/2 returns :ok or {:error, err}. Retry transient failures
# with exponential backoff, raising once retries are exhausted.
defmodule Retry do
  def send_with_retry(client, event, max_retries \\ 3, backoff \\ 100, attempt \\ 0) do
    case KubeMQ.Client.send_event(client, event) do
      :ok ->
        :ok

      {:error, err} when attempt >= max_retries ->
        raise "Failed after #{max_retries} retries: #{err.message}"

      {:error, err} ->
        IO.puts("Attempt #{attempt + 1} failed: #{err.message} — retrying in #{backoff}ms")
        Process.sleep(backoff)
        send_with_retry(client, event, max_retries, backoff * 2, attempt + 1)
    end
  end
end

package main

import (
    "context"
    "log"
    "os"
    "os/signal"
    "syscall"

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

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    client, err := kubemq.NewClient(ctx,
        kubemq.WithAddress("localhost", 50000),
    )
    if err != nil {
        log.Fatal(err)
    }

    sig := make(chan os.Signal, 1)
    signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)

    go func() {
        <-sig
        log.Println("Shutting down...")
        cancel()
        client.Close()
        os.Exit(0)
    }()

    log.Println("Running — press Ctrl+C to stop")
    <-ctx.Done()
}

import signal
import sys
from kubemq.pubsub import Client as PubSubClient

client = PubSubClient(address="localhost:50000")

def shutdown(signum, frame):
    print("Shutting down...")
    client.close()
    sys.exit(0)

signal.signal(signal.SIGINT, shutdown)
signal.signal(signal.SIGTERM, shutdown)

print("Running — press Ctrl+C to stop")
signal.pause()

const { KubeMQClient } = require("kubemq-js");

const client = new KubeMQClient({ address: "localhost:50000" });

async function shutdown() {
  console.log("Shutting down...");
  await client.close();
  process.exit(0);
}

process.on("SIGINT", shutdown);
process.on("SIGTERM", shutdown);

console.log("Running — press Ctrl+C to stop");

PubSubClient client = PubSubClient.builder()
    .address("localhost:50000")
    .clientId("shutdown-demo")
    .build();

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    System.out.println("Shutting down...");
    client.close();
}));

System.out.println("Running — press Ctrl+C to stop");
Thread.currentThread().join();

await using var client = new KubeMQClient(new KubeMQClientOptions
{
    Address = "localhost:50000",
});
await client.ConnectAsync();

var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) =>
{
    e.Cancel = true;
    Console.WriteLine("Shutting down...");
    cts.Cancel();
};

Console.WriteLine("Running — press Ctrl+C to stop");
try { await Task.Delay(Timeout.Infinite, cts.Token); }
catch (OperationCanceledException) { }

val client = PubSubClient("localhost:50000")

Runtime.getRuntime().addShutdownHook(Thread {
    println("Shutting down...")
    client.close()
})

println("Running — press Ctrl+C to stop")
Thread.currentThread().join()

#include <kubemq/client.h>
#include <csignal>
#include <iostream>
#include <atomic>

std::atomic<bool> running{true};
kubemq::PubSubClient* globalClient = nullptr;

void signalHandler(int) {
    std::cout << "Shutting down..." << std::endl;
    running = false;
    if (globalClient) globalClient->close();
}

int main() {
    kubemq::PubSubClient client("localhost:50000");
    globalClient = &client;

    std::signal(SIGINT, signalHandler);
    std::signal(SIGTERM, signalHandler);

    std::cout << "Running — press Ctrl+C to stop" << std::endl;
    while (running) {
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    return 0;
}

use kubemq::prelude::*;

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

    println!("Running — press Ctrl+C to stop");

    // Wait for a shutdown signal, then close the client cleanly.
    tokio::signal::ctrl_c().await.ok();
    println!("Shutting down...");

    // close() flushes pending work and cancels active subscriptions.
    client.close().await?;
    println!("Client closed — shutdown complete");

    Ok(())
}

require 'kubemq'

client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: 'shutdown-demo')

shutdown = proc do
  puts 'Shutting down...'
  client.close
  exit 0
end

Signal.trap('INT', &shutdown)
Signal.trap('TERM', &shutdown)

puts 'Running — press Ctrl+C to stop'
sleep

{:ok, client} =
  KubeMQ.Client.start_link(address: "localhost:50000", client_id: "shutdown-demo")

# Trap exits so the process can close the client before terminating.
Process.flag(:trap_exit, true)
IO.puts("Running — press Ctrl+C to stop")

receive do
  {:EXIT, _from, _reason} ->
    IO.puts("Shutting down...")
    KubeMQ.Client.close(client)
    IO.puts("Client closed — shutdown complete")
end