Handle Command

Overview

A command handler is the receiving side of KubeMQ's Commands pattern — the code that actually does the work a caller is blocked waiting on. Instead of building your own request-routing layer on top of a queue, you register a handler once via client.subscribeToCommands({ onCommand }), and KubeMQ delivers every matching command on that channel to it as a long-lived, server-streamed subscription, turning the channel into a synchronous RPC endpoint.

Handling happens inside the onCommand async callback: you read the command's id, channel, and body, run your business logic, then send a reply with client.sendCommandResponse({ id: cmd.id, replyChannel: cmd.replyChannel, executed: success }). Passing back cmd.id and cmd.replyChannel — a server-generated correlation channel — is what lets the broker route the response to the exact caller blocked on sendCommand; nothing else identifies which request the response belongs to.

Gotchas: the reply must be sent before the sender's timeout expires or the caller sees a timeout even if you eventually respond; set executed: false with an error string to signal a business-logic rejection rather than a successful run; and onCommand runs per incoming command, so slow or blocking work inside it head-of-line blocks subsequent commands on the same subscription.

Prerequisites

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

Code

/**
 * Example: Subscribe to and Handle Commands
 *
 * Demonstrates subscribing to a command channel and responding to incoming
 * commands. The handler processes each command and sends a response back
 * to the sender.
 *
 * Prerequisites:
 *   - KubeMQ server running on localhost:50000
 *
 * Run: npx tsx examples/rpc/handle-command.ts
 */
import { KubeMQClient } from 'kubemq-js';

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

  try {
    const subscription = client.subscribeToCommands({
      channel: 'js-rpc.handle-command',
      onCommand: async (cmd) => {
        console.log('Received command:', cmd.id);
        console.log('  Channel:', cmd.channel);
        console.log('  Body:', new TextDecoder().decode(cmd.body));

        const payload = JSON.parse(new TextDecoder().decode(cmd.body));

        // Process the command and send a response.
        const success =
          payload.action === 'set-temperature' && payload.value >= 16 && payload.value <= 30;

        await client.sendCommandResponse({
          id: cmd.id,
          replyChannel: cmd.replyChannel,
          executed: success,
          error: success ? undefined : 'Temperature out of range (16-30°C)',
        });

        console.log('  Response sent:', success ? 'executed' : 'rejected');
      },
      onError: (err) => {
        console.error('Subscription error:', err.message);
      },
    });

    console.log('Listening for commands on "js-rpc.handle-command"...');
    console.log('Press Ctrl+C to stop');

    // Keep the process running until interrupted.
    await new Promise((resolve) => process.on('SIGINT', resolve));

    subscription.cancel();
  } finally {
    await client.close();
  }
}

main().catch(console.error);

How It Works

• subscribeToCommands() registers an onCommand async callback; each incoming command is delivered to it and the handler must call sendCommandResponse() before the sender's timeout expires.
• cmd.replyChannel is a server-generated correlation channel — it must be passed back in sendCommandResponse() so KubeMQ can route the response to the correct caller.
• await client.sendCommandResponse() sends the response over the same gRPC connection; executed: false with an error string signals a business-logic rejection to the sender.
• The process stays alive with process.on('SIGINT', resolve) so the handler continues serving until Ctrl+C; subscription.cancel() tears down the gRPC stream cleanly on exit.

Related

• Pattern overview
• Node.js SDK Reference
• Send Command
• Command Timeout