Add support for parsing local emulator slug (#40483)

* Add parsing for UseDevelopmentEmulator

* Use unreleased_azure-core-amqp.

* Use development storage in builder.

* Add option to enable/disable ssl.

* Fix naming on web socket test.

* Fix NPE when peerDetails are not set.

* Add connection option to enable ssl.

* Update EHCB to use enableSsl.

* Update Reactor provider to use peer details or not if SSL enabled.

* Fix ConnectionOption breaks.

* Fix build break in ServiceBusClientBuilder

* Add changelog message.

Author: conniey

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

@@ -4,6 +4,8 @@
 
 ### Features Added
 
+- Added feature to enable/disable SSL when initially creating connection to support amqp calls on port 5672.
+
 ### Breaking Changes
 
 ### Bugs Fixed

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

@@ -38,6 +38,7 @@ public class ConnectionOptions {
     private final String clientVersion;
     private final SslDomain.VerifyMode verifyMode;
     private final String hostname;
+    private final boolean enableSsl;
     private final int port;
 
     /**
@@ -67,7 +68,7 @@ public ConnectionOptions(String fullyQualifiedNamespace, TokenCredential tokenCr
         SslDomain.VerifyMode verifyMode, String product, String clientVersion) {
         this(fullyQualifiedNamespace, tokenCredential, authorizationType, authorizationScope, transport, retryOptions,
             proxyOptions, scheduler, clientOptions, verifyMode, product, clientVersion, fullyQualifiedNamespace,
-            getPort(transport));
+            getPort(transport), true);
     }
 
     /**
@@ -91,6 +92,7 @@ public ConnectionOptions(String fullyQualifiedNamespace, TokenCredential tokenCr
      *     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.
+     * @param enableSsl true to enable SSL when creating AMQP connection.
      *
      * @throws NullPointerException in the case that {@code fullyQualifiedNamespace}, {@code tokenCredential},
      *     {@code authorizationType}, {@code transport}, {@code retryOptions}, {@code scheduler},
@@ -99,7 +101,8 @@ public ConnectionOptions(String fullyQualifiedNamespace, TokenCredential tokenCr
     public ConnectionOptions(String fullyQualifiedNamespace, TokenCredential tokenCredential,
         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) {
+        SslDomain.VerifyMode verifyMode, String product, String clientVersion, String hostname, int port,
+        boolean enableSsl) {
 
         this.fullyQualifiedNamespace
             = Objects.requireNonNull(fullyQualifiedNamespace, "'fullyQualifiedNamespace' is required.");
@@ -112,6 +115,7 @@ public ConnectionOptions(String fullyQualifiedNamespace, TokenCredential tokenCr
         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.enableSsl = enableSsl;
         this.port = port != -1 ? port : getPort(transport);
         this.proxyOptions = proxyOptions;
 
@@ -202,7 +206,8 @@ public Scheduler getScheduler() {
     }
 
     /**
-     * Gets the verification mode for the SSL certificate.
+     * Gets the verification mode for the SSL certificate.  The verification mode used when {@link #isEnableSsl()} is
+     * true.
      *
      * @return The verification mode for the SSL certificate.
      */
@@ -248,6 +253,15 @@ public int getPort() {
         return port;
     }
 
+    /**
+     * Check if SSL should be enabled when creating the AMQP connection.
+     *
+     * @return {@code true} to enable SSL when creating AMQP connection, false otherwise.
+     */
+    public boolean isEnableSsl() {
+        return enableSsl;
+    }
+
     private static int getPort(AmqpTransportType transport) {
         switch (transport) {
             case AMQP:

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

@@ -28,6 +28,7 @@ public class ConnectionStringProperties {
     private static final String SHARED_ACCESS_SIGNATURE = "SharedAccessSignature";
     private static final String SAS_VALUE_PREFIX = "sharedaccesssignature ";
     private static final String ENTITY_PATH = "EntityPath";
+    private static final String USE_DEVELOPMENT_EMULATOR = "UseDevelopmentEmulator";
     private static final String CONNECTION_STRING_WITH_ACCESS_KEY = "Endpoint={endpoint};"
         + "SharedAccessKeyName={sharedAccessKeyName};SharedAccessKey={sharedAccessKey};EntityPath={entityPath}";
     private static final String CONNECTION_STRING_WITH_SAS = "Endpoint={endpoint};SharedAccessSignature="
@@ -41,6 +42,7 @@ public class ConnectionStringProperties {
     private final String sharedAccessKeyName;
     private final String sharedAccessKey;
     private final String sharedAccessSignature;
+    private final boolean useDevelopmentEmulator;
 
     /**
      * Creates a new instance by parsing the {@code connectionString} into its components.
@@ -62,6 +64,7 @@ public ConnectionStringProperties(String connectionString) {
         String sharedAccessKeyName = null;
         String sharedAccessKeyValue = null;
         String sharedAccessSignature = null;
+        Boolean useDevelopmentEmulator = null;
 
         for (String tokenValuePair : tokenValuePairs) {
             final String[] pair = tokenValuePair.split(TOKEN_VALUE_SEPARATOR, 2);
@@ -87,6 +90,8 @@ public ConnectionStringProperties(String connectionString) {
                 sharedAccessKeyValue = value;
             } else if (key.equalsIgnoreCase(ENTITY_PATH)) {
                 entityPath = value;
+            } else if (key.equalsIgnoreCase(USE_DEVELOPMENT_EMULATOR)) {
+                useDevelopmentEmulator = Boolean.valueOf(value);
             } else if (key.equalsIgnoreCase(SHARED_ACCESS_SIGNATURE)
                 && value.toLowerCase(Locale.ROOT).startsWith(SAS_VALUE_PREFIX)) {
                 sharedAccessSignature = value;
@@ -112,6 +117,8 @@ public ConnectionStringProperties(String connectionString) {
         this.sharedAccessKeyName = sharedAccessKeyName;
         this.sharedAccessKey = sharedAccessKeyValue;
         this.sharedAccessSignature = sharedAccessSignature;
+
+        this.useDevelopmentEmulator = useDevelopmentEmulator != null ? useDevelopmentEmulator : false;
     }
 
     /**
@@ -155,6 +162,15 @@ public String getSharedAccessSignature() {
         return sharedAccessSignature;
     }
 
+    /**
+     * Gets whether the connection string points to a development emulator.
+     *
+     * @return true if the connection string's endpoint is a development emulator.
+     */
+    public boolean useDevelopmentEmulator() {
+        return useDevelopmentEmulator;
+    }
+
     /*
      * The function checks for pre existing scheme of "sb://" , "http://" or "https://". If the scheme is not provided
      * in endpoint, it will set the default scheme to "sb://".

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

@@ -68,20 +68,26 @@ public ReactorHandlerProvider(ReactorProvider provider, Meter meter) {
      *
      * @param connectionId Identifier associated with this connection.
      * @param options Options for the connection.
+     *
      * @return A new {@link ConnectionHandler}.
      *
      * @throws NullPointerException If {@code connectionId}, {@code productName}, {@code clientVersion},
      *      {@code options} is {@code null}.
      */
     public ConnectionHandler createConnectionHandler(String connectionId, ConnectionOptions options) {
+
         Objects.requireNonNull(connectionId, "'connectionId' cannot be null.");
         Objects.requireNonNull(options, "'options' cannot be null.");
 
-        AmqpMetricsProvider metricsProvider = getMetricProvider(options.getFullyQualifiedNamespace(), null);
-        if (options.getTransportType() == AmqpTransportType.AMQP) {
-            final SslPeerDetails peerDetails = Proton.sslPeerDetails(options.getHostname(), options.getPort());
+        final AmqpMetricsProvider metricsProvider = getMetricProvider(options.getFullyQualifiedNamespace(), null);
 
-            return new ConnectionHandler(connectionId, options, peerDetails, metricsProvider);
+        if (options.getTransportType() == AmqpTransportType.AMQP) {
+            if (options.isEnableSsl()) {
+                final SslPeerDetails peerDetails = Proton.sslPeerDetails(options.getHostname(), options.getPort());
+                return new ConnectionHandler(connectionId, options, peerDetails, metricsProvider);
+            } else {
+                return new ConnectionHandler(connectionId, options, metricsProvider);
+            }
         }
 
         if (options.getTransportType() != AmqpTransportType.AMQP_WEB_SOCKETS) {

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

@@ -11,6 +11,7 @@
 import com.azure.core.util.ClientOptions;
 import com.azure.core.util.CoreUtils;
 import com.azure.core.util.UserAgentUtil;
+import com.azure.core.util.logging.LoggingEventBuilder;
 import org.apache.qpid.proton.Proton;
 import org.apache.qpid.proton.amqp.Symbol;
 import org.apache.qpid.proton.amqp.transport.ErrorCondition;
@@ -68,6 +69,21 @@ public class ConnectionHandler extends Handler {
     private final ConnectionOptions connectionOptions;
     private final SslPeerDetails peerDetails;
     private final AmqpMetricsProvider metricProvider;
+    private final boolean enableSsl;
+
+    /**
+     * Creates a handler that handles proton-j's connection events without using SSL.
+     *
+     * @param connectionId Identifier for this connection.
+     * @param connectionOptions Options used when creating the AMQP connection.
+     * @param metricProvider The AMQP metrics provider.
+     * @throws NullPointerException if {@code connectionOptions}, {@code peerDetails}, or {@code metricProvider} is
+     * null.
+     */
+    public ConnectionHandler(final String connectionId, final ConnectionOptions connectionOptions,
+        AmqpMetricsProvider metricProvider) {
+        this(connectionId, connectionOptions, null, metricProvider, false);
+    }
 
     /**
      * Creates a handler that handles proton-j's connection events.
@@ -81,6 +97,11 @@ public class ConnectionHandler extends Handler {
      */
     public ConnectionHandler(final String connectionId, final ConnectionOptions connectionOptions,
         SslPeerDetails peerDetails, AmqpMetricsProvider metricProvider) {
+        this(connectionId, connectionOptions, peerDetails, metricProvider, true);
+    }
+
+    ConnectionHandler(String connectionId, ConnectionOptions connectionOptions, SslPeerDetails peerDetails,
+        AmqpMetricsProvider metricProvider, boolean enableSsl) {
         super(connectionId,
             Objects.requireNonNull(connectionOptions, "'connectionOptions' cannot be null.").getHostname());
         add(new Handshaker());
@@ -100,8 +121,14 @@ public ConnectionHandler(final String connectionId, final ConnectionOptions conn
 
         this.connectionProperties.put(USER_AGENT.toString(), userAgent);
 
-        this.peerDetails = Objects.requireNonNull(peerDetails, "'peerDetails' cannot be null.");
+        if (enableSsl) {
+            this.peerDetails = Objects.requireNonNull(peerDetails, "'peerDetails' cannot be null.");
+        } else {
+            this.peerDetails = peerDetails;
+        }
+
         this.metricProvider = Objects.requireNonNull(metricProvider, "'metricProvider' cannot be null.");
+        this.enableSsl = enableSsl;
     }
 
     /**
@@ -143,6 +170,10 @@ protected void addTransportLayers(Event event, TransportInternal transport) {
         // Refer to http://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-transport-v1.0-os.html#doc-doc-idle-time-out
         transport.setIdleTimeout(CONNECTION_IDLE_TIMEOUT);
 
+        if (!enableSsl) {
+            return;
+        }
+
         final SslDomain sslDomain = Proton.sslDomain();
         sslDomain.init(SslDomain.Mode.CLIENT);
 
@@ -214,10 +245,12 @@ public void onConnectionInit(Event event) {
     public void onConnectionBound(Event event) {
         final Transport transport = event.getTransport();
 
-        logger.atInfo()
-            .addKeyValue(HOSTNAME_KEY, getHostname())
-            .addKeyValue("peerDetails", () -> peerDetails.getHostname() + ":" + peerDetails.getPort())
-            .log("onConnectionBound");
+        final LoggingEventBuilder builder = logger.atInfo().addKeyValue(HOSTNAME_KEY, getHostname());
+        if (peerDetails != null) {
+            builder.addKeyValue("peerDetails", () -> peerDetails.getHostname() + ":" + peerDetails.getPort());
+        }
+
+        builder.log("onConnectionBound");
 
         this.addTransportLayers(event, (TransportInternal) transport);
 

sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ConnectionOptionsTest.java

@@ -19,7 +19,9 @@
 import reactor.core.scheduler.Scheduler;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
  * Tests for {@link ConnectionOptions}.
@@ -78,6 +80,8 @@ public void propertiesSet() {
         assertEquals(scope, actual.getAuthorizationScope());
         assertEquals(retryOptions, actual.getRetry());
         assertEquals(verifyMode, actual.getSslVerifyMode());
+
+        assertTrue(actual.isEnableSsl());
     }
 
     /**
@@ -101,7 +105,7 @@ public void propertiesAndPortSet() {
         final ConnectionOptions actual
             = new ConnectionOptions(fullyQualifiedNamespace, tokenCredential, CbsAuthorizationType.JSON_WEB_TOKEN,
                 scope, AmqpTransportType.AMQP, retryOptions, ProxyOptions.SYSTEM_DEFAULTS, scheduler, clientOptions,
-                verifyMode, productName, clientVersion, actualHostname, port);
+                verifyMode, productName, clientVersion, actualHostname, port, false);
 
         // Assert
         assertEquals(fullyQualifiedNamespace, actual.getFullyQualifiedNamespace());
@@ -121,5 +125,7 @@ public void propertiesAndPortSet() {
 
         assertEquals(actualHostname, actual.getHostname());
         assertEquals(port, actual.getPort());
+
+        assertFalse(actual.isEnableSsl());
     }
 }

sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ConnectionStringPropertiesTest.java

@@ -7,6 +7,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.util.Locale;
 import java.util.stream.Stream;
@@ -151,6 +152,7 @@ public void namespaceConnectionString() {
         final ConnectionStringProperties properties = new ConnectionStringProperties(connectionString);
 
         // Assert
+        Assertions.assertFalse(properties.useDevelopmentEmulator());
         Assertions.assertEquals(HOST, properties.getEndpoint().getHost());
         Assertions.assertEquals(SAS_KEY, properties.getSharedAccessKeyName());
         Assertions.assertEquals(SAS_VALUE, properties.getSharedAccessKey());
@@ -169,6 +171,28 @@ public void parseConnectionString() {
         final ConnectionStringProperties properties = new ConnectionStringProperties(connectionString);
 
         // Assert
+        Assertions.assertFalse(properties.useDevelopmentEmulator());
+        Assertions.assertEquals(HOST, properties.getEndpoint().getHost());
+        Assertions.assertEquals(SAS_KEY, properties.getSharedAccessKeyName());
+        Assertions.assertEquals(SAS_VALUE, properties.getSharedAccessKey());
+        Assertions.assertEquals(EVENT_HUB, properties.getEntityPath());
+    }
+
+    /**
+     * Verifies we can create ConnectionStringProperties with "UseDevelopmentEmulator" specified.
+     */
+    @ValueSource(booleans = { true, false })
+    @ParameterizedTest
+    public void useDevelopmentConnectionString(boolean useEmulator) {
+        // Arrange
+        final String connectionString = getConnectionString(HOSTNAME_URI, EVENT_HUB, SAS_KEY, SAS_VALUE)
+            + "UseDevelopmentEmulator=" + useEmulator + ";";
+
+        // Act
+        final ConnectionStringProperties properties = new ConnectionStringProperties(connectionString);
+
+        // Assert
+        Assertions.assertEquals(useEmulator, properties.useDevelopmentEmulator());
         Assertions.assertEquals(HOST, properties.getEndpoint().getHost());
         Assertions.assertEquals(SAS_KEY, properties.getSharedAccessKeyName());
         Assertions.assertEquals(SAS_VALUE, properties.getSharedAccessKey());

sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorConnectionTest.java

@@ -713,7 +713,7 @@ void setsPropertiesUsingCustomEndpoint() throws IOException {
         final ConnectionOptions connectionOptions = new ConnectionOptions(CREDENTIAL_INFO.getEndpoint().getHost(),
             tokenCredential, CbsAuthorizationType.SHARED_ACCESS_SIGNATURE, "scope", AmqpTransportType.AMQP,
             new AmqpRetryOptions(), ProxyOptions.SYSTEM_DEFAULTS, SCHEDULER, CLIENT_OPTIONS, VERIFY_MODE, PRODUCT,
-            CLIENT_VERSION, hostname, port);
+            CLIENT_VERSION, hostname, port, true);
 
         final ConnectionHandler connectionHandler
             = new ConnectionHandler(connectionId, connectionOptions, peerDetails, AmqpMetricsProvider.noop());

sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/ReactorHandlerProviderTest.java

@@ -159,7 +159,7 @@ public void getsConnectionHandlerAMQP(String hostname, int port, String expected
         final ConnectionOptions connectionOptions = new ConnectionOptions(FULLY_QUALIFIED_DOMAIN_NAME, tokenCredential,
             CbsAuthorizationType.SHARED_ACCESS_SIGNATURE, "scope", AmqpTransportType.AMQP, new AmqpRetryOptions(),
             ProxyOptions.SYSTEM_DEFAULTS, scheduler, CLIENT_OPTIONS, VERIFY_MODE, PRODUCT, CLIENT_VERSION, hostname,
-            port);
+            port, true);
 
         final ConnectionHandler handler = provider.createConnectionHandler(CONNECTION_ID, connectionOptions);
 
@@ -235,7 +235,7 @@ public void getsConnectionHandlerSystemProxy(String hostname, Integer port, Stri
         final ConnectionOptions connectionOptions = new ConnectionOptions(fullyQualifiedDomainName, tokenCredential,
             CbsAuthorizationType.SHARED_ACCESS_SIGNATURE, "scope", AmqpTransportType.AMQP_WEB_SOCKETS,
             new AmqpRetryOptions(), null, scheduler, CLIENT_OPTIONS, VERIFY_MODE, PRODUCT, CLIENT_VERSION, hostname,
-            port);
+            port, true);
 
         when(proxySelector.select(any())).thenAnswer(invocation -> {
             final URI uri = invocation.getArgument(0);
@@ -366,7 +366,7 @@ public void correctPeerDetailsCustomEndpoint() throws MalformedURLException {
             = new ConnectionOptions(HOSTNAME, tokenCredential, CbsAuthorizationType.SHARED_ACCESS_SIGNATURE, "scope",
                 AmqpTransportType.AMQP_WEB_SOCKETS, new AmqpRetryOptions(), ProxyOptions.SYSTEM_DEFAULTS, scheduler,
                 CLIENT_OPTIONS, SslDomain.VerifyMode.VERIFY_PEER_NAME, PRODUCT, CLIENT_VERSION,
-                customEndpoint.getHost(), customEndpoint.getDefaultPort());
+                customEndpoint.getHost(), customEndpoint.getDefaultPort(), true);
 
         final Connection connection = mock(Connection.class);
         when(connection.getRemoteState()).thenReturn(EndpointState.UNINITIALIZED);

sdk/core/azure-core-amqp/src/test/java/com/azure/core/amqp/implementation/handler/WebSocketsConnectionHandlerTest.java

@@ -198,7 +198,7 @@ public void onConnectionInitDifferentEndpoint() {
         final ConnectionOptions connectionOptions = new ConnectionOptions(fullyQualifiedNamespace, tokenCredential,
             CbsAuthorizationType.SHARED_ACCESS_SIGNATURE, "authorization-scope", AmqpTransportType.AMQP_WEB_SOCKETS,
             new AmqpRetryOptions(), ProxyOptions.SYSTEM_DEFAULTS, scheduler, CLIENT_OPTIONS, VERIFY_MODE, PRODUCT,
-            CLIENT_VERSION, customEndpoint, port);
+            CLIENT_VERSION, customEndpoint, port, true);
 
         try (WebSocketsConnectionHandler handler = new WebSocketsConnectionHandler(CONNECTION_ID, connectionOptions,
             peerDetails, AmqpMetricsProvider.noop())) {
@@ -252,7 +252,7 @@ public void websocketConfigureUsesCustomEndpointHostnameAsHostname() {
         final ConnectionOptions connectionOptionsWithCustomEndpoint
             = new ConnectionOptions(HOSTNAME, tokenCredential, CbsAuthorizationType.SHARED_ACCESS_SIGNATURE, "scope",
                 AmqpTransportType.AMQP_WEB_SOCKETS, new AmqpRetryOptions(), ProxyOptions.SYSTEM_DEFAULTS, scheduler,
-                CLIENT_OPTIONS, VERIFY_MODE, PRODUCT, CLIENT_VERSION, customEndpointHostname, 200);
+                CLIENT_OPTIONS, VERIFY_MODE, PRODUCT, CLIENT_VERSION, customEndpointHostname, 200, true);
 
         try (WebSocketsConnectionHandler handler = new WebSocketsConnectionHandler(CONNECTION_ID,
             connectionOptionsWithCustomEndpoint, peerDetails, AmqpMetricsProvider.noop())) {