Add ServiceBus Session Receiver Client

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ReceiverOptions.java

@@ -3,7 +3,6 @@
 
 package com.azure.messaging.servicebus;
 
-import com.azure.core.util.CoreUtils;
 import com.azure.messaging.servicebus.models.ReceiveMode;
 
 import java.time.Duration;
@@ -16,28 +15,22 @@ class ReceiverOptions {
     private final int prefetchCount;
     private final boolean enableAutoComplete;
     private final String sessionId;
-    private final boolean isRollingSessionReceiver;
     private final Integer maxConcurrentSessions;
-    private final boolean isSessionReceiver;
     private final Duration maxLockRenewDuration;
 
     ReceiverOptions(ReceiveMode receiveMode, int prefetchCount, Duration maxLockRenewDuration,
         boolean enableAutoComplete) {
-        this(receiveMode, prefetchCount, maxLockRenewDuration, enableAutoComplete, null, false, null);
+        this(receiveMode, prefetchCount, maxLockRenewDuration, enableAutoComplete, null, null);
     }
 
     ReceiverOptions(ReceiveMode receiveMode, int prefetchCount, Duration maxLockRenewDuration,
-        boolean enableAutoComplete, String sessionId, boolean isRollingSessionReceiver,
-        Integer maxConcurrentSessions) {
-
+        boolean enableAutoComplete, String sessionId, Integer maxConcurrentSessions) {
         this.receiveMode = receiveMode;
         this.prefetchCount = prefetchCount;
         this.enableAutoComplete = enableAutoComplete;
         this.sessionId = sessionId;
-        this.isRollingSessionReceiver = isRollingSessionReceiver;
         this.maxConcurrentSessions = maxConcurrentSessions;
         this.maxLockRenewDuration = maxLockRenewDuration;
-        this.isSessionReceiver = !CoreUtils.isNullOrEmpty(sessionId) || isRollingSessionReceiver;
     }
 
     /**
@@ -90,7 +83,7 @@ boolean isAutoLockRenewEnabled() {
      * @return true if it is a session-aware receiver; false otherwise.
      */
     boolean isSessionReceiver() {
-        return isSessionReceiver;
+        return sessionId != null || maxConcurrentSessions != null;
     }
 
     /**
@@ -100,7 +93,7 @@ boolean isSessionReceiver() {
      *     false} otherwise.
      */
     public boolean isRollingSessionReceiver() {
-        return isRollingSessionReceiver;
+        return maxConcurrentSessions != null && maxConcurrentSessions > 0 && sessionId == null;
     }
 
     /**

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusClientBuilder.java

@@ -624,11 +624,9 @@ public ServiceBusSenderClient buildClient() {
     @ServiceClientBuilder(serviceClients = {ServiceBusReceiverClient.class, ServiceBusReceiverAsyncClient.class})
     public final class ServiceBusSessionReceiverClientBuilder {
         private boolean enableAutoComplete = true;
-        private Integer maxConcurrentSessions = null;
         private int prefetchCount = DEFAULT_PREFETCH_COUNT;
         private String queueName;
         private ReceiveMode receiveMode = ReceiveMode.PEEK_LOCK;
-        private String sessionId;
         private String subscriptionName;
         private String topicName;
         private Duration maxAutoLockRenewDuration = MAX_LOCK_RENEW_DEFAULT_DURATION;
@@ -666,24 +664,6 @@ public ServiceBusSessionReceiverClientBuilder maxAutoLockRenewDuration(Duration
             return this;
         }
 
-        /**
-         * Enables session processing roll-over by processing at most {@code maxConcurrentSessions}.
-         *
-         * @param maxConcurrentSessions Maximum number of concurrent sessions to process at any given time.
-         *
-         * @return The modified {@link ServiceBusSessionReceiverClientBuilder} object.
-         * @throws IllegalArgumentException if {@code maxConcurrentSessions} is less than 1.
-         */
-        public ServiceBusSessionReceiverClientBuilder maxConcurrentSessions(int maxConcurrentSessions) {
-            if (maxConcurrentSessions < 1) {
-                throw logger.logExceptionAsError(new IllegalArgumentException(
-                    "maxConcurrentSessions cannot be less than 1."));
-            }
-
-            this.maxConcurrentSessions = maxConcurrentSessions;
-            return this;
-        }
-
         /**
          * Sets the prefetch count of the receiver. For both {@link ReceiveMode#PEEK_LOCK PEEK_LOCK} and {@link
          * ReceiveMode#RECEIVE_AND_DELETE RECEIVE_AND_DELETE} modes the default value is 1.
@@ -728,18 +708,6 @@ public ServiceBusSessionReceiverClientBuilder receiveMode(ReceiveMode receiveMod
             return this;
         }
 
-        /**
-         * Sets the session id.
-         *
-         * @param sessionId session id.
-         *
-         * @return The modified {@link ServiceBusSessionReceiverClientBuilder} object.
-         */
-        public ServiceBusSessionReceiverClientBuilder sessionId(String sessionId) {
-            this.sessionId = sessionId;
-            return this;
-        }
-
         /**
          * Sets the name of the subscription in the topic to listen to. <b>{@link #topicName(String)} must also be set.
          * </b>
@@ -771,7 +739,7 @@ public ServiceBusSessionReceiverClientBuilder topicName(String topicName) {
          * Creates an <b>asynchronous</b>, <b>session-aware</b> Service Bus receiver responsible for reading {@link
          * ServiceBusMessage messages} from a specific queue or topic.
          *
-         * @return An new {@link ServiceBusReceiverAsyncClient} that receives messages from a queue or topic.
+         * @return An new {@link ServiceBusSessionReceiverAsyncClient} that receives messages from a queue or topic.
          * @throws IllegalStateException if {@link #queueName(String) queueName} or {@link #topicName(String)
          *     topicName} are not set or, both of these fields are set. It is also thrown if the Service Bus {@link
          *     #connectionString(String) connectionString} contains an {@code EntityPath} that does not match one set in
@@ -780,7 +748,7 @@ public ServiceBusSessionReceiverClientBuilder topicName(String topicName) {
          * @throws IllegalArgumentException Queue or topic name are not set via {@link #queueName(String)
          *     queueName()} or {@link #topicName(String) topicName()}, respectively.
          */
-        public ServiceBusReceiverAsyncClient buildAsyncClient() {
+        public ServiceBusSessionReceiverAsyncClient buildAsyncClient() {
             return buildAsyncClient(true);
         }
 
@@ -797,11 +765,11 @@ public ServiceBusReceiverAsyncClient buildAsyncClient() {
          * @throws IllegalArgumentException Queue or topic name are not set via {@link #queueName(String)
          *     queueName()} or {@link #topicName(String) topicName()}, respectively.
          */
-        public ServiceBusReceiverClient buildClient() {
-            return new ServiceBusReceiverClient(buildAsyncClient(false), retryOptions.getTryTimeout());
+        public ServiceBusSessionReceiverClient buildClient() {
+            return new ServiceBusSessionReceiverClient(buildAsyncClient(false));
         }
 
-        private ServiceBusReceiverAsyncClient buildAsyncClient(boolean isAutoCompleteAllowed) {
+        private ServiceBusSessionReceiverAsyncClient buildAsyncClient(boolean isAutoCompleteAllowed) {
             final MessagingEntityType entityType = validateEntityPaths(logger, connectionStringEntityName, topicName,
                 queueName);
             final String entityPath = getEntityPath(logger, entityType, queueName, topicName, subscriptionName,
@@ -822,34 +790,11 @@ private ServiceBusReceiverAsyncClient buildAsyncClient(boolean isAutoCompleteAll
 
             final ServiceBusConnectionProcessor connectionProcessor = getOrCreateConnectionProcessor(messageSerializer);
             final ReceiverOptions receiverOptions = new ReceiverOptions(receiveMode, prefetchCount,
-                maxAutoLockRenewDuration, enableAutoComplete, sessionId, isRollingSessionReceiver(),
-                maxConcurrentSessions);
-
-            final ServiceBusSessionManager sessionManager = new ServiceBusSessionManager(entityPath, entityType,
-                connectionProcessor, tracerProvider, messageSerializer, receiverOptions);
-
-            return new ServiceBusReceiverAsyncClient(connectionProcessor.getFullyQualifiedNamespace(), entityPath,
-                entityType, receiverOptions, connectionProcessor, ServiceBusConstants.OPERATION_TIMEOUT,
-                tracerProvider, messageSerializer, ServiceBusClientBuilder.this::onClientClose, sessionManager);
-        }
-
-        /**
-         * This is a rolling session receiver only if maxConcurrentSessions is > 0 AND sessionId is null or empty. If
-         * there is a sessionId, this is going to be a single, named session receiver.
-         *
-         * @return {@code true} if this is an unnamed rolling session receiver; {@code false} otherwise.
-         */
-        private boolean isRollingSessionReceiver() {
-            if (maxConcurrentSessions == null) {
-                return false;
-            }
-
-            if (maxConcurrentSessions < 1) {
-                throw logger.logExceptionAsError(
-                    new IllegalArgumentException("Maximum number of concurrent sessions must be positive."));
-            }
+                maxAutoLockRenewDuration, enableAutoComplete, null, null);
 
-            return CoreUtils.isNullOrEmpty(sessionId);
+            return new ServiceBusSessionReceiverAsyncClient(connectionProcessor.getFullyQualifiedNamespace(),
+                entityPath, entityType, receiverOptions, connectionProcessor, tracerProvider, messageSerializer,
+                ServiceBusClientBuilder.this::onClientClose);
         }
     }
 

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java

@@ -63,21 +63,17 @@
  * {@codesnippet com.azure.messaging.servicebus.servicebusasyncreceiverclient.receiveWithReceiveAndDeleteMode}
  *
  * <p><strong>Receive messages from a specific session</strong></p>
- * <p>To fetch messages from a specific session, set {@link ServiceBusSessionReceiverClientBuilder#sessionId(String)}.
+ * <p>To fetch messages from a specific session, switch to {@link ServiceBusSessionReceiverClientBuilder} and
+ * build the session receiver client. Use {@link ServiceBusSessionReceiverAsyncClient#acceptSession(String)} to create a
+ * session-bound {@link ServiceBusReceiverAsyncClient}.
  * </p>
  * {@codesnippet com.azure.messaging.servicebus.servicebusasyncreceiverclient.instantiation#sessionId}
  *
- * <p><strong>Process messages from multiple sessions</strong></p>
- * <p>To process messages from multiple sessions, set
- * {@link ServiceBusSessionReceiverClientBuilder#maxConcurrentSessions(int)}. This will process in parallel at most
- * {@code maxConcurrentSessions}. In addition, when all the messages in a session have been consumed, it will find the
- * next available session to process.</p>
- * {@codesnippet com.azure.messaging.servicebus.servicebusasyncreceiverclient.instantiation#multiplesessions}
- *
  * <p><strong>Process messages from the first available session</strong></p>
  * <p>To process messages from the first available session, switch to {@link ServiceBusSessionReceiverClientBuilder} and
- * build the receiver client. It will find the first available session to process messages from.</p>
- * {@codesnippet com.azure.messaging.servicebus.servicebusasyncreceiverclient.instantiation#singlesession}
+ * build the session receiver client. Use {@link ServiceBusSessionReceiverAsyncClient#acceptNextSession()} to
+ * find the first available session to process messages from.</p>
+ * {@codesnippet com.azure.messaging.servicebus.servicebusasyncreceiverclient.instantiation#nextsession}
  *
  * <p><strong>Rate limiting consumption of messages from Service Bus resource</strong></p>
  * <p>For message receivers that need to limit the number of messages they receive at a given time, they can use
@@ -390,6 +386,11 @@ public Mono<byte[]> getSessionState(String sessionId) {
         }
 
         if (sessionManager != null) {
+            if (validateSessionId(sessionId)) {
+                return monoError(logger, new IllegalArgumentException(String.format(
+                    "This receiver client is tied to session %s. It can't be used for another session(%s).",
+                    receiverOptions.getSessionId(), sessionId)));
+            }
             return sessionManager.getSessionState(sessionId);
         } else {
             return connectionProcessor
@@ -426,7 +427,11 @@ public Mono<ServiceBusReceivedMessage> peekMessage(String sessionId) {
             return monoError(logger, new IllegalStateException(
                 String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "peek")));
         }
-
+        if (validateSessionId(sessionId)) {
+            return monoError(logger, new IllegalArgumentException(String.format(
+                "This receiver client is tied to session %s. It can't be used for another session(%s).",
+                receiverOptions.getSessionId(), sessionId)));
+        }
         return connectionProcessor
             .flatMap(connection -> connection.getManagementNode(entityPath, entityType))
             .flatMap(channel -> {
@@ -472,7 +477,11 @@ public Mono<ServiceBusReceivedMessage> peekMessageAt(long sequenceNumber, String
             return monoError(logger, new IllegalStateException(
                 String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "peekAt")));
         }
-
+        if (validateSessionId(sessionId)) {
+            return monoError(logger, new IllegalArgumentException(String.format(
+                "This receiver client is tied to session %s. It can't be used for another session(%s).",
+                receiverOptions.getSessionId(), sessionId)));
+        }
         return connectionProcessor
             .flatMap(connection -> connection.getManagementNode(entityPath, entityType))
             .flatMap(node -> node.peek(sequenceNumber, sessionId, getLinkName(sessionId)));
@@ -506,7 +515,11 @@ public Flux<ServiceBusReceivedMessage> peekMessages(int maxMessages, String sess
             return fluxError(logger, new IllegalStateException(
                 String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "peekBatch")));
         }
-
+        if (validateSessionId(sessionId)) {
+            return fluxError(logger, new IllegalArgumentException(String.format(
+                "This receiver client is tied to session %s. It can't be used for another session(%s).",
+                receiverOptions.getSessionId(), sessionId)));
+        }
         return connectionProcessor
             .flatMap(connection -> connection.getManagementNode(entityPath, entityType))
             .flatMapMany(node -> {
@@ -569,7 +582,11 @@ public Flux<ServiceBusReceivedMessage> peekMessagesAt(int maxMessages, long sequ
             return fluxError(logger, new IllegalStateException(
                 String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "peekBatchAt")));
         }
-
+        if (validateSessionId(sessionId)) {
+            return fluxError(logger, new IllegalArgumentException(String.format(
+                "This receiver client is tied to session %s. It can't be used for another session(%s).",
+                receiverOptions.getSessionId(), sessionId)));
+        }
         return connectionProcessor
             .flatMap(connection -> connection.getManagementNode(entityPath, entityType))
             .flatMapMany(node -> node.peek(sequenceNumber, sessionId, getLinkName(sessionId), maxMessages));
@@ -638,6 +655,11 @@ public Mono<ServiceBusReceivedMessage> receiveDeferredMessage(long sequenceNumbe
      * @return A deferred message with the matching {@code sequenceNumber}.
      */
     public Mono<ServiceBusReceivedMessage> receiveDeferredMessage(long sequenceNumber, String sessionId) {
+        if (validateSessionId(sessionId)) {
+            return monoError(logger, new IllegalArgumentException(String.format(
+                "This receiver client is tied to session %s. It can't be used for another session(%s).",
+                receiverOptions.getSessionId(), sessionId)));
+        }
         return connectionProcessor
             .flatMap(connection -> connection.getManagementNode(entityPath, entityType))
             .flatMap(node -> node.receiveDeferredMessages(receiverOptions.getReceiveMode(),
@@ -683,7 +705,11 @@ public Flux<ServiceBusReceivedMessage> receiveDeferredMessages(Iterable<Long> se
             return fluxError(logger, new IllegalStateException(
                 String.format(INVALID_OPERATION_DISPOSED_RECEIVER, "receiveDeferredMessageBatch")));
         }
-
+        if (validateSessionId(sessionId)) {
+            return fluxError(logger, new IllegalArgumentException(String.format(
+                "This receiver client is tied to session %s. It can't be used for another session(%s).",
+                receiverOptions.getSessionId(), sessionId)));
+        }
         return connectionProcessor
             .flatMap(connection -> connection.getManagementNode(entityPath, entityType))
             .flatMapMany(node -> node.receiveDeferredMessages(receiverOptions.getReceiveMode(),
@@ -808,7 +834,11 @@ public Mono<OffsetDateTime> renewSessionLock(String sessionId) {
         } else if (!receiverOptions.isSessionReceiver()) {
             return monoError(logger, new IllegalStateException("Cannot renew session lock on a non-session receiver."));
         }
-
+        if (validateSessionId(sessionId)) {
+            return monoError(logger, new IllegalArgumentException(String.format(
+                "This receiver client is tied to session %s. It can't be used for another session(%s).",
+                receiverOptions.getSessionId(), sessionId)));
+        }
         final String linkName = sessionManager != null
             ? sessionManager.getLinkName(sessionId)
             : null;
@@ -846,7 +876,11 @@ public Mono<Void> renewSessionLock(String sessionId, Duration maxLockRenewalDura
         } else if (sessionId.isEmpty()) {
             return monoError(logger, new IllegalArgumentException("'sessionId' cannot be empty."));
         }
-
+        if (validateSessionId(sessionId)) {
+            return monoError(logger, new IllegalArgumentException(String.format(
+                "This receiver client is tied to session %s. It can't be used for another session(%s).",
+                receiverOptions.getSessionId(), sessionId)));
+        }
         final LockRenewalOperation operation = new LockRenewalOperation(sessionId, maxLockRenewalDuration, true,
             this::renewSessionLock);
 
@@ -870,7 +904,11 @@ public Mono<Void> setSessionState(String sessionId, byte[] sessionState) {
         } else if (!receiverOptions.isSessionReceiver()) {
             return monoError(logger, new IllegalStateException("Cannot set session state on a non-session receiver."));
         }
-
+        if (validateSessionId(sessionId)) {
+            return monoError(logger, new IllegalArgumentException(String.format(
+                "This receiver client is tied to session %s. It can't be used for another session(%s).",
+                receiverOptions.getSessionId(), sessionId)));
+        }
         final String linkName = sessionManager != null
             ? sessionManager.getLinkName(sessionId)
             : null;
@@ -1136,4 +1174,9 @@ private String getLinkName(String sessionId) {
             return existing != null ? existing.getLinkName() : null;
         }
     }
+
+    private boolean validateSessionId(String sessionId) {
+        String optionSessionId = receiverOptions.getSessionId();
+        return optionSessionId != null && !optionSessionId.equals(sessionId);
+    }
 }

sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverClient.java

@@ -53,6 +53,14 @@ public final class ServiceBusReceiverClient implements AutoCloseable {
         this.operationTimeout = Objects.requireNonNull(operationTimeout, "'operationTimeout' cannot be null.");
     }
 
+    // TODO: ServiceBusReceiverClient will remove the above constructor that accepts operationTimeout.
+    //  This one is added because ServiceBusSessionReceiverClient need this constructor right now.
+    //  this.operationTimeout won't be needed after timeout is moved to methods.
+    ServiceBusReceiverClient(ServiceBusReceiverAsyncClient asyncClient) {
+        this.asyncClient = Objects.requireNonNull(asyncClient, "'asyncClient' cannot be null.");
+        this.operationTimeout = Duration.ofDays(10000); // TODO: remove after moving timeout to methods.
+    }
+
     /**
      * Gets the fully qualified Service Bus namespace that the connection is associated with. This is likely similar to
      * {@code {yournamespace}.servicebus.windows.net}.