Feature: Publisher.NotifyPublishWithReturn() for ordered return+confirmation handling

[joec4i]

Problem

Currently, NotifyPublish() and NotifyReturn() handlers run in separate goroutines, which can cause them to be processed out of order. This makes it challenging to properly link a returned message with their corresponding confirmation when handling unrouted messages.

For mandatory publishes, the publisher might need to make a decision based on both the confirmation and message return. In such cases, the pairing is crucial.

Solution

According to the RabbitMQ documentation:

For unroutable messages, the broker will issue a confirm once the exchange verifies a message won't route to any queue (returns an empty list of queues). If the message is also published as mandatory, the basic.return is sent to the client before basic.ack. The same is true for negative acknowledgements (basic.nack).

That was implemented in github.com/rabbitmq/amqp091-go channel.go

Unroutable mandatory or immediate messages are acknowledged immediately after
any Channel.NotifyReturn listeners have been notified. Other messages are
acknowledged when all queues that should have the message routed to them have
either received acknowledgment of delivery or have enqueued the message,
persisting the message if necessary.

This PR introduces Publisher.NotifyPublishWithReturn(), a new method that ensures proper pairing of returns and confirmations according to RabbitMQ protocol ordering.

Handler behavior:

• Successfully routed messages: Handler receives confirmation with empty Return{}
• Unrouted mandatory messages: Handler receives both return information and confirmation in correct order
• Unrouted non-mandatory messages: Handler receives confirmation only with empty Return{}

Testing

I've added integration tests to cover the above behaviours

examples/publishers_confirms is also updated with a NotifyPublishWithReturn() call. Below is the test output:

awaiting signal
message published
message publishing confirmed
2025/07/30 10:44:46 message confirmed and returned from server: hello, world, deliveryTag=1, ACK=true
2025/07/30 10:44:46 message returned from server: hello, world
message published
message publishing confirmed
2025/07/30 10:44:47 message returned from server: hello, world
2025/07/30 10:44:47 message confirmed and returned from server: hello, world, deliveryTag=2, ACK=true
message published
message publishing confirmed
2025/07/30 10:44:48 message confirmed and returned from server: hello, world, deliveryTag=3, ACK=true
2025/07/30 10:44:48 message returned from server: hello, world
message published
message publishing confirmed
2025/07/30 10:44:49 message returned from server: hello, world
2025/07/30 10:44:49 message confirmed and returned from server: hello, world, deliveryTag=4, ACK=true
message published
message publishing confirmed
2025/07/30 10:44:50 message returned from server: hello, world
2025/07/30 10:44:50 message confirmed and returned from server: hello, world, deliveryTag=5, ACK=true
message published
message publishing confirmed
2025/07/30 10:44:51 message returned from server: hello, world
2025/07/30 10:44:51 message confirmed and returned from server: hello, world, deliveryTag=6, ACK=true

interrupt
stopping publisher
2025/07/30 10:44:51 gorabbit INFO: closing channel manager...
2025/07/30 10:44:51 gorabbit INFO: closing publisher...
2025/07/30 10:44:51 gorabbit INFO: closing connection manager...
2025/07/30 10:44:51 gorabbit INFO: amqp channel closed gracefully
2025/07/30 10:44:51 gorabbit WARN: returns channel closed, stopping return handler
2025/07/30 10:44:51 gorabbit WARN: confirmations channel closed
2025/07/30 10:44:51 gorabbit INFO: amqp connection closed gracefully

Process finished with the exit code 0

@wagslane Thoughts? 🙏

[joec4i]

This is no longer needed since //go:build integration is used in integration_test.go