diff --git a/sdk/core/azure-core-amqp/CHANGELOG.md b/sdk/core/azure-core-amqp/CHANGELOG.md index 95f3244eaa11..db88e271f4f7 100644 --- a/sdk/core/azure-core-amqp/CHANGELOG.md +++ b/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. diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java index 416286ace200..0c38ed2b5d3c 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java +++ b/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 createRequestResponseChan new ClientLogger(RequestResponseChannel.class + ":" + entityPath))); } - Mono dispose(AmqpShutdownSignal shutdownSignal) { + Mono 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(); diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSession.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSession.java index 8ec8656e9eee..5d58e2bd7592 100644 --- a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSession.java +++ b/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 isClosed() { return isClosedMono.asMono(); } - Mono dispose(String message, ErrorCondition errorCondition, boolean disposeLinks) { + Mono 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(); } /** diff --git a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionTest.java b/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionTest.java index 6f2083faeeec..e2016cca0998 100644 --- a/sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionTest.java +++ b/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(); } diff --git a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ManagementChannel.java b/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ManagementChannel.java index e9a3e1853893..9fd59529d17e 100644 --- a/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ManagementChannel.java +++ b/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,9 +159,9 @@ private Mono getProperties(Map properties, Class 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); @@ -169,7 +169,7 @@ private Mono getProperties(Map properties, Class respo final Throwable error = ExceptionUtil.amqpResponseCodeToException(statusCode.getValue(), statusDescription, channel.getErrorContext()); - throw logger.logExceptionAsWarning(Exceptions.propagate(error)); + sink.error(logger.logExceptionAsWarning(Exceptions.propagate(error))); })); }); }