azure-core-amqp: Consolidate getEndpointStates and naming changes.

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

@@ -3,15 +3,16 @@
 
 package com.azure.core.amqp;
 
+import com.azure.core.amqp.exception.AmqpException;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import java.io.Closeable;
 import java.util.Map;
 
 /**
  * Represents a TCP connection between the client and a service that uses the AMQP protocol.
  */
-public interface AmqpConnection extends EndpointStateNotifier, Closeable {
+public interface AmqpConnection extends AutoCloseable {
     /**
      * Gets the connection identifier.
      *
@@ -62,4 +63,25 @@ public interface AmqpConnection extends EndpointStateNotifier, Closeable {
      * @return {@code true} if a session with the name was removed; {@code false} otherwise.
      */
     boolean removeSession(String sessionName);
+
+    /**
+     * Gets the endpoint states for the AMQP connection. {@link AmqpException AmqpExceptions} that occur on the link are
+     * reported in the connection state. When the stream terminates, the connection is closed.
+     *
+     * @return A stream of endpoint states for the AMQP connection.
+     */
+    Flux<AmqpEndpointState> getEndpointStates();
+
+    /**
+     * Gets any shutdown signals that occur in the AMQP endpoint.
+     *
+     * @return A stream of shutdown signals that occur in the AMQP endpoint.
+     */
+    Flux<AmqpShutdownSignal> getShutdownSignals();
+
+    /**
+     * Closes the AMQP connection.
+     */
+    @Override
+    void close();
 }

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

@@ -3,12 +3,13 @@
 
 package com.azure.core.amqp;
 
-import java.io.Closeable;
+import com.azure.core.amqp.exception.AmqpException;
+import reactor.core.publisher.Flux;
 
 /**
  * Represents a unidirectional AMQP link.
  */
-public interface AmqpLink extends EndpointStateNotifier, Closeable {
+public interface AmqpLink extends AutoCloseable {
     /**
      * Gets the name of the link.
      *
@@ -29,4 +30,18 @@ public interface AmqpLink extends EndpointStateNotifier, Closeable {
      * @return The host name of the message broker that this link that is connected to.
      */
     String getHostname();
+
+    /**
+     * Gets the endpoint states for the AMQP link. {@link AmqpException AmqpExceptions} that occur on the link are
+     * reported in the connection state. When the stream terminates, the link is closed.
+     *
+     * @return A stream of endpoint states for the AMQP link.
+     */
+    Flux<AmqpEndpointState> getEndpointStates();
+
+    /**
+     * Closes the AMQP link.
+     */
+    @Override
+    void close();
 }

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

@@ -120,13 +120,13 @@ public String getValue() {
     /**
      * Parses an header value to its message constant.
      *
-     * @param headerValue the messaging header value to parse.
+     * @param value the messaging header value to parse.
      * @return the parsed MessageConstant object, or {@code null} if unable to parse.
      * @throws NullPointerException if {@code constant} is {@code null}.
      */
-    public static AmqpMessageConstant fromString(String headerValue) {
-        Objects.requireNonNull(headerValue, "'headerValue' cannot be null.");
+    public static AmqpMessageConstant fromString(String value) {
+        Objects.requireNonNull(value, "'value' cannot be null.");
 
-        return RESERVED_CONSTANTS_MAP.get(headerValue);
+        return RESERVED_CONSTANTS_MAP.get(value);
     }
 }

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

@@ -3,15 +3,16 @@
 
 package com.azure.core.amqp;
 
+import com.azure.core.amqp.exception.AmqpException;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import java.io.Closeable;
 import java.time.Duration;
 
 /**
  * An AMQP session representing bidirectional communication that supports multiple {@link AmqpLink AMQP links}.
  */
-public interface AmqpSession extends EndpointStateNotifier, Closeable {
+public interface AmqpSession extends AutoCloseable {
     /**
      * Gets the name for this AMQP session.
      *
@@ -55,4 +56,18 @@ public interface AmqpSession extends EndpointStateNotifier, Closeable {
      * @return {@code true} if the link was removed; {@code false} otherwise.
      */
     boolean removeLink(String linkName);
+
+    /**
+     * Gets the endpoint states for the AMQP session. {@link AmqpException AmqpExceptions} that occur on the link are
+     * reported in the connection state. When the stream terminates, the session is closed.
+     *
+     * @return A stream of endpoint states for the AMQP session.
+     */
+    Flux<AmqpEndpointState> getEndpointStates();
+
+    /**
+     * Closes the AMQP session.
+     */
+    @Override
+    void close();
 }

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

@@ -6,7 +6,6 @@
 import com.azure.core.credential.TokenCredential;
 import reactor.core.publisher.Mono;
 
-import java.io.Closeable;
 import java.time.OffsetDateTime;
 
 /**
@@ -15,7 +14,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 EndpointStateNotifier, Closeable {
+public interface ClaimsBasedSecurityNode extends AutoCloseable {
     /**
      * Authorizes the caller with the CBS node to access resources for the {@code audience}.
      *
@@ -26,4 +25,10 @@ public interface ClaimsBasedSecurityNode extends EndpointStateNotifier, Closeabl
      * CBS node.
      */
     Mono<OffsetDateTime> authorize(String audience, String scopes);
+
+    /**
+     * Closes session to the claims-based security node.
+     */
+    @Override
+    void close();
 }

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

@@ -0,0 +1,31 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.amqp.implementation;
+
+import com.azure.core.amqp.AmqpEndpointState;
+import org.apache.qpid.proton.engine.EndpointState;
+
+/**
+ * Helper class for managing endpoint states from proton-j.
+ */
+class AmqpEndpointStateUtil {
+    /**
+     * Translates proton-j endpoint states into an AMQP endpoint state.
+     * @param state proton-j endpoint state.
+     * @return The corresponding {@link AmqpEndpointState}.
+     * @throws IllegalArgumentException if {@code state} is not a supported {@link AmqpEndpointState}.
+     */
+    static AmqpEndpointState getConnectionState(EndpointState state) {
+        switch (state) {
+            case ACTIVE:
+                return AmqpEndpointState.ACTIVE;
+            case UNINITIALIZED:
+                return AmqpEndpointState.UNINITIALIZED;
+            case CLOSED:
+                return AmqpEndpointState.CLOSED;
+            default:
+                throw new IllegalArgumentException("This endpoint state is not supported. State:" + state);
+        }
+    }
+}

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

@@ -1,28 +1,29 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
-package com.azure.core.amqp;
+package com.azure.core.amqp.implementation;
 
+import com.azure.core.amqp.AmqpShutdownSignal;
 import com.azure.core.util.logging.ClientLogger;
 
 /**
  * Handles exceptions generated by AMQP connections, sessions, and/or links.
  */
-public abstract class AmqpExceptionHandler {
+abstract class AmqpExceptionHandler {
     private final ClientLogger logger = new ClientLogger(AmqpExceptionHandler.class);
 
     /**
      * Creates a new instance of the exception handler.
      */
-    protected AmqpExceptionHandler() {
+    AmqpExceptionHandler() {
     }
 
     /**
      * Notifies the exception handler of an exception.
      *
      * @param exception The exception that caused the connection error.
      */
-    public void onConnectionError(Throwable exception) {
+    void onConnectionError(Throwable exception) {
         logger.warning("Connection exception encountered: " + exception.toString(), exception);
     }
 
@@ -31,7 +32,7 @@ public void onConnectionError(Throwable exception) {
      *
      * @param shutdownSignal The shutdown signal that was received.
      */
-    public void onConnectionShutdown(AmqpShutdownSignal shutdownSignal) {
+    void onConnectionShutdown(AmqpShutdownSignal shutdownSignal) {
         logger.info("Shutdown received: {}", shutdownSignal);
     }
 }

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

@@ -7,7 +7,6 @@
 import org.apache.qpid.proton.message.Message;
 import reactor.core.publisher.Flux;
 
-import java.io.Closeable;
 import java.util.function.Supplier;
 
 /**
@@ -21,7 +20,7 @@ public interface AmqpReceiveLink extends AmqpLink {
      * Initialises the link from the client to the message broker and begins to receive messages from the broker.
      *
      * @return A Flux of AMQP messages which completes when the client calls
-     * {@link Closeable#close() AmqpReceiveLink.close()} or an unrecoverable error occurs on the AMQP link.
+     * {@link AutoCloseable#close() AmqpReceiveLink.close()} or an unrecoverable error occurs on the AMQP link.
      */
     Flux<Message> receive();
 

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

@@ -17,7 +17,7 @@ public class AzureTokenManagerProvider implements TokenManagerProvider {
     static final String TOKEN_AUDIENCE_FORMAT = "amqp://%s/%s";
 
     private final ClientLogger logger = new ClientLogger(AzureTokenManagerProvider.class);
-    private final CBSAuthorizationType authorizationType;
+    private final CbsAuthorizationType authorizationType;
     private final String fullyQualifiedNamespace;
     private final String activeDirectoryScope;
 
@@ -29,7 +29,7 @@ public class AzureTokenManagerProvider implements TokenManagerProvider {
      * @param fullyQualifiedNamespace Fully-qualified namespace of the message broker.
      * @param activeDirectoryScope Scope used to access AD resources for the Azure service.
      */
-    public AzureTokenManagerProvider(CBSAuthorizationType authorizationType, String fullyQualifiedNamespace,
+    public AzureTokenManagerProvider(CbsAuthorizationType authorizationType, String fullyQualifiedNamespace,
                                      String activeDirectoryScope) {
         this.activeDirectoryScope = Objects.requireNonNull(activeDirectoryScope,
             "'activeDirectoryScope' cannot be null.");

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/CBSAuthorizationType.java → sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/CbsAuthorizationType.java

@@ -8,7 +8,7 @@
 /**
  * An enumeration of supported authorization methods with the {@link ClaimsBasedSecurityNode}.
  */
-public enum CBSAuthorizationType {
+public enum CbsAuthorizationType {
     /**
      * Authorize with CBS through a shared access signature.
      */
@@ -23,7 +23,7 @@ public enum CBSAuthorizationType {
 
     private final String scheme;
 
-    CBSAuthorizationType(String scheme) {
+    CbsAuthorizationType(String scheme) {
         this.scheme = scheme;
     }
 

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

@@ -7,7 +7,6 @@
 import com.azure.core.amqp.ClaimsBasedSecurityNode;
 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;
@@ -19,20 +18,19 @@
 import java.util.Map;
 import java.util.Objects;
 
-public class ClaimsBasedSecurityChannel extends EndpointStateNotifierBase implements ClaimsBasedSecurityNode {
-    static final String PUT_TOKEN_OPERATION = "operation";
-    static final String PUT_TOKEN_OPERATION_VALUE = "put-token";
+public class ClaimsBasedSecurityChannel implements ClaimsBasedSecurityNode {
     static final String PUT_TOKEN_TYPE = "type";
     static final String PUT_TOKEN_AUDIENCE = "name";
+    private static final String PUT_TOKEN_OPERATION = "operation";
+    private static final String PUT_TOKEN_OPERATION_VALUE = "put-token";
 
     private final TokenCredential credential;
     private final Mono<RequestResponseChannel> cbsChannelMono;
-    private final CBSAuthorizationType authorizationType;
+    private final CbsAuthorizationType authorizationType;
     private final AmqpRetryOptions retryOptions;
 
     public ClaimsBasedSecurityChannel(Mono<RequestResponseChannel> responseChannelMono, TokenCredential tokenCredential,
-               CBSAuthorizationType authorizationType, AmqpRetryOptions retryOptions) {
-        super(new ClientLogger(ClaimsBasedSecurityChannel.class));
+               CbsAuthorizationType authorizationType, AmqpRetryOptions retryOptions) {
 
         this.authorizationType = Objects.requireNonNull(authorizationType, "'authorizationType' cannot be null.");
         this.retryOptions = Objects.requireNonNull(retryOptions, "'retryOptions' cannot be null.");

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

@@ -24,10 +24,10 @@ public class ConnectionOptions {
     private final Scheduler scheduler;
     private final String fullyQualifiedNamespace;
     private final String entityPath;
-    private final CBSAuthorizationType authorizationType;
+    private final CbsAuthorizationType authorizationType;
 
     public ConnectionOptions(String fullyQualifiedNamespace, String entityPath, TokenCredential tokenCredential,
-            CBSAuthorizationType authorizationType, AmqpTransportType transport, AmqpRetryOptions retryOptions,
+            CbsAuthorizationType authorizationType, AmqpTransportType transport, AmqpRetryOptions retryOptions,
             ProxyOptions proxyOptions, Scheduler scheduler) {
         this.fullyQualifiedNamespace = Objects.requireNonNull(fullyQualifiedNamespace,
             "'fullyQualifiedNamespace' is required.");
@@ -52,7 +52,7 @@ public TokenCredential getTokenCredential() {
         return tokenCredential;
     }
 
-    public CBSAuthorizationType getAuthorizationType() {
+    public CbsAuthorizationType getAuthorizationType() {
         return authorizationType;
     }