Send-Via / TransferSender to send messages across entities in a transaction (#215)

This feature lets a user send messages across entities in a single transaction.

A single transaction cannot theoretically span across entities. So as to support cross-entity transaction, you can now send a message to a `destination-queue` via another queue. The transaction will be performed on the `via-queue`, and once successful, the message will be forwarded/transferred to its intended destination.

Sample:
```java
IMessageSender viaQueueSender = ClientFactory.createMessageSenderFromEntityPath(factory, viaQueueName);
IMessageReceiver viaQueueReceiver = ClientFactory.createMessageReceiverFromEntityPath(factory, viaQueueName, ReceiveMode.PEEKLOCK);
IMessageSender destinationViaSender = ClientFactory.createTransferMessageSenderFromEntityPath(factory, destinationQueueName, viaQueueName);

Message message1 = new Message("message");
message1.setMessageId("1");
message1.setPartitionKey("pk1");
Message message2 = new Message("message");
message2.setMessageId("2");
message2.setPartitionKey("pk2");
message2.setViaPartitionKey("pk1");

viaQueueSender.send(message1);
IMessage receivedMessage = viaQueueReceiver.receive();

TransactionContext transaction = this.factory.startTransaction();
viaQueueReceiver.complete(receivedMessage.getLockToken(), transaction);
destinationViaSender.send(message2, transaction);
transaction.commit();
```

In the above example, operations on both `message1` and `message2` (intended for different entities) are being performed within the same entity (i.e., the `via-queue`)

Author: nemakam

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/ClientFactory.java

@@ -86,6 +86,25 @@ public static IMessageSender createMessageSenderFromEntityPath(MessagingFactory
         return Utils.completeFuture(createMessageSenderFromEntityPathAsync(messagingFactory, entityPath));
     }
 
+    /**
+     * Creates a transfer message sender. This sender sends message to destination entity via another entity.
+     *
+     * This is mainly to be used when sending messages in a transaction.
+     * When messages need to be sent across entities in a single transaction, this can be used to ensure
+     * all the messages land initially in the same entity/partition for local transactions, and then
+     * let service bus handle transferring the message to the actual destination.
+     * @param messagingFactory messaging factory (which represents a connection) on which sender needs to be created.
+     * @param entityPath path of the final destination of the message.
+     * @param viaEntityPath The initial destination of the message.
+     * @return IMessageSender instance
+     * @throws InterruptedException if the current thread was interrupted while waiting
+     * @throws ServiceBusException if the sender cannot be created
+     */
+    public static IMessageSender createTransferMessageSenderFromEntityPath(MessagingFactory messagingFactory, String entityPath, String viaEntityPath)  throws InterruptedException, ServiceBusException
+    {
+        return Utils.completeFuture(createTransferMessageSenderFromEntityPathAsync(messagingFactory, entityPath, viaEntityPath));
+    }
+
     /**
      * Create message sender asynchronously from connection string with <a href="https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-sas">Shared Access Signatures</a>
      *
@@ -132,7 +151,7 @@ public static CompletableFuture<IMessageSender> createMessageSenderFromEntityPat
     public static CompletableFuture<IMessageSender> createMessageSenderFromEntityPathAsync(URI namespaceEndpointURI, String entityPath, ClientSettings clientSettings)
     {
         Utils.assertNonNull("namespaceEndpointURI", namespaceEndpointURI);
-        MessageSender sender = new MessageSender(namespaceEndpointURI, entityPath, clientSettings);
+        MessageSender sender = new MessageSender(namespaceEndpointURI, entityPath, null, clientSettings);
         return sender.initializeAsync().thenApply((v) -> sender);
     }
 
@@ -148,6 +167,25 @@ public static CompletableFuture<IMessageSender> createMessageSenderFromEntityPat
         return sender.initializeAsync().thenApply((v) -> sender);
     }
 
+    /**
+     * Creates a transfer message sender asynchronously. This sender sends message to destination entity via another entity.
+     *
+     * This is mainly to be used when sending messages in a transaction.
+     * When messages need to be sent across entities in a single transaction, this can be used to ensure
+     * all the messages land initially in the same entity/partition for local transactions, and then
+     * let service bus handle transferring the message to the actual destination.
+     * @param messagingFactory messaging factory (which represents a connection) on which sender needs to be created.
+     * @param entityPath path of the final destination of the message.
+     * @param viaEntityPath The initial destination of the message.
+     * @return a CompletableFuture representing the pending creating of IMessageSender instance.
+     */
+    public static CompletableFuture<IMessageSender> createTransferMessageSenderFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, String viaEntityPath)
+    {
+        Utils.assertNonNull("messagingFactory", messagingFactory);
+        MessageSender sender = new MessageSender(messagingFactory, viaEntityPath, entityPath);
+        return sender.initializeAsync().thenApply((v) -> sender);
+    }
+
     /**
      * Create {@link IMessageReceiver} in default {@link ReceiveMode#PEEKLOCK} mode from service bus connection string with <a href="https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-sas">Shared Access Signatures</a>
      *

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/IMessage.java

@@ -332,6 +332,26 @@ public interface IMessage {
      */
     public void setPartitionKey(String partitionKey);
 
+    /**
+     * Gets the partition key for sending a message to a entity via another partitioned transfer entity.
+     *
+     * If a message is sent via a transfer queue in the scope of a transaction, this value selects the
+     * transfer queue partition: This is functionally equivalent to {@link #getPartitionKey()} and ensures that
+     * messages are kept together and in order as they are transferred.
+     *
+     * @return partition key on the via queue.
+     * @see <a href="https://docs.microsoft.com/azure/service-bus-messaging/service-bus-transactions#transfers-and-send-via">Transfers and Send Via</a>.
+     */
+    public String getViaPartitionKey();
+
+    /**
+     * Sets a via-partition key for sending a message to a destination entity via another partitioned entity
+     *
+     * @param viaPartitionKey via-partition key of this message
+     * @see #getViaPartitionKey()
+     */
+    public void setViaPartitionKey(String viaPartitionKey);
+
     /**
      * Gets the name of the queue or subscription that this message was enqueued on, before it was deadlettered.
      * 

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/IMessageSender.java

@@ -122,15 +122,6 @@ public interface IMessageSender extends IMessageEntityClient {
      */
     CompletableFuture<Void> cancelScheduledMessageAsync(long sequenceNumber);
 
-    /**
-     * Cancels the enqueuing of an already sent scheduled message, if it was not already enqueued. This is an asynchronous method returning a CompletableFuture which completes when the message is cancelled.
-     *
-     * @param sequenceNumber sequence number of the scheduled message
-     * @param transaction {@link TransactionContext} which this operation should enlist to.
-     * @return a CompletableFuture representing the pending cancellation
-     */
-    CompletableFuture<Void> cancelScheduledMessageAsync(long sequenceNumber, TransactionContext transaction);
-
     /**
      * Sends a scheduled message to the Azure Service Bus entity this sender is connected to. A scheduled message is enqueued and made available to receivers only at the scheduled enqueue time.
      * This method blocks until the message is sent to the entity. Calling this method is equivalent to calling <code>scheduleMessageAsync(message, scheduledEnqueueTimeUtc).get()</code>. For better performance, use async methods.
@@ -165,15 +156,4 @@ public interface IMessageSender extends IMessageEntityClient {
      * @throws ServiceBusException  if scheduled message couldn't be cancelled
      */
     void cancelScheduledMessage(long sequenceNumber) throws InterruptedException, ServiceBusException;
-
-    /**
-     * Cancels the enqueuing of an already sent scheduled message, if it was not already enqueued. This method blocks until the message is sent to the entity. Calling this method is equivalent to calling <code>cancelScheduledMessageAsync(sequenceNumber).get()</code>.
-     * For better performance, use async methods.
-     *
-     * @param sequenceNumber sequence number of the scheduled message
-     * @param transaction {@link TransactionContext} which this operation should enlist to.
-     * @throws InterruptedException if the current thread was interrupted while waiting
-     * @throws ServiceBusException  if scheduled message couldn't be cancelled
-     */
-    void cancelScheduledMessage(long sequenceNumber, TransactionContext transaction) throws InterruptedException, ServiceBusException;
 }

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/Message.java

@@ -58,6 +58,8 @@ final public class Message implements Serializable, IMessage {
 	private String replyTo;
 
 	private String partitionKey;
+
+	private String viaPartitionKey;
 
 	private String deadLetterSource;
 
@@ -272,7 +274,17 @@ public String getPartitionKey() {
 	@Override
 	public void setPartitionKey(String partitionKey) {
 		this.partitionKey = partitionKey;		
-	}	
+	}
+
+	@Override
+	public String getViaPartitionKey() {
+        return this.viaPartitionKey;
+    }
+
+    @Override
+    public void setViaPartitionKey(String partitionKey) {
+        this.viaPartitionKey = partitionKey;
+    }
 
 	@Override
 	public String getDeadLetterSource() {

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java

@@ -61,7 +61,7 @@ public void registerMessageHandler(IMessageHandler handler) throws InterruptedEx
 
     @Override
     public void registerMessageHandler(IMessageHandler handler, MessageHandlerOptions handlerOptions) throws InterruptedException, ServiceBusException {
-        TRACE_LOGGER.info("Registering message handler on entity '{}' with '{}'", this.entityPath, handlerOptions);
+        TRACE_LOGGER.info("Registering message handler on entity '{}' with '{}'.", this.entityPath, handlerOptions);
         this.setHandlerRegistered();
         this.messageHandler = handler;
         this.messageHandlerOptions = handlerOptions;

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageConverter.java

@@ -59,6 +59,11 @@ public static org.apache.qpid.proton.message.Message convertBrokeredMessageToAmq
 		{
 			messageAnnotationsMap.put(Symbol.valueOf(ClientConstants.PARTITIONKEYNAME), brokeredMessage.getPartitionKey());
 		}
+
+		if(!StringUtil.isNullOrEmpty(brokeredMessage.getViaPartitionKey()))
+		{
+			messageAnnotationsMap.put(Symbol.valueOf(ClientConstants.VIAPARTITIONKEYNAME), brokeredMessage.getViaPartitionKey());
+		}
 
 		amqpMessage.setMessageAnnotations(new MessageAnnotations(messageAnnotationsMap));
 
@@ -169,6 +174,9 @@ public static Message convertAmqpMessageToBrokeredMessage(org.apache.qpid.proton
 	                    case ClientConstants.PARTITIONKEYNAME:
 	                        brokeredMessage.setPartitionKey((String)entry.getValue());
 	                        break;
+						case ClientConstants.VIAPARTITIONKEYNAME:
+							brokeredMessage.setViaPartitionKey((String)entry.getValue());
+							break;
 	                    case ClientConstants.DEADLETTERSOURCENAME:
 	                        brokeredMessage.setDeadLetterSource((String)entry.getValue());
 	                        break;

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSender.java

@@ -22,6 +22,7 @@ final class MessageSender extends InitializableEntity implements IMessageSender
     private static final Logger TRACE_LOGGER = LoggerFactory.getLogger(MessageSender.class);
     private boolean ownsMessagingFactory;
     private String entityPath = null;
+    private String transferDestinationPath = null;
     private MessagingFactory messagingFactory = null;
     private CoreMessageSender internalSender = null;
     private boolean isInitialized = false;
@@ -32,24 +33,30 @@ private MessageSender() {
         super(StringUtil.getShortRandomString());
     }
 
-    MessageSender(URI namespaceEndpointURI, String entityPath, ClientSettings clientSettings) {
+    MessageSender(URI namespaceEndpointURI, String entityPath, String transferDestinationPath, ClientSettings clientSettings) {
         this();
 
         this.namespaceEndpointURI = namespaceEndpointURI;
+        this.transferDestinationPath = transferDestinationPath;
         this.entityPath = entityPath;
         this.clientSettings = clientSettings;
         this.ownsMessagingFactory = true;
     }
 
     MessageSender(MessagingFactory messagingFactory, String entityPath) {
-        this(messagingFactory, entityPath, false);
+        this(messagingFactory, entityPath, null, false);
     }
 
-    private MessageSender(MessagingFactory messagingFactory, String entityPath, boolean ownsMessagingFactory) {
+    MessageSender(MessagingFactory messagingFactory, String entityPath, String transferDestinationPath) {
+        this(messagingFactory, entityPath, transferDestinationPath, false);
+    }
+
+    private MessageSender(MessagingFactory messagingFactory, String entityPath, String transferDestinationPath, boolean ownsMessagingFactory) {
         this();
 
         this.messagingFactory = messagingFactory;
         this.entityPath = entityPath;
+        this.transferDestinationPath = transferDestinationPath;
         this.ownsMessagingFactory = ownsMessagingFactory;
     }
 
@@ -77,7 +84,7 @@ synchronized CompletableFuture<Void> initializeAsync() {
             return factoryFuture.thenComposeAsync((v) ->
             {
                 TRACE_LOGGER.info("Creating MessageSender to entity '{}'", this.entityPath);
-                CompletableFuture<CoreMessageSender> senderFuture = CoreMessageSender.create(this.messagingFactory, StringUtil.getShortRandomString(), this.entityPath);
+                CompletableFuture<CoreMessageSender> senderFuture = CoreMessageSender.create(this.messagingFactory, StringUtil.getShortRandomString(), this.entityPath, this.transferDestinationPath);
                 CompletableFuture<Void> postSenderCreationFuture = new CompletableFuture<Void>();
                 senderFuture.handleAsync((s, coreSenderCreationEx) -> {
                     if (coreSenderCreationEx == null) {
@@ -195,14 +202,8 @@ public CompletableFuture<Long> scheduleMessageAsync(IMessage message, Instant sc
 
     @Override
     public CompletableFuture<Void> cancelScheduledMessageAsync(long sequenceNumber) {
-        return this.cancelScheduledMessageAsync(sequenceNumber, TransactionContext.NULL_TXN);
-    }
-
-    @Override
-    public CompletableFuture<Void> cancelScheduledMessageAsync(long sequenceNumber, TransactionContext transaction) {
         return this.internalSender.cancelScheduledMessageAsync(
                 new Long[]{sequenceNumber},
-                transaction,
                 this.messagingFactory.getClientSetttings().getOperationTimeout());
     }
 
@@ -221,11 +222,6 @@ public void cancelScheduledMessage(long sequenceNumber) throws InterruptedExcept
         Utils.completeFuture(this.cancelScheduledMessageAsync(sequenceNumber));
     }
 
-    @Override
-    public void cancelScheduledMessage(long sequenceNumber, TransactionContext transaction) throws InterruptedException, ServiceBusException {
-        Utils.completeFuture(this.cancelScheduledMessageAsync(sequenceNumber, transaction));
-    }
-
     MessagingFactory getMessagingFactory() {
         return this.messagingFactory;
     }

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/QueueClient.java

@@ -207,14 +207,9 @@ public CompletableFuture<Long> scheduleMessageAsync(IMessage message, Instant sc
 
     @Override
     public CompletableFuture<Void> cancelScheduledMessageAsync(long sequenceNumber) {
-        return this.cancelScheduledMessageAsync(sequenceNumber, TransactionContext.NULL_TXN);
-    }
-
-    @Override
-    public CompletableFuture<Void> cancelScheduledMessageAsync(long sequenceNumber, TransactionContext transaction) {
         return this.createSenderAsync().thenComposeAsync((v) ->
         {
-            return this.sender.cancelScheduledMessageAsync(sequenceNumber, transaction);
+            return this.sender.cancelScheduledMessageAsync(sequenceNumber);
         });
     }
 
@@ -233,11 +228,6 @@ public void cancelScheduledMessage(long sequenceNumber) throws InterruptedExcept
         Utils.completeFuture(this.cancelScheduledMessageAsync(sequenceNumber));
     }
 
-    @Override
-    public void cancelScheduledMessage(long sequenceNumber, TransactionContext transaction) throws InterruptedException, ServiceBusException {
-        Utils.completeFuture(this.cancelScheduledMessageAsync(sequenceNumber, transaction));
-    }
-
     @Override
     public String getEntityPath() {
         return this.queuePath;

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/TopicClient.java

@@ -115,11 +115,6 @@ public CompletableFuture<Void> cancelScheduledMessageAsync(long sequenceNumber)
         return this.sender.cancelScheduledMessageAsync(sequenceNumber);
     }
 
-    @Override
-    public CompletableFuture<Void> cancelScheduledMessageAsync(long sequenceNumber, TransactionContext transaction) {
-        return this.sender.cancelScheduledMessageAsync(sequenceNumber, transaction);
-    }
-
     @Override
     public long scheduleMessage(IMessage message, Instant scheduledEnqueueTimeUtc) throws InterruptedException, ServiceBusException {
         return this.sender.scheduleMessage(message, scheduledEnqueueTimeUtc);
@@ -135,11 +130,6 @@ public void cancelScheduledMessage(long sequenceNumber) throws InterruptedExcept
         this.sender.cancelScheduledMessage(sequenceNumber);
     }
 
-    @Override
-    public void cancelScheduledMessage(long sequenceNumber, TransactionContext transaction) throws InterruptedException, ServiceBusException {
-        this.sender.cancelScheduledMessage(sequenceNumber, transaction);
-    }
-
     @Override
     public String getEntityPath() {
         return this.sender.getEntityPath();

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/ClientConstants.java

@@ -37,6 +37,7 @@ private ClientConstants() { }
 	//public static final String LOCKTOKENNAME = "x-opt-lock-token";
 	public static final String LOCKEDUNTILNAME = "x-opt-locked-until";
 	public static final String PARTITIONKEYNAME = "x-opt-partition-key";
+	public static final String VIAPARTITIONKEYNAME = "x-opt-via-partition-key";
 	public static final String DEADLETTERSOURCENAME = "x-opt-deadletter-source";
 	public static final UUID ZEROLOCKTOKEN = new UUID(0l, 0l);	
 
@@ -52,6 +53,7 @@ private ClientConstants() { }
 	public final static Symbol PUBLISHER_REVOKED_ERROR = Symbol.getSymbol(AmqpConstants.VENDOR + ":publisher-revoked");
 	public final static Symbol TIMEOUT_ERROR = Symbol.getSymbol(AmqpConstants.VENDOR + ":timeout");
 	public final static Symbol LINK_TIMEOUT_PROPERTY = Symbol.getSymbol(AmqpConstants.VENDOR + ":timeout");
+    public final static Symbol LINK_TRANSFER_DESTINATION_PROPERTY = Symbol.getSymbol(AmqpConstants.VENDOR + ":transfer-destination-address");
 	public final static Symbol LINK_PEEKMODE_PROPERTY = Symbol.getSymbol(AmqpConstants.VENDOR + ":peek-mode");
 	public final static Symbol TRACKING_ID_PROPERTY = Symbol.getSymbol(AmqpConstants.VENDOR + ":tracking-id");
 	public static final Symbol DEADLETTERNAME = Symbol.valueOf(AmqpConstants.VENDOR + ":dead-letter");
@@ -118,6 +120,7 @@ private ClientConstants() { }
 	public static final String REQUEST_RESPONSE_MESSAGE_ID = "message-id";
 	public static final String REQUEST_RESPONSE_SESSION_ID = "session-id";
 	public static final String REQUEST_RESPONSE_PARTITION_KEY = "partition-key";
+    public static final String REQUEST_RESPONSE_VIA_PARTITION_KEY = "via-partition-key";
 	public static final String REQUEST_RESPONSE_FROM_SEQUENCE_NUMER = "from-sequence-number";
 	public static final String REQUEST_RESPONSE_MESSAGE_COUNT = "message-count";
 	public static final String REQUEST_RESPONSE_STATUS_CODE = "statusCode";