Exposes management node in azure-core-amqp (#22095)

* Update AmqpConnection to have a getManagementNode.

* Adding AmqpManagementNode.

* Update AmqpConnection, AmqpManagementNode, AmqpSession to use AsyncCloseable.

* Adding AsyncCloseable to AmqpLink.

* ClaimsBasedSecurityNode.java uses AsyncCloseable.

* Implements CbsNode's closeAsync() and adds tests.

* ReactorSession implements closeAsync()

* ReactorConnection uses closeAsync(). Renames dispose() to closeAsync(). Fixes errors where some close operations were not subscribed to.

* RequestResponseChannel. Remove close operation with message.

* Adding DeliveryOutcome models and DeliveryState enum.

* Add authorization scope to connection options.

* Add MessageUtils to serialize and deserialize AmqpAnnotatedMessage

* Update AmqpManagementNode to expose delivery outcomes because they can be associated with messages.

* Adding MessageUtil support for converting DeliveryOutcome and Outcomes.

* Fixing build breaks from ConnectionOptions.

* Adding management channel class.

* Adding management channel into ReactorConnection.

* Update ExceptionUtil to return instead of throwing on unknown amqp error codes.

* Moving ManagementChannel formatting.

* Add javadocs to ReceivedDeliveryOutcome.

* Add tests for ManagementChannel

* Adding tests for message utils.

* Fix javadoc on ModifiedDeliveryOutcome

* ReactorConnection: Hook up dispose method.

* EventHubs: Fixing instances of ConnectionOptions.

* ServiceBus: Fix build errors using ConnectionOptions.

* Adding MessageUtilsTests.

* Updating CHANGELOG.

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

@@ -4,6 +4,9 @@
 
 ### New Features
 - Exposing CbsAuthorizationType.
+- Exposing ManagementNode that can perform management and metadata operations on an AMQP message broker.
+- AmqpConnection, AmqpSession, AmqpSendLink, and AmqpReceiveLink extend from AsyncCloseable.
+- Delivery outcomes and delivery states are added.
 
 ### Bug Fixes
 - Fixed a bug where connection and sessions would not be disposed when their endpoint closed.

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/AmqpConnection.java

@@ -4,6 +4,7 @@
 package com.azure.core.amqp;
 
 import com.azure.core.amqp.exception.AmqpException;
+import com.azure.core.util.AsyncCloseable;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -13,7 +14,7 @@
 /**
  * Represents a TCP connection between the client and a service that uses the AMQP protocol.
  */
-public interface AmqpConnection extends Disposable {
+public interface AmqpConnection extends Disposable, AsyncCloseable {
     /**
      * Gets the connection identifier.
      *
@@ -53,6 +54,7 @@ public interface AmqpConnection extends Disposable {
      * Creates a new session with the given session name.
      *
      * @param sessionName Name of the session.
+     *
      * @return The AMQP session that was created.
      */
     Mono<AmqpSession> createSession(String sessionName);
@@ -61,6 +63,7 @@ public interface AmqpConnection extends Disposable {
      * Removes a session with the {@code sessionName} from the AMQP connection.
      *
      * @param sessionName Name of the session to remove.
+     *
      * @return {@code true} if a session with the name was removed; {@code false} otherwise.
      */
     boolean removeSession(String sessionName);
@@ -79,4 +82,26 @@ public interface AmqpConnection extends Disposable {
      * @return A stream of shutdown signals that occur in the AMQP endpoint.
      */
     Flux<AmqpShutdownSignal> getShutdownSignals();
+
+    /**
+     * Gets or creates the management node.
+     *
+     * @param entityPath Entity for which to get the management node of.
+     *
+     * @return A Mono that completes with the management node.
+     *
+     * @throws UnsupportedOperationException if there is no implementation of fetching a management node.
+     */
+    default Mono<AmqpManagementNode> getManagementNode(String entityPath) {
+        return Mono.error(new UnsupportedOperationException("This has not been implemented."));
+    }
+
+    /**
+     * Disposes of the AMQP connection.
+     *
+     * @return Mono that completes when the close operation is complete.
+     */
+    default Mono<Void> closeAsync() {
+        return Mono.fromRunnable(this::dispose);
+    }
 }

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/AmqpLink.java

@@ -4,13 +4,16 @@
 package com.azure.core.amqp;
 
 import com.azure.core.amqp.exception.AmqpException;
+import com.azure.core.util.AsyncCloseable;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 /**
  * Represents a unidirectional AMQP link.
  */
-public interface AmqpLink extends Disposable {
+public interface AmqpLink extends Disposable, AsyncCloseable {
+
     /**
      * Gets the name of the link.
      *
@@ -39,4 +42,13 @@ public interface AmqpLink extends Disposable {
      * @return A stream of endpoint states for the AMQP link.
      */
     Flux<AmqpEndpointState> getEndpointStates();
+
+    /**
+     * Disposes of the AMQP link.
+     *
+     * @return A mono that completes when the link is disposed.
+     */
+    default Mono<Void> closeAsync() {
+        return Mono.fromRunnable(() -> dispose());
+    }
 }

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/AmqpManagementNode.java

@@ -0,0 +1,33 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.amqp;
+
+import com.azure.core.amqp.models.AmqpAnnotatedMessage;
+import com.azure.core.amqp.models.DeliveryOutcome;
+import com.azure.core.util.AsyncCloseable;
+import reactor.core.publisher.Mono;
+
+/**
+ * An AMQP endpoint that allows users to perform management and metadata operations on it.
+ */
+public interface AmqpManagementNode extends AsyncCloseable {
+    /**
+     * Sends a message to the management node.
+     *
+     * @param message Message to send.
+     *
+     * @return Response from management node.
+     */
+    Mono<AmqpAnnotatedMessage> send(AmqpAnnotatedMessage message);
+
+    /**
+     * Sends a message to the management node and associates the {@code deliveryOutcome} with that message.
+     *
+     * @param message Message to send.
+     * @param deliveryOutcome Delivery outcome to associate with the message.
+     *
+     * @return Response from management node.
+     */
+    Mono<AmqpAnnotatedMessage> send(AmqpAnnotatedMessage message, DeliveryOutcome deliveryOutcome);
+}

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/AmqpSession.java

@@ -4,6 +4,7 @@
 package com.azure.core.amqp;
 
 import com.azure.core.amqp.exception.AmqpException;
+import com.azure.core.util.AsyncCloseable;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -13,7 +14,7 @@
 /**
  * An AMQP session representing bidirectional communication that supports multiple {@link AmqpLink AMQP links}.
  */
-public interface AmqpSession extends Disposable {
+public interface AmqpSession extends Disposable, AsyncCloseable {
     /**
      * Gets the name for this AMQP session.
      *
@@ -91,4 +92,9 @@ public interface AmqpSession extends Disposable {
      * @return A completable mono.
      */
     Mono<Void> rollbackTransaction(AmqpTransaction transaction);
+
+    @Override
+    default Mono<Void> closeAsync() {
+        return Mono.fromRunnable(() -> dispose());
+    }
 }

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/ClaimsBasedSecurityNode.java

@@ -4,6 +4,7 @@
 package com.azure.core.amqp;
 
 import com.azure.core.credential.TokenCredential;
+import com.azure.core.util.AsyncCloseable;
 import reactor.core.publisher.Mono;
 
 import java.time.OffsetDateTime;
@@ -14,7 +15,7 @@
  * @see <a href="https://www.oasis-open.org/committees/download.php/62097/amqp-cbs-v1.0-wd05.doc">
  * AMPQ Claims-based Security v1.0</a>
  */
-public interface ClaimsBasedSecurityNode extends AutoCloseable {
+public interface ClaimsBasedSecurityNode extends AutoCloseable, AsyncCloseable {
     /**
      * Authorizes the caller with the CBS node to access resources for the {@code audience}.
      *
@@ -31,4 +32,9 @@ public interface ClaimsBasedSecurityNode extends AutoCloseable {
      */
     @Override
     void close();
+
+    @Override
+    default Mono<Void> closeAsync() {
+        return Mono.fromRunnable(() -> close());
+    }
 }

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

@@ -10,7 +10,6 @@
 import com.azure.core.amqp.models.CbsAuthorizationType;
 import com.azure.core.credential.TokenCredential;
 import com.azure.core.credential.TokenRequestContext;
-import com.azure.core.util.logging.ClientLogger;
 import org.apache.qpid.proton.Proton;
 import org.apache.qpid.proton.amqp.messaging.AmqpValue;
 import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
@@ -35,7 +34,6 @@ public class ClaimsBasedSecurityChannel implements ClaimsBasedSecurityNode {
     private static final String PUT_TOKEN_OPERATION = "operation";
     private static final String PUT_TOKEN_OPERATION_VALUE = "put-token";
 
-    private final ClientLogger logger = new ClientLogger(ClaimsBasedSecurityChannel.class);
     private final TokenCredential credential;
     private final Mono<RequestResponseChannel> cbsChannelMono;
     private final CbsAuthorizationType authorizationType;
@@ -87,9 +85,11 @@ public Mono<OffsetDateTime> authorize(String tokenAudience, String scopes) {
 
     @Override
     public void close() {
-        final RequestResponseChannel channel = cbsChannelMono.block(retryOptions.getTryTimeout());
-        if (channel != null) {
-            channel.closeAsync().block();
-        }
+        closeAsync().block(retryOptions.getTryTimeout());
+    }
+
+    @Override
+    public Mono<Void> closeAsync() {
+        return cbsChannelMono.flatMap(channel -> channel.closeAsync());
     }
 }

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

@@ -23,18 +23,14 @@
  */
 @Immutable
 public class ConnectionOptions {
-    // These name version keys are used in our properties files to specify client product and version information.
-    static final String NAME_KEY = "name";
-    static final String VERSION_KEY = "version";
-    static final String UNKNOWN = "UNKNOWN";
-
     private final TokenCredential tokenCredential;
     private final AmqpTransportType transport;
     private final AmqpRetryOptions retryOptions;
     private final ProxyOptions proxyOptions;
     private final Scheduler scheduler;
     private final String fullyQualifiedNamespace;
     private final CbsAuthorizationType authorizationType;
+    private final String authorizationScope;
     private final ClientOptions clientOptions;
     private final String product;
     private final String clientVersion;
@@ -62,10 +58,10 @@ public class ConnectionOptions {
      *     {@code proxyOptions} or {@code verifyMode} is null.
      */
     public ConnectionOptions(String fullyQualifiedNamespace, TokenCredential tokenCredential,
-        CbsAuthorizationType authorizationType, AmqpTransportType transport, AmqpRetryOptions retryOptions,
-        ProxyOptions proxyOptions, Scheduler scheduler, ClientOptions clientOptions,
+        CbsAuthorizationType authorizationType, String authorizationScope, AmqpTransportType transport,
+        AmqpRetryOptions retryOptions, ProxyOptions proxyOptions, Scheduler scheduler, ClientOptions clientOptions,
         SslDomain.VerifyMode verifyMode, String product, String clientVersion) {
-        this(fullyQualifiedNamespace, tokenCredential, authorizationType, transport, retryOptions,
+        this(fullyQualifiedNamespace, tokenCredential, authorizationType, authorizationScope, transport, retryOptions,
             proxyOptions, scheduler, clientOptions, verifyMode, product, clientVersion, fullyQualifiedNamespace,
             getPort(transport));
     }
@@ -94,14 +90,15 @@ public ConnectionOptions(String fullyQualifiedNamespace, TokenCredential tokenCr
      *     {@code clientOptions}, {@code hostname}, or {@code verifyMode} is null.
      */
     public ConnectionOptions(String fullyQualifiedNamespace, TokenCredential tokenCredential,
-        CbsAuthorizationType authorizationType, AmqpTransportType transport, AmqpRetryOptions retryOptions,
-        ProxyOptions proxyOptions, Scheduler scheduler, ClientOptions clientOptions,
+        CbsAuthorizationType authorizationType, String authorizationScope, AmqpTransportType transport,
+        AmqpRetryOptions retryOptions, ProxyOptions proxyOptions, Scheduler scheduler, ClientOptions clientOptions,
         SslDomain.VerifyMode verifyMode, String product, String clientVersion, String hostname, int port) {
 
         this.fullyQualifiedNamespace = Objects.requireNonNull(fullyQualifiedNamespace,
             "'fullyQualifiedNamespace' is required.");
         this.tokenCredential = Objects.requireNonNull(tokenCredential, "'tokenCredential' is required.");
         this.authorizationType = Objects.requireNonNull(authorizationType, "'authorizationType' is required.");
+        this.authorizationScope = Objects.requireNonNull(authorizationScope, "'authorizationScope' is required.");
         this.transport = Objects.requireNonNull(transport, "'transport' is required.");
         this.retryOptions = Objects.requireNonNull(retryOptions, "'retryOptions' is required.");
         this.scheduler = Objects.requireNonNull(scheduler, "'scheduler' is required.");
@@ -115,6 +112,15 @@ public ConnectionOptions(String fullyQualifiedNamespace, TokenCredential tokenCr
         this.clientVersion = Objects.requireNonNull(clientVersion, "'clientVersion' cannot be null.");
     }
 
+    /**
+     * Gets the scope to use when authorizing.
+     *
+     * @return The scope to use when authorizing.
+     */
+    public String getAuthorizationScope() {
+        return authorizationScope;
+    }
+
     /**
      * Gets the authorisation type for the CBS node.
      *

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

@@ -8,7 +8,6 @@
 import com.azure.core.amqp.exception.AmqpException;
 import com.azure.core.amqp.exception.AmqpResponseCode;
 
-import java.util.Locale;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -78,8 +77,9 @@ public static Exception toException(String errorCondition, String description, A
             case NOT_FOUND:
                 return distinguishNotFound(description, errorContext);
             default:
-                throw new IllegalArgumentException(String.format(Locale.ROOT, "This condition '%s' is not known.",
-                    condition));
+                return new AmqpException(false, condition, String.format("errorCondition[%s]. description[%s] "
+                        + "Condition could not be mapped to a transient condition.",
+                    errorCondition, description), errorContext);
         }
 
         return new AmqpException(isTransient, condition, description, errorContext);