[Event Hubs] Use AwaitableSender in lieu of Sender

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

@@ -481,14 +481,24 @@ export const retryableErrors: string[] = [
   "ServerBusyError",
   "ServiceUnavailableError",
   "OperationCancelledError",
+
+  // OperationTimeoutError occurs when the service fails to respond within a given timeframe.
+  // Since reasons for such failures can be transient, this is treated as a retryable error.
   "OperationTimeoutError",
+
   "SenderBusyError",
   "MessagingError",
   "DetachForcedError",
   "ConnectionForcedError",
-  "TransferLimitExceededError"
+  "TransferLimitExceededError",
+
+  // InsufficientCreditError occurs when the number of credits available on Rhea link is insufficient.
+  // Since reasons for such shortage can be transient such as for pending delivery of messages, this is treated as a retryable error.
+  "InsufficientCreditError"
 ];
 
+export const nonRetryableErrors: string[] = ["AbortError"];
+
 /**
  * Maps some SytemErrors to amqp error conditions
  * @enum SystemErrorConditionMapper
@@ -539,6 +549,29 @@ function isBrowserWebsocketError(err: any): boolean {
   return result;
 }
 
+/**
+ * @internal
+ * Checks if given object maps to a valid custom error. If yes, configures and returns the appropriate error instance, else returns `undefined`.
+ * @param err
+ */
+function getCustomError(err: AmqpError | Error): MessagingError | undefined {
+  const error: MessagingError = err as MessagingError;
+  if (
+    // instanceof checks on custom Errors doesn't work without manually setting the prototype within the error.
+    // Must do a name check until the custom error is updated, and that doesn't break compatibility
+    // https://github.com/Microsoft/TypeScript/wiki/Breaking-Changes#extending-built-ins-like-error-array-and-map-may-no-longer-work
+    retryableErrors.indexOf((err as Error).name) > 0
+  ) {
+    error.retryable = true;
+    return error;
+  } else if (nonRetryableErrors.indexOf((err as Error).name) > 0) {
+    error.retryable = false;
+    return error;
+  } else {
+    return undefined;
+  }
+}
+
 /**
  * Translates the AQMP error received at the protocol layer or a generic Error into a MessagingError.
  *
@@ -551,30 +584,16 @@ export function translate(err: AmqpError | Error): MessagingError {
     return err as MessagingError;
   }
 
-  let error: MessagingError = err as MessagingError;
-
-  // OperationTimeoutError occurs when the service fails to respond within a given timeframe.
-  // Since reasons for such failures can be transient, this is treated as a retryable error.
-  if (
-    // instanceof checks on custom Errors doesn't work without manually setting the prototype within the error.
-    // Must do a name check until OperationTimeoutError is updated, and that doesn't break compatibility
-    // https://github.com/Microsoft/TypeScript/wiki/Breaking-Changes#extending-built-ins-like-error-array-and-map-may-no-longer-work
-    (err as Error).name === "OperationTimeoutError"
-  ) {
-    error.retryable = true;
-    return error;
+  const customError = getCustomError(err);
+  if (customError) {
+    return customError;
   }
 
+  let error: MessagingError = err as MessagingError;
+
   // Built-in errors like TypeError and RangeError should not be retryable as these indicate issues
   // with user input and not an issue with the Messaging process.
-  if (
-    err instanceof TypeError ||
-    err instanceof RangeError ||
-    // instanceof checks on custom Errors doesn't work without manually setting the prototype within the error.
-    // Must do a name check until AbortError is updated, and that doesn't break compatibility
-    // https://github.com/Microsoft/TypeScript/wiki/Breaking-Changes#extending-built-ins-like-error-array-and-map-may-no-longer-work
-    (err as Error).name === "AbortError"
-  ) {
+  if (err instanceof TypeError || err instanceof RangeError) {
     error.retryable = false;
     return error;
   }

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

@@ -6,6 +6,7 @@
 
 import { AbortSignalLike } from '@azure/abort-controller';
 import { AmqpError } from 'rhea-promise';
+import { AwaitableSender } from 'rhea-promise';
 import { ConnectionContextBase } from '@azure/core-amqp';
 import { DataTransformer } from '@azure/core-amqp';
 import { DefaultDataTransformer } from '@azure/core-amqp';
@@ -15,7 +16,6 @@ import { EventHubConnectionConfig } from '@azure/core-amqp';
 import { MessagingError } from '@azure/core-amqp';
 import { Receiver } from 'rhea-promise';
 import { ReceiverOptions } from 'rhea-promise';
-import { Sender } from 'rhea-promise';
 import { SharedKeyCredential } from '@azure/core-amqp';
 import { TokenCredential } from '@azure/core-amqp';
 import { TokenType } from '@azure/core-amqp';

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

@@ -4,17 +4,15 @@
 import uuid from "uuid/v4";
 import * as log from "./log";
 import {
-  Sender,
+  AwaitableSender,
   EventContext,
   OnAmqpEvent,
-  SenderOptions as RheaSenderOptions,
-  SenderEvents,
+  AwaitableSenderOptions,
   message,
   AmqpError
 } from "rhea-promise";
 import {
   defaultLock,
-  Func,
   retry,
   translate,
   AmqpMessage,
@@ -36,6 +34,7 @@ import { getRetryAttemptTimeoutInMs, RetryOptions } from "./eventHubClient";
  */
 interface CreateSenderOptions {
   newName?: boolean;
+  sendTimeoutInSeconds?: number;
 }
 
 /**
@@ -79,7 +78,7 @@ export class EventHubSender extends LinkEntity {
    * @property [_sender] The AMQP sender link.
    * @private
    */
-  private _sender?: Sender;
+  private _sender?: AwaitableSender;
 
   /**
    * Creates a new EventHubSender instance.
@@ -278,7 +277,7 @@ export class EventHubSender extends LinkEntity {
       }
       if (shouldReopen) {
         await defaultLock.acquire(this.senderLock, () => {
-          const options: RheaSenderOptions = this._createSenderOptions({
+          const options: AwaitableSenderOptions = this._createSenderOptions({
             newName: true
           });
           // shall retry forever at an interval of 15 seconds if the error is a retryable error
@@ -524,17 +523,18 @@ export class EventHubSender extends LinkEntity {
     );
   }
 
-  private _createSenderOptions(options: CreateSenderOptions): RheaSenderOptions {
+  private _createSenderOptions(options: CreateSenderOptions): AwaitableSenderOptions {
     if (options.newName) this.name = `${uuid()}`;
-    const srOptions: RheaSenderOptions = {
+    const srOptions: AwaitableSenderOptions = {
       name: this.name,
       target: {
         address: this.address
       },
       onError: this._onAmqpError,
       onClose: this._onAmqpClose,
       onSessionError: this._onSessionError,
-      onSessionClose: this._onSessionClose
+      onSessionClose: this._onSessionClose,
+      sendTimeoutInSeconds: options.sendTimeoutInSeconds
     };
     log.sender("Creating sender with options: %O", srOptions);
     return srOptions;
@@ -557,101 +557,35 @@ export class EventHubSender extends LinkEntity {
     const abortSignal: AbortSignalLike | undefined = options.abortSignal;
     const sendEventPromise = () =>
       new Promise<void>(async (resolve, reject) => {
-        let waitTimer: any;
-
-        let onRejected: Func<EventContext, void>;
-        let onReleased: Func<EventContext, void>;
-        let onModified: Func<EventContext, void>;
-        let onAccepted: Func<EventContext, void>;
-        let onAborted: () => void;
-
         const rejectOnAbort = () => {
           const desc: string =
-            `[${this._context.connectionId}] The send operation on the Sender "${
-              this.name
-            }" with ` + `address "${this.address}" has been cancelled by the user.`;
+            `[${this._context.connectionId}] The send operation on the Sender "${this.name}" with ` +
+            `address "${this.address}" has been cancelled by the user.`;
           log.error(desc);
           return reject(new AbortError("The send operation has been cancelled by the user."));
         };
 
         if (abortSignal && abortSignal.aborted) {
           // operation has been cancelled, so exit quickly
-          return rejectOnAbort();
-        }
-
-        onAborted = () => {
-          removeListeners();
           rejectOnAbort();
-        };
-
-        onAccepted = (context: EventContext) => {
-          // Since we will be adding listener for accepted and rejected event every time
-          // we send a message, we need to remove listener for both the events.
-          // This will ensure duplicate listeners are not added for the same event.
-          removeListeners();
-          log.sender(
-            "[%s] Sender '%s', got event accepted.",
-            this._context.connectionId,
-            this.name
-          );
-          resolve();
-        };
-
-        onRejected = (context: EventContext) => {
-          removeListeners();
-          log.error("[%s] Sender '%s', got event rejected.", this._context.connectionId, this.name);
-          const err = translate(context!.delivery!.remote_state!.error);
-          log.error(err);
-          reject(err);
-        };
-
-        onReleased = (context: EventContext) => {
-          removeListeners();
-          log.error("[%s] Sender '%s', got event released.", this._context.connectionId, this.name);
-          let err: Error;
-          if (context!.delivery!.remote_state!.error) {
-            err = translate(context!.delivery!.remote_state!.error);
-          } else {
-            err = new Error(
-              `[${this._context.connectionId}] Sender '${this.name}', ` +
-                `received a release disposition.Hence we are rejecting the promise.`
-            );
-          }
-          log.error(err);
-          reject(err);
-        };
-
-        onModified = (context: EventContext) => {
-          removeListeners();
-          log.error("[%s] Sender '%s', got event modified.", this._context.connectionId, this.name);
-          let err: Error;
-          if (context!.delivery!.remote_state!.error) {
-            err = translate(context!.delivery!.remote_state!.error);
-          } else {
-            err = new Error(
-              `[${this._context.connectionId}] Sender "${this.name}", ` +
-                `received a modified disposition.Hence we are rejecting the promise.`
-            );
-          }
-          log.error(err);
-          reject(err);
-        };
+        }
 
         const removeListeners = (): void => {
           clearTimeout(waitTimer);
-          // When `removeListeners` is called on timeout, the sender might be closed and cleared
-          // So, check if it exists, before removing listeners from it.
           if (abortSignal) {
             abortSignal.removeEventListener("abort", onAborted);
           }
-          if (this._sender) {
-            this._sender.removeListener(SenderEvents.rejected, onRejected);
-            this._sender.removeListener(SenderEvents.accepted, onAccepted);
-            this._sender.removeListener(SenderEvents.released, onReleased);
-            this._sender.removeListener(SenderEvents.modified, onModified);
-          }
         };
 
+        const onAborted = () => {
+          removeListeners();
+          rejectOnAbort();
+        };
+
+        if (abortSignal) {
+          abortSignal.addEventListener("abort", onAborted);
+        }
+
         const actionAfterTimeout = () => {
           removeListeners();
           const desc: string =
@@ -666,11 +600,7 @@ export class EventHubSender extends LinkEntity {
           return reject(translate(e));
         };
 
-        if (abortSignal) {
-          abortSignal.addEventListener("abort", onAborted);
-        }
-
-        waitTimer = setTimeout(
+        const waitTimer = setTimeout(
           actionAfterTimeout,
           getRetryAttemptTimeoutInMs(options.retryOptions)
         );
@@ -684,13 +614,14 @@ export class EventHubSender extends LinkEntity {
 
           try {
             await defaultLock.acquire(this.senderLock, () => {
-              return this._init();
+              return this._init(
+                this._createSenderOptions({
+                  sendTimeoutInSeconds: getRetryAttemptTimeoutInMs(options.retryOptions)
+                })
+              );
             });
           } catch (err) {
-            if (abortSignal) {
-              abortSignal.removeEventListener("abort", onAborted);
-            }
-            clearTimeout(waitTimer);
+            removeListeners();
             err = translate(err);
             log.error(
               "[%s] An error occurred while creating the sender %s",
@@ -716,18 +647,26 @@ export class EventHubSender extends LinkEntity {
             this.name
           );
 
-          this._sender!.on(SenderEvents.accepted, onAccepted);
-          this._sender!.on(SenderEvents.rejected, onRejected);
-          this._sender!.on(SenderEvents.modified, onModified);
-          this._sender!.on(SenderEvents.released, onReleased);
-
-          const delivery = this._sender!.send(message, undefined, 0x80013700);
-          log.sender(
-            "[%s] Sender '%s', sent message with delivery id: %d",
-            this._context.connectionId,
-            this.name,
-            delivery.id
-          );
+          try {
+            const delivery = await this._sender!.send(message, undefined, 0x80013700);
+            log.sender(
+              "[%s] Sender '%s', sent message with delivery id: %d",
+              this._context.connectionId,
+              this.name,
+              delivery.id
+            );
+            return resolve();
+          } catch (err) {
+            err = translate(err);
+            log.error(
+              "[%s] An error occurred while sending the message",
+              this._context.connectionId,
+              err
+            );
+            return reject(err);
+          } finally {
+            removeListeners();
+          }
         } else {
           // let us retry to send the message after some time.
           const msg =
@@ -764,7 +703,7 @@ export class EventHubSender extends LinkEntity {
    * @ignore
    * @returns
    */
-  private async _init(options?: RheaSenderOptions): Promise<void> {
+  private async _init(options?: AwaitableSenderOptions): Promise<void> {
     try {
       // isOpen isConnecting  Should establish
       // true     false          No
@@ -785,7 +724,7 @@ export class EventHubSender extends LinkEntity {
         if (!options) {
           options = this._createSenderOptions({});
         }
-        this._sender = await this._context.connection.createSender(options);
+        this._sender = await this._context.connection.createAwaitableSender(options);
         this.isConnecting = false;
         log.error(
           "[%s] Sender '%s' with address '%s' has established itself.",

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

@@ -10,7 +10,7 @@ import {
   TokenType
 } from "@azure/core-amqp";
 import { ConnectionContext } from "./connectionContext";
-import { Sender, Receiver } from "rhea-promise";
+import { AwaitableSender, Receiver } from "rhea-promise";
 import * as log from "./log";
 
 /**
@@ -230,7 +230,7 @@ export class LinkEntity {
    * @param [link] The Sender or Receiver link that needs to be closed and
    * removed.
    */
-  protected async _closeLink(link?: Sender | Receiver): Promise<void> {
+  protected async _closeLink(link?: AwaitableSender | Receiver): Promise<void> {
     clearTimeout(this._tokenRenewalTimer as NodeJS.Timer);
     if (link) {
       try {