Delay Policy

Overview

A delay policy defers when a queued message becomes visible to consumers — you send it now, but nothing can receive it until a countdown you set expires. That's the mechanism behind retry-after-backoff, rate-limited notifications, "remind me in an hour" workflows, and staggering a burst of work so it doesn't hit downstream consumers all at once, all without standing up a separate scheduler.

It works entirely at send time: SetDelaySeconds attaches a delay to the message itself before you hand it to upstream.Send. The broker starts the countdown the moment it accepts the message and simply excludes it from delivery until the timer elapses — after that it behaves like any other queued message, available to whichever consumer polls next.

Gotchas: the delay is a floor, not a guarantee — the message becomes eligible when the timer expires, but actual delivery still waits for a consumer to poll, so don't rely on it for precise scheduling. It's one-shot: there's no recurrence or cron-like behavior, so long or repeating delays need application logic on top. And it's independent of redelivery — a delayed message that's later nacked or times out after delivery follows normal visibility-timeout/retry rules, not the original send-time delay.

Prerequisites

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

Code

// Example: queues-stream/delay-policy
//
// Demonstrates sending queue messages with a delay policy via stream.
// Delayed messages are not available for consumption until the delay expires.
//
// Channel: go-queues-stream.delay-policy
// Client ID: go-queues-stream-delay-policy-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),
		kubemq.WithClientId("go-queues-stream-delay-policy-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-queues-stream.delay-policy"

	// Send a message with 10-second delay via upstream stream.
	upstream, err := client.QueueUpstream(ctx)
	if err != nil {
		log.Fatal(err)
	}
	defer upstream.Close()

	delayedMsg := kubemq.NewQueueMessage().
		SetChannel(channel).
		SetBody([]byte("delayed by 10s")).
		SetDelaySeconds(10)

	if err := upstream.Send("req-delay", []*kubemq.QueueMessage{delayedMsg}); err != nil {
		log.Fatal(err)
	}

	select {
	case res := <-upstream.Results:
		if res != nil && res.IsError {
			log.Printf("Error: %s", res.Error)
		} else {
			fmt.Println("Sent message with 10s delay policy")
		}
	case <-time.After(3 * time.Second):
		fmt.Println("Sent message (no result confirmation within timeout)")
	}
}

How It Works

1. kubemq.NewQueueMessage().SetDelaySeconds(10) embeds a 10-second delivery delay directly on the message; the broker holds it invisibly until the delay expires.
2. upstream.Send("req-delay", []*kubemq.QueueMessage{delayedMsg}) sends the single delayed message via the persistent upstream stream; the broker accepts it and starts the delay timer.
3. The select on upstream.Results confirms the batch was accepted; a time.After(3*time.Second) fallback handles the case where the broker does not send a result confirmation within the window.
4. Using QueueUpstream here mirrors a production pattern where many delayed messages are queued in a single stream session for better throughput than individual SendQueueMessage calls.

Related

• Pattern overview
• Go SDK Reference
• Send & Receive
• Ack All