Cancel Subscription

Overview

Every events store subscription opens a long-lived stream to the broker — a background thread that keeps pulling delivered events until you tell it to stop. Calling cancel on a CancellationToken is how you release that thread deliberately: shutting down a worker, rotating consumers, or tearing down a script without leaking connections or leaving a dangling stream on the server.

Internally, cancel.cancel signals the subscription thread to unwind its receive loop; thread.join then blocks until that thread has actually exited, giving you a clean, deterministic shutdown point instead of guessing with a sleep.

Gotchas: cancelling only stops this subscriber — 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. Skipping thread.join after calling cancel means the process (or the next step in your script) can race the shutdown, since cancellation is asynchronous relative to the calling thread.

Prerequisites

• KubeMQ server running on localhost:50000
• Ruby SDK installed (gem install kubemq)

Code

require 'kubemq'

address = ENV.fetch('KUBEMQ_ADDRESS', 'localhost:50000')
channel = 'ruby-events-store.cancel-subscription'

begin
  client = KubeMQ::PubSubClient.new(address: address, client_id: 'es-cancel-example')
  puts "Connected to #{address}"

  cancel = KubeMQ::CancellationToken.new

  sub = KubeMQ::PubSub::EventsStoreSubscription.new(
    channel: channel,
    start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_FIRST
  )
  thread = client.subscribe_to_events_store(sub, cancellation_token: cancel, on_error: lambda { |e|
    puts "Error: #{e.message}"
  }) do |event|
    puts "Received: seq=#{event.sequence}, body=#{event.body}"
  end
  sleep 1

  msg = KubeMQ::PubSub::EventStoreMessage.new(channel: channel, metadata: 'test', body: 'before-cancel')
  client.send_event_store(msg)
  sleep 1

  puts 'Cancelling subscription...'
  cancel.cancel
  thread.join(3)
  puts 'Subscription cancelled'

  msg = KubeMQ::PubSub::EventStoreMessage.new(channel: channel, metadata: 'test', body: 'after-cancel')
  client.send_event_store(msg)
  sleep 1
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  client&.close
  puts 'Done'
end

How It Works

• CancellationToken.cancel gracefully stops the subscription thread.
• Events sent after cancellation are still persisted but not delivered to the cancelled subscriber.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Persistent Pub/Sub