Overview

A graceful shutdown stops a KubeMQ client without dropping in-flight messages or leaking server-side subscription state. Killing a process outright, or closing the connection mid-callback, can truncate a handler or leave the server thinking a consumer is still there. In a container platform that sends SIGTERM before force-killing a pod, handling that signal turns a rolling deploy into a clean handoff instead of a burst of errors.

The pattern has a fixed order: stop new work by cancelling every tracked Subscription, then close the client with a drain timeout so in-flight operations get a bounded window to finish before the gRPC connection is torn down. process.on('SIGINT' | 'SIGTERM', ...) wires OS signals into the handler, each subscription's sub.cancel() stops new deliveries, and client.close({ timeoutSeconds, callbackTimeoutSeconds }) bounds the gRPC drain and callback drain separately.

Gotchas: cancelling subscriptions after calling client.close(), instead of before, can race the connection teardown. Timeouts too short cut off the message you were protecting; too long and Kubernetes SIGKILLs the pod anyway. Also cancel subscriptions in the error path, not just the happy path.

Prerequisites

KubeMQ server running on localhost:50000
Node.js SDK installed (npm install kubemq-js)

Code

graceful-shutdown.ts

/**
 * Example: Graceful Shutdown
 *
 * Demonstrates how to cleanly shut down a KubeMQ client by cancelling
 * all active subscriptions, allowing in-flight operations to drain,
 * and closing the connection with configurable timeouts.
 *
 * Prerequisites:
 *   - KubeMQ server running on localhost:50000
 *
 * Run: npx tsx examples/error-handling/graceful-shutdown.ts
 */
import { KubeMQClient, createEventMessage, createQueueMessage } from 'kubemq-js';
import type { Subscription } from 'kubemq-js';

async function main(): Promise<void> {
  const client = await KubeMQClient.create({
    address: 'localhost:50000',
    clientId: 'js-error-handling-graceful-shutdown-client',
  });

  // Track all active subscriptions for cleanup.
  const subscriptions: Subscription[] = [];

  try {
    // Set up multiple subscriptions.
    const eventSub = client.subscribeToEvents({
      channel: 'js-error-handling.graceful-shutdown-events',
      onEvent: (event) => {
        console.log('[Events] Received:', new TextDecoder().decode(event.body));
      },
      onError: (err) => {
        console.error('[Events] Error:', err.message);
      },
    });
    subscriptions.push(eventSub);
    console.log('Started event subscription');

    const commandSub = client.subscribeToCommands({
      channel: 'js-error-handling.graceful-shutdown-cmds',
      onCommand: async (cmd) => {
        console.log('[Commands] Received:', cmd.id);
        await client.sendCommandResponse({
          id: cmd.id,
          replyChannel: cmd.replyChannel,
          executed: true,
        });
      },
      onError: (err) => {
        console.error('[Commands] Error:', err.message);
      },
    });
    subscriptions.push(commandSub);
    console.log('Started command subscription');

    // Simulate some work.
    await client.sendEvent(
      createEventMessage({
        channel: 'js-error-handling.graceful-shutdown-events',
        body: 'processing started',
      }),
    );

    await client.sendQueueMessage(
      createQueueMessage({
        channel: 'js-error-handling.graceful-shutdown-queue',
        body: 'queued task',
      }),
    );

    console.log('Published event and queued message');

    // Register shutdown handler.
    const shutdown = async (signal: string) => {
      console.log(`\nReceived ${signal} — starting graceful shutdown...`);

      // Step 1: Cancel all subscriptions so no new messages arrive.
      console.log('Step 1: Cancelling subscriptions...');
      for (const sub of subscriptions) {
        sub.cancel();
      }
      console.log(`  Cancelled ${subscriptions.length} subscription(s)`);

      // Step 2: Close the client with drain timeout.
      // This waits for in-flight operations to complete.
      console.log('Step 2: Closing client (draining in-flight operations)...');
      await client.close({
        timeoutSeconds: 5, // max 5s for gRPC operations to drain
        callbackTimeoutSeconds: 10, // max 10s for callbacks to finish
      });

      console.log('Step 3: Shutdown complete. State:', client.state);
      process.exit(0);
    };

    process.on('SIGINT', () => shutdown('SIGINT'));
    process.on('SIGTERM', () => shutdown('SIGTERM'));

    console.log('\nRunning... Press Ctrl+C for graceful shutdown');
    console.log('(Auto-shutting down in 5 seconds for demo purposes)\n');

    await new Promise((resolve) => setTimeout(resolve, 5000));

    // Demonstrate programmatic shutdown.
    await shutdown('auto');
  } catch (err) {
    console.error('Unexpected error:', (err as Error).message);

    // Emergency cleanup — cancel subs even on error.
    for (const sub of subscriptions) {
      sub.cancel();
    }
    await client.close();
  }
}

main().catch(console.error);

How It Works

A Subscription[] array is maintained so all active subscribeToEvents/subscribeToCommands handles can be cancelled atomically in the shutdown() function — stopping new message delivery before closing.
client.close({ timeoutSeconds: 5, callbackTimeoutSeconds: 10 }) waits up to 5 s for in-flight gRPC calls and 10 s for user callbacks to finish, then disconnects regardless.
process.on('SIGINT', ...) / process.on('SIGTERM', ...) wire the shutdown handler to OS signals so container orchestrators (Kubernetes) get a clean exit.
The emergency catch block also calls cancel() on all subscriptions in case of unexpected errors, preventing subscription leaks if close() is never reached normally.

Graceful Shutdown

Overview

Prerequisites

Code

How It Works

Related

On this page