Add ConfirmationChannel for async publisher confirmations

[lukebakken]

Traditional publisher confirms in the Java client require manual
tracking of sequence numbers and correlation of Basic.Return messages.
This makes per-message error handling complex and provides no built-in
async pattern, backpressure mechanism, or message correlation support.

This change introduces ConfirmationChannel, a wrapper that provides
automatic publisher confirmation tracking with a
CompletableFuture-based API, optional throttling, and generic context
parameter for message correlation. The implementation uses
listener-based integration with existing Channel instances, requiring
no modifications to ChannelN.

New API components:

• ConfirmationChannel interface - Extends Channel and adds
  basicPublishAsync() methods that return CompletableFuture<T>
• ConfirmationChannelN implementation - Wraps any Channel instance
  and tracks confirmations via return/confirm/shutdown listeners
• PublishException - Exception thrown when message is nack'd or
  returned, with sequence number, routing details, and user context

The wrapper maintains independent sequence numbers using AtomicLong
and stores confirmation state in a ConcurrentHashMap. Each entry holds
the future, rate limiter permit, and user-provided context. Messages
include an x-seq-no header for correlating Basic.Return responses.

Rate limiting is optional via RateLimiter parameter. The
ThrottlingRateLimiter implementation uses progressive delays
(0-1000ms) based on capacity usage, applying backpressure when available
permits fall below a threshold (default 50%).

The basicPublish() and waitForConfirms() methods throw
UnsupportedOperationException on ConfirmationChannel to prevent
mixing synchronous and asynchronous patterns. All other Channel
methods delegate to the wrapped instance.

[lukebakken]

Hey @acogoluegnes!

Here's a proposal for implementing "automatic" publisher confirmation tracking in a similar manner to the .NET client. As you're the expert here, please suggest better names, a better implementation, etc, as I'm just barely familiar enough with Java and this project to implement this feature with the help of a genie. I did, of course, review the code.

If you like the way this is going, I thought I'd also add rate-throttling in a similar manner to the .NET client, i.e. as the outstanding confirmation window closes, increase delay between publishes to (hopefully) allow the broker to catch up.

[acogoluegnes]

Thanks for this contributon @lukebakken. I think it is on the right track. I added some comments in the code. Just a few more remarks:

• the new methods in the public interfaces should have a default implementation if we back port the PR to 5.x. This is for backward compatibility. Throwing an exception is good enough.
• synchronized blocks and Object-based synchronization is considered old-style Java concurrency. We could see if there are more modern utilities in the Java concurrency toolkit to handle what we need in ChannelN. I'm not super opinionated on this topic though and this is an implementation details we can polish later.
• the bulk of the PR is in ChannelN, it is not rocket science but add some non-trivial logic to a somewhat already complex class. I was wondering if it could be possible to externalize all the logic in a PublishConfirmState class used from ChannelN. This state class would register regular return and confirm listeners on the channel. This could even be an interface with a no-op implementation when the feature is not activated, minimalizing the impact on ChannelN. We can discuss this design when all the features are added.

[michaelklishin]

@lukebakken I have sent you an external contributor invite for this repo.

[lukebakken]

@michaelklishin @acogoluegnes I took this comment to heart and re-implemented this feature as a ConfirmationChannel class that wraps ChannelN and does its own confirmation tracking. That way, no modifications to the existing Channel interface or ChannelN class are necessary.

[lukebakken]

Alrighty gang, all set I think. Small tutorial here - rabbitmq/rabbitmq-tutorials#707

[michaelklishin]

@lukebakken since this is a feature by most definitions, we now would have to go through a special approval process on our end :(