[Event Hubs] Minor edits (Azure#25040)

Simplifies some code and threads the abort signal to more promises in
link opening. It is part of my work to simplify link opening logic to
get ready to add receiver redirect.

Author: deyaaeldeen

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

@@ -33,6 +33,8 @@ import { EventHubSender } from "./eventHubSender";
 import { getRuntimeInfo } from "./util/runtimeInfo";
 import { isCredential } from "./util/typeGuards";
 import { packageJsonInfo } from "./util/constants";
+import { AbortSignalLike } from "@azure/abort-controller";
+import { createAbortablePromise } from "@azure/core-util";
 
 /**
  * @internal
@@ -77,7 +79,7 @@ export interface ConnectionContext extends ConnectionContextBase {
    * An AMQP link cannot be opened if the AMQP connection
    * is in the process of closing or disconnecting.
    */
-  readyToOpenLink(): Promise<void>;
+  readyToOpenLink(options?: { abortSignal?: AbortSignalLike }): Promise<void>;
   /**
    * Closes all AMQP links, sessions and connection.
    */
@@ -100,7 +102,7 @@ export interface ConnectionContextInternalMembers extends ConnectionContext {
   /**
    * Resolves once the context's connection emits a `disconnected` event.
    */
-  waitForDisconnectedEvent(): Promise<void>;
+  waitForDisconnectedEvent(options?: { abortSignal?: AbortSignalLike }): Promise<void>;
   /**
    * Resolves once the connection has finished being reset.
    * Connections are reset as part of reacting to a `disconnected` event.
@@ -203,27 +205,27 @@ export namespace ConnectionContext {
         // then the rhea connection is in the process of terminating.
         return Boolean(!this.connection.isOpen() && this.connection.isRemoteOpen());
       },
-      async readyToOpenLink() {
+      async readyToOpenLink(optionsArg?: { abortSignal?: AbortSignalLike }) {
         // Check that the connection isn't in the process of closing.
         // This can happen when the idle timeout has been reached but
         // the underlying socket is waiting to be destroyed.
         if (this.isConnectionClosing()) {
           // Wait for the disconnected event that indicates the underlying socket has closed.
-          await this.waitForDisconnectedEvent();
+          await this.waitForDisconnectedEvent(optionsArg);
         }
 
         // Wait for the connection to be reset.
         await this.waitForConnectionReset();
       },
-      waitForDisconnectedEvent() {
-        return new Promise((resolve) => {
+      waitForDisconnectedEvent(optionsArg?: { abortSignal?: AbortSignalLike }) {
+        return createAbortablePromise((resolve) => {
           logger.verbose(
             `[${this.connectionId}] Attempting to reinitialize connection` +
               ` but the connection is in the process of closing.` +
               ` Waiting for the disconnect event before continuing.`
           );
           this.connection.once(ConnectionEvents.disconnected, resolve);
-        });
+        }, optionsArg);
       },
       waitForConnectionReset() {
         // Check if the connection is currently in the process of disconnecting.

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

@@ -415,7 +415,7 @@ export class EventHubReceiver extends LinkEntity {
       );
 
       // Wait for the connectionContext to be ready to open the link.
-      await this._context.readyToOpenLink();
+      await this._context.readyToOpenLink({ abortSignal });
       await this._negotiateClaim({ setTokenRenewal: false, abortSignal, timeoutInMs });
 
       const receiverOptions: CreateReceiverOptions = {
@@ -466,7 +466,7 @@ export class EventHubReceiver extends LinkEntity {
    * Creates the options that need to be specified while creating an AMQP receiver link.
    */
   private _createReceiverOptions(options: CreateReceiverOptions): RheaReceiverOptions {
-    const rcvrOptions: RheaReceiverOptions = {
+    const receiverOptions: RheaReceiverOptions = {
       name: this.name,
       autoaccept: true,
       source: {
@@ -481,26 +481,26 @@ export class EventHubReceiver extends LinkEntity {
     };
 
     if (typeof this.ownerLevel === "number") {
-      rcvrOptions.properties = {
+      receiverOptions.properties = {
         [Constants.attachEpoch]: types.wrap_long(this.ownerLevel),
       };
     }
 
     if (this.options.trackLastEnqueuedEventProperties) {
-      rcvrOptions.desired_capabilities = Constants.enableReceiverRuntimeMetricName;
+      receiverOptions.desired_capabilities = Constants.enableReceiverRuntimeMetricName;
     }
 
     const eventPosition = options.eventPosition || this.eventPosition;
     if (eventPosition) {
       // Set filter on the receiver if event position is specified.
       const filterClause = getEventPositionFilter(eventPosition);
       if (filterClause) {
-        (rcvrOptions.source as any).filter = {
+        (receiverOptions.source as any).filter = {
           "apache.org:selector-filter:string": types.wrap_described(filterClause, 0x468c00000004),
         };
       }
     }
-    return rcvrOptions;
+    return receiverOptions;
   }
 
   /**
@@ -740,7 +740,5 @@ export function waitForEvents(
           .then(() => delay(readIntervalWaitTimeInMs, updatedOptions))
           .then(receivedAfterWait),
         delay(maxWaitTimeInMs, updatedOptions).then(receivedNone),
-      ]).finally(() => {
-        aborter.abort();
-      });
+      ]).finally(() => aborter.abort());
 }

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

@@ -116,8 +116,8 @@ export class LinkEntity {
     setTokenRenewal,
     timeoutInMs,
   }: {
-    setTokenRenewal: boolean | undefined;
-    abortSignal: AbortSignalLike | undefined;
+    setTokenRenewal?: boolean;
+    abortSignal?: AbortSignalLike;
     timeoutInMs: number;
   }): Promise<void> {
     // Acquire the lock and establish a cbs session if it does not exist on the connection.
@@ -137,12 +137,11 @@ export class LinkEntity {
     if (!this._context.cbsSession.isOpen()) {
       await defaultCancellableLock.acquire(
         this._context.cbsSession.cbsLock,
-        () => {
-          return this._context.cbsSession.init({
+        () =>
+          this._context.cbsSession.init({
             abortSignal,
             timeoutInMs: timeoutInMs - (Date.now() - startTime),
-          });
-        },
+          }),
         {
           abortSignal,
           timeoutInMs,
@@ -184,14 +183,11 @@ export class LinkEntity {
     );
     await defaultCancellableLock.acquire(
       this._context.negotiateClaimLock,
-      () => {
-        return this._context.cbsSession.negotiateClaim(
-          this.audience,
-          tokenObject.token,
-          tokenType,
-          { abortSignal, timeoutInMs: timeoutInMs - (Date.now() - startTime) }
-        );
-      },
+      () =>
+        this._context.cbsSession.negotiateClaim(this.audience, tokenObject.token, tokenType, {
+          abortSignal,
+          timeoutInMs: timeoutInMs - (Date.now() - startTime),
+        }),
       {
         abortSignal,
         timeoutInMs: timeoutInMs - (Date.now() - startTime),
@@ -205,7 +201,7 @@ export class LinkEntity {
       this.address
     );
     if (setTokenRenewal) {
-      await this._ensureTokenRenewal();
+      this._ensureTokenRenewal();
     }
   }
 
@@ -222,24 +218,23 @@ export class LinkEntity {
     if (this._tokenRenewalTimer) {
       clearTimeout(this._tokenRenewalTimer);
     }
-    this._tokenRenewalTimer = setTimeout(async () => {
-      try {
-        await this._negotiateClaim({
+    this._tokenRenewalTimer = setTimeout(
+      () =>
+        this._negotiateClaim({
           setTokenRenewal: true,
-          abortSignal: undefined,
-          timeoutInMs: getRetryAttemptTimeoutInMs(undefined),
-        });
-      } catch (err: any) {
-        logger.verbose(
-          "[%s] %s '%s' with address %s, an error occurred while renewing the token: %O",
-          this._context.connectionId,
-          this._type,
-          this.name,
-          this.address,
-          err
-        );
-      }
-    }, this._tokenTimeoutInMs);
+          timeoutInMs: getRetryAttemptTimeoutInMs(),
+        }).catch((err) =>
+          logger.verbose(
+            "[%s] %s '%s' with address %s, an error occurred while renewing the token: %O",
+            this._context.connectionId,
+            this._type,
+            this.name,
+            this.address,
+            err
+          )
+        ),
+      this._tokenTimeoutInMs
+    );
     logger.verbose(
       "[%s] %s '%s' with address %s, has next token renewal in %d milliseconds @(%s).",
       this._context.connectionId,
@@ -289,10 +284,6 @@ export class LinkEntity {
    * @returns The entity type.
    */
   private get _type(): string {
-    let result = "LinkEntity";
-    if ((this as any).constructor && (this as any).constructor.name) {
-      result = (this as any).constructor.name;
-    }
-    return result;
+    return this.constructor.name ?? "LinkEntity";
   }
 }

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

@@ -359,7 +359,7 @@ export class ManagementClient extends LinkEntity {
           this._mgmtReqResLink.sender.name,
           this._mgmtReqResLink.receiver.name
         );
-        await this._ensureTokenRenewal();
+        this._ensureTokenRenewal();
       }
     } catch (err: any) {
       const translatedError = translate(err);

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

@@ -2,18 +2,15 @@
 // Licensed under the MIT license.
 
 import { Constants, RetryOptions } from "@azure/core-amqp";
-import { isDefined } from "@azure/core-util";
 
 /**
  * @internal
  */
-export function getRetryAttemptTimeoutInMs(retryOptions: RetryOptions | undefined): number {
-  const timeoutInMs =
-    !isDefined(retryOptions) ||
-    typeof retryOptions.timeoutInMs !== "number" ||
-    !isFinite(retryOptions.timeoutInMs) ||
-    retryOptions.timeoutInMs < Constants.defaultOperationTimeoutInMs
-      ? Constants.defaultOperationTimeoutInMs
-      : retryOptions.timeoutInMs;
-  return timeoutInMs;
+export function getRetryAttemptTimeoutInMs(retryOptions: RetryOptions = {}): number {
+  const { timeoutInMs } = retryOptions;
+  return typeof timeoutInMs !== "number" ||
+    !isFinite(timeoutInMs) ||
+    timeoutInMs < Constants.defaultOperationTimeoutInMs
+    ? Constants.defaultOperationTimeoutInMs
+    : timeoutInMs;
 }