[core-amqp] set the max listener limit to 1000 for sender and receiver

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

@@ -1,7 +1,17 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
-import { Connection, ConnectionOptions, generate_uuid } from "rhea-promise";
+import {
+  AwaitableSender,
+  Connection,
+  ConnectionOptions,
+  CreateAwaitableSenderOptions,
+  CreateReceiverOptions,
+  CreateSenderOptions,
+  Receiver,
+  Sender,
+  generate_uuid,
+} from "rhea-promise";
 import { getFrameworkInfo, getPlatformInfo } from "./util/runtimeInfo";
 import { CbsClient } from "./cbs";
 import { ConnectionConfig } from "./connectionConfig/connectionConfig";
@@ -100,6 +110,51 @@ export interface CreateConnectionContextBaseParameters {
   operationTimeoutInMs?: number;
 }
 
+class CoreAmqpConnection extends Connection {
+  /**
+   * Creates an amqp sender link. Max listener limit on the sender is set to 1000 because the
+   * default value of 10 in NodeJS is too low.
+   * @param options - Optional parameters to create a sender link.
+   * @returns Promise<Sender>.
+   */
+  async createSender(options?: CreateSenderOptions): Promise<Sender> {
+    const sender = await super.createSender(options);
+    sender.setMaxListeners(1000);
+    return sender;
+  }
+
+  /**
+   * Creates an awaitable amqp sender. Max listener limit on the sender is set to 1000 because the
+   * default value of 10 in NodeJS is too low.
+   * @param options - Optional parameters to create an awaitable sender link.
+   * - If `onError` and `onSessionError` handlers are not provided then the `AwaitableSender` will
+   * clear the timer and reject the Promise for all the entries of inflight send operation in its
+   * `deliveryDispositionMap`.
+   * - If the user is handling the reconnection of sender link or the underlying connection in it's
+   * app, then the `onError` and `onSessionError` handlers must be provided by the user and (s)he
+   * shall be responsible of clearing the `deliveryDispositionMap` of inflight `send()` operation.
+   *
+   * @returns Promise<AwaitableSender>.
+   */
+  async createAwaitableSender(options?: CreateAwaitableSenderOptions): Promise<AwaitableSender> {
+    const sender = await super.createAwaitableSender(options);
+    sender.setMaxListeners(1000);
+    return sender;
+  }
+
+  /**
+   * Creates an amqp receiver link. Max listener limit on the sender is set to 1000 because the
+   * default value of 10 in NodeJS is too low.
+   * @param options - Optional parameters to create a receiver link.
+   * @returns Promise<Receiver>.
+   */
+  async createReceiver(options?: CreateReceiverOptions): Promise<Receiver> {
+    const receiver = await super.createReceiver(options);
+    receiver.setMaxListeners(1000);
+    return receiver;
+  }
+}
+
 // eslint-disable-next-line @typescript-eslint/no-redeclare -- renaming constant would be a breaking change.
 export const ConnectionContextBase = {
   /**
@@ -157,7 +212,7 @@ export const ConnectionContextBase = {
       };
     }
 
-    const connection = new Connection(connectionOptions);
+    const connection = new CoreAmqpConnection(connectionOptions);
     const connectionLock = `${Constants.establishConnection}-${generate_uuid()}`;
     const connectionContextBase: ConnectionContextBase = {
       wasConnectionCloseCalled: false,
@@ -168,7 +223,7 @@ export const ConnectionContextBase = {
       cbsSession: new CbsClient(connection, connectionLock),
       config: parameters.config,
       refreshConnection() {
-        const newConnection = new Connection(connectionOptions);
+        const newConnection = new CoreAmqpConnection(connectionOptions);
         const newConnectionLock = `${Constants.establishConnection}-${generate_uuid()}`;
         this.wasConnectionCloseCalled = false;
         this.connectionLock = newConnectionLock;