Handle Query
Register a handler for incoming KubeMQ queries in Node.js, processing each request and returning a data response to the caller.
Overview
A query handler is the answering side of KubeMQ's request/response RPC pattern — the code that does real work and sends back data, unlike a Command handler, which only acknowledges receipt. Reach for it whenever a caller needs an actual answer — a lookup result, a computed value, a status object — not just confirmation that a message arrived.
Registering a handler with subscribeToQueries and an onQuery callback opens a subscription; the broker delivers every matching query to your callback as it arrives. The callback builds a response carrying the query's replyChannel back to the broker, so the answer routes to the specific caller blocked waiting, and sets body with the real result (or executed: false with an error) before calling sendQueryResponse.
Gotchas: if the handler never sends a response, the caller blocks until its own timeout elapses and fails with a timeout, not a fast error. An exception thrown inside onQuery doesn't automatically become a failure reply, so uncaught errors can leave the sender hanging. And because every matching query lands on the same callback, slow or blocking handler code delays every other in-flight caller.
Prerequisites
- KubeMQ server running on
localhost:50000 - Node.js SDK installed (
npm install kubemq-js)
Code
/**
* Example: Subscribe to and Handle Queries
*
* Demonstrates subscribing to a query channel and responding with data.
* The handler processes each query, looks up the requested data, and
* sends a response containing the result.
*
* Prerequisites:
* - KubeMQ server running on localhost:50000
*
* Run: npx tsx examples/rpc/handle-query.ts
*/
import { KubeMQClient } from 'kubemq-js';
const inventory: Record<string, { inStock: number; price: number }> = {
'WIDGET-42': { inStock: 150, price: 12.99 },
'GADGET-7': { inStock: 0, price: 24.5 },
'GIZMO-99': { inStock: 42, price: 8.75 },
};
async function main(): Promise<void> {
const client = await KubeMQClient.create({
address: 'localhost:50000',
clientId: 'js-rpc-handle-query-client',
});
try {
const subscription = client.subscribeToQueries({
channel: 'js-rpc.handle-query',
onQuery: async (query) => {
console.log('Received query:', query.id);
const request = JSON.parse(new TextDecoder().decode(query.body));
const item = inventory[request.sku];
if (item) {
await client.sendQueryResponse({
id: query.id,
replyChannel: query.replyChannel,
executed: true,
body: new TextEncoder().encode(
JSON.stringify({
sku: request.sku,
...item,
warehouse: request.warehouse,
}),
),
});
console.log(' Responded with inventory data for', request.sku);
} else {
await client.sendQueryResponse({
id: query.id,
replyChannel: query.replyChannel,
executed: false,
error: `SKU not found: ${request.sku}`,
});
console.log(' SKU not found:', request.sku);
}
},
onError: (err) => {
console.error('Subscription error:', err.message);
},
});
console.log('Listening for queries on "js-rpc.handle-query"...');
console.log('Press Ctrl+C to stop');
await new Promise((resolve) => process.on('SIGINT', resolve));
subscription.cancel();
} finally {
await client.close();
}
}
main().catch(console.error);
How It Works
subscribeToQueries()registers anonQueryasync callback; unlike commands, query responses must include abodypayload (the data returned to the caller).query.replyChannelis a server-generated per-request correlation channel — pass it back insendQueryResponse()so KubeMQ routes the response to the correct caller.- The handler returns two response shapes:
executed: truewithbody(found SKU) andexecuted: falsewitherror(SKU not found) — the caller checksresponse.executedto distinguish them. - The
inventorymap is a simple in-memory lookup; in production replace it with a database query, cache lookup, or microservice call.
Related
Was this page helpful?