Add PublishAsync to Publisher for non-blocking message publishing

Introduce PublishAsync(ctx, message, callback) on Publisher as a
fire-and-forget alternative to Publish. The send itself is synchronous
(SendWithReceipt), but the broker-confirmation wait (SendReceipt.Wait)
runs in a background goroutine and delivers its outcome via a
PublishAsyncCallback.

Back-pressure is enforced through a channel-based semaphore (inFlight)
whose capacity is controlled by PublisherOptions.MaxInFlight (default
256): PublishAsync blocks the caller until a slot is available or the
context is cancelled. Each confirmation goroutine respects a configurable
PublishTimeout (default 30 s) before surfacing a timeout error through
the callback.

Also add integration tests covering the happy path, bulk sending,
MaxInFlight throttling, context-cancellation back-pressure, validation
errors, StateReleased outcomes, and custom timeouts, together with a
runnable example in docs/examples/publish_async/.

Signed-off-by: Gabriele Santomaggio <G.santomaggio@gmail.com>

Author: Gsantomaggio

docs/examples/publish_async/main.go

@@ -0,0 +1,169 @@
+// RabbitMQ AMQP 1.0 Go Client: https://github.com/rabbitmq/rabbitmq-amqp-go-client
+// RabbitMQ AMQP 1.0 documentation: https://www.rabbitmq.com/docs/amqp
+// The example demonstrates how to use PublishAsync to send messages without blocking
+// the caller while waiting for broker confirmation.
+//
+// Key concepts shown:
+//   - PublishAsync fires the send immediately and delivers the outcome via a callback.
+//   - MaxInFlight limits how many confirmation goroutines can run concurrently,
+//     providing back-pressure: PublishAsync blocks the caller when the limit is reached.
+//   - PublishTimeout controls how long each confirmation goroutine waits for the broker
+//     before the callback receives a timeout error.
+//   - A sync.WaitGroup is used to wait for all callbacks before shutting down.
+//
+// example path: https://github.com/rabbitmq/rabbitmq-amqp-go-client/tree/main/docs/examples/publish_async/main.go
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
+)
+
+func main() {
+	const (
+		queueName      = "publish-async-go-queue"
+		totalMessages  = 2_000_000
+		maxInFlight    = 100
+		publishTimeout = 10 * time.Second
+	)
+
+	rmq.Info("PublishAsync example starting")
+
+	// Counters updated from the callback goroutines.
+	var accepted, released, rejected, failed atomic.Int32
+
+	// Track how long the whole publish loop takes.
+	startTime := time.Now()
+
+	// stateChanged receives connection lifecycle events.
+	stateChanged := make(chan *rmq.StateChanged, 1)
+	go func(ch chan *rmq.StateChanged) {
+		for sc := range ch {
+			rmq.Info("[connection] status changed", "from", sc.From, "to", sc.To)
+		}
+	}(stateChanged)
+
+	env := rmq.NewEnvironment("amqp://guest:guest@localhost:5672/", nil)
+	amqpConnection, err := env.NewConnection(context.Background())
+	if err != nil {
+		rmq.Error("Error opening connection", "error", err)
+		return
+	}
+	amqpConnection.NotifyStatusChange(stateChanged)
+	rmq.Info("AMQP connection opened")
+
+	management := amqpConnection.Management()
+	queueInfo, err := management.DeclareQueue(context.Background(), &rmq.QuorumQueueSpecification{
+		Name: queueName,
+	})
+	if err != nil {
+		rmq.Error("Error declaring queue", "error", err)
+		return
+	}
+
+	// Create the publisher with async-specific options:
+	//   MaxInFlight  – at most 100 confirmation goroutines run concurrently.
+	//                  When the limit is reached, PublishAsync blocks the caller
+	//                  until a slot becomes free.
+	//   PublishTimeout – each confirmation goroutine waits at most 30 s for a
+	//                    broker acknowledgement before the callback receives an error.
+	publisher, err := amqpConnection.NewPublisher(context.Background(),
+		&rmq.QueueAddress{Queue: queueName},
+		&rmq.PublisherOptions{
+			MaxInFlight:    maxInFlight,
+			PublishTimeout: publishTimeout,
+		})
+	if err != nil {
+		rmq.Error("Error creating publisher", "error", err)
+		return
+	}
+
+	// wg is decremented once per callback invocation.
+	var wg sync.WaitGroup
+	wg.Add(totalMessages)
+
+	for i := 0; i < totalMessages; i++ {
+		msgBody := fmt.Sprintf("hello async %d", i)
+		msg := rmq.NewMessage([]byte(msgBody))
+
+		err = publisher.PublishAsync(context.Background(), msg,
+			func(result *rmq.PublishResult, cbErr error) {
+				defer wg.Done()
+
+				if cbErr != nil {
+					// Timeout or send-level error.
+					rmq.Error("[Publisher] async callback error", "error", cbErr)
+					failed.Add(1)
+					return
+				}
+
+				switch result.Outcome.(type) {
+				case *rmq.StateAccepted:
+					accepted.Add(1)
+				case *rmq.StateReleased:
+					rmq.Warn("[Publisher] message not routed", "body", result.Message.Data[0])
+					released.Add(1)
+				case *rmq.StateRejected:
+					s := result.Outcome.(*rmq.StateRejected)
+					rmq.Warn("[Publisher] message rejected", "error", s.Error)
+					rejected.Add(1)
+				}
+			})
+
+		if err != nil {
+			// PublishAsync only returns an error for validation failures or when the
+			// caller's context is cancelled while waiting for a free in-flight slot.
+			rmq.Error("[Publisher] PublishAsync error", "error", err)
+			wg.Done()
+		}
+	}
+
+	// Block until every callback has been invoked.
+	rmq.Info("Waiting for all confirmations…")
+	wg.Wait()
+
+	elapsed := time.Since(startTime)
+	rmq.Info("All messages confirmed",
+		"accepted", accepted.Load(),
+		"released", released.Load(),
+		"rejected", rejected.Load(),
+		"failed", failed.Load(),
+		"elapsed", elapsed.Round(time.Millisecond),
+		"msg/s", fmt.Sprintf("%.0f", float64(totalMessages)/elapsed.Seconds()),
+	)
+
+	if err = publisher.Close(context.Background()); err != nil {
+		rmq.Error("Error closing publisher", "error", err)
+	}
+
+	// press any key to close
+	println("press any key to close and clean up")
+
+	var input string
+	_, _ = fmt.Scanln(&input)
+
+	purged, err := management.PurgeQueue(context.Background(), queueInfo.Name())
+	if err != nil {
+		rmq.Error("Error purging queue", "error", err)
+	} else {
+		rmq.Info("Queue purged", "messages", purged)
+	}
+
+	if err = management.DeleteQueue(context.Background(), queueInfo.Name()); err != nil {
+		rmq.Error("Error deleting queue", "error", err)
+	}
+
+	if err = env.CloseConnections(context.Background()); err != nil {
+		rmq.Error("Error closing connection", "error", err)
+	}
+
+	rmq.Info("AMQP connection closed")
+	time.Sleep(100 * time.Millisecond)
+	close(stateChanged)
+}

pkg/rabbitmqamqp/amqp_publisher.go

@@ -6,6 +6,7 @@ import (
 	"net/url"
 	"strings"
 	"sync/atomic"
+	"time"
 
 	"github.com/Azure/go-amqp"
 	"github.com/google/uuid"
@@ -23,6 +24,8 @@ type Publisher struct {
 	linkName       string
 	destinationAdd string
 	id             string
+	inFlight       chan struct{}
+	publishTimeout time.Duration
 }
 
 func (m *Publisher) Id() string {
@@ -35,14 +38,27 @@ func newPublisher(ctx context.Context, connection *AmqpConnection, destinationAd
 		id = options.id()
 	}
 
-	r := &Publisher{connection: connection, linkName: getLinkName(options), destinationAdd: destinationAdd, id: id}
+	maxInFlight := DefaultMaxInFlight
+	publishTimeout := DefaultPublishTimeout
+	if options != nil {
+		maxInFlight = options.maxInFlight()
+		publishTimeout = options.publishTimeout()
+	}
+
+	r := &Publisher{
+		connection:     connection,
+		linkName:       getLinkName(options),
+		destinationAdd: destinationAdd,
+		id:             id,
+		inFlight:       make(chan struct{}, maxInFlight),
+		publishTimeout: publishTimeout,
+	}
 	connection.entitiesTracker.storeOrReplaceProducer(r)
 	err := r.createSender(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	// Record the publisher opening metric
 	connection.metricsCollector.OpenPublisher()
 
 	return r, nil
@@ -147,6 +163,76 @@ func (m *Publisher) Publish(ctx context.Context, message *amqp.Message) (*Publis
 	}, err
 }
 
+/*
+PublishAsync sends a message asynchronously and delivers the result via a callback.
+
+The message is sent to the broker immediately (same as Publish), but the confirmation
+wait (SendReceipt.Wait) happens in a background goroutine. When the broker confirms
+the message or the timeout expires, the callback is invoked with the result or an error.
+
+Back-pressure: the number of concurrent in-flight confirmations is bounded by
+MaxInFlight (configurable via PublisherOptions, default 256). When the limit is
+reached, PublishAsync blocks on the caller's goroutine until a slot becomes
+available or the context is cancelled.
+
+The timeout for each confirmation wait is controlled by PublisherOptions.PublishTimeout
+(default 10s).
+*/
+func (m *Publisher) PublishAsync(ctx context.Context, message *amqp.Message, callback PublishAsyncCallback) error {
+	if m.destinationAdd == "" {
+		if message.Properties == nil || message.Properties.To == nil {
+			return fmt.Errorf("message properties TO is required to send a message to a dynamic target address")
+		}
+		if err := validateAddress(*message.Properties.To); err != nil {
+			return err
+		}
+	}
+
+	if message.Header == nil {
+		message.Header = &amqp.MessageHeader{
+			Durable: true,
+		}
+	}
+
+	// Acquire an in-flight slot; blocks if MaxInFlight goroutines are already waiting.
+	select {
+	case m.inFlight <- struct{}{}:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+
+	receipt, err := m.sender.Load().SendWithReceipt(ctx, message, nil)
+	if err != nil {
+		<-m.inFlight
+		return err
+	}
+
+	publishCtx := m.buildPublishContext(message)
+	m.connection.metricsCollector.Publish(publishCtx)
+
+	go func() {
+		defer func() { <-m.inFlight }()
+
+		waitCtx, cancel := context.WithTimeout(context.Background(), m.publishTimeout)
+		defer cancel()
+
+		state, waitErr := receipt.Wait(waitCtx)
+		if waitErr != nil {
+			callback(nil, fmt.Errorf("publish confirmation failed: %w", waitErr))
+			return
+		}
+
+		m.recordPublishDisposition(state, publishCtx)
+
+		callback(&PublishResult{
+			Message: message,
+			Outcome: state,
+		}, nil)
+	}()
+
+	return nil
+}
+
 // recordPublishDisposition records the publish disposition metric based on the delivery state.
 func (m *Publisher) recordPublishDisposition(state DeliveryState, ctx PublishContext) {
 	switch state.(type) {