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: client.SendQueueMessage enqueues a message on a channel, and client.PollQueue pulls it back with a bounded WaitTimeoutSeconds. The AutoAck option controls settlement — set true and the broker removes the message the instant it's delivered; set false and you must acknowledge (or reject) it yourself once processing succeeds.

Gotchas: auto-ack means "delivered," not "processed" — if your handler crashes after receiving but before finishing the work, the message is already gone with no chance to retry. Polling an empty queue isn't an error; it just returns zero messages once the wait timeout elapses. And any message that's received but never acknowledged reappears on the queue after its visibility timeout, so a crashed or slow consumer causes redelivery — write handlers that tolerate seeing the same message twice.

Prerequisites

• KubeMQ server running on localhost:50000
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: queues/send-receive
//
// Demonstrates basic queue send and receive operations.
// A message is sent to a queue and then consumed (pulled) from it.
//
// Channel: go-queues.send-receive
// Client ID: go-queues-send-receive-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/kubemq-io/kubemq-go/v2"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := kubemq.NewClient(ctx,
		kubemq.WithAddress("localhost", 50000), // TODO: Replace with your KubeMQ server address
		kubemq.WithClientId("go-queues-send-receive-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-queues.send-receive"

	// Send a single queue message.
	msg := kubemq.NewQueueMessage().
		SetChannel(channel).
		SetBody([]byte("hello queue")).
		SetMetadata("greeting")

	result, err := client.SendQueueMessage(ctx, msg)
	if err != nil {
		log.Fatal(err)
	}
	if result.IsError {
		log.Fatalf("Send failed: %s", result.Error)
	}
	fmt.Printf("Sent: id=%s\n", result.MessageID)

	// Receive (consume) messages from the queue via PollQueue.
	resp, err := client.PollQueue(ctx, &kubemq.PollRequest{
		Channel:            channel,
		MaxItems:           10,
		WaitTimeoutSeconds: 5,
		AutoAck:            true,
	})
	if err != nil {
		log.Fatal(err)
	}
	if resp.IsError {
		log.Fatalf("Receive failed: %s", resp.Error)
	}
	fmt.Printf("Received: %d messages\n", len(resp.Messages))
	for _, dsMsg := range resp.Messages {
		fmt.Printf("  body=%s metadata=%s\n", dsMsg.Message.Body, dsMsg.Message.Metadata)
	}
}

// Expected output:
// Sent: id=<message-id>
// Received: 1 messages
//   body=hello queue metadata=greeting

How It Works

1. client.SendQueueMessage(ctx, kubemq.NewQueueMessage()...) durably enqueues a message on go-queues.send-receive; result.IsError and result.MessageID indicate success and the assigned broker ID.
2. client.PollQueue(ctx, &kubemq.PollRequest{...}) is a convenience method that opens a downstream receiver internally, polls up to MaxItems messages within WaitTimeoutSeconds, and returns them.
3. With AutoAck: true the broker automatically marks messages as acknowledged when they are delivered, so they are removed from the queue without a separate ack call.
4. Each element of resp.Messages is a *QueueDownstreamMessage wrapping a *kubemq.QueueMessage; the original payload is accessed via dsMsg.Message.Body.

Related

• Pattern overview
• Go SDK Reference
• Ack All
• Ack & Reject