Use AutoClosable for to ReactorSender, ReactorReceiver and ServiceBus…

…SessionReceiver (Azure#21834)
AzureSDKAutomation · May 26, 2021 · 9d6deb6 · 9d6deb6
1 parent 9201432
commit 9d6deb6
Show file tree

Hide file tree

Showing 5 changed files with 9 additions and 6 deletions.
diff --git a/...e/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java b/...e/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorConnection.java
@@ -108,7 +108,7 @@ public ReactorConnection(String connectionId, ConnectionOptions connectionOption
                 final Mono<AmqpEndpointState> activeEndpoint = getEndpointStates()
                     .filter(state -> state == AmqpEndpointState.ACTIVE)
                     .next()
-                    .timeout(operationTimeout, Mono.error(new AmqpException(false, String.format(
+                    .timeout(operationTimeout, Mono.error(new AmqpException(true, String.format(
                         "Connection '%s' not opened within AmqpRetryOptions.tryTimeout(): %s", connectionId,
                         operationTimeout), handler.getErrorContext())));
                 return activeEndpoint.thenReturn(reactorConnection);

diff --git a/...ore/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorReceiver.java b/...ore/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorReceiver.java
@@ -37,7 +37,7 @@
 /**
  * Handles receiving events from Event Hubs service and translating them to proton-j messages.
  */
-public class ReactorReceiver implements AmqpReceiveLink, AsyncAutoCloseable {
+public class ReactorReceiver implements AmqpReceiveLink, AsyncAutoCloseable, AutoCloseable {
     private final String entityPath;
     private final Receiver receiver;
     private final ReceiveLinkHandler handler;
@@ -225,6 +225,7 @@ public void dispose() {
         close();
     }
 
+    @Override
     public void close() {
         closeAsync().block(retryOptions.getTryTimeout());
     }

diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSender.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSender.java
@@ -62,7 +62,7 @@
 /**
  * Handles scheduling and transmitting events through proton-j to Event Hubs service.
  */
-class ReactorSender implements AmqpSendLink, AsyncAutoCloseable {
+class ReactorSender implements AmqpSendLink, AsyncAutoCloseable, AutoCloseable {
     private final String entityPath;
     private final Sender sender;
     private final SendLinkHandler handler;
@@ -337,6 +337,7 @@ public void dispose() {
         close();
     }
 
+    @Override
     public void close() {
         closeAsync().block(retryOptions.getTryTimeout());
     }

diff --git a/...ure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionTest.java b/...ure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionTest.java
@@ -422,7 +422,7 @@ void createCBSNodeTimeoutException() throws IOException {
                 assertTrue(error instanceof AmqpException);
 
                 final AmqpException amqpException = (AmqpException) error;
-                assertFalse(amqpException.isTransient());
+                assertTrue(amqpException.isTransient());
                 assertNull(amqpException.getErrorCondition());
 
                 assertNotNull(amqpException.getMessage());
@@ -583,7 +583,7 @@ void cannotCreateResourcesOnFailure() {
                 Assertions.fail("Exception was not the correct type: " + error);
             }
 
-            assertFalse(amqpException.isTransient());
+            assertTrue(amqpException.isTransient());
         };
 
         when(event.getTransport()).thenReturn(transport);

diff --git a/...ng-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiver.java b/...ng-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusSessionReceiver.java
@@ -29,7 +29,7 @@
 /**
  * Represents an session that is received when "any" session is accepted from the service.
  */
-class ServiceBusSessionReceiver implements AsyncAutoCloseable {
+class ServiceBusSessionReceiver implements AsyncAutoCloseable, AutoCloseable {
     private final AtomicBoolean isDisposed = new AtomicBoolean();
     private final LockContainer<OffsetDateTime> lockContainer;
     private final AtomicReference<OffsetDateTime> sessionLockedUntil = new AtomicReference<>();
@@ -212,6 +212,7 @@ public Mono<Void> closeAsync() {
         return receiveLink.closeAsync().doFinally(signal -> subscriptions.dispose());
     }
 
+    @Override
     public void close() {
         closeAsync().block(retryOptions.getTryTimeout());
     }