Fix subscription bugs in ReactorSession and ReactorConnection

sdk/core/azure-core-amqp/CHANGELOG.md

@@ -2,6 +2,12 @@
 
 ## 2.1.0-beta.1 (Unreleased)
 
+### New Features
+- Exposing CbsAuthorizationType.
+
+### Bug Fixes
+- Fixed a bug where connection and sessions would not be disposed when their endpoint closed.
+
 ## 2.0.6 (2021-05-24)
 ### Bug Fixes
 - Fixed a bug that caused amqp connection not to retry when network error happened.

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java

@@ -120,7 +120,7 @@ public ReactorConnection(String connectionId, ConnectionOptions connectionOption
                 if (isDisposed.getAndSet(true)) {
                     logger.verbose("connectionId[{}] was already disposed. {}", connectionId, message);
                 } else {
-                    dispose(new AmqpShutdownSignal(false, false, message));
+                    closeAsync(new AmqpShutdownSignal(false, false, message)).subscribe();
                 }
             });
 
@@ -133,7 +133,7 @@ public ReactorConnection(String connectionId, ConnectionOptions connectionOption
             .onErrorResume(error -> {
                 if (!isDisposed.getAndSet(true)) {
                     logger.verbose("connectionId[{}]: Disposing of active sessions due to error.", connectionId);
-                    return dispose(new AmqpShutdownSignal(false, false,
+                    return closeAsync(new AmqpShutdownSignal(false, false,
                         error.getMessage())).then(Mono.error(error));
                 } else {
                     return Mono.error(error);
@@ -144,7 +144,7 @@ public ReactorConnection(String connectionId, ConnectionOptions connectionOption
                     logger.verbose("connectionId[{}]: Disposing of active sessions due to connection close.",
                         connectionId);
 
-                    dispose(new AmqpShutdownSignal(false, false,
+                    closeAsync(new AmqpShutdownSignal(false, false,
                         "Connection handler closed.")).subscribe();
                 }
             })
@@ -310,7 +310,7 @@ public void dispose() {
         // Because the reactor executor schedules the pending close after the timeout, we want to give sufficient time
         // for the rest of the tasks to run.
         final Duration timeout = operationTimeout.plus(operationTimeout);
-        dispose(new AmqpShutdownSignal(false, true, "Disposed by client."))
+        closeAsync(new AmqpShutdownSignal(false, true, "Disposed by client."))
             .publishOn(Schedulers.boundedElastic())
             .block(timeout);
     }
@@ -356,7 +356,7 @@ protected AmqpChannelProcessor<RequestResponseChannel> createRequestResponseChan
                 new ClientLogger(RequestResponseChannel.class + ":" + entityPath)));
     }
 
-    Mono<Void> dispose(AmqpShutdownSignal shutdownSignal) {
+    Mono<Void> closeAsync(AmqpShutdownSignal shutdownSignal) {
         logger.info("connectionId[{}] signal[{}]: Disposing of ReactorConnection.", connectionId, shutdownSignal);
 
         if (cbsChannelProcessor != null) {
@@ -494,7 +494,7 @@ public void onConnectionError(Throwable exception) {
             if (!isDisposed.getAndSet(true)) {
                 logger.verbose("onReactorError connectionId[{}], hostName[{}]: Disposing.", connectionId,
                     getFullyQualifiedNamespace());
-                dispose(new AmqpShutdownSignal(false, false,
+                closeAsync(new AmqpShutdownSignal(false, false,
                     "onReactorError: " + exception.toString()))
                     .subscribe();
             }
@@ -508,7 +508,7 @@ void onConnectionShutdown(AmqpShutdownSignal shutdownSignal) {
 
             if (!isDisposed.getAndSet(true)) {
                 logger.verbose("onConnectionShutdown connectionId[{}], hostName[{}]: disposing.");
-                dispose(shutdownSignal).subscribe();
+                closeAsync(shutdownSignal).subscribe();
             }
         }
     }
@@ -533,7 +533,7 @@ private void dispose() {
             }
 
             if (session instanceof ReactorSession) {
-                ((ReactorSession) session).dispose("Closing session.", null, true)
+                ((ReactorSession) session).closeAsync("Closing session.", null, true)
                     .subscribe();
             } else {
                 session.dispose();

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSession.java

@@ -128,7 +128,7 @@ public ReactorSession(AmqpConnection amqpConnection, Session session, SessionHan
         connectionSubscriptions = Disposables.composite(
             this.endpointStates.subscribe(),
 
-            shutdownSignals.flatMap(signal ->  dispose("Shutdown signal received", null, false)).subscribe());
+            shutdownSignals.flatMap(signal ->  closeAsync("Shutdown signal received", null, false)).subscribe());
 
         session.open();
     }
@@ -152,7 +152,7 @@ public boolean isDisposed() {
      */
     @Override
     public void dispose() {
-        dispose("Dispose called.", null, true)
+        closeAsync("Dispose called.", null, true)
             .block(retryOptions.getTryTimeout());
     }
 
@@ -240,7 +240,7 @@ Mono<Void> isClosed() {
         return isClosedMono.asMono();
     }
 
-    Mono<Void> dispose(String message, ErrorCondition errorCondition, boolean disposeLinks) {
+    Mono<Void> closeAsync(String message, ErrorCondition errorCondition, boolean disposeLinks) {
         if (isDisposed.getAndSet(true)) {
             return isClosedMono.asMono();
         }
@@ -596,7 +596,7 @@ private void handleClose() {
             "connectionId[{}] sessionName[{}] Disposing of active send and receive links due to session close.",
             sessionHandler.getConnectionId(), sessionName);
 
-        dispose("", null, true);
+        closeAsync("", null, true).subscribe();
     }
 
     private void handleError(Throwable error) {
@@ -610,12 +610,12 @@ private void handleError(Throwable error) {
 
             condition = new ErrorCondition(Symbol.getSymbol(errorCondition), exception.getMessage());
 
-            dispose(exception.getMessage(), condition, true);
+            closeAsync(exception.getMessage(), condition, true).subscribe();
         } else {
             condition = null;
         }
 
-        dispose(error.getMessage(), condition, true);
+        closeAsync(error.getMessage(), condition, true).subscribe();
     }
 
     /**

sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionTest.java

@@ -680,12 +680,12 @@ void disposeAsync() throws IOException {
         connection2.getReactorConnection().subscribe();
 
         // Act and Assert
-        StepVerifier.create(connection2.dispose(signal))
+        StepVerifier.create(connection2.closeAsync(signal))
             .verifyComplete();
 
         assertTrue(connection2.isDisposed());
 
-        StepVerifier.create(connection2.dispose(signal))
+        StepVerifier.create(connection2.closeAsync(signal))
             .verifyComplete();
     }
 

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ManagementChannel.java

@@ -95,7 +95,7 @@ public class ManagementChannel implements EventHubManagementNode {
 
         //@formatter:off
         this.subscription = responseChannelMono
-            .flatMapMany(e -> e.getEndpointStates().distinct())
+            .flatMapMany(e -> e.getEndpointStates().distinctUntilChanged())
             .subscribe(e -> {
                 logger.info("Management endpoint state: {}", e);
                 endpointStateSink.next(e);
@@ -159,17 +159,17 @@ private <T> Mono<T> getProperties(Map<String, Object> properties, Class<T> respo
             request.setApplicationProperties(applicationProperties);
 
             return channelMono.flatMap(channel -> channel.sendWithAck(request)
-                .map(message -> {
+                .handle((message, sink) -> {
                     if (RequestResponseUtils.isSuccessful(message)) {
-                        return messageSerializer.deserialize(message, responseType);
+                        sink.next(messageSerializer.deserialize(message, responseType));
                     }
 
                     final AmqpResponseCode statusCode = RequestResponseUtils.getStatusCode(message);
                     final String statusDescription = RequestResponseUtils.getStatusDescription(message);
                     final Throwable error = ExceptionUtil.amqpResponseCodeToException(statusCode.getValue(),
                         statusDescription, channel.getErrorContext());
 
-                    throw logger.logExceptionAsWarning(Exceptions.propagate(error));
+                    sink.error(logger.logExceptionAsWarning(Exceptions.propagate(error)));
                 }));
         });
     }