Request-Reply

Overview

Request-reply gives you synchronous RPC on top of KubeMQ's messaging fabric: a caller sends a query and blocks until the handler actually processing the request sends back a real answer — not just an acknowledgment. Reach for it whenever the caller needs a return value to proceed — a lookup, a computed result, a status check — the same shape as an HTTP call, but routed by KubeMQ instead of a service mesh or DNS.

A responder subscribes with subscribeToQueries() and, inside its onQuery callback, calls client.sendQueryResponse() with the original query.id and query.replyChannel — copying those fields is what lets KubeMQ route the response to the one caller waiting, not broadcast it. The caller's client.sendQuery() blocks until that response arrives or timeoutInSeconds elapses, returning a QueryResponse with executed and body.

Gotchas: if no subscriber is listening — or the handler crashes before replying — sendQuery() simply times out; there's no way to distinguish "no handler" from "handler is slow" from the timeout alone. id and replyChannel on the response must echo back the incoming query's values unchanged, or the reply is silently dropped or misrouted. If you don't actually need a return value, use commands instead — they only need an ack, so they don't tie up a caller waiting on a round trip.

Prerequisites

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

Code

/**
 * Example: Request-Reply Pattern
 *
 * Demonstrates the request-reply messaging pattern using KubeMQ queries.
 * A responder subscribes to a channel, processes incoming requests, and
 * sends back a data response to the caller.
 *
 * Prerequisites:
 *   - KubeMQ server running on localhost:50000
 *
 * Run: npx tsx examples/patterns/request-reply.ts
 */
import { KubeMQClient, createQuery } from 'kubemq-js';

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

  try {
    const querySub = client.subscribeToQueries({
      channel: 'js-patterns.request-reply',
      onQuery: async (query) => {
        const request = JSON.parse(new TextDecoder().decode(query.body));
        console.log('Query handler received:', request);

        const result = { userId: request.userId, name: 'Alice', role: 'admin' };

        await client.sendQueryResponse({
          id: query.id,
          replyChannel: query.replyChannel,
          executed: true,
          body: new TextEncoder().encode(JSON.stringify(result)),
        });
      },
      onError: (err) => {
        console.error('Query subscription error:', err.message);
      },
    });

    await new Promise((r) => setTimeout(r, 500));

    const queryResponse = await client.sendQuery(
      createQuery({
        channel: 'js-patterns.request-reply',
        body: JSON.stringify({ userId: 'u-1001' }),
        timeoutInSeconds: 5,
      }),
    );

    if (queryResponse.executed && queryResponse.body) {
      const data = JSON.parse(new TextDecoder().decode(queryResponse.body));
      console.log('Query response:', data);
    }

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

main().catch(console.error);

How It Works

• The responder uses subscribeToQueries() with an async onQuery callback; inside the callback it calls client.sendQueryResponse() with the original query.id and query.replyChannel to route the response back to the correct caller.
• The 500ms wait after subscribing ensures the query handler is registered on the server before sendQuery() is called — without it the server would reject the query for having no handler.
• client.sendQuery() blocks until the responder's response arrives or timeoutInSeconds elapses, returning a QueryResponse with executed, body, and optional cacheHit fields.
• This pattern is synchronous from the caller's perspective but fully asynchronous over gRPC; the SDK manages the correlation between outgoing query IDs and incoming responses.

Related

• Pattern overview
• Node.js SDK Reference
• Fan-Out
• Work Queue