Publishing the scheduled message to NATS

[Santosh1608]

Observed behavior

I noticed that NATS v2.12 introduces support for scheduled (delayed) message delivery, where a message can be published now and delivered to consumers after a specified delay.

However, I’m unable to find any official documentation or sample code showing how to publish a scheduled message using the nats.go client.

Specifically:

Does the nats.go client currently support publishing scheduled/delayed messages?

If yes, could you point me to the relevant documentation or example code demonstrating how to publish a message with a delivery delay?

Any clarification or sample usage would be really helpful.

For the reference: https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-51.md

Expected behavior

I need a support for publishing the scheduled message

Server and client version

nats-server Version: 2.12.2
nats Version: v1.47.0

Host environment

No response

Steps to reproduce

No response

[markpeterfejes]

I was experimenting with @thedevtoni today to see if nats is able to perhaps throttle messages via this way, perhaps this gives you an idea in the meantime:

package main

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

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

func main() {
	ctx := context.Background()
	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Close()
	js, _ := jetstream.New(nc)

	// Create stream with delayed messages enabled
	js.DeleteStream(ctx, "TEST")
	nc.Request("$JS.API.STREAM.CREATE.TEST", []byte(`{
		"name": "TEST",
		"subjects": ["otel.>"],
		"storage": "memory",
		"allow_msg_schedules": true
	}`), 5*time.Second)

	// Subscribe to target subject
	stream, _ := js.Stream(ctx, "TEST")
	consumer, _ := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
		FilterSubject: "otel.target.>",
	})

	go func() {
		for {
			msg, err := consumer.Next()
			if err != nil {
				continue
			}
			log.Printf("📩 Received: %s", msg.Data())
			msg.Ack()
		}
	}()

	// Publish 100 delayed messages to same subject (simulating span burst for one trace)
	traceID := "abc123"
	schedule := "@at " + time.Now().Add(5*time.Second).Format(time.RFC3339)

	log.Printf("Publishing 100 messages with 5s delay to trace %s...", traceID)
	for i := 1; i <= 100; i++ {
		hdr := nats.Header{}
		hdr.Set("Nats-Schedule", schedule)
		hdr.Set("Nats-Schedule-Target", "otel.target."+traceID)

		js.PublishMsg(ctx, &nats.Msg{
			Subject: "otel.schedule." + traceID,
			Data:    []byte(fmt.Sprintf("span-%d", i)),
			Header:  hdr,
		})
	}

	log.Println("Waiting for delivery...")
	time.Sleep(8 * time.Second)

	// Expected: only 1 message received (the last one: span-100)
	js.DeleteStream(ctx, "TEST")
}

[piotrpio]

@markpeterfejes Thank you for creating the issue and thanks for the feedback. We'll add basic support (such as header constants) and docs to the client, apologies for the delay. In the upcoming v2.14 on nats-server scheduling will also support more advanced patterns (such as @every keyword and cron scheduling), at that point we'll also possibly add more tooling around that (although that will probably land in orbit.go, not in nats.go)