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 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 — but a transient error during an in-flight call still needs to be retried by the caller, since reconnection restores the transport, not the operation that was in progress.

It works by passing a reconnect_policy (initial_delay, max_attempts) to KubeMQ.Client.start_link/1, which handles reconnection at the transport level automatically. KubeMQ.Client.connection_state/1 returns the current state as an atom for monitoring. For operations that must succeed despite a mid-call disconnect, pair transport-level reconnection with application-level retry, as the RetryHelper module does here — pattern-matching on error codes like :transient and :timeout and retrying with its own exponential backoff. Gotchas: the transport's reconnect_policy and the application's retry helper are two independent layers — reconnecting the socket doesn't replay a call that failed mid-flight, so you still need retry logic around individual operations; max_attempts on the reconnect policy is a hard cap, so a long-lived outage can exhaust it while the application-level retry keeps looping separately; and retrying on every error code (not just :transient/:timeout) risks retrying non-idempotent operations that already partially succeeded.

Prerequisites

• KubeMQ server running on localhost:50000
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

{:ok, client} =
  KubeMQ.Client.start_link(
    address: "localhost:50000",
    client_id: "elixir-reconnect-example",
    reconnect_policy: [initial_delay: 2_000, max_attempts: 5]
  )

IO.puts("Connected. Monitoring connection state...")

state = KubeMQ.Client.connection_state(client)
IO.puts("Current state: #{state}")

defmodule RetryHelper do
  def with_retry(fun, retries \\ 3, delay \\ 1_000) do
    case fun.() do
      {:error, %{code: code}} when code in [:transient, :timeout] and retries > 0 ->
        IO.puts("  Transient error, retrying in #{delay}ms... (#{retries} left)")
        Process.sleep(delay)
        with_retry(fun, retries - 1, delay * 2)

      result ->
        result
    end
  end
end

result =
  RetryHelper.with_retry(fn ->
    KubeMQ.Client.send_event(client,
      KubeMQ.Event.new(channel: "elixir-error-handling.reconnection", body: "resilient message"))
  end)

case result do
  :ok -> IO.puts("Event sent successfully (possibly after reconnect)")
  {:error, err} -> IO.puts("Failed after retries: #{err.message}")
end

IO.puts("Final connection state: #{KubeMQ.Client.connection_state(client)}")
KubeMQ.Client.close(client)

How It Works

• reconnect_policy configures automatic reconnection at the transport level
• connection_state/1 returns the current connection state atom
• The RetryHelper module implements application-level retry with exponential backoff
• Pattern matching on error codes :transient and :timeout determines retry eligibility

Related

• Custom Timeouts
• Connection Error