Cancel Subscription

Overview

Every events store subscription opens a long-lived stream to the broker — a linked process that keeps pulling delivered events until you tell it to stop. Calling KubeMQ.Subscription.cancel/1 is how you release that process deliberately: shutting down a worker, rotating consumers, or tearing down a supervision tree without leaking connections or leaving a dangling stream on the server.

Internally, cancel/1 sends a close signal that unwinds the subscription process and detaches from the broker-side subscription registration; KubeMQ.Subscription.active?/1 reports whether that process is still receiving events, so you can confirm shutdown instead of assuming it.

Gotchas: cancelling only stops this subscription — the channel keeps storing every event published afterward, so nothing is lost, and a fresh subscription with a replay start position picks up exactly where this one left off. cancel/1 returning :ok means the request was accepted, not that the process has necessarily finished tearing down; check active?/1 if you need to be certain no more events will land.

Prerequisites

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

Code

channel = "elixir-events-store.cancel-subscription"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-es-cancel")

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events_store(client, channel,
    start_at: :start_new_only,
    on_event: fn event -> IO.puts("Got: #{event.body}") end
  )

IO.puts("Subscription active? #{KubeMQ.Subscription.active?(sub)}")

{:ok, _} = KubeMQ.Client.send_event_store(client,
  KubeMQ.EventStore.new(channel: channel, body: "Before cancel"))
Process.sleep(500)

:ok = KubeMQ.Subscription.cancel(sub)
IO.puts("Cancelled. Active? #{KubeMQ.Subscription.active?(sub)}")

{:ok, _} = KubeMQ.Client.send_event_store(client,
  KubeMQ.EventStore.new(channel: channel, body: "After cancel"))
Process.sleep(500)
IO.puts("Post-cancel event sent but not received.")

KubeMQ.Client.close(client)

How It Works

• Subscription.active?/1 reports whether the subscription is still receiving events
• Subscription.cancel/1 stops the subscription and releases server resources
• Events published after cancellation are stored but not delivered to the cancelled subscriber

Related

• Persistent Pub/Sub
• Graceful Shutdown