[event-hubs] re-add support for custom endpoint address (#13287)

We had reverted the custom endpoint address changes in #13096 so that we could do a core-amqp release.

This PR reverts that revert and removes the 'beta' moniker from the event hubs version number.

Original PR: #12909

sdk/core/core-amqp/CHANGELOG.md

@@ -1,7 +1,13 @@
 # Release History
 
-## 2.0.2 (Unreleased)
+## 2.1.0 (Unreleased)
 
+- Adds the ability to configure the `amqpHostname` and `port` that a `ConnectionContextBase` will use when connecting to a service.
+  The `host` field refers to the DNS host or IP address of the service, whereas the `amqpHostname`
+  is the fully qualified host name of the service. Normally `host` and `amqpHostname` 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 `host` field and continue
+  using the public host as the `amqpHostname`.
 
 ## 2.0.1 (2021-01-07)
 

sdk/core/core-amqp/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@azure/core-amqp",
   "sdk-type": "client",
-  "version": "2.0.2",
+  "version": "2.1.0",
   "description": "Common library for amqp based azure sdks like @azure/event-hubs.",
   "author": "Microsoft Corporation",
   "license": "MIT",

sdk/core/core-amqp/review/core-amqp.api.md

@@ -159,10 +159,12 @@ export enum ConditionErrorNameMapper {
 
 // @public
 export interface ConnectionConfig {
+    amqpHostname?: string;
     connectionString: string;
     endpoint: string;
     entityPath?: string;
     host: string;
+    port?: number;
     sharedAccessKey: string;
     sharedAccessKeyName: string;
     webSocket?: WebSocketImpl;

sdk/core/core-amqp/src/ConnectionContextBase.ts

@@ -122,9 +122,9 @@ export const ConnectionContextBase = {
     const connectionOptions: ConnectionOptions = {
       transport: Constants.TLS,
       host: parameters.config.host,
-      hostname: parameters.config.host,
+      hostname: parameters.config.amqpHostname ?? parameters.config.host,
       username: parameters.config.sharedAccessKeyName,
-      port: 5671,
+      port: parameters.config.port ?? 5671,
       reconnect: false,
       properties: {
         product: parameters.connectionProperties.product,
@@ -147,10 +147,11 @@ export const ConnectionContextBase = {
       const host = parameters.config.host;
       const endpoint = parameters.config.webSocketEndpointPath || "";
       const socketOptions = parameters.config.webSocketConstructorOptions || {};
+      const port = parameters.config.port ?? 443;
 
       connectionOptions.webSocketOptions = {
         webSocket: socket,
-        url: `wss://${host}:443/${endpoint}`,
+        url: `wss://${host}:${port}/${endpoint}`,
         protocol: ["AMQPWSB10"],
         options: socketOptions
       };

sdk/core/core-amqp/src/connectionConfig/connectionConfig.ts

@@ -26,9 +26,22 @@ export interface ConnectionConfig {
    */
   endpoint: string;
   /**
-   * @property {string} host - The host "<yournamespace>.servicebus.windows.net".
+   * The DNS hostname or IP address of the service.
+   * Typically of the form "<yournamespace>.servicebus.windows.net" unless connecting
+   * to the service through an intermediary.
    */
   host: string;
+  /**
+   * The fully qualified name of the host to connect to.
+   * This field can be used by AMQP proxies to determine the correct back-end service to
+   * connect the client to.
+   * Typically of the form "<yournamespace>.servicebus.windows.net".
+   */
+  amqpHostname?: string;
+  /**
+   * The port number.
+   */
+  port?: number;
   /**
    * @property {string} connectionString - The connection string.
    */
@@ -135,7 +148,7 @@ export const ConnectionConfig = {
       throw new TypeError("Missing 'entityPath' in configuration");
     }
     if (config.entityPath != undefined) {
-       config.entityPath = String(config.entityPath);
+      config.entityPath = String(config.entityPath);
     }
 
     if (!isSharedAccessSignature(config.connectionString)) {

sdk/core/core-amqp/test/context.spec.ts

@@ -35,6 +35,215 @@ describe("ConnectionContextBase", function() {
     done();
   });
 
+  it("should set host and hostname to the same value by default", function() {
+    const connectionString =
+      "Endpoint=sb://hostname.servicebus.windows.net/;SharedAccessKeyName=sakName;SharedAccessKey=sak;EntityPath=ep";
+    const path = "mypath";
+    const config = ConnectionConfig.create(connectionString, path);
+    const context = ConnectionContextBase.create({
+      config: config,
+      connectionProperties: {
+        product: "MSJSClient",
+        userAgent: "/js-amqp-client",
+        version: "1.0.0"
+      }
+    });
+    should.exist(context.config);
+    should.exist(context.connection);
+    should.exist(context.connectionId);
+    should.exist(context.connectionLock);
+    should.exist(context.negotiateClaimLock);
+    context.connection.options.hostname!.should.equal("hostname.servicebus.windows.net");
+    context.connection.options.host!.should.equal("hostname.servicebus.windows.net");
+    context.wasConnectionCloseCalled.should.equal(false);
+    context.connection.should.instanceOf(Connection);
+    context.connection.options.properties!.product.should.equal("MSJSClient");
+    context.connection.options.properties!["user-agent"].should.equal("/js-amqp-client");
+    context.connection.options.properties!.version.should.equal("1.0.0");
+    context.cbsSession.should.instanceOf(CbsClient);
+  });
+
+  it("should allow setting host and hostname to different values", function() {
+    const connectionString =
+      "Endpoint=sb://hostname.servicebus.windows.net/;SharedAccessKeyName=sakName;SharedAccessKey=sak;EntityPath=ep";
+    const path = "mypath";
+    const config = ConnectionConfig.create(connectionString, path);
+    config.amqpHostname = "127.0.0.1";
+    const context = ConnectionContextBase.create({
+      config: config,
+      connectionProperties: {
+        product: "MSJSClient",
+        userAgent: "/js-amqp-client",
+        version: "1.0.0"
+      }
+    });
+    should.exist(context.config);
+    should.exist(context.connection);
+    should.exist(context.connectionId);
+    should.exist(context.connectionLock);
+    should.exist(context.negotiateClaimLock);
+    context.connection.options.hostname!.should.equal("127.0.0.1");
+    context.connection.options.host!.should.equal("hostname.servicebus.windows.net");
+    context.wasConnectionCloseCalled.should.equal(false);
+    context.connection.should.instanceOf(Connection);
+    context.connection.options.properties!.product.should.equal("MSJSClient");
+    context.connection.options.properties!["user-agent"].should.equal("/js-amqp-client");
+    context.connection.options.properties!.version.should.equal("1.0.0");
+    context.cbsSession.should.instanceOf(CbsClient);
+  });
+
+  it("should allow specifying a port", function() {
+    const connectionString =
+      "Endpoint=sb://hostname.servicebus.windows.net/;SharedAccessKeyName=sakName;SharedAccessKey=sak;EntityPath=ep";
+    const path = "mypath";
+    const config = ConnectionConfig.create(connectionString, path);
+    config.port = 1111;
+    const context = ConnectionContextBase.create({
+      config: config,
+      connectionProperties: {
+        product: "MSJSClient",
+        userAgent: "/js-amqp-client",
+        version: "1.0.0"
+      }
+    });
+    should.exist(context.config);
+    should.exist(context.connection);
+    should.exist(context.connectionId);
+    should.exist(context.connectionLock);
+    should.exist(context.negotiateClaimLock);
+    context.connection.options.port!.should.equal(1111);
+    context.wasConnectionCloseCalled.should.equal(false);
+    context.connection.should.instanceOf(Connection);
+    context.connection.options.properties!.product.should.equal("MSJSClient");
+    context.connection.options.properties!["user-agent"].should.equal("/js-amqp-client");
+    context.connection.options.properties!.version.should.equal("1.0.0");
+    context.cbsSession.should.instanceOf(CbsClient);
+  });
+
+  it("should have a default port (5671)", function() {
+    const connectionString =
+      "Endpoint=sb://hostname.servicebus.windows.net/;SharedAccessKeyName=sakName;SharedAccessKey=sak;EntityPath=ep";
+    const path = "mypath";
+    const config = ConnectionConfig.create(connectionString, path);
+    const context = ConnectionContextBase.create({
+      config: config,
+      connectionProperties: {
+        product: "MSJSClient",
+        userAgent: "/js-amqp-client",
+        version: "1.0.0"
+      }
+    });
+    should.exist(context.config);
+    should.exist(context.connection);
+    should.exist(context.connectionId);
+    should.exist(context.connectionLock);
+    should.exist(context.negotiateClaimLock);
+    context.connection.options.port!.should.equal(5671);
+    context.wasConnectionCloseCalled.should.equal(false);
+    context.connection.should.instanceOf(Connection);
+    context.connection.options.properties!.product.should.equal("MSJSClient");
+    context.connection.options.properties!["user-agent"].should.equal("/js-amqp-client");
+    context.connection.options.properties!.version.should.equal("1.0.0");
+    context.cbsSession.should.instanceOf(CbsClient);
+  });
+
+  it("should allow setting host and hostname to different values when using websockets", function() {
+    const websockets: any = () => {};
+    const connectionString =
+      "Endpoint=sb://hostname.servicebus.windows.net/;SharedAccessKeyName=sakName;SharedAccessKey=sak;EntityPath=ep";
+    const path = "mypath";
+    const config = ConnectionConfig.create(connectionString, path);
+    config.webSocket = websockets;
+    config.amqpHostname = config.host;
+    config.host = "127.0.0.1";
+    const context = ConnectionContextBase.create({
+      config: config,
+      connectionProperties: {
+        product: "MSJSClient",
+        userAgent: "/js-amqp-client",
+        version: "1.0.0"
+      }
+    });
+    should.exist(context.config);
+    should.exist(context.connection);
+    should.exist(context.connectionId);
+    should.exist(context.connectionLock);
+    should.exist(context.negotiateClaimLock);
+    context.connection.options.host!.should.equal("127.0.0.1");
+    context.connection.options.hostname!.should.equal("hostname.servicebus.windows.net");
+    context.wasConnectionCloseCalled.should.equal(false);
+    context.connection.should.instanceOf(Connection);
+    context.connection.options.properties!.product.should.equal("MSJSClient");
+    context.connection.options.properties!["user-agent"].should.equal("/js-amqp-client");
+    context.connection.options.properties!.version.should.equal("1.0.0");
+    context.connection.options.webSocketOptions!.url.should.equal(`wss://127.0.0.1:443/`);
+    context.cbsSession.should.instanceOf(CbsClient);
+  });
+
+  it("should have a default port when using websockets (443)", function() {
+    const websockets: any = () => {};
+    const connectionString =
+      "Endpoint=sb://hostname.servicebus.windows.net/;SharedAccessKeyName=sakName;SharedAccessKey=sak;EntityPath=ep";
+    const path = "mypath";
+    const config = ConnectionConfig.create(connectionString, path);
+    config.webSocket = websockets;
+    const context = ConnectionContextBase.create({
+      config: config,
+      connectionProperties: {
+        product: "MSJSClient",
+        userAgent: "/js-amqp-client",
+        version: "1.0.0"
+      }
+    });
+    should.exist(context.config);
+    should.exist(context.connection);
+    should.exist(context.connectionId);
+    should.exist(context.connectionLock);
+    should.exist(context.negotiateClaimLock);
+    context.wasConnectionCloseCalled.should.equal(false);
+    context.connection.should.instanceOf(Connection);
+    context.connection.options.properties!.product.should.equal("MSJSClient");
+    context.connection.options.properties!["user-agent"].should.equal("/js-amqp-client");
+    context.connection.options.properties!.version.should.equal("1.0.0");
+    context.connection.options.webSocketOptions!.url.should.equal(
+      `wss://hostname.servicebus.windows.net:443/`
+    );
+    context.cbsSession.should.instanceOf(CbsClient);
+  });
+
+  it("should allow specifying a port when using websockets", function() {
+    const websockets: any = () => {};
+    const connectionString =
+      "Endpoint=sb://hostname.servicebus.windows.net/;SharedAccessKeyName=sakName;SharedAccessKey=sak;EntityPath=ep";
+    const path = "mypath";
+    const config = ConnectionConfig.create(connectionString, path);
+    config.webSocket = websockets;
+    config.port = 1111;
+    const context = ConnectionContextBase.create({
+      config: config,
+      connectionProperties: {
+        product: "MSJSClient",
+        userAgent: "/js-amqp-client",
+        version: "1.0.0"
+      }
+    });
+    should.exist(context.config);
+    should.exist(context.connection);
+    should.exist(context.connectionId);
+    should.exist(context.connectionLock);
+    should.exist(context.negotiateClaimLock);
+    context.connection.options.port!.should.equal(1111);
+    context.wasConnectionCloseCalled.should.equal(false);
+    context.connection.should.instanceOf(Connection);
+    context.connection.options.properties!.product.should.equal("MSJSClient");
+    context.connection.options.properties!["user-agent"].should.equal("/js-amqp-client");
+    context.connection.options.properties!.version.should.equal("1.0.0");
+    context.connection.options.webSocketOptions!.url.should.equal(
+      `wss://hostname.servicebus.windows.net:1111/`
+    );
+    context.cbsSession.should.instanceOf(CbsClient);
+  });
+
   if (isNode) {
     it("should accept a websocket constructor in Node", async () => {
       const connectionString =

sdk/eventhub/event-hubs/CHANGELOG.md

@@ -1,6 +1,12 @@
 # Release History
 
-## 5.3.2 (Unreleased)
+## 5.4.0 (Unreleased)
+
+- Adds the `customEndpointAddress` field to `EventHubClientOptions`.
+  This allows for specifying a custom endpoint to use when communicating
+  with the Event Hubs service, which is useful when your network does not
+  allow communicating to the standard Event Hubs endpoint.
+  Resolves [#12901](https://github.com/Azure/azure-sdk-for-js/issues/12901).
 
 - Updates documentation for `EventData` to call out that the `body` field
   must be converted to a byte array or `Buffer` when cross-language

sdk/eventhub/event-hubs/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@azure/event-hubs",
   "sdk-type": "client",
-  "version": "5.3.2",
+  "version": "5.4.0",
   "description": "Azure Event Hubs SDK for JS.",
   "author": "Microsoft Corporation",
   "license": "MIT",
@@ -89,7 +89,7 @@
   },
   "dependencies": {
     "@azure/abort-controller": "^1.0.0",
-    "@azure/core-amqp": "^2.0.0",
+    "@azure/core-amqp": "^2.1.0",
     "@azure/core-asynciterator-polyfill": "^1.0.0",
     "@azure/core-tracing": "1.0.0-preview.9",
     "@azure/core-auth": "^1.1.3",

sdk/eventhub/event-hubs/review/event-hubs.api.md

@@ -74,6 +74,7 @@ export interface EventDataBatch {
 
 // @public
 export interface EventHubClientOptions {
+    customEndpointAddress?: string;
     retryOptions?: RetryOptions;
     userAgent?: string;
     webSocketOptions?: WebSocketOptions;

sdk/eventhub/event-hubs/rollup.base.config.js

@@ -19,7 +19,7 @@ const input = "dist-esm/src/index.js";
 const production = process.env.NODE_ENV === "production";
 
 export function nodeConfig(test = false) {
-  const externalNodeBuiltins = ["events", "util", "os"];
+  const externalNodeBuiltins = ["events", "util", "os", "url"];
   const baseConfig = {
     input: input,
     external: depNames.concat(externalNodeBuiltins),

sdk/eventhub/event-hubs/src/connectionContext.ts

@@ -501,6 +501,10 @@ export function createConnectionContext(
     config = EventHubConnectionConfig.create(connectionString);
   }
 
+  if (options?.customEndpointAddress) {
+    EventHubConnectionConfig.setCustomEndpointAddress(config, options.customEndpointAddress);
+  }
+
   ConnectionConfig.validate(config);
 
   return ConnectionContext.create(config, credential, options);

sdk/eventhub/event-hubs/src/eventhubConnectionConfig.ts

@@ -3,11 +3,14 @@
 /* eslint-disable eqeqeq */
 
 import { ConnectionConfig } from "@azure/core-amqp";
+import { parseEndpoint } from "./util/parseEndpoint";
 
 /**
  * Describes the connection config object that is created after parsing an EventHub connection
  * string. It also provides some convenience methods for getting the address and audience for
  * different entities.
+ * @internal
+ * @ignore
  */
 export interface EventHubConnectionConfig extends ConnectionConfig {
   /**
@@ -65,7 +68,8 @@ export interface EventHubConnectionConfig extends ConnectionConfig {
  * Describes the connection config object that is created after parsing an EventHub connection
  * string. It also provides some convenience methods for getting the address and audience for
  * different entities.
- * @module EventHubConnectionConfig
+ * @internal
+ * @ignore
  */
 export const EventHubConnectionConfig = {
   /**
@@ -141,6 +145,22 @@ export const EventHubConnectionConfig = {
     return config as EventHubConnectionConfig;
   },
 
+  /**
+   * Updates the provided EventHubConnectionConfig to use the custom endpoint address.
+   * @param config An existing connection configuration to be updated.
+   * @param customEndpointAddress The custom endpoint address to use.
+   */
+  setCustomEndpointAddress(config: EventHubConnectionConfig, customEndpointAddress: string): void {
+    // The amqpHostname should match the host prior to using the custom endpoint.
+    config.amqpHostname = config.host;
+    const { hostname, port } = parseEndpoint(customEndpointAddress);
+    // Since we specify the port separately, set host to the customEndpointAddress hostname.
+    config.host = hostname;
+    if (port) {
+      config.port = parseInt(port, 10);
+    }
+  },
+
   /**
    * Validates the properties of connection config.
    * @param {ConnectionConfig} config The connection config to be validated.

sdk/eventhub/event-hubs/src/models/public.ts

@@ -111,18 +111,24 @@ export enum CloseReason {
  */
 export interface EventHubClientOptions {
   /**
-   * @property
+   * A custom endpoint to use when connecting to the Event Hubs service.
+   * This can be useful when your network does not allow connecting to the
+   * standard Azure Event Hubs endpoint address, but does allow connecting
+   * through an intermediary.
+   *
+   * Example: "https://my.custom.endpoint:100/"
+   */
+  customEndpointAddress?: string;
+  /**
    * Options to configure the retry policy for all the operations on the client.
    * For example, `{ "maxRetries": 4 }` or `{ "maxRetries": 4, "retryDelayInMs": 30000 }`.
    */
   retryOptions?: RetryOptions;
   /**
-   * @property
    * Options to configure the channelling of the AMQP connection over Web Sockets.
    */
   webSocketOptions?: WebSocketOptions;
   /**
-   * @property
    * Value that is appended to the built in user agent string that is passed to the Event Hubs service.
    */
   userAgent?: string;

sdk/eventhub/event-hubs/src/util/constants.ts

@@ -6,5 +6,5 @@
  */
 export const packageJsonInfo = {
   name: "@azure/event-hubs",
-  version: "5.3.2"
+  version: "5.4.0"
 };

sdk/eventhub/event-hubs/src/util/parseEndpoint.ts

@@ -0,0 +1,20 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+/**
+ * Parses the host, hostname, and port from an endpoint.
+ * @param endpoint And endpoint to parse.
+ * @hidden
+ * @internal
+ */
+export function parseEndpoint(endpoint: string): { host: string; hostname: string; port?: string } {
+  const hostMatch = endpoint.match(/.*:\/\/([^/]*)/);
+  if (!hostMatch) {
+    throw new TypeError(`Invalid endpoint missing host: ${endpoint}`);
+  }
+
+  const [, host] = hostMatch;
+  const [hostname, port] = host.split(":");
+
+  return { host, hostname, port };
+}

sdk/eventhub/event-hubs/test/client.spec.ts

@@ -89,6 +89,40 @@ describe("Create EventHubConsumerClient", function(): void {
     should.equal(client.eventHubName, "my-event-hub-name");
     should.equal(client.fullyQualifiedNamespace, "test.servicebus.windows.net");
   });
+
+  it("respects customEndpointAddress when using connection string", () => {
+    const client = new EventHubConsumerClient(
+      "dummy",
+      "Endpoint=sb://test.servicebus.windows.net;SharedAccessKeyName=b;SharedAccessKey=c;EntityPath=my-event-hub-name",
+      { customEndpointAddress: "sb://foo.private.bar:111" }
+    );
+    client.should.be.an.instanceof(EventHubConsumerClient);
+    client["_context"].config.host.should.equal("foo.private.bar");
+    client["_context"].config.amqpHostname!.should.equal("test.servicebus.windows.net");
+    client["_context"].config.port!.should.equal(111);
+  });
+
+  it("respects customEndpointAddress when using credentials", () => {
+    const dummyCredential: TokenCredential = {
+      getToken: async () => {
+        return {
+          token: "boo",
+          expiresOnTimestamp: 12324
+        };
+      }
+    };
+    const client = new EventHubConsumerClient(
+      "dummy",
+      "test.servicebus.windows.net",
+      "my-event-hub-name",
+      dummyCredential,
+      { customEndpointAddress: "sb://foo.private.bar:111" }
+    );
+    client.should.be.an.instanceof(EventHubConsumerClient);
+    client["_context"].config.host.should.equal("foo.private.bar");
+    client["_context"].config.amqpHostname!.should.equal("test.servicebus.windows.net");
+    client["_context"].config.port!.should.equal(111);
+  });
 });
 
 describe("Create EventHubProducerClient", function(): void {
@@ -155,6 +189,38 @@ describe("Create EventHubProducerClient", function(): void {
     should.equal(client.eventHubName, "my-event-hub-name");
     should.equal(client.fullyQualifiedNamespace, "test.servicebus.windows.net");
   });
+
+  it("respects customEndpointAddress when using connection string", () => {
+    const client = new EventHubProducerClient(
+      "Endpoint=sb://test.servicebus.windows.net;SharedAccessKeyName=b;SharedAccessKey=c;EntityPath=my-event-hub-name",
+      { customEndpointAddress: "sb://foo.private.bar:111" }
+    );
+    client.should.be.an.instanceof(EventHubProducerClient);
+    client["_context"].config.host.should.equal("foo.private.bar");
+    client["_context"].config.amqpHostname!.should.equal("test.servicebus.windows.net");
+    client["_context"].config.port!.should.equal(111);
+  });
+
+  it("respects customEndpointAddress when using credentials", () => {
+    const dummyCredential: TokenCredential = {
+      getToken: async () => {
+        return {
+          token: "boo",
+          expiresOnTimestamp: 12324
+        };
+      }
+    };
+    const client = new EventHubProducerClient(
+      "test.servicebus.windows.net",
+      "my-event-hub-name",
+      dummyCredential,
+      { customEndpointAddress: "sb://foo.private.bar:111" }
+    );
+    client.should.be.an.instanceof(EventHubProducerClient);
+    client["_context"].config.host.should.equal("foo.private.bar");
+    client["_context"].config.amqpHostname!.should.equal("test.servicebus.windows.net");
+    client["_context"].config.port!.should.equal(111);
+  });
 });
 
 describe("EventHubConsumerClient with non existent namespace", function(): void {

sdk/eventhub/event-hubs/test/config.spec.ts

@@ -75,5 +75,112 @@ describe("ConnectionConfig", function() {
 
       done();
     });
+
+    describe("setCustomEndpointAddress", () => {
+      it("overwrites host", () => {
+        const config = EventHubConnectionConfig.create(
+          "Endpoint=sb://hostname.servicebus.windows.net/;SharedAccessKeyName=sakName;SharedAccessKey=sak;EntityPath=ep"
+        );
+        config.should.have.property("host").that.equals("hostname.servicebus.windows.net");
+        config.should.have.property("sharedAccessKeyName").that.equals("sakName");
+        config.should.have.property("sharedAccessKey").that.equals("sak");
+        config.should.have.property("entityPath").that.equals("ep");
+
+        config.getManagementAddress().should.equal("ep/$management");
+        config.getSenderAddress().should.equal("ep");
+        config.getSenderAddress("0").should.equal("ep/Partitions/0");
+        config.getSenderAddress(0).should.equal("ep/Partitions/0");
+        config.getReceiverAddress("0").should.equal("ep/ConsumerGroups/$default/Partitions/0");
+        config.getReceiverAddress(0).should.equal("ep/ConsumerGroups/$default/Partitions/0");
+        config.getReceiverAddress("0", "cg").should.equal("ep/ConsumerGroups/cg/Partitions/0");
+        config.getReceiverAddress(0, "cg").should.equal("ep/ConsumerGroups/cg/Partitions/0");
+
+        config
+          .getManagementAudience()
+          .should.equal("sb://hostname.servicebus.windows.net/ep/$management");
+        config.getSenderAudience().should.equal("sb://hostname.servicebus.windows.net/ep");
+        config
+          .getSenderAudience("0")
+          .should.equal("sb://hostname.servicebus.windows.net/ep/Partitions/0");
+        config
+          .getSenderAudience(0)
+          .should.equal("sb://hostname.servicebus.windows.net/ep/Partitions/0");
+        config
+          .getReceiverAudience("0")
+          .should.equal(
+            "sb://hostname.servicebus.windows.net/ep/ConsumerGroups/$default/Partitions/0"
+          );
+        config
+          .getReceiverAudience(0)
+          .should.equal(
+            "sb://hostname.servicebus.windows.net/ep/ConsumerGroups/$default/Partitions/0"
+          );
+        config
+          .getReceiverAudience("0", "cg")
+          .should.equal("sb://hostname.servicebus.windows.net/ep/ConsumerGroups/cg/Partitions/0");
+        config
+          .getReceiverAudience(0, "cg")
+          .should.equal("sb://hostname.servicebus.windows.net/ep/ConsumerGroups/cg/Partitions/0");
+
+        EventHubConnectionConfig.setCustomEndpointAddress(config, "https://foo.private.endpoint");
+        config.should.have.property("amqpHostname").that.equals("hostname.servicebus.windows.net");
+        config.should.have.property("host").that.equals("foo.private.endpoint");
+        config.should.not.have.property("port");
+      });
+
+      it("overwrites host and port", () => {
+        const config = EventHubConnectionConfig.create(
+          "Endpoint=sb://hostname.servicebus.windows.net/;SharedAccessKeyName=sakName;SharedAccessKey=sak;EntityPath=ep"
+        );
+        config.should.have.property("host").that.equals("hostname.servicebus.windows.net");
+        config.should.have.property("sharedAccessKeyName").that.equals("sakName");
+        config.should.have.property("sharedAccessKey").that.equals("sak");
+        config.should.have.property("entityPath").that.equals("ep");
+
+        config.getManagementAddress().should.equal("ep/$management");
+        config.getSenderAddress().should.equal("ep");
+        config.getSenderAddress("0").should.equal("ep/Partitions/0");
+        config.getSenderAddress(0).should.equal("ep/Partitions/0");
+        config.getReceiverAddress("0").should.equal("ep/ConsumerGroups/$default/Partitions/0");
+        config.getReceiverAddress(0).should.equal("ep/ConsumerGroups/$default/Partitions/0");
+        config.getReceiverAddress("0", "cg").should.equal("ep/ConsumerGroups/cg/Partitions/0");
+        config.getReceiverAddress(0, "cg").should.equal("ep/ConsumerGroups/cg/Partitions/0");
+
+        config
+          .getManagementAudience()
+          .should.equal("sb://hostname.servicebus.windows.net/ep/$management");
+        config.getSenderAudience().should.equal("sb://hostname.servicebus.windows.net/ep");
+        config
+          .getSenderAudience("0")
+          .should.equal("sb://hostname.servicebus.windows.net/ep/Partitions/0");
+        config
+          .getSenderAudience(0)
+          .should.equal("sb://hostname.servicebus.windows.net/ep/Partitions/0");
+        config
+          .getReceiverAudience("0")
+          .should.equal(
+            "sb://hostname.servicebus.windows.net/ep/ConsumerGroups/$default/Partitions/0"
+          );
+        config
+          .getReceiverAudience(0)
+          .should.equal(
+            "sb://hostname.servicebus.windows.net/ep/ConsumerGroups/$default/Partitions/0"
+          );
+        config
+          .getReceiverAudience("0", "cg")
+          .should.equal("sb://hostname.servicebus.windows.net/ep/ConsumerGroups/cg/Partitions/0");
+        config
+          .getReceiverAudience(0, "cg")
+          .should.equal("sb://hostname.servicebus.windows.net/ep/ConsumerGroups/cg/Partitions/0");
+
+        EventHubConnectionConfig.setCustomEndpointAddress(
+          config,
+          "https://foo.private.endpoint:1111"
+        );
+        config.should.have.property("amqpHostname").that.equals("hostname.servicebus.windows.net");
+        config.should.have.property("host").that.equals("foo.private.endpoint");
+        config.should.have.property("port").that.equals(1111);
+      });
+    });
   });
 });

sdk/eventhub/event-hubs/test/impl/parseEndpoint.spec.ts

@@ -0,0 +1,40 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+import { parseEndpoint } from "../../src/util/parseEndpoint";
+import chai from "chai";
+const should = chai.should();
+
+describe("parseEndpoint", () => {
+  it("throws an error for invalid inputs", () => {
+    should.throw(() => parseEndpoint(""), /Invalid endpoint/);
+    should.throw(() => parseEndpoint("missing-protocol"), /Invalid endpoint/);
+    should.throw(() => parseEndpoint("//missing-protocol"), /Invalid endpoint/);
+  });
+
+  it("extracts host, hostname, and port", () => {
+    parseEndpoint("sb://test.servicebus.windows.net:5671").should.eql({
+      host: "test.servicebus.windows.net:5671",
+      hostname: "test.servicebus.windows.net",
+      port: "5671"
+    } as ReturnType<typeof parseEndpoint>);
+
+    parseEndpoint("https://127.0.0.1:5671").should.eql({
+      host: "127.0.0.1:5671",
+      hostname: "127.0.0.1",
+      port: "5671"
+    } as ReturnType<typeof parseEndpoint>);
+
+    parseEndpoint("amqps://127.0.0.1:5671/path/?query=foo").should.eql({
+      host: "127.0.0.1:5671",
+      hostname: "127.0.0.1",
+      port: "5671"
+    } as ReturnType<typeof parseEndpoint>);
+
+    parseEndpoint("wss://127.0.0.1/path/?query=foo").should.eql({
+      host: "127.0.0.1",
+      hostname: "127.0.0.1",
+      port: undefined
+    } as ReturnType<typeof parseEndpoint>);
+  });
+});

sdk/servicebus/service-bus/package.json

@@ -95,7 +95,7 @@
   },
   "dependencies": {
     "@azure/abort-controller": "^1.0.0",
-    "@azure/core-amqp": "^2.0.0",
+    "@azure/core-amqp": "^2.1.0",
     "@azure/core-asynciterator-polyfill": "^1.0.0",
     "@azure/core-http": "^1.2.0",
     "@azure/core-tracing": "1.0.0-preview.9",