-
Notifications
You must be signed in to change notification settings - Fork 49
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add an optional callback to the RMQConnectionFactory to be notified of confirmed/nack-ed published messages. Setting this callback will enable publisher confirms globally (for all the children objects created by the RMQConnectionFactory instance). Publisher confirms use a publishing sequence number to correlate messages. As there is no such concept in JMS, the JMS client takes care of mapping sequence numbers with original messages. The confirm callback has one method with one context parameter. This allows adding more information about published messages later (e.g. the session or producer instance), without breaking changes. The context consists now of the original message and a flag for confirmed or nacked. Fixes #115
- Loading branch information
1 parent
5be302b
commit 48382b2
Showing
14 changed files
with
508 additions
and
32 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
33 changes: 33 additions & 0 deletions
33
src/main/java/com/rabbitmq/jms/client/ConfirmListener.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| /* Copyright (c) 2019 Pivotal Software, Inc. All rights reserved. */ | ||
| package com.rabbitmq.jms.client; | ||
|
|
||
| /** | ||
| * Listener to be notified of publisher confirms. | ||
| * <p> | ||
| * Publisher Confirms are a RabbitMQ extension to the AMQP protocol. | ||
| * In the context of the JMS client, they allow to be asynchronously | ||
| * notified when an outbound message has been confirmed by the broker | ||
| * (meaning the message made it safely to the broker). | ||
| * <p> | ||
| * Note publisher confirms are enabled at the {@link com.rabbitmq.jms.admin.RMQConnectionFactory} level, | ||
| * by setting a {@link ConfirmListener} instance with | ||
| * {@link com.rabbitmq.jms.admin.RMQConnectionFactory#setConfirmListener(ConfirmListener)}. This | ||
| * means all the AMQP {@link com.rabbitmq.client.Channel} created from the {@link com.rabbitmq.jms.admin.RMQConnectionFactory} | ||
| * instance will have publisher confirms enabled. | ||
| * | ||
| * @see <a href="https://www.rabbitmq.com/confirms.html#publisher-confirms">Publisher Confirms</a> | ||
| * @see <a href="https://www.rabbitmq.com/publishers.html#data-safety">Publisher Guide</a> | ||
| * @see com.rabbitmq.jms.admin.RMQConnectionFactory#setConfirmListener(ConfirmListener) | ||
| * @since 1.13.0 | ||
| */ | ||
| @FunctionalInterface | ||
| public interface ConfirmListener { | ||
|
|
||
| /** | ||
| * Callback invoked when an outbound message is confirmed. | ||
| * | ||
| * @param context information about the confirmed message | ||
| */ | ||
| void handle(PublisherConfirmContext context); | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
39 changes: 39 additions & 0 deletions
39
src/main/java/com/rabbitmq/jms/client/PublisherConfirmContext.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| /* Copyright (c) 2019 Pivotal Software, Inc. All rights reserved. */ | ||
| package com.rabbitmq.jms.client; | ||
|
|
||
| import javax.jms.Message; | ||
|
|
||
| /** | ||
| * Information an outbound message being confirmed. | ||
| * | ||
| * @see ConfirmListener | ||
| * @since 1.13.0 | ||
| */ | ||
| public class PublisherConfirmContext { | ||
|
|
||
| private final Message message; | ||
| private final boolean ack; | ||
|
|
||
| PublisherConfirmContext(Message message, boolean ack) { | ||
| this.message = message; | ||
| this.ack = ack; | ||
| } | ||
|
|
||
| /** | ||
| * The message being confirmed. | ||
| * | ||
| * @return the confirmed message | ||
| */ | ||
| public Message getMessage() { | ||
| return message; | ||
| } | ||
|
|
||
| /** | ||
| * Whether the message is confirmed or nack-ed (considered lost). | ||
| * | ||
| * @return true if confirmed, false if nack-ed | ||
| */ | ||
| public boolean isAck() { | ||
| return ack; | ||
| } | ||
| } |
90 changes: 90 additions & 0 deletions
90
src/main/java/com/rabbitmq/jms/client/PublisherConfirmsUtils.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,90 @@ | ||
| /* Copyright (c) 2019 Pivotal Software, Inc. All rights reserved. */ | ||
| package com.rabbitmq.jms.client; | ||
|
|
||
| import com.rabbitmq.client.Channel; | ||
|
|
||
| import javax.jms.Message; | ||
| import java.util.Map; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
| import java.util.concurrent.atomic.AtomicLong; | ||
| import java.util.function.Consumer; | ||
|
|
||
| /** | ||
| * Utility class to handle publisher confirms. | ||
| * | ||
| * @since 1.13.0 | ||
| */ | ||
| class PublisherConfirmsUtils { | ||
|
|
||
| /** | ||
| * Enables publisher confirms support. | ||
| * <p> | ||
| * Adds a {@link com.rabbitmq.client.ConfirmListener} to the AMQP {@link Channel} | ||
| * and notifies the user-provided {@link ConfirmListener} when confirms notified | ||
| * arrive. | ||
| * <p> | ||
| * Returns a {@link PublishingListener} that must be called whenever a message is published. | ||
| * | ||
| * @param channel | ||
| * @param confirmListener | ||
| * @return | ||
| */ | ||
| static PublishingListener configurePublisherConfirmsSupport(Channel channel, ConfirmListener confirmListener) { | ||
| final Map<Long, Message> outstandingConfirms = new ConcurrentHashMap<>(); | ||
| final AtomicLong multipleLowerBound = new AtomicLong(1); | ||
| PublishingListener publishingListener = (message, sequenceNumber) -> { | ||
| outstandingConfirms.put(sequenceNumber, message); | ||
| }; | ||
| channel.addConfirmListener(new com.rabbitmq.client.ConfirmListener() { | ||
| @Override | ||
| public void handleAck(long deliveryTag, boolean multiple) { | ||
| cleanPublisherConfirmsCorrelation( | ||
| outstandingConfirms, multipleLowerBound, | ||
| deliveryTag, multiple, message -> confirmListener.handle(new PublisherConfirmContext(message, true)) | ||
| ); | ||
| } | ||
|
|
||
| @Override | ||
| public void handleNack(long deliveryTag, boolean multiple) { | ||
| cleanPublisherConfirmsCorrelation( | ||
| outstandingConfirms, multipleLowerBound, | ||
| deliveryTag, multiple, message -> confirmListener.handle(new PublisherConfirmContext(message, false)) | ||
| ); | ||
| } | ||
| }); | ||
| return publishingListener; | ||
| } | ||
|
|
||
| /** | ||
| * Cleans the data structure used to correlate publishing sequence numbers to messages when a confirm comes in. | ||
| * <p> | ||
| * Invoke a provided callback for each message confirmed/nack-ed. | ||
| * | ||
| * @param outstandingConfirms | ||
| * @param multipleLowerBound | ||
| * @param deliveryTag | ||
| * @param multiple | ||
| * @param messageConsumer | ||
| */ | ||
| private static void cleanPublisherConfirmsCorrelation(Map<Long, Message> outstandingConfirms, AtomicLong multipleLowerBound, | ||
| long deliveryTag, boolean multiple, Consumer<Message> messageConsumer) { | ||
| Long lowerBound = multipleLowerBound.get(); | ||
| if (multiple) { | ||
| for (long i = lowerBound; i <= deliveryTag; i++) { | ||
| Message message = outstandingConfirms.remove(i); | ||
| if (message != null) { | ||
| messageConsumer.accept(message); | ||
| } | ||
| } | ||
| } else { | ||
| Message message = outstandingConfirms.remove(deliveryTag); | ||
| if (message != null) { | ||
| messageConsumer.accept(message); | ||
| } | ||
| if (deliveryTag == lowerBound + 1) { | ||
| multipleLowerBound.compareAndSet(lowerBound, deliveryTag); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } |
15 changes: 15 additions & 0 deletions
15
src/main/java/com/rabbitmq/jms/client/PublishingListener.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| /* Copyright (c) 2019 Pivotal Software, Inc. All rights reserved. */ | ||
| package com.rabbitmq.jms.client; | ||
|
|
||
| import javax.jms.Message; | ||
|
|
||
| /** | ||
| * Internal interface to notify about a published message when publisher confirms are enabled. | ||
| * | ||
| * @since 1.13.0 | ||
| */ | ||
| interface PublishingListener { | ||
|
|
||
| void publish(Message message, long sequenceNumber); | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.