Send & Receive

Overview

Queue send/receive is the foundational operation for guaranteed-delivery, point-to-point messaging: you reach for it whenever work needs to survive past the moment it's created and be handled by exactly one consumer, not broadcast to every subscriber. Unlike pub/sub, a queued message sits durably on the broker until something pulls it, so the producer and consumer never need to be online at the same time — a slow or offline worker adds latency, it doesn't drop the message.

This tutorial builds the smallest possible version of that round trip: sendQueueMessage() enqueues a message on a channel, and receiveQueueMessages() pulls it back within a bounded waitTimeoutSeconds. Settlement here is manual — each received message is locked (invisible to other consumers) until you call msg.ack(), which confirms processing and removes it from the queue for good.

Gotchas: if the client disconnects or the handler crashes before msg.ack() runs, the lock expires and the message reappears for redelivery once its visibility timeout elapses — write handlers that tolerate seeing the same message twice. Calling receiveQueueMessages() against an empty queue isn't an error; it just waits out waitTimeoutSeconds and returns an empty array. And receiveQueueMessages() defaults to pulling one message per call, so don't assume it drains the whole queue in one shot.

Prerequisites

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

Code

/**
 * Example: Basic Queue Send/Receive
 *
 * Demonstrates guaranteed delivery messaging with queues. A producer
 * sends messages and a consumer polls for them. Each message is delivered
 * to exactly one consumer and must be acknowledged.
 *
 * Prerequisites:
 *   - KubeMQ server running on localhost:50000
 *
 * Run: npx tsx examples/queues/send-receive.ts
 */
import { KubeMQClient, createQueueMessage } from 'kubemq-js';

async function main(): Promise<void> {
  // TODO: Replace with your KubeMQ server address
  const client = await KubeMQClient.create({
    address: 'localhost:50000',
    clientId: 'js-queues-send-receive-client',
  });

  try {
    // Send a message to the queue.
    const result = await client.sendQueueMessage(
      createQueueMessage({
        channel: 'js-queues.send-receive',
        body: 'Resize image: /uploads/photo-001.jpg',
        tags: { format: 'jpeg', width: '800' },
      }),
    );
    console.log('Sent message:', result.messageId);

    // Receive messages from the queue.
    const messages = await client.receiveQueueMessages({
      channel: 'js-queues.send-receive',
      waitTimeoutSeconds: 5,
    });

    for (const msg of messages) {
      console.log('Received:', new TextDecoder().decode(msg.body));
      console.log('  Tags:', msg.tags);
      await msg.ack();
      console.log('  Acknowledged');
    }
  } finally {
    await client.close();
  }
}

main().catch(console.error);

// Expected output:
// Sent message: <message-id>
// Received: Resize image: /uploads/photo-001.jpg
//   Tags: { format: 'jpeg', width: '800' }
//   Acknowledged

How It Works

• sendQueueMessage() persists the message on the server and returns a QueueSendResult with the server-assigned messageId.
• receiveQueueMessages() does a blocking poll: it waits up to waitTimeoutSeconds (5 s) for messages and returns up to maxMessages (default 1) per call.
• Each received message is locked (invisible to other consumers) until msg.ack() is called; if the client disconnects without acking, the lock expires and the message reappears.
• await msg.ack() sends the acknowledgment back to the server over the same gRPC connection; unacknowledged messages reappear after the visibility timeout.

Related

• Pattern overview
• Node.js SDK Reference
• Ack All
• Ack & Reject