diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/ClientFactory.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/ClientFactory.java index a7f206ccff6b..1b8c2e36f3b0 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/ClientFactory.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/ClientFactory.java @@ -113,10 +113,29 @@ static IMessageSender createMessageSenderFromEntityPath(MessagingFactory messagi * @throws InterruptedException if the current thread was interrupted while waiting * @throws ServiceBusException if the sender cannot be created */ + @Deprecated public static IMessageSender createTransferMessageSenderFromEntityPath(MessagingFactory messagingFactory, String entityPath, String viaEntityPath) throws InterruptedException, ServiceBusException { return Utils.completeFuture(createTransferMessageSenderFromEntityPathAsync(messagingFactory, entityPath, viaEntityPath)); } + /** + * Creates a transacted 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 transactionContext the TransactionContext that this sender will be a part of. + * @return IMessageSender instance + * @throws InterruptedException if the current thread was interrupted while waiting + * @throws ServiceBusException if the sender cannot be created + */ + public static IMessageSender createTransactedMessageSenderFromEntityPath(MessagingFactory messagingFactory, String entityPath, TransactionContext transactionContext) throws InterruptedException, ServiceBusException { + return Utils.completeFuture(createTransactedMessageSenderFromEntityPathAsync(messagingFactory, entityPath, transactionContext)); + } + /** * Create message sender asynchronously from connection string with Shared Access Signatures * @@ -170,7 +189,7 @@ public static CompletableFuture createMessageSenderFromEntityPat static CompletableFuture createMessageSenderFromEntityPathAsync(URI namespaceEndpointURI, String entityPath, MessagingEntityType entityType, ClientSettings clientSettings) { Utils.assertNonNull("namespaceEndpointURI", namespaceEndpointURI); - MessageSender sender = new MessageSender(namespaceEndpointURI, entityPath, null, entityType, clientSettings); + MessageSender sender = new MessageSender(namespaceEndpointURI, entityPath, null, null, entityType, clientSettings); return sender.initializeAsync().thenApply((v) -> sender); } @@ -202,9 +221,30 @@ static CompletableFuture createMessageSenderFromEntityPathAsync( * @param viaEntityPath The initial destination of the message. * @return a CompletableFuture representing the pending creating of IMessageSender instance. */ + @Deprecated public static CompletableFuture createTransferMessageSenderFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, String viaEntityPath) { Utils.assertNonNull("messagingFactory", messagingFactory); - MessageSender sender = new MessageSender(messagingFactory, viaEntityPath, entityPath, null); + MessageSender sender = new MessageSender(messagingFactory, viaEntityPath, entityPath, null, null); + return sender.initializeAsync().thenApply((v) -> sender); + } + + /** + * Creates a transacted message sender asynchronously. + * This sender sends message to destination entity via the queue/topic of the first transacted sender/receiver attached with the transaction. + * + * 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 transactionContext The TransactionContext that this sender will be a part of. + * @return a CompletableFuture representing the pending creating of IMessageSender instance. + */ + public static CompletableFuture createTransactedMessageSenderFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, TransactionContext transactionContext) { + Utils.assertNonNull("messagingFactory", messagingFactory); + Utils.assertNonNull("transactionContext", transactionContext); + MessageSender sender = new MessageSender(messagingFactory, entityPath, null, transactionContext, null); return sender.initializeAsync().thenApply((v) -> sender); } @@ -342,9 +382,36 @@ public static IMessageReceiver createMessageReceiverFromEntityPath(MessagingFact public static IMessageReceiver createMessageReceiverFromEntityPath(MessagingFactory messagingFactory, String entityPath, ReceiveMode receiveMode) throws InterruptedException, ServiceBusException { return Utils.completeFuture(createMessageReceiverFromEntityPathAsync(messagingFactory, entityPath, receiveMode)); } + + /** + * Creates a message receiver to the entity. + * @param messagingFactory messaging factory (which represents a connection) on which receiver needs to be created + * @param entityPath path of the entity + * @param transactionContext the TransactionContext that this receiver will be a part of + * @return IMessageReceiver instance + * @throws InterruptedException if the current thread was interrupted while waiting + * @throws ServiceBusException if the receiver cannot be created + */ + public static IMessageReceiver createTransactedMessageReceiverFromEntityPath(MessagingFactory messagingFactory, String entityPath, TransactionContext transactionContext) throws InterruptedException, ServiceBusException { + return Utils.completeFuture(createTransactedMessageReceiverFromEntityPathAsync(messagingFactory, entityPath, transactionContext, DEFAULTRECEIVEMODE)); + } + + /** + * Creates a message receiver to the entity. + * @param messagingFactory messaging factory (which represents a connection) on which receiver needs to be created + * @param entityPath path of the entity + * @param transactionContext the TransactionContext that this receiver will be a part of + * @param receiveMode PeekLock or ReceiveAndDelete + * @return IMessageReceiver instance + * @throws InterruptedException if the current thread was interrupted while waiting + * @throws ServiceBusException if the receiver cannot be created + */ + public static IMessageReceiver createTransactedMessageReceiverFromEntityPath(MessagingFactory messagingFactory, String entityPath, TransactionContext transactionContext, ReceiveMode receiveMode) throws InterruptedException, ServiceBusException { + return Utils.completeFuture(createTransactedMessageReceiverFromEntityPathAsync(messagingFactory, entityPath, transactionContext, receiveMode)); + } - static IMessageReceiver createMessageReceiverFromEntityPath(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType, ReceiveMode receiveMode) throws InterruptedException, ServiceBusException { - return Utils.completeFuture(createMessageReceiverFromEntityPathAsync(messagingFactory, entityPath, entityType, receiveMode)); + static IMessageReceiver createMessageReceiverFromEntityPath(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType, TransactionContext transactionContext, ReceiveMode receiveMode) throws InterruptedException, ServiceBusException { + return Utils.completeFuture(createMessageReceiverFromEntityPathAsync(messagingFactory, entityPath, entityType, transactionContext, receiveMode)); } /** @@ -437,7 +504,7 @@ public static CompletableFuture createMessageReceiverFromEntit public static CompletableFuture createMessageReceiverFromEntityPathAsync(URI namespaceEndpointURI, String entityPath, ClientSettings clientSettings, ReceiveMode receiveMode) { Utils.assertNonNull("namespaceEndpointURI", namespaceEndpointURI); Utils.assertNonNull("entityPath", entityPath); - MessageReceiver receiver = new MessageReceiver(namespaceEndpointURI, entityPath, null, clientSettings, receiveMode); + MessageReceiver receiver = new MessageReceiver(namespaceEndpointURI, entityPath, null, clientSettings, null, receiveMode); return receiver.initializeAsync().thenApply((v) -> receiver); } @@ -459,12 +526,37 @@ public static CompletableFuture createMessageReceiverFromEntit * @return a CompletableFuture representing the pending creation of message receiver */ public static CompletableFuture createMessageReceiverFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, ReceiveMode receiveMode) { - return createMessageReceiverFromEntityPathAsync(messagingFactory, entityPath, null, receiveMode); + return createMessageReceiverFromEntityPathAsync(messagingFactory, entityPath, null, null, receiveMode); + } + + /** + * Asynchronously creates a new transacted message receiver to the entity on the messagingFactory. + * @param messagingFactory messaging factory (which represents a connection) on which receiver needs to be created. + * @param entityPath path of entity + * @param transactionContext the TransactionContext that this receiver will be a part of. + * @return a CompletableFuture representing the pending creation of message receiver + */ + public static CompletableFuture createTransactedMessageReceiverFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, TransactionContext transactionContext) { + Utils.assertNonNull("transactionContext", transactionContext); + return createMessageReceiverFromEntityPathAsync(messagingFactory, entityPath, null, transactionContext, DEFAULTRECEIVEMODE); + } + + /** + * Asynchronously creates a new transacted message receiver to the entity on the messagingFactory. + * @param messagingFactory messaging factory (which represents a connection) on which receiver needs to be created. + * @param entityPath path of entity + * @param transactionContext the TransactionContext that this receiver will be a part of. + * @param receiveMode PeekLock or ReceiveAndDelete + * @return a CompletableFuture representing the pending creation of message receiver + */ + public static CompletableFuture createTransactedMessageReceiverFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, TransactionContext transactionContext, ReceiveMode receiveMode) { + Utils.assertNonNull("transactionContext", transactionContext); + return createMessageReceiverFromEntityPathAsync(messagingFactory, entityPath, null, transactionContext, receiveMode); } - static CompletableFuture createMessageReceiverFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType, ReceiveMode receiveMode) { + static CompletableFuture createMessageReceiverFromEntityPathAsync(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType, TransactionContext transactionContext, ReceiveMode receiveMode) { Utils.assertNonNull("messagingFactory", messagingFactory); - MessageReceiver receiver = new MessageReceiver(messagingFactory, entityPath, entityType, receiveMode); + MessageReceiver receiver = new MessageReceiver(messagingFactory, entityPath, entityType, transactionContext, receiveMode); return receiver.initializeAsync().thenApply((v) -> receiver); } diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java index 6604c793e4fd..2305db283674 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java @@ -86,7 +86,7 @@ public void registerMessageHandler(IMessageHandler handler, MessageHandlerOption this.messageHandlerOptions = handlerOptions; this.customCodeExecutor = executorService; - this.innerReceiver = ClientFactory.createMessageReceiverFromEntityPath(this.factory, this.entityPath, this.entityType, this.receiveMode); + this.innerReceiver = ClientFactory.createMessageReceiverFromEntityPath(this.factory, this.entityPath, this.entityType, null, this.receiveMode); TRACE_LOGGER.info("Created MessageReceiver to entity '{}'", this.entityPath); if (this.prefetchCount != UNSET_PREFETCH_COUNT) { this.innerReceiver.setPrefetchCount(this.prefetchCount); diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageReceiver.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageReceiver.java index a35aefe4a7f7..c5f6e3252844 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageReceiver.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageReceiver.java @@ -54,6 +54,7 @@ class MessageReceiver extends InitializableEntity implements IMessageReceiver, I private MessageBrowser browser = null; private int messagePrefetchCount; private ScheduledFuture requestResponseLockTokenPruner = null; + private TransactionContext transactionContext; private final ConcurrentHashMap requestResponseLockTokensToLockTimesMap; @@ -68,27 +69,33 @@ private MessageReceiver(ReceiveMode receiveMode) { } } - private MessageReceiver(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType, boolean ownsMessagingFactory, ReceiveMode receiveMode) { + private MessageReceiver(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType, TransactionContext transactionContext, boolean ownsMessagingFactory, ReceiveMode receiveMode) { this(receiveMode); this.messagingFactory = messagingFactory; this.entityPath = entityPath; this.entityType = entityType; + this.transactionContext = transactionContext; this.ownsMessagingFactory = ownsMessagingFactory; } - MessageReceiver(URI namespaceEndpointURI, String entityPath, MessagingEntityType entityType, ClientSettings clientSettings, ReceiveMode receiveMode) { + MessageReceiver(URI namespaceEndpointURI, String entityPath, MessagingEntityType entityType, ClientSettings clientSettings, TransactionContext transactionContext, ReceiveMode receiveMode) { this(receiveMode); this.namespaceEndpointURI = namespaceEndpointURI; this.clientSettings = clientSettings; this.entityPath = entityPath; this.entityType = entityType; + this.transactionContext = transactionContext; this.ownsMessagingFactory = true; } MessageReceiver(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType, ReceiveMode receiveMode) { - this(messagingFactory, entityPath, entityType, false, receiveMode); + this(messagingFactory, entityPath, entityType, null, false, receiveMode); + } + + MessageReceiver(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType, TransactionContext transactionContext, ReceiveMode receiveMode) { + this(messagingFactory, entityPath, entityType, transactionContext, false, receiveMode); } @Override @@ -118,10 +125,10 @@ synchronized CompletableFuture initializeAsync() { CompletableFuture receiverFuture; if (MessageReceiver.this.isSessionReceiver()) { TRACE_LOGGER.info("Creating SessionReceiver to entity '{}', requestedSessionId '{}', browsable session '{}', ReceiveMode '{}'", this.entityPath, this.getRequestedSessionId(), this.isBrowsableSession(), this.receiveMode); - receiverFuture = CoreMessageReceiver.create(this.messagingFactory, StringUtil.getShortRandomString(), this.entityPath, this.getRequestedSessionId(), this.isBrowsableSession(), this.messagePrefetchCount, getSettleModePairForRecevieMode(this.receiveMode), this.entityType); + receiverFuture = CoreMessageReceiver.create(this.messagingFactory, StringUtil.getShortRandomString(), this.entityPath, this.getRequestedSessionId(), this.isBrowsableSession(), this.messagePrefetchCount, this.transactionContext, getSettleModePairForRecevieMode(this.receiveMode), this.entityType); } else { TRACE_LOGGER.info("Creating MessageReceiver to entity '{}', ReceiveMode '{}'", this.entityPath, this.receiveMode); - receiverFuture = CoreMessageReceiver.create(this.messagingFactory, StringUtil.getShortRandomString(), this.entityPath, this.messagePrefetchCount, getSettleModePairForRecevieMode(this.receiveMode), this.entityType); + receiverFuture = CoreMessageReceiver.create(this.messagingFactory, StringUtil.getShortRandomString(), this.entityPath, this.messagePrefetchCount, this.transactionContext, getSettleModePairForRecevieMode(this.receiveMode), this.entityType); } acceptReceiverFuture = receiverFuture.whenCompleteAsync((r, coreReceiverCreationEx) -> { @@ -205,7 +212,7 @@ public void abandon(UUID lockToken, Map propertiesToModify, Tran @Override public CompletableFuture abandonAsync(UUID lockToken) { - return this.abandonAsync(lockToken, TransactionContext.NULL_TXN); + return this.abandonAsync(lockToken, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override @@ -215,7 +222,7 @@ public CompletableFuture abandonAsync(UUID lockToken, TransactionContext t @Override public CompletableFuture abandonAsync(UUID lockToken, Map propertiesToModify) { - return this.abandonAsync(lockToken, propertiesToModify, TransactionContext.NULL_TXN); + return this.abandonAsync(lockToken, propertiesToModify, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override @@ -234,7 +241,7 @@ public CompletableFuture abandonAsync(UUID lockToken, Map @Override public void complete(UUID lockToken) throws InterruptedException, ServiceBusException { - this.complete(lockToken, TransactionContext.NULL_TXN); + this.complete(lockToken, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override @@ -250,7 +257,7 @@ public void completeBatch(Collection messages) { @Override public CompletableFuture completeAsync(UUID lockToken) { - return this.completeAsync(lockToken, TransactionContext.NULL_TXN); + return this.completeAsync(lockToken, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override @@ -297,7 +304,7 @@ public void defer(UUID lockToken, Map propertiesToModify, Transa @Override public CompletableFuture deferAsync(UUID lockToken) { - return this.deferAsync(lockToken, null, TransactionContext.NULL_TXN); + return this.deferAsync(lockToken, null, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override @@ -307,7 +314,7 @@ public CompletableFuture deferAsync(UUID lockToken, TransactionContext tra @Override public CompletableFuture deferAsync(UUID lockToken, Map propertiesToModify) { - return this.deferAsync(lockToken, propertiesToModify, TransactionContext.NULL_TXN); + return this.deferAsync(lockToken, propertiesToModify, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override @@ -366,7 +373,7 @@ public void deadLetter(UUID lockToken, String deadLetterReason, String deadLette @Override public CompletableFuture deadLetterAsync(UUID lockToken) { - return this.deadLetterAsync(lockToken, null, null, null, TransactionContext.NULL_TXN); + return this.deadLetterAsync(lockToken, null, null, null, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override @@ -376,7 +383,7 @@ public CompletableFuture deadLetterAsync(UUID lockToken, TransactionContex @Override public CompletableFuture deadLetterAsync(UUID lockToken, Map propertiesToModify) { - return this.deadLetterAsync(lockToken, null, null, propertiesToModify, TransactionContext.NULL_TXN); + return this.deadLetterAsync(lockToken, null, null, propertiesToModify, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override @@ -386,17 +393,17 @@ public CompletableFuture deadLetterAsync(UUID lockToken, Map deadLetterAsync(UUID lockToken, String deadLetterReason, String deadLetterErrorDescription) { - return this.deadLetterAsync(lockToken, deadLetterReason, deadLetterErrorDescription, null, TransactionContext.NULL_TXN); + return this.deadLetterAsync(lockToken, deadLetterReason, deadLetterErrorDescription, null, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override public CompletableFuture deadLetterAsync(UUID lockToken, String deadLetterReason, String deadLetterErrorDescription, TransactionContext transaction) { - return this.deadLetterAsync(lockToken, deadLetterReason, deadLetterErrorDescription, null, TransactionContext.NULL_TXN); + return this.deadLetterAsync(lockToken, deadLetterReason, deadLetterErrorDescription, null, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override public CompletableFuture deadLetterAsync(UUID lockToken, String deadLetterReason, String deadLetterErrorDescription, Map propertiesToModify) { - return this.deadLetterAsync(lockToken, deadLetterReason, deadLetterErrorDescription, propertiesToModify, TransactionContext.NULL_TXN); + return this.deadLetterAsync(lockToken, deadLetterReason, deadLetterErrorDescription, propertiesToModify, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSender.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSender.java index 6123741da4bb..b615f6d0ae3e 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSender.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSender.java @@ -30,31 +30,33 @@ final class MessageSender extends InitializableEntity implements IMessageSender private boolean isInitialized = false; private URI namespaceEndpointURI; private ClientSettings clientSettings; + private TransactionContext transactionContext; private MessageSender() { super(StringUtil.getShortRandomString()); } - MessageSender(URI namespaceEndpointURI, String entityPath, String transferDestinationPath, MessagingEntityType entityType, ClientSettings clientSettings) { + MessageSender(URI namespaceEndpointURI, String entityPath, String transferDestinationPath, TransactionContext transaction, MessagingEntityType entityType, ClientSettings clientSettings) { this(); this.namespaceEndpointURI = namespaceEndpointURI; - this.transferDestinationPath = transferDestinationPath; this.entityPath = entityPath; + this.transferDestinationPath = transferDestinationPath; this.clientSettings = clientSettings; this.ownsMessagingFactory = true; this.entityType = entityType; + this.transactionContext = transaction; } MessageSender(MessagingFactory messagingFactory, String entityPath, MessagingEntityType entityType) { - this(messagingFactory, entityPath, null, entityType, false); + this(messagingFactory, entityPath, null, null, entityType, false); } - MessageSender(MessagingFactory messagingFactory, String entityPath, String transferDestinationPath, MessagingEntityType entityType) { - this(messagingFactory, entityPath, transferDestinationPath, entityType, false); + MessageSender(MessagingFactory messagingFactory, String entityPath, String transferDestinationPath, TransactionContext transaction, MessagingEntityType entityType) { + this(messagingFactory, entityPath, null, transaction, entityType, false); } - private MessageSender(MessagingFactory messagingFactory, String entityPath, String transferDestinationPath, MessagingEntityType entityType, boolean ownsMessagingFactory) { + private MessageSender(MessagingFactory messagingFactory, String entityPath, String transferDestinationPath, TransactionContext transaction, MessagingEntityType entityType, boolean ownsMessagingFactory) { this(); this.messagingFactory = messagingFactory; @@ -62,6 +64,7 @@ private MessageSender(MessagingFactory messagingFactory, String entityPath, Stri this.transferDestinationPath = transferDestinationPath; this.ownsMessagingFactory = ownsMessagingFactory; this.entityType = entityType; + this.transactionContext = transaction; } @Override @@ -86,7 +89,7 @@ synchronized CompletableFuture initializeAsync() { return factoryFuture.thenComposeAsync((v) -> { TRACE_LOGGER.info("Creating MessageSender to entity '{}'", this.entityPath); - CompletableFuture senderFuture = CoreMessageSender.create(this.messagingFactory, StringUtil.getShortRandomString(), this.entityPath, this.transferDestinationPath, this.entityType); + CompletableFuture senderFuture = CoreMessageSender.create(this.messagingFactory, StringUtil.getShortRandomString(), this.entityPath, this.transactionContext, this.entityType); CompletableFuture postSenderCreationFuture = new CompletableFuture(); senderFuture.handleAsync((s, coreSenderCreationEx) -> { if (coreSenderCreationEx == null) { @@ -116,7 +119,7 @@ CoreMessageSender getInternalSender() { @Override public void send(IMessage message) throws InterruptedException, ServiceBusException { - this.send(message, TransactionContext.NULL_TXN); + this.send(message, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override @@ -126,7 +129,7 @@ public void send(IMessage message, TransactionContext transaction) throws Interr @Override public void sendBatch(Collection message) throws InterruptedException, ServiceBusException { - Utils.completeFuture(this.sendBatchAsync(message)); + Utils.completeFuture(this.sendBatchAsync(message, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext)); } @Override @@ -136,7 +139,7 @@ public void sendBatch(Collection message, TransactionContext @Override public CompletableFuture sendAsync(IMessage message) { - return this.sendAsync(message, TransactionContext.NULL_TXN); + return this.sendAsync(message, this.transactionContext == null ? TransactionContext.NULL_TXN : this.transactionContext); } @Override diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSession.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSession.java index a48ea7a5848d..1cb09d080605 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSession.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageSession.java @@ -15,7 +15,7 @@ public class MessageSession extends MessageReceiver implements IMessageSession { private String requestedSessionId; MessageSession(URI namespaceEndpointURI, String entityPath, MessagingEntityType entityType, String requestedSessionId, ClientSettings clientSettings, ReceiveMode receiveMode) { - super(namespaceEndpointURI, entityPath, entityType, clientSettings, receiveMode); + super(namespaceEndpointURI, entityPath, entityType, clientSettings, null, receiveMode); this.requestedSessionId = requestedSessionId; } diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/amqp/BaseLinkHandler.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/amqp/BaseLinkHandler.java index 02cf1f700e58..708dc53e1103 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/amqp/BaseLinkHandler.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/amqp/BaseLinkHandler.java @@ -25,7 +25,6 @@ public void onLinkLocalClose(Event event) { final Link link = event.getLink(); if (link != null) { TRACE_LOGGER.debug("local link close. linkName:{}", link.getName()); - closeSession(link); } freeClosedLink(link); } @@ -41,7 +40,6 @@ public void onLinkRemoteClose(Event event) { ErrorCondition condition = link.getRemoteCondition(); this.processOnClose(link, condition); - closeSession(link); } freeClosedLink(link); } diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/AsyncUtil.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/AsyncUtil.java index ee3ea00b826d..3d79db42731a 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/AsyncUtil.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/AsyncUtil.java @@ -36,6 +36,12 @@ public static boolean completeFutureExceptionallyAndGetStatus(CompletableFut public static void completeFutureExceptionally(CompletableFuture future, Throwable exception) { MessagingFactory.INTERNAL_THREAD_POOL.submit(new CompleteExceptionallyCallable<>(future, exception)); } + + public static CompletableFuture completedFutureFromException(Throwable exception) { + CompletableFuture future = new CompletableFuture(); + future.completeExceptionally(exception); + return future; + } public static void run(Runnable runnable) { MessagingFactory.INTERNAL_THREAD_POOL.submit(runnable); diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/Controller.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/Controller.java index 47748b1e66d6..c5a60c8d4bc6 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/Controller.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/Controller.java @@ -5,6 +5,8 @@ import com.microsoft.azure.servicebus.ClientSettings; import com.microsoft.azure.servicebus.TransactionContext; +import com.microsoft.azure.servicebus.amqp.SessionHandler; + import org.apache.qpid.proton.amqp.Binary; import org.apache.qpid.proton.amqp.Symbol; import org.apache.qpid.proton.amqp.UnsignedInteger; @@ -17,10 +19,14 @@ import org.apache.qpid.proton.amqp.transaction.Discharge; import org.apache.qpid.proton.amqp.transport.SenderSettleMode; import org.apache.qpid.proton.amqp.transport.Target; +import org.apache.qpid.proton.engine.BaseHandler; +import org.apache.qpid.proton.engine.Connection; +import org.apache.qpid.proton.engine.Session; import org.apache.qpid.proton.message.Message; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.IOException; import java.net.URI; import java.util.HashMap; import java.util.Map; @@ -34,22 +40,41 @@ class Controller { private AtomicBoolean isInitialized = new AtomicBoolean(false); private URI namespaceEndpointURI; private ClientSettings clientSettings; + private Session session; Controller(URI namespaceEndpointURI, MessagingFactory factory, ClientSettings clientSettings) { this.namespaceEndpointURI = namespaceEndpointURI; this.messagingFactory = factory; this.clientSettings = clientSettings; } + + Session getSession() { + return this.session; + } synchronized CompletableFuture initializeAsync() { if (this.isInitialized.get()) { return CompletableFuture.completedFuture(null); } else { + TRACE_LOGGER.info("Creating new Session to coordinator"); + Connection connection = this.messagingFactory.getActiveConnectionOrNothing(); + if (connection == null) { + return AsyncUtil.completedFutureFromException(new IOException("The connection is closed.")); + } + + this.session = connection.session(); + this.session.setOutgoingWindow(Integer.MAX_VALUE); + this.session.open(); + BaseHandler.setHandler(this.session, new SessionHandler("Controller")); + TRACE_LOGGER.info("Created new Session to coordinator"); + TRACE_LOGGER.info("Creating MessageSender to coordinator"); CompletableFuture senderFuture = CoreMessageSender.create( this.messagingFactory, StringUtil.getShortRandomString(), null, + null, + this.session, Controller.getControllerLinkSettings(this.messagingFactory)); CompletableFuture postSenderCreationFuture = new CompletableFuture(); senderFuture.handleAsync((s, coreSenderCreationEx) -> { @@ -115,7 +140,7 @@ public CompletableFuture dischargeAsync(Binary txnId, boolean isCommit) { } protected CompletableFuture closeAsync() { - return null; + return CompletableFuture.completedFuture(null); } private static SenderLinkSettings getControllerLinkSettings(MessagingFactory underlyingFactory) { diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java index d3cb55f12559..9d82d57251e3 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java @@ -90,6 +90,7 @@ public class CoreMessageReceiver extends ClientEntity implements IAmqpReceiver, private Instant sessionLockedUntilUtc; private boolean isSessionLockLost; private ConcurrentLinkedQueue prefetchedMessages; + private Session receiveSession; private Receiver receiveLink; private RequestResponseLink requestResponseLink; private WorkItem linkOpen; @@ -116,6 +117,7 @@ private CoreMessageReceiver(final MessagingFactory factory, final String recvPath, final String sessionId, final int prefetchCount, + final TransactionContext transactionContext, final SettleModePair settleModePair, final MessagingEntityType entityType) { super(name); @@ -145,6 +147,10 @@ private CoreMessageReceiver(final MessagingFactory factory, this.currentPrefetechedMessagesCount = new AtomicInteger(); this.entityType = entityType; + if (transactionContext != null) { + this.receiveSession = this.underlyingFactory.getOrCreateController(transactionContext.getTransactionId()).join().getSession(); + } + this.timedOutUpdateStateRequestsDaemon = () -> { try { if (CoreMessageReceiver.this.getIsClosed()) { @@ -231,6 +237,17 @@ public static CompletableFuture create( final int prefetchCount, final SettleModePair settleModePair, final MessagingEntityType entityType) { + return create(factory, name, recvPath, prefetchCount, null, settleModePair, entityType); + } + + public static CompletableFuture create( + final MessagingFactory factory, + final String name, + final String recvPath, + final int prefetchCount, + final TransactionContext transactionContext, + final SettleModePair settleModePair, + final MessagingEntityType entityType) { TRACE_LOGGER.info("Creating core message receiver to '{}'", recvPath); CoreMessageReceiver msgReceiver = new CoreMessageReceiver( factory, @@ -238,6 +255,7 @@ public static CompletableFuture create( recvPath, null, prefetchCount, + transactionContext, settleModePair, entityType); return msgReceiver.createLink(); @@ -252,6 +270,19 @@ public static CompletableFuture create( final int prefetchCount, final SettleModePair settleModePair, final MessagingEntityType entityType) { + return create(factory, name, recvPath, sessionId, isBrowsableSession, prefetchCount, null, settleModePair, entityType); + } + + public static CompletableFuture create( + final MessagingFactory factory, + final String name, + final String recvPath, + final String sessionId, + final boolean isBrowsableSession, + final int prefetchCount, + final TransactionContext transactionContext, + final SettleModePair settleModePair, + final MessagingEntityType entityType) { TRACE_LOGGER.info("Creating core session receiver to '{}', sessionId '{}', browseonly session '{}'", recvPath, sessionId, isBrowsableSession); CoreMessageReceiver msgReceiver = new CoreMessageReceiver( factory, @@ -259,10 +290,12 @@ public static CompletableFuture create( recvPath, sessionId, prefetchCount, + transactionContext, settleModePair, entityType); msgReceiver.isSessionReceiver = true; msgReceiver.isBrowsableSession = isBrowsableSession; + return msgReceiver.createLink(); } @@ -352,16 +385,12 @@ private void createReceiveLink() { return; } - final Session session = connection.session(); - session.setIncomingCapacity(Integer.MAX_VALUE); - session.open(); - BaseHandler.setHandler(session, new SessionHandler(this.receivePath)); - + Session linkSession = this.receiveSession == null ? this.underlyingFactory.getSession() : this.receiveSession; final String receiveLinkNamePrefix = "Receiver".concat(TrackingUtil.TRACKING_ID_TOKEN_SEPARATOR).concat(StringUtil.getShortRandomString()); final String receiveLinkName = !StringUtil.isNullOrEmpty(connection.getRemoteContainer()) ? receiveLinkNamePrefix.concat(TrackingUtil.TRACKING_ID_TOKEN_SEPARATOR).concat(connection.getRemoteContainer()) : receiveLinkNamePrefix; - final Receiver receiver = session.receiver(receiveLinkName); + final Receiver receiver = linkSession.receiver(receiveLinkName); Source source = new Source(); source.setAddress(receivePath); diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageSender.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageSender.java index e026a164e28a..c48ecbbf4d7c 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageSender.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageSender.java @@ -93,7 +93,6 @@ public class CoreMessageSender extends ClientEntity implements IAmqpSender, IErr private SenderLinkSettings linkSettings; private String transferDestinationPath; private String transferSasTokenAudienceURI; - private boolean isSendVia; private int maxMessageSize; private boolean shouldRetryLinkOpenIfConnectionIsClosedAfterCBSTokenSent = true; @@ -101,25 +100,45 @@ public class CoreMessageSender extends ClientEntity implements IAmqpSender, IErr public static CompletableFuture create( final MessagingFactory factory, final String clientId, - final String senderPath, - final String transferDestinationPath) { - return CoreMessageSender.create(factory, clientId, senderPath, transferDestinationPath, null); + final String senderPath) { + return CoreMessageSender.create(factory, clientId, senderPath, null); } public static CompletableFuture create( final MessagingFactory factory, final String clientId, final String senderPath, - final String transferDestinationPath, final MessagingEntityType entityType) { - return CoreMessageSender.create(factory, clientId, entityType, CoreMessageSender.getDefaultLinkProperties(senderPath, transferDestinationPath, factory, entityType)); + + return CoreMessageSender.create(factory, clientId, null, entityType, CoreMessageSender.getDefaultLinkProperties(senderPath, factory, entityType)); + } + + public static CompletableFuture create( + final MessagingFactory factory, + final String clientId, + final String senderPath, + final TransactionContext transactionContext, + final MessagingEntityType entityType) { + + return CoreMessageSender.create(factory, clientId, transactionContext, entityType, CoreMessageSender.getDefaultLinkProperties(senderPath, factory, entityType)); } static CompletableFuture create( final MessagingFactory factory, final String clientId, + final TransactionContext transactionContext, final MessagingEntityType entityType, final SenderLinkSettings linkSettings) { + return create(factory, clientId, transactionContext, entityType, null, linkSettings); + } + + static CompletableFuture create( + final MessagingFactory factory, + final String clientId, + final TransactionContext transactionContext, + final MessagingEntityType entityType, + final Session session, + final SenderLinkSettings linkSettings) { TRACE_LOGGER.info("Creating core message sender to '{}'", linkSettings.linkPath); final Connection connection = factory.getActiveConnectionCreateIfNecessary(); @@ -131,7 +150,7 @@ static CompletableFuture create( final CoreMessageSender msgSender = new CoreMessageSender(factory, clientId, entityType, linkSettings); TimeoutTracker openLinkTracker = TimeoutTracker.create(factory.getOperationTimeout()); msgSender.initializeLinkOpen(openLinkTracker); - + CompletableFuture authenticationFuture = null; if (linkSettings.requiresAuthentication) { authenticationFuture = msgSender.sendTokenAndSetRenewTimer(false); @@ -146,10 +165,16 @@ static CompletableFuture create( msgSender.linkFirstOpen.completeExceptionally(cause); } else { try { + Session linkSession = session; + if (linkSession == null && transactionContext != null) { + linkSession = msgSender.underlyingFactory.getOrCreateController(transactionContext.getTransactionId()).join().getSession(); + } + + Session linkSessionFinal = linkSession; // Must be an effectively final variable to be used in the scope below msgSender.underlyingFactory.scheduleOnReactorThread(new DispatchHandler() { @Override public void onEvent() { - msgSender.createSendLink(msgSender.linkSettings); + msgSender.createSendLink(linkSessionFinal, msgSender.linkSettings); } }); } catch (IOException ioException) { @@ -209,7 +234,6 @@ private CoreMessageSender(final MessagingFactory factory, final String sendLinkN String transferPath = (String) linkSettings.linkProperties.getOrDefault(ClientConstants.LINK_TRANSFER_DESTINATION_PROPERTY, null); if (transferPath != null && !transferPath.isEmpty()) { this.transferDestinationPath = transferPath; - this.isSendVia = true; this.transferSasTokenAudienceURI = String.format(ClientConstants.SAS_TOKEN_AUDIENCE_FORMAT, factory.getHostName(), transferDestinationPath); } else { // Ensure it is null. @@ -549,7 +573,7 @@ private void cleanupFailedSend(final SendWorkItem failedSend, fin ExceptionUtil.completeExceptionally(failedSend.getWork(), exception, this, true); } - private static SenderLinkSettings getDefaultLinkProperties(String sendPath, String transferDestinationPath, MessagingFactory underlyingFactory, MessagingEntityType entityType) { + private static SenderLinkSettings getDefaultLinkProperties(String sendPath, MessagingFactory underlyingFactory, MessagingEntityType entityType) { SenderLinkSettings linkSettings = new SenderLinkSettings(); linkSettings.linkPath = sendPath; @@ -566,9 +590,6 @@ private static SenderLinkSettings getDefaultLinkProperties(String sendPath, Stri if (entityType != null) { linkProperties.put(ClientConstants.ENTITY_TYPE_PROPERTY, entityType.getIntValue()); } - if (transferDestinationPath != null && !transferDestinationPath.isEmpty()) { - linkProperties.put(ClientConstants.LINK_TRANSFER_DESTINATION_PROPERTY, transferDestinationPath); - } linkSettings.linkProperties = linkProperties; @@ -576,6 +597,10 @@ private static SenderLinkSettings getDefaultLinkProperties(String sendPath, Stri } private void createSendLink(SenderLinkSettings linkSettings) { + createSendLink(null, linkSettings); + } + + private void createSendLink(Session session, SenderLinkSettings linkSettings) { TRACE_LOGGER.info("Creating send link to '{}'", this.sendPath); Connection connection = this.underlyingFactory.getActiveConnectionOrNothing(); if (connection == null) { @@ -599,13 +624,9 @@ private void createSendLink(SenderLinkSettings linkSettings) { return; } - - final Session session = connection.session(); - session.setOutgoingWindow(Integer.MAX_VALUE); - session.open(); - BaseHandler.setHandler(session, new SessionHandler(sendPath)); - final Sender sender = session.sender(linkSettings.linkName); + Session linkSession = session == null ? this.underlyingFactory.getSession() : session; + final Sender sender = linkSession.sender(linkSettings.linkName); sender.setTarget(linkSettings.target); sender.setSource(linkSettings.source); sender.setProperties(linkSettings.linkProperties); diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/MessagingFactory.java b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/MessagingFactory.java index 1f03d1c38feb..add79ba1c5d3 100644 --- a/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/MessagingFactory.java +++ b/sdk/servicebus/microsoft-azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/MessagingFactory.java @@ -5,13 +5,16 @@ import java.io.IOException; import java.net.URI; +import java.nio.ByteBuffer; import java.nio.channels.UnresolvedAddressException; import java.time.Duration; import java.time.Instant; import java.util.LinkedList; import java.util.Locale; +import java.util.Map; import java.util.UUID; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -26,6 +29,8 @@ import com.microsoft.azure.servicebus.amqp.ProtonUtil; import com.microsoft.azure.servicebus.amqp.ReactorDispatcher; import com.microsoft.azure.servicebus.amqp.ReactorHandler; +import com.microsoft.azure.servicebus.amqp.SessionHandler; + import org.apache.qpid.proton.amqp.Binary; import org.apache.qpid.proton.amqp.transport.ErrorCondition; import org.apache.qpid.proton.engine.BaseHandler; @@ -35,6 +40,7 @@ import org.apache.qpid.proton.engine.Handler; import org.apache.qpid.proton.engine.HandlerException; import org.apache.qpid.proton.engine.Link; +import org.apache.qpid.proton.engine.Session; import org.apache.qpid.proton.reactor.Reactor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -59,6 +65,7 @@ public class MessagingFactory extends ClientEntity implements IAmqpConnection { private final String hostName; private final CompletableFuture connetionCloseFuture; private final ConnectionHandler connectionHandler; + private final Map controllers; private final ReactorHandler reactorHandler; private final LinkedList registeredLinks; private final Object reactorLock; @@ -67,8 +74,8 @@ public class MessagingFactory extends ClientEntity implements IAmqpConnection { private Reactor reactor; private ReactorDispatcher reactorScheduler; private Connection connection; - private Controller controller; - + private Session session; + private CompletableFuture factoryOpenFuture; private CompletableFuture cbsLinkCreationFuture; private RequestResponseLink cbsLink; @@ -86,7 +93,8 @@ private MessagingFactory(URI namespaceEndpointUri, ClientSettings clientSettings this.registeredLinks = new LinkedList(); this.connetionCloseFuture = new CompletableFuture(); this.reactorLock = new Object(); - this.connectionHandler = ConnectionHandler.create(clientSettings.getTransportType(), this); + this.connectionHandler = ConnectionHandler.create(clientSettings.getTransportType(), this); + this.controllers = new ConcurrentHashMap(); this.factoryOpenFuture = new CompletableFuture(); this.cbsLinkCreationFuture = new CompletableFuture(); this.managementLinksCache = new RequestResponseLinkCache(this); @@ -108,6 +116,10 @@ public void onReactorInit(Event e) { Timer.register(this.getClientId()); } + Session getSession() { + return this.session; + } + /** * Starts a new service side transaction. The {@link TransactionContext} should be passed to all operations that * needs to be in this transaction. @@ -125,9 +137,13 @@ public TransactionContext startTransaction() throws ServiceBusException, Interru * @return A CompletableFuture which returns a new transaction */ public CompletableFuture startTransactionAsync() { - return this.getController() + return this.getOrCreateController(null) .thenCompose(controller -> controller.declareAsync() - .thenApply(binary -> new TransactionContext(binary.asByteBuffer(), this))); + .thenApply(binary -> { + ByteBuffer txnId = binary.asByteBuffer(); + this.controllers.put(txnId, controller); + return new TransactionContext(txnId, this); + })); } /** @@ -156,27 +172,37 @@ public CompletableFuture endTransactionAsync(TransactionContext transactio return exceptionCompletion; } - return this.getController() + return this.getOrCreateController(transaction.getTransactionId()) .thenCompose(controller -> controller.dischargeAsync(new Binary(transaction.getTransactionId().array()), commit) - .thenRun(() -> transaction.notifyTransactionCompletion(commit))); + .thenApply($null -> { + transaction.notifyTransactionCompletion(commit); + this.controllers.remove(transaction.getTransactionId()); + return controller; + })) + .thenCompose(controller -> controller.closeAsync()); } - private CompletableFuture getController() { - if (this.controller != null) { - return CompletableFuture.completedFuture(this.controller); + /** + * Attempt to create the Controller if the txnId is null, else look for the Controller from the map. + * @param txnId the transaction ID that the Controller is associated with. + * @return the created or found Controller object. + */ + CompletableFuture getOrCreateController(ByteBuffer txnId) { + if (txnId == null) { + return createController(); + } + + Controller controller = this.controllers.get(txnId); + if (controller != null) { + return CompletableFuture.completedFuture(controller); + } else { + throw new RuntimeException("Cannot find the transaction controller associated with the txnId " + new String(txnId.array())); } - - return createController(); } private synchronized CompletableFuture createController() { - if (this.controller != null) { - return CompletableFuture.completedFuture(this.controller); - } - Controller controller = new Controller(this.namespaceEndpointUri, this, this.clientSettings); return controller.initializeAsync().thenApply(v -> { - this.controller = controller; return controller; }); } @@ -390,6 +416,11 @@ public static MessagingFactory createFromConnectionString(final String connectio @Override public void onConnectionOpen() { if (!factoryOpenFuture.isDone()) { + this.session = connection.session(); + session.setOutgoingWindow(Integer.MAX_VALUE); + session.open(); + BaseHandler.setHandler(session, new SessionHandler("my session")); + TRACE_LOGGER.info("MessagingFactory opened."); AsyncUtil.completeFuture(this.factoryOpenFuture, this); } diff --git a/sdk/servicebus/microsoft-azure-servicebus/src/test/java/com/microsoft/azure/servicebus/TransactionTests.java b/sdk/servicebus/microsoft-azure-servicebus/src/test/java/com/microsoft/azure/servicebus/TransactionTests.java new file mode 100644 index 000000000000..4851c8975a48 --- /dev/null +++ b/sdk/servicebus/microsoft-azure-servicebus/src/test/java/com/microsoft/azure/servicebus/TransactionTests.java @@ -0,0 +1,419 @@ +package com.microsoft.azure.servicebus; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.time.Duration; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; + +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import com.microsoft.azure.servicebus.management.ManagementClientAsync; +import com.microsoft.azure.servicebus.management.QueueDescription; +import com.microsoft.azure.servicebus.management.SubscriptionDescription; +import com.microsoft.azure.servicebus.management.TopicDescription; +import com.microsoft.azure.servicebus.primitives.ConnectionStringBuilder; +import com.microsoft.azure.servicebus.primitives.MessagingFactory; +import com.microsoft.azure.servicebus.primitives.ServiceBusException; + +public class TransactionTests { + private static final Duration DEFAULT_TIMEOUT = Duration.ofMillis(2000); + private static final Duration DEFAULT_MESSAGE_TTL = Duration.ofMinutes(1); + private static final ConnectionStringBuilder builder = new ConnectionStringBuilder("Endpoint=sb://contoso.servicebus.onebox.windows-int.net/;SharedAccessKeyName=DefaultNamespaceSasAllKeyName;SharedAccessKey=8864/auVd3qDC75iTjBL1GJ4D2oXC6bIttRd0jzDZ+g="); + private static MessagingFactory factory; + private static ManagementClientAsync managementClient; + private String guid; + private TransactionContext transaction; + private IMessageSender viaEntityTestSender; // Not transacted, used only to send test messages + private IMessageReceiver viaEntityTestReceiver; // Not transacted, used only to verify messages are received + private IMessageReceiver destinationEntityTestReceiver; // Not transacted, used only to verify messages are received + private String viaEntitySendPath; + private String viaEntityReceivePath; + private String destinationEntitySendPath; + private String destinationEntityReceivePath; + private IMessageSender viaEntitySender; + private IMessageSender destinationEntitySender; + private IMessageReceiver viaEntityReceiver; + + @BeforeClass + public static void classInit() throws InterruptedException, ExecutionException { + managementClient = new ManagementClientAsync(builder); + factory = MessagingFactory.createFromConnectionStringBuilder(builder); + } + + @Before + public void testInit() throws InterruptedException, ServiceBusException, ExecutionException { + guid = UUID.randomUUID().toString().substring(0, 10); + System.out.println("Creating transaction"); + this.transaction = factory.startTransactionAsync().get(); + System.out.println("Created transaction"); + } + + @AfterClass + public static void classCleanup() throws IOException, ServiceBusException { + factory.close(); + managementClient.close(); + } + + @Test + public void viaQueueSenderDestinationQueueCommitTest() throws Exception { + sendviaTest("viaQueueSenderDestinationQueueSenderCommitTest", false, false, true, true, true, true, true); + } + + @Test + public void viaQueueSenderDestinationQueueRollbackTest() throws Exception { + sendviaTest("viaQueueSenderDestinationQueueSenderRollbackTest", false, false, true, true, true, true, false); + } + + @Test + public void viaQueueReceiverDestinationQueueCommitTest() throws Exception { + sendviaTest("viaQueueReceiverDestinationQueueCommitTest", false, false, true, true, false, true, true); + } + + @Test + public void viaQueueReceiverDestinationQueueRollbackTest() throws Exception { + sendviaTest("viaQueueReceiverDestinationQueueRollbackTest", false, false, true, true, false, true, false); + } + + @Test + public void viaQueueSenderDestinationTopicCommitTest() throws Exception { + sendviaTest("viaQueueSenderDestinationTopicCommitTest", false, false, true, false, true, true, true); + } + + @Test + public void viaQueueSenderDestinationTopicRollbackTest() throws Exception { + sendviaTest("viaQueueSenderDestinationTopicRollbackTest", false, false, true, false, true, true, false); + } + + @Test + public void viaQueueReceiverDestinationTopicCommitTest() throws Exception { + sendviaTest("viaQueueReceiverDestinationTopicCommitTest", false, false, true, false, false, true, true); + } + + @Test + public void viaQueueReceiverDestinationTopicRollbackTest() throws Exception { + sendviaTest("viaQueueReceiverDestinationTopicRollbackTest", false, false, true, false, false, true, false); + } + + @Test + public void viaTopicSenderDestinationQueueCommitTest() throws Exception { + sendviaTest("viaTopicSenderDestinationQueueCommitTest", false, false, false, true, true, true, true); + } + + @Test + public void viaTopicSenderDestinationQueueRollbackTest() throws Exception { + sendviaTest("viaTopicSenderDestinationQueueRollbackTest", false, false, false, true, true, true, false); + } + + @Test + public void viaTopicReceiverDestinationQueueCommitTest() throws Exception { + sendviaTest("viaTopicReceiverDestinationQueueCommitTest", false, false, false, true, false, true, true); + } + + @Test + public void viaTopicReceiverDestinationQueueRollbackTest() throws Exception { + sendviaTest("viaTopicReceiverDestinationQueueRollbackTest", false, false, false, true, false, true, false); + } + + @Test + public void viaTopicSenderDestinationTopicCommitTest() throws Exception { + sendviaTest("viaTopicSenderDestinationTopicCommitTest", false, false, false, false, true, true, true); + } + + @Test + public void viaTopicSenderDestinationTopicRollbackTest() throws Exception { + sendviaTest("viaTopicSenderDestinationTopicRollbackTest", false, false, false, false, true, true, false); + } + + @Test + public void viaTopicReceiverDestinationTopicCommitTest() throws Exception { + sendviaTest("viaTopicReceiverDestinationTopicCommitTest", false, false, false, false, false, true, true); + } + + @Test + public void viaTopicReceiverDestinationTopicRollbackTest() throws Exception { + sendviaTest("viaTopicReceiverDestinationTopicRollbackTest", false, false, false, false, true, true, false); + } + + @Test + public void viaPartitionedQueueSenderDestinationQueueTest() throws Exception { + sendviaTest("viaPartitionedQueueSenderDestinationQueueTest", true, false, true, true, true, true, true); + } + + @Test + public void viaPartitionedTopicSenderDestinationQueueTest() throws Exception { + sendviaTest("viaPartitionedTopicSenderDestinationQueueTest", true, false, false, true, true, true, true); + } + + @Test + public void viaPartitionedQueueSenderDestinationTopicTest() throws Exception { + sendviaTest("viaPartitionedQueueSenderDestinationTopicTest", true, false, true, false, true, true, true); + } + + @Test + public void viaPartitionedTopicSenderDestinationTopicTest() throws Exception { + sendviaTest("viaPartitionedTopicSenderDestinationTopicTest", true, false, false, false, true, true, true); + } + + @Test + public void viaQueueSenderPartitionedDestinationQueueTest() throws Exception { + sendviaTest("viaQueueSenderPartitionedDestinationQueueTest", false, true, true, true, true, true, true); + } + + @Test + public void viaTopicSenderPartitionedDestinationQueueTest() throws Exception { + sendviaTest("viaTopicSenderPartitionedDestinationQueueTest", false, true, false, true, true, true, true); + } + + @Test + public void viaQueueSenderPartitionedDestinationTopicTest() throws Exception { + sendviaTest("viaQueueSenderPartitionedDestinationTopicTest", false, true, true, false, true, true, true); + } + + @Test + public void viaTopicSenderPartitionedDestinationTopicTest() throws Exception { + sendviaTest("viaTopicSenderPartitionedDestinationTopicTest", false, true, false, false, true, true, true); + } + + @Test + public void viaPartitionedQueueSenderPartitionedDestinationQueueTest() throws Exception { + sendviaTest("viaPartitionedQueueSenderPartitionedDestinationQueueTest", true, true, true, true, true, true, true); + } + + @Test + public void viaPartitionedTopicSenderPartitionedDestinationQueueTest() throws Exception { + sendviaTest("viaPartitionedTopicSenderPartitionedDestinationQueueTest", true, true, false, true, true, true, true); + } + + @Test + public void viaPartitionedQueueSenderPartitionedDestinationTopicTest() throws Exception { + sendviaTest("viaPartitionedQueueSenderPartitionedDestinationTopicTest", true, true, true, false, true, true, true); + } + + @Test + public void viaPartitionedTopicSenderPartitionedDestinationTopicTest() throws Exception { + sendviaTest("viaPartitionedTopicSenderPartitionedDestinationTopicTest", true, true, false, false, true, true, true); + } + + @Test + public void sameSubscriptionsTransactionalReceiversTest() throws Exception { + sameEntityTransactionalReceiversTest("sameSubscriptionsTransactionalReceiversTest", true); + } + + @Test + public void differentSubscriptionsOnSameTopicTransactionalReceiversTest() throws Exception { + sameEntityTransactionalReceiversTest("sameSubscriptionsTransactionalReceiversTest", false); + } + + // Test all scenarios of creating a receiver to the destination destination after the send via entity has been already establshed. + @Test + public void DestinationEntityReceiverTest() throws Exception { + String expectedErrMsg = "Local transactions cannot span multiple top-level entities such as queue or topic."; + + for (int i = 0; i < 8; i++) { + boolean isViaEntityQueue = (i & 4) > 0; + boolean isDestinationEntityQueue = (i & 2) > 0; + boolean isViaSender = (i & 1) > 0; + + System.out.println("Creating transaction"); + this.transaction = factory.startTransactionAsync().get(); + System.out.println("Created transaction"); + try { + System.out.println(String.format( + "Running DestinationReceiverTest with isViaEntityQueue=%s, isDestinationEntityQueue=%s, isViaSender=%s", + String.valueOf(isViaEntityQueue), + String.valueOf(isDestinationEntityQueue), + String.valueOf(isViaSender))); + + sendviaTest("DestinationReceiverTest-" + i, false, false, isViaEntityQueue, isDestinationEntityQueue, isViaSender, false, true); + fail(String.format("Should have thrown exception saying '%s'", expectedErrMsg)); + } catch (Exception e) { + assertTrue(String.format("Did not get the expected error message. Expected: %s. Actual: %s", expectedErrMsg, e.getMessage()), e.getMessage().contains(expectedErrMsg)); + System.out.println("Failed with the expected exception."); + } + } + } + + private void createEntities(boolean isViaEntityQueue, boolean isDestinationEntityQueue, boolean isViaEntityPartitioned, boolean isDestinationEntityPartitioned) { + if (isViaEntityQueue) { + QueueDescription qd = new QueueDescription(this.viaEntitySendPath); + qd.setDefaultMessageTimeToLive(DEFAULT_MESSAGE_TTL); + qd.setEnablePartitioning(isViaEntityPartitioned); + managementClient.createQueueAsync(qd).join(); + this.viaEntityReceivePath = this.viaEntitySendPath; + } else { + TopicDescription td = new TopicDescription(this.viaEntitySendPath); + td.setDefaultMessageTimeToLive(DEFAULT_MESSAGE_TTL); + td.setEnablePartitioning(isViaEntityPartitioned); + managementClient.createTopicAsync(td).join(); + SubscriptionDescription sd = managementClient.createSubscriptionAsync(this.viaEntitySendPath, "ViaSubscription-" + guid).join(); + this.viaEntityReceivePath = sd.getPath(); + } + + if (isDestinationEntityQueue) { + QueueDescription qd = new QueueDescription(this.destinationEntitySendPath); + qd.setDefaultMessageTimeToLive(DEFAULT_MESSAGE_TTL); + qd.setEnablePartitioning(isDestinationEntityPartitioned); + managementClient.createQueueAsync(qd).join(); + this.destinationEntityReceivePath = this.destinationEntitySendPath; + } else { + TopicDescription td = new TopicDescription(this.destinationEntitySendPath); + td.setDefaultMessageTimeToLive(DEFAULT_MESSAGE_TTL); + td.setEnablePartitioning(isDestinationEntityPartitioned); + managementClient.createTopicAsync(td).join(); + SubscriptionDescription sd = managementClient.createSubscriptionAsync(destinationEntitySendPath, "DestinationSubscription-" + guid).join(); + this.destinationEntityReceivePath = sd.getPath(); + } + } + + // No need to create test sender for destination entities because transacted receivers are not allowed for destination entities + private void createTestSendersAndReceivers(boolean isViaSender, boolean isDestinationSender) { + System.out.println("Creating test senders and receivers"); + this.destinationEntityTestReceiver = ClientFactory.createMessageReceiverFromEntityPathAsync(factory, this.destinationEntityReceivePath, ReceiveMode.PEEKLOCK).join(); + this.viaEntityTestReceiver = ClientFactory.createMessageReceiverFromEntityPathAsync(factory, this.viaEntityReceivePath, ReceiveMode.PEEKLOCK).join(); + this.viaEntityTestSender = ClientFactory.createMessageSenderFromEntityPathAsync(factory, this.viaEntitySendPath, null).join(); + System.out.println("Created test senders and receivers"); + } + + private void createTransactedSendersAndReceivers(boolean isViaSender, boolean isDestinationSender) { + if (isViaSender) { + System.out.println("Creating transacted sender on the via entity"); + this.viaEntitySender = ClientFactory.createTransactedMessageSenderFromEntityPathAsync(factory, this.viaEntitySendPath, this.transaction).join(); + System.out.println("Created transacted sender on the via entity"); + } else { + System.out.println("Creating transacted receiver on the via entity"); + this.viaEntityReceiver = ClientFactory.createTransactedMessageReceiverFromEntityPathAsync(factory, this.viaEntityReceivePath, this.transaction, ReceiveMode.PEEKLOCK).join(); + System.out.println("Created transacted receiver on the via entity"); + } + + if (isDestinationSender) { + System.out.println("Creating transacted sender on the destination entity"); + this.destinationEntitySender = ClientFactory.createTransactedMessageSenderFromEntityPathAsync(factory, this.destinationEntitySendPath, this.transaction).join(); + System.out.println("Created transacted sender on the destination entity"); + } else { + System.out.println("Creating transacted receiver on the destination entity. Should fail."); + ClientFactory.createTransactedMessageReceiverFromEntityPathAsync(factory, this.destinationEntityReceivePath, this.transaction, ReceiveMode.PEEKLOCK).join(); + System.out.println("Created transacted receiver on the destination entity"); + } + } + + public void sameEntityTransactionalReceiversTest(String testName, boolean sameSubscriptions) throws Exception { + TopicDescription td = new TopicDescription(testName + guid); + td.setDefaultMessageTimeToLive(DEFAULT_MESSAGE_TTL); + managementClient.createTopicAsync(td).join(); + SubscriptionDescription sd = managementClient.createSubscriptionAsync(td.getPath(), testName).join(); + SubscriptionDescription sd2 = sameSubscriptions ? sd : managementClient.createSubscriptionAsync(td.getPath(), testName + "2").join(); + + System.out.println("Creating test sender"); + IMessageSender testSender = ClientFactory.createMessageSenderFromEntityPathAsync(factory, td.getPath(), null).join(); + System.out.println("Creating first receiver"); + IMessageReceiver transactedReceiver = ClientFactory.createTransactedMessageReceiverFromEntityPathAsync(factory, sd.getPath(), this.transaction, ReceiveMode.PEEKLOCK).join(); + System.out.println("Creating second receiver"); + IMessageReceiver transactedReceiver2 = ClientFactory.createTransactedMessageReceiverFromEntityPathAsync(factory, sd2.getPath(), this.transaction, ReceiveMode.PEEKLOCK).join(); + + com.microsoft.azure.servicebus.Message message = new com.microsoft.azure.servicebus.Message("my message"); + message.setTimeToLive(DEFAULT_MESSAGE_TTL); + testSender.send(message); + if (sameSubscriptions) { + testSender.send(message); + } + + IMessage received = transactedReceiver.receive(DEFAULT_TIMEOUT); + assertNotNull(received); + transactedReceiver.complete(received.getLockToken()); + IMessage received2 = transactedReceiver2.receive(DEFAULT_TIMEOUT); + assertNotNull(received2); + transactedReceiver2.complete(received2.getLockToken()); + + System.out.println("Committing transaction"); + this.transaction.commit(); + System.out.println("Committed transaction"); + + // Just test that there is no more messages + assertNull(transactedReceiver.peek()); + assertNull(transactedReceiver2.peek()); + } + + private void sendviaTest( + String testName, + boolean isViaEntityPartitioned, + boolean isDestinationEntityPartitioned, + boolean isViaEntityQueue, + boolean isDestinationEntityQueue, + boolean isViaSender, + boolean isDestinationSender, + boolean isCommit) throws Exception { + + this.viaEntitySendPath = testName + "-Via-" + guid; + this.destinationEntitySendPath = testName + "-Destination-" + guid; + + try { + this.createEntities(isViaEntityQueue, isDestinationEntityQueue, isViaEntityPartitioned, isDestinationEntityPartitioned); + this.createTestSendersAndReceivers(isViaSender, isDestinationSender); + this.createTransactedSendersAndReceivers(isViaSender, isDestinationSender); + + com.microsoft.azure.servicebus.Message destinationEntityMessage = new com.microsoft.azure.servicebus.Message("message for destination entity"); + com.microsoft.azure.servicebus.Message viaEntityMessage = new com.microsoft.azure.servicebus.Message("message for via entity"); + destinationEntityMessage.setTimeToLive(DEFAULT_MESSAGE_TTL); + viaEntityMessage.setTimeToLive(DEFAULT_MESSAGE_TTL); + destinationEntityMessage.setViaPartitionKey(new String(transaction.getTransactionId().array())); + viaEntityMessage.setViaPartitionKey(new String(transaction.getTransactionId().array())); + + this.destinationEntitySender.send(destinationEntityMessage); + if (isViaSender) { + this.viaEntitySender.send(viaEntityMessage); + } else { + this.viaEntityTestSender.send(viaEntityMessage); + IMessage received = this.viaEntityReceiver.receive(DEFAULT_TIMEOUT); + assertNotNull("The via entity did not receive the message.", received); + this.viaEntityReceiver.complete(received.getLockToken()); + } + + if (isCommit) { + System.out.println("Committing transaction"); + this.transaction.commit(); + System.out.println("Committed transaction"); + + assertNotNull("Should have received message from destination entity.", this.destinationEntityTestReceiver.receive(DEFAULT_TIMEOUT)); + if (isViaSender) { + assertNotNull("Should have received message from via entity.", this.viaEntityTestReceiver.receive(DEFAULT_TIMEOUT)); + } else { + assertNull("Should not have received message from via entity.", this.viaEntityTestReceiver.receive(DEFAULT_TIMEOUT)); + } + } else { + System.out.println("Rolling back transaction"); + this.transaction.rollback(); + System.out.println("Rolled back transaction"); + + assertNull("Should not have received message from destination entity.", this.destinationEntityTestReceiver.receive(DEFAULT_TIMEOUT)); + if (isViaSender) { + assertNull("Should not have received message from via entity.", this.viaEntityTestReceiver.receive(DEFAULT_TIMEOUT)); + } else { + assertNotNull("Should have received message from via entity.", this.viaEntityTestReceiver.peek()); + } + } + } finally { + // No need to wait for the delete CompletableFutures, if exception happens nothing we can do about them. + if (isViaEntityQueue) { + managementClient.deleteQueueAsync(this.viaEntitySendPath); + } else { + managementClient.deleteTopicAsync(this.viaEntitySendPath); + } + + if (isDestinationEntityQueue) { + managementClient.deleteQueueAsync(this.destinationEntitySendPath); + } else { + managementClient.deleteQueueAsync(this.destinationEntitySendPath); + } + } + } +}