Add ability to specify different endpoint. (#18996)

* Add null checks to base Handler class.

* Update ConnectionHandler to take connection options.

* Update WebSocketsConnectionHandler.

* Update ConnectionOptions.

* Update Handler providers to take ConnectionOptions instead of an entire slew of things.

* Update ReactorHandlerProvider to use connectionOptions.

* Update ReactorConnection.

* Add changelog entry

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

@@ -2,23 +2,33 @@
 
 ## 2.1.0-beta.1 (Unreleased)
 
+### New Features
+
+- Updates RetryUtil to use RetrySpec.
+- Adds the ability to configure the `hostname` and `port` that will be used when connecting to a service via
+  `ConnectionOptions`. The `hostname` field refers to the DNS host or IP address of the service, whereas the
+  `fullyQualifiedNamespace` is the fully qualified host name of the service.
+  Normally `hostname` and `fullyQualifiedNamespace` will be the same. However, if your network does not allow
+  connecting to the service via the public host, you can specify a custom host (e.g. an application gateway) via the
+  `hostname` field and continue using the public host as the `fullyQualifiedNamespace`.
 
 ## 2.0.1 (2021-01-11)
 
 ### New Features
 
-- Changed connections from sharing the global `Schedulers.single()` to having a `Scheduler.newSingle()` per connection to improve performance.
+- Changed connections from sharing the global `Schedulers.single()` to having a `Scheduler.newSingle()` per connection
+  to improve performance.
 
 ## 2.0.0 (2020-11-30)
 ### New Features
 - Added 'AmqpAddress' as a type to support 'AmqpMessageProperties#replyTo' and 'AmqpMessageProperties#to' properties.
-- Added 'AmqpMessageId' as a type to support 'AmqpMessageProperties#correlationId' and 'AmqpMessageProperties#messageId' 
+- Added 'AmqpMessageId' as a type to support 'AmqpMessageProperties#correlationId' and 'AmqpMessageProperties#messageId'
   properties.
 - Added static methods to instantiate 'AmqpMessageBody' for example 'AmqpMessageBody#fromData(byte[])'.
 
 ### Breaking Changes
-- Changed  'AmqpMessageBody' from interface to a class. User can use 'getBodyType()' to know what is the 'AmqpBodyType' 
-  of the message.
+- Changed  'AmqpMessageBody' from an interface to a class. User can use 'getBodyType()' to know what is the
+  'AmqpBodyType' of the message.
 - Changed type of 'correlationId' and 'messageId' in type 'AmqpMessageProperties' from 'String' to 'AmqpMessageId'.
 - Changed type of 'replyTo' and 'to' in type 'AmqpMessageProperties' from 'String' to 'AmqpAddress'.
 - Removed copy constructor for 'AmqpAnnotatedMessage'.
@@ -29,8 +39,8 @@
 
 ## 1.7.0-beta.2 (2020-11-10)
 ### New Features
-- Optionally enable idempotency of a send link to send AMQP messages with producer group id, producer owner level and 
-producer sequence number in the message annotations.
+- Optionally enable idempotency of a send link to send AMQP messages with producer group id, producer owner level and
+  producer sequence number in the message annotations.
 
 ## 1.7.0-beta.1 (2020-11-03)
 ### Dependency Updates
@@ -40,7 +50,7 @@ producer sequence number in the message annotations.
 ### New Features
 - Added peer certificate verification options when connecting to an AMQP endpoint.
 ### Breaking Changes
-- Removed `BinaryData` type which was used for `AmqpAnnotatedMessage`. 
+- Removed `BinaryData` type which was used for `AmqpAnnotatedMessage`.
 ### Dependency Updates
 - Upgraded `azure-core` dependency to `1.9.0`.
 
@@ -154,4 +164,3 @@ producer sequence number in the message annotations.
 
 This package's
 [documentation](https://github.com/Azure/azure-sdk-for-java/blob/azure-core-amqp_1.0.0-preview.1/core/azure-core-amqp/README.md)
-

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

@@ -6,9 +6,12 @@
 import com.azure.core.amqp.AmqpRetryOptions;
 import com.azure.core.amqp.AmqpTransportType;
 import com.azure.core.amqp.ProxyOptions;
+import com.azure.core.amqp.implementation.handler.ConnectionHandler;
+import com.azure.core.amqp.implementation.handler.WebSocketsConnectionHandler;
 import com.azure.core.annotation.Immutable;
 import com.azure.core.credential.TokenCredential;
 import com.azure.core.util.ClientOptions;
+import com.azure.core.util.logging.ClientLogger;
 import org.apache.qpid.proton.engine.SslDomain;
 import reactor.core.scheduler.Scheduler;
 
@@ -28,32 +31,102 @@ public class ConnectionOptions {
     private final CbsAuthorizationType authorizationType;
     private final ClientOptions clientOptions;
     private final SslDomain.VerifyMode verifyMode;
+    private final String hostname;
+    private final int port;
 
+    /**
+     * Creates an instance with the following options set. The AMQP connection is created to the
+     * {@code fullyQualifiedNamespace} using a port based on the {@code transport}.
+     *
+     * @param fullyQualifiedNamespace Fully qualified namespace for the AMQP broker. (ie.
+     *     namespace.servicebus.windows.net)
+     * @param tokenCredential The credential for connecting to the AMQP broker.
+     * @param authorizationType The authorisation type used for authorizing with the CBS node.
+     * @param transport The type connection used for the AMQP connection.
+     * @param retryOptions Retry options for the connection.
+     * @param proxyOptions Any proxy options to set.
+     * @param scheduler Scheduler for async operations.
+     * @param clientOptions Client options for the connection.
+     * @param verifyMode How to verify SSL information.
+     *
+     * @throws NullPointerException in the case that {@code fullyQualifiedNamespace}, {@code tokenCredential},
+     *     {@code authorizationType}, {@code transport}, {@code retryOptions}, {@code scheduler}, {@code clientOptions}
+     *     {@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,
-            SslDomain.VerifyMode verifyMode) {
+        CbsAuthorizationType authorizationType, AmqpTransportType transport, AmqpRetryOptions retryOptions,
+        ProxyOptions proxyOptions, Scheduler scheduler, ClientOptions clientOptions,
+        SslDomain.VerifyMode verifyMode) {
+        this(fullyQualifiedNamespace, tokenCredential, authorizationType, transport, retryOptions,
+            proxyOptions, scheduler, clientOptions, verifyMode, fullyQualifiedNamespace, getPort(transport));
+    }
+
+    /**
+     * Creates an instance with the connection options set. Used when an alternative address should be made for the
+     * connection rather than through the fullyQualifiedNamespace.
+     *
+     * @param fullyQualifiedNamespace Fully qualified namespace for the AMQP broker. (ie.
+     *     namespace.servicebus.windows.net)
+     * @param tokenCredential The credential for connecting to the AMQP broker.
+     * @param authorizationType The authorisation type used for authorizing with the CBS node.
+     * @param transport The type connection used for the AMQP connection.
+     * @param retryOptions Retry options for the connection.
+     * @param proxyOptions (Optional) Any proxy options to set.
+     * @param scheduler Scheduler for async operations.
+     * @param clientOptions Client options for the connection.
+     * @param verifyMode How to verify SSL information.
+     * @param hostname Connection hostname. Used to create the connection to in the case that we cannot
+     *     connect directly to the AMQP broker.
+     * @param port Connection port. Used to create the connection to in the case we cannot connect directly
+     *     to the AMQP broker.
+     *
+     * @throws NullPointerException in the case that {@code fullyQualifiedNamespace}, {@code tokenCredential},
+     *     {@code authorizationType}, {@code transport}, {@code retryOptions}, {@code scheduler},
+     *     {@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,
+        SslDomain.VerifyMode verifyMode, 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.transport = Objects.requireNonNull(transport, "'transport' is required.");
         this.retryOptions = Objects.requireNonNull(retryOptions, "'retryOptions' is required.");
-        this.proxyOptions = Objects.requireNonNull(proxyOptions, "'proxyConfiguration' is required.");
         this.scheduler = Objects.requireNonNull(scheduler, "'scheduler' is required.");
         this.clientOptions = Objects.requireNonNull(clientOptions, "'clientOptions' is required.");
         this.verifyMode = Objects.requireNonNull(verifyMode, "'verifyMode' is required.");
+        this.hostname = Objects.requireNonNull(hostname, "'hostname' cannot be null.");
+        this.port = port != -1 ? port : getPort(transport);
+        this.proxyOptions = proxyOptions;
     }
 
+    /**
+     * Gets the authorisation type for the CBS node.
+     *
+     * @return The authorisation type for the CBS node.
+     */
     public CbsAuthorizationType getAuthorizationType() {
         return authorizationType;
     }
 
+    /**
+     * Gets the client options.
+     *
+     * @return The client options.
+     */
     public ClientOptions getClientOptions() {
         return clientOptions;
     }
 
+    /**
+     * The fully qualified domain name for the AMQP broker. Typically of the form
+     * {@literal "<your-namespace>.service.windows.net"}.
+     *
+     * @return The fully qualified domain name for the AMQP broker.
+     */
     public String getFullyQualifiedNamespace() {
         return fullyQualifiedNamespace;
     }
@@ -62,23 +135,80 @@ public AmqpRetryOptions getRetry() {
         return retryOptions;
     }
 
+    /**
+     * Gets the proxy options set.
+     *
+     * @return The proxy options set. {@code null} if there are no options set.
+     */
     public ProxyOptions getProxyOptions() {
         return proxyOptions;
     }
 
+    /**
+     * Gets the scheduler to execute tasks on.
+     *
+     * @return The scheduler to execute tasks on.
+     */
     public Scheduler getScheduler() {
         return scheduler;
     }
 
+    /**
+     * Gets the verification mode for the SSL certificate.
+     *
+     * @return The verification mode for the SSL certificate.
+     */
     public SslDomain.VerifyMode getSslVerifyMode() {
         return verifyMode;
     }
 
+    /**
+     * Gets the credential for authorising with Event Hubs.
+     *
+     * @return The credential for authorising with Event Hubs.
+     */
     public TokenCredential getTokenCredential() {
         return tokenCredential;
     }
 
+    /**
+     * Gets the transport type for the AMQP connection.
+     *
+     * @return The transport type for the AMQP connection.
+     */
     public AmqpTransportType getTransportType() {
         return transport;
     }
+
+    /**
+     * Gets the DNS hostname or IP address of the service. Typically of the form
+     * {@literal "<your-namespace>.service.windows.net"}, unless connecting to the service through an intermediary.
+     *
+     * @return The DNS hostname or IP address to connect to.
+     */
+    public String getHostname() {
+        return hostname;
+    }
+
+    /**
+     * Gets the port to connect to when creating the connection to the service. This is usually the port for the
+     * AMQP protocol or 443 for web sockets, but can differ if connecting through an intermediary.
+     *
+     * @return The port to connect to when creating the connection to the service.
+     */
+    public int getPort() {
+        return port;
+    }
+
+    private static int getPort(AmqpTransportType transport) {
+        switch (transport) {
+            case AMQP:
+                return ConnectionHandler.AMQPS_PORT;
+            case AMQP_WEB_SOCKETS:
+                return WebSocketsConnectionHandler.HTTPS_PORT;
+            default:
+                throw new ClientLogger(ConnectionOptions.class).logThrowableAsError(
+                    new IllegalArgumentException("Transport Type is not supported: " + transport));
+        }
+    }
 }

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

@@ -97,10 +97,8 @@ public ReactorConnection(String connectionId, ConnectionOptions connectionOption
         this.tokenManagerProvider = Objects.requireNonNull(tokenManagerProvider,
             "'tokenManagerProvider' cannot be null.");
         this.messageSerializer = messageSerializer;
-        this.handler = handlerProvider.createConnectionHandler(connectionId,
-            connectionOptions.getFullyQualifiedNamespace(), connectionOptions.getTransportType(),
-            connectionOptions.getProxyOptions(), product, clientVersion, connectionOptions.getSslVerifyMode(),
-            connectionOptions.getClientOptions());
+        this.handler = handlerProvider.createConnectionHandler(connectionId, product, clientVersion,
+            connectionOptions);
 
         this.retryPolicy = RetryUtil.getRetryPolicy(connectionOptions.getRetry());
         this.senderSettleMode = senderSettleMode;