Cancel Subscription

Overview

A live Events subscription runs on a background thread indefinitely, so a long-running service needs an explicit way to tear one down without closing the whole client connection — for example when a feature flag disables a channel, a worker is draining before shutdown, or a subscription needs to be re-created with different options. Signaling a CancellationToken stops delivery cleanly and lets the subscription thread exit on its own terms.

client.subscribe_to_events takes a cancellation_token: and starts the subscription on its own thread, returning immediately. Calling cancel on that token tells the thread to stop gracefully; calling join on the subscription then blocks until that thread actually exits (or a timeout elapses), which is how you know teardown is complete rather than merely requested.

Gotchas: cancelling the token only affects this one subscription — other subscribers on the same channel keep receiving events. Events already in flight when you cancel may still be delivered briefly afterward. And because Events are fire-and-forget, anything published after cancellation is simply dropped for this subscriber — there's no queue to catch up from later.

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.cancel-subscription'

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

  cancel = KubeMQ::CancellationToken.new

  sub = KubeMQ::PubSub::EventsSubscription.new(channel: channel)
  subscription = client.subscribe_to_events(sub, cancellation_token: cancel, on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
    puts "Received: #{event.metadata}"
  end
  sleep 1

  msg = KubeMQ::PubSub::EventMessage.new(channel: channel, metadata: 'before-cancel', body: 'data')
  client.send_event(msg)
  sleep 1

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

  msg = KubeMQ::PubSub::EventMessage.new(channel: channel, metadata: 'after-cancel', body: 'data')
  client.send_event(msg)
  sleep 1
  puts 'Event sent after cancel should not be received'
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  client&.close
  puts 'Done'
end

How It Works

• CancellationToken.cancel signals the subscription thread to stop gracefully.
• The subscription's join method blocks until the thread exits or the timeout elapses.
• Events sent after cancellation are not delivered to the cancelled subscriber.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Basic Pub/Sub
• Graceful Shutdown