[Event Hubs] Support Development Emulator (Azure#29195)

### Packages impacted by this PR
@azure/core-amqp & @azure/event-hubs

### Issues associated with this PR
Azure#28548

### Describe the problem that is addressed by this PR
Supporting local development without hitting the service.

### What are the possible designs available to address the problem? If
there are more than one possible design, why was the one in this PR
chosen?
The service is rolling out their own emulator and this PR adds support
for connecting to it. Design is documented in
https://gist.github.com/conniey/680afa1cb66fb3ca56f202241911ba91

### Are there test cases added in this PR? _(If not, why?)_
Yes

### Provide a list of related PRs _(if any)_
N/A

### Command used to generate this PR:**_(Applicable only to SDK release
request PRs)_

### Checklists
- [x] Added impacted package name to the issue description
- [ ] Does this PR needs any fixes in the SDK Generator?** _(If so,
create an Issue in the
[Autorest/typescript](https://github.com/Azure/autorest.typescript)
repository and link it here)_
- [x] Added a changelog (if necessary)

Author: deyaaeldeen

sdk/core/core-amqp/CHANGELOG.md

@@ -2,11 +2,14 @@
 
 ## 4.3.0 (Unreleased)
 
-### Other Changes
-
+### Breaking Changes
 - Moved to ESM core with builds for ESM, CommonJS, React-Native and Browser.
 - Moved unit tests from mocha to vitest.
 
+### Other Changes
+
+- Adds support for connecting to the development emulator. The connection string for the development emulator should have the `";UseDevelopmentEmulator=true"` slug.
+
 ## 4.2.2 (2024-05-02)
 
 ### Bugs Fixed

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

@@ -190,6 +190,7 @@ export interface ConnectionConfig {
     port?: number;
     sharedAccessKey: string;
     sharedAccessKeyName: string;
+    useDevelopmentEmulator?: boolean;
     webSocket?: WebSocketImpl;
     webSocketConstructorOptions?: any;
     webSocketEndpointPath?: string;
@@ -259,6 +260,7 @@ export const Constants: {
     readonly partitionId: "partitionId";
     readonly readOperation: "READ";
     readonly TLS: "tls";
+    readonly TCP: "tcp";
     readonly establishConnection: "establishConnection";
     readonly defaultConsumerGroup: "$default";
     readonly eventHub: "eventhub";
@@ -425,6 +427,9 @@ export enum ErrorNameConditionMapper {
     UnauthorizedError = "amqp:unauthorized-access"
 }
 
+// @public
+export function isLoopbackAddress(address: string): boolean;
+
 // @public
 export function isMessagingError(error: Error | MessagingError): error is MessagingError;
 

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

@@ -177,11 +177,11 @@ export const ConnectionContextBase = {
     }
 
     const connectionOptions: ConnectionOptions = {
-      transport: Constants.TLS,
+      transport: (parameters.config.useDevelopmentEmulator ? Constants.TCP : Constants.TLS) as any,
       host: parameters.config.host,
       hostname: parameters.config.amqpHostname ?? parameters.config.host,
       username: parameters.config.sharedAccessKeyName,
-      port: parameters.config.port ?? 5671,
+      port: parameters.config.port ?? (parameters.config.useDevelopmentEmulator ? 5672 : 5671),
       reconnect: false,
       properties: {
         product: parameters.connectionProperties.product,

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

@@ -3,7 +3,7 @@
 
 import { WebSocketImpl } from "rhea-promise";
 import { isDefined } from "@azure/core-util";
-import { parseConnectionString } from "../util/utils.js";
+import { isLoopbackAddress, parseConnectionString } from "../util/utils.js";
 
 /**
  * Describes the options that can be provided while creating a connection config.
@@ -78,6 +78,20 @@ export interface ConnectionConfig {
    * Options to be passed to the WebSocket constructor
    */
   webSocketConstructorOptions?: any;
+  /**
+   * This should be true only if the connection string contains the slug ";UseDevelopmentEmulator=true"
+   * and the endpoint is a loopback address.
+   */
+  useDevelopmentEmulator?: boolean;
+}
+
+function getHost(endpoint: string): string {
+  const matches = /.*:\/\/([^/]*)/.exec(endpoint);
+  const match = matches?.[1];
+  if (!match) {
+    return isLoopbackAddress(endpoint) ? endpoint : "";
+  }
+  return match;
 }
 
 /**
@@ -102,6 +116,7 @@ export const ConnectionConfig = {
       SharedAccessKeyName: string;
       SharedAccessKey: string;
       EntityPath?: string;
+      UseDevelopmentEmulator?: string;
     }>(connectionString);
     if (!parsedCS.Endpoint) {
       throw new TypeError("Missing Endpoint in Connection String.");
@@ -112,9 +127,10 @@ export const ConnectionConfig = {
     const result: ConnectionConfig = {
       connectionString: connectionString,
       endpoint: parsedCS.Endpoint,
-      host: parsedCS && parsedCS.Endpoint ? (parsedCS.Endpoint.match(".*://([^/]*)") || [])[1] : "",
+      host: getHost(parsedCS.Endpoint),
       sharedAccessKeyName: parsedCS.SharedAccessKeyName,
       sharedAccessKey: parsedCS.SharedAccessKey,
+      useDevelopmentEmulator: parsedCS.UseDevelopmentEmulator === "true",
     };
 
     if (path || parsedCS.EntityPath) {
@@ -140,6 +156,12 @@ export const ConnectionConfig = {
     }
     config.endpoint = String(config.endpoint);
 
+    if (config.useDevelopmentEmulator && !isLoopbackAddress(config.endpoint)) {
+      throw new TypeError(
+        `When using the development environment, the endpoint should be a localhost. Given endpoint is "${config.endpoint}".`,
+      );
+    }
+
     if (!config.host) {
       throw new TypeError("Missing 'host' in configuration");
     }

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

@@ -30,6 +30,7 @@ export {
 export {
   delay,
   parseConnectionString,
+  isLoopbackAddress,
   defaultCancellableLock,
   ParsedOutput,
   WebSocketOptions,

sdk/core/core-amqp/src/util/constants.ts

@@ -29,6 +29,7 @@ export const Constants = {
   partitionId: "partitionId",
   readOperation: "READ",
   TLS: "tls",
+  TCP: "tcp",
   establishConnection: "establishConnection",
   defaultConsumerGroup: "$default",
   eventHub: "eventhub",

sdk/core/core-amqp/src/util/utils.ts

@@ -180,6 +180,15 @@ export async function delay<T>(
   }
 }
 
+/**
+ * Checks if an address is localhost.
+ * @param address - The address to check.
+ * @returns true if the address is localhost, false otherwise.
+ */
+export function isLoopbackAddress(address: string): boolean {
+  return /^(.*:\/\/)?(127\.[\d.]+|[0:]+1|localhost)/.test(address.toLowerCase());
+}
+
 /**
  * @internal
  *

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

@@ -86,6 +86,15 @@ describe("ConnectionConfig", function () {
       }, /Missing Endpoint/);
     });
 
+    it("Parses the connection string for the development emulator", async function () {
+      const config = ConnectionConfig.create(
+        "Endpoint=sb://localhost:6765;SharedAccessKeyName=<< REDACTED >>;SharedAccessKey=<< REDACTED >>;UseDevelopmentEmulator=true",
+      );
+      assert.equal(config.endpoint, "sb://localhost:6765/");
+      assert.equal(config.host, "localhost:6765");
+      assert.isTrue(config.useDevelopmentEmulator);
+    });
+
     describe("Throws error if required connection config properties are not present", function () {
       const connectionString = `
         Endpoint=sb://hostname.servicebus.windows.net/;
@@ -211,6 +220,94 @@ describe("ConnectionConfig", function () {
         assert.equal(config.entityPath, "entityPath", `EntityPath is not a string`);
       });
     });
+
+    describe("Development Emulator Validation", function () {
+      it("Accepts localhost", async function () {
+        const connectionString =
+          "Endpoint=sb://localhost:6765;SharedAccessKeyName=<< REDACTED >>;SharedAccessKey=<< REDACTED >>;UseDevelopmentEmulator=true";
+        const config: ConnectionConfig = {
+          connectionString,
+          endpoint: "localhost:6765/",
+          host: "sb://localhost:6765/",
+          sharedAccessKeyName: "sakName",
+          sharedAccessKey: "abcd",
+          useDevelopmentEmulator: true,
+        };
+        ConnectionConfig.validate(config);
+      });
+
+      it("Accepts 127.0.0.1", async function () {
+        const connectionString =
+          "Endpoint=sb://127.0.0.1:6765;SharedAccessKeyName=<< REDACTED >>;SharedAccessKey=<< REDACTED >>;UseDevelopmentEmulator=true";
+        const config: ConnectionConfig = {
+          connectionString,
+          endpoint: "127.0.0.1:6765/",
+          host: "sb://127.0.0.1:6765",
+          sharedAccessKeyName: "sakName",
+          sharedAccessKey: "abcd",
+          useDevelopmentEmulator: true,
+        };
+        ConnectionConfig.validate(config);
+      });
+
+      it("Accepts 0:0:0:0:0:0:0:1", async function () {
+        const connectionString =
+          "Endpoint=sb://0:0:0:0:0:0:0:1;SharedAccessKeyName=<< REDACTED >>;SharedAccessKey=<< REDACTED >>;UseDevelopmentEmulator=true";
+        const config: ConnectionConfig = {
+          connectionString,
+          endpoint: "0:0:0:0:0:0:0:1/",
+          host: "sb://0:0:0:0:0:0:0:1",
+          sharedAccessKeyName: "sakName",
+          sharedAccessKey: "abcd",
+          useDevelopmentEmulator: true,
+        };
+        ConnectionConfig.validate(config);
+      });
+
+      it("Accepts ::1", async function () {
+        const connectionString =
+          "Endpoint=sb://::1;SharedAccessKeyName=<< REDACTED >>;SharedAccessKey=<< REDACTED >>;UseDevelopmentEmulator=true";
+        const config: ConnectionConfig = {
+          connectionString,
+          endpoint: "::1/",
+          host: "sb://::1",
+          sharedAccessKeyName: "sakName",
+          sharedAccessKey: "abcd",
+          useDevelopmentEmulator: true,
+        };
+        ConnectionConfig.validate(config);
+      });
+
+      it("Accepts localhost with missing scheme", async function () {
+        const connectionString =
+          "Endpoint=localhost:6765;SharedAccessKeyName=<< REDACTED >>;SharedAccessKey=<< REDACTED >>;UseDevelopmentEmulator=true";
+        const config: ConnectionConfig = {
+          connectionString,
+          endpoint: "localhost:6765/",
+          host: "localhost:6765/",
+          sharedAccessKeyName: "sakName",
+          sharedAccessKey: "abcd",
+          useDevelopmentEmulator: true,
+        };
+        ConnectionConfig.validate(config);
+      });
+
+      it("Rejects non-local host", async function () {
+        const connectionString =
+          "Endpoint=sb://hostname.servicebus.windows.net/;SharedAccessKeyName=<< REDACTED >>;SharedAccessKey=<< REDACTED >>;UseDevelopmentEmulator=true";
+        const config: ConnectionConfig = {
+          connectionString,
+          endpoint: "sb://hostname.servicebus.windows.net/",
+          host: "hostname.servicebus.windows.net",
+          sharedAccessKeyName: "sakName",
+          sharedAccessKey: "abcd",
+          useDevelopmentEmulator: true,
+        };
+        assert.throw(() => {
+          ConnectionConfig.validate(config);
+        }, new RegExp("When using the development environment"));
+      });
+    });
   });
 
   describe("SharedAccessSignature", () => {

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

@@ -2,7 +2,7 @@
 // Licensed under the MIT license.
 
 import { describe, it, assert } from "vitest";
-import { CbsClient, ConnectionConfig, ConnectionContextBase } from "../src/index.js";
+import { CbsClient, ConnectionConfig, ConnectionContextBase, Constants } from "../src/index.js";
 import { Connection } from "rhea-promise";
 import type { ConnectionOptions as TlsConnectionOptions } from "node:tls";
 
@@ -27,6 +27,7 @@ describe("ConnectionContextBase", function () {
     assert.isDefined(context.negotiateClaimLock);
     assert.isFalse(context.wasConnectionCloseCalled);
     assert.instanceOf(context.connection, Connection);
+    assert.equal(context.connection.options.transport, "tls");
     assert.equal(context.connection.options.properties!.product, "MSJSClient");
     assert.equal(context.connection.options.properties!["user-agent"], "/js-amqp-client");
     assert.equal(context.connection.options.properties!.version, "1.0.0");
@@ -283,6 +284,25 @@ describe("ConnectionContextBase", function () {
     }, /user-agent string cannot be more than 512 characters/);
   });
 
+  it("disables tls when connecting to the development emulator", async function () {
+    const connectionString =
+      "Endpoint=sb://localhost;SharedAccessKeyName=sakName;SharedAccessKey=sak;EntityPath=ep;UseDevelopmentEmulator=true";
+    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",
+      },
+    });
+    assert.isDefined(context.connection);
+    assert.instanceOf(context.connection, Connection);
+    assert.equal(context.connection.options.transport, Constants.TCP);
+    assert.equal((context.connection.options as TlsConnectionOptions).port, 5672);
+  });
+
   describe("#refreshConnection", function () {
     it("should update fields on the context", function () {
       const connectionString =

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

@@ -0,0 +1,32 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+import { describe, it, assert } from "vitest";
+import { isLoopbackAddress } from "../src/util/utils.js";
+
+describe("isLoopbackAddress", () => {
+  it("returns true for localhost", () => {
+    assert.isTrue(isLoopbackAddress("sb://localhost"));
+  });
+
+  it("returns true for 127-prefix addresses", () => {
+    assert.isTrue(isLoopbackAddress("sb://127.0.0.1"));
+    assert.isTrue(isLoopbackAddress("sb://127.0.0.2"));
+  });
+
+  it("returns true for 0:0:0:0:0:1", () => {
+    assert.isTrue(isLoopbackAddress("sb://0:0:0:0:0:1"));
+  });
+
+  it("returns true for ::1", () => {
+    assert.isTrue(isLoopbackAddress("sb://::1"));
+  });
+
+  it("returns true for localhost with missing scheme", () => {
+    assert.isTrue(isLoopbackAddress("localhost"));
+  });
+
+  it("returns false for other addresses", () => {
+    assert.isFalse(isLoopbackAddress("sb://test.servicebus.windows.net"));
+  });
+});