[Event Hubs] Use AwaitableSender in lieu of Sender

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

@@ -481,12 +481,20 @@ 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"
 ];
 
 /**
@@ -553,28 +561,9 @@ export function translate(err: AmqpError | Error): 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;
-  }
-
   // 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;
   }
@@ -602,7 +591,10 @@ export function translate(err: AmqpError | Error): MessagingError {
       // not found
       error.retryable = false;
     }
-  } else if (isSystemError(err)) {
+    return error;
+  }
+
+  if (isSystemError(err)) {
     // translate
     const condition = (err as any).code;
     const description = (err as Error).message;
@@ -617,15 +609,32 @@ export function translate(err: AmqpError | Error): MessagingError {
       // not found
       error.retryable = false;
     }
-  } else if (isBrowserWebsocketError(err)) {
+    return error;
+  }
+
+  if (isBrowserWebsocketError(err)) {
     // Translate browser communication errors during opening handshake to generic SeviceCommunicationError
     error = new MessagingError("Websocket connection failed.");
     error.name = ConditionErrorNameMapper[ErrorNameConditionMapper.ServiceCommunicationError];
     error.retryable = false;
-  } else {
-    // Translate a generic error into MessagingError.
-    error = new MessagingError((err as Error).message);
-    error.stack = (err as Error).stack;
+    return error;
   }
+
+  // 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
+  const errorName = (err as Error).name;
+  if (retryableErrors.indexOf(errorName) > -1) {
+    error.retryable = true;
+    return error;
+  }
+  if (errorName === "AbortError") {
+    error.retryable = false;
+    return error;
+  }
+
+  // Translate a generic error into MessagingError.
+  error = new MessagingError((err as Error).message);
+  error.stack = (err as Error).stack;
   return error;
 }

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

@@ -5,6 +5,7 @@ import * as chai from "chai";
 const should = chai.should();
 
 import * as Errors from "../src/errors";
+import { AbortError } from "@azure/abort-controller";
 
 class AMQPError {
   name = "AmqpProtocolError";
@@ -30,6 +31,35 @@ describe("Errors", function() {
       translatedError.message.should.equal(ehError.message);
     });
 
+    it("Sets retryable to true, if input is custom error and name is OperationTimeoutError", function() {
+      const err = new Error("error message");
+      err.name = "OperationTimeoutError";
+      const translatedError = Errors.translate(err);
+      should.equal(translatedError.name === "OperationTimeoutError", true);
+      translatedError.message.should.equal(err.message);
+      translatedError.stack!.should.equal(err.stack);
+      translatedError.retryable.should.equal(true);
+    });
+
+    it("Sets retryable to true, if input is custom error and name is InsufficientCreditError", function() {
+      const err = new Error("error message");
+      err.name = "InsufficientCreditError";
+      const translatedError = Errors.translate(err);
+      should.equal(translatedError.name === "InsufficientCreditError", true);
+      translatedError.message.should.equal(err.message);
+      translatedError.stack!.should.equal(err.stack);
+      translatedError.retryable.should.equal(true);
+    });
+
+    it("Sets retryable to false, if input is the custom AbortError", function() {
+      const err = new AbortError("error message");
+      const translatedError = Errors.translate(err);
+      should.equal(translatedError.name === "AbortError", true);
+      translatedError.message.should.equal(err.message);
+      translatedError.stack!.should.equal(err.stack);
+      translatedError.retryable.should.equal(false);
+    });
+
     it("Sets retryable to false, and acts as a passthrough if the input is TypeError", function() {
       const err = new TypeError("This is a wrong type!!");
       const translatedError = Errors.translate(err);
@@ -67,10 +97,7 @@ describe("Errors", function() {
       { from: "<unknown>", to: "MessagingError" }
     ].forEach(function(mapping) {
       it("translates " + mapping.from + " into " + mapping.to, function() {
-        const err: any = new AMQPError(
-          mapping.from as any,
-          mapping.message as any
-        );
+        const err: any = new AMQPError(mapping.from as any, mapping.message as any);
         const translatedError = <Errors.MessagingError>Errors.translate(err);
         translatedError.name.should.equal(mapping.to);
         if (
@@ -113,18 +140,13 @@ describe("Errors", function() {
         code: "ESOMETHINGRANDOM",
         errno: "ESOMETHINGRANDOM",
         syscall: "read",
-        message:
-          "code: ESOMETHINGRANDOM, errno: ESOMETHINGRANDOM, syscall: read"
+        message: "code: ESOMETHINGRANDOM, errno: ESOMETHINGRANDOM, syscall: read"
       }
     ].forEach(function(mapping) {
       it(
-        "SystemError from node.js  with code: '" +
-          mapping.code +
-          "' to a MessagingError",
+        "SystemError from node.js  with code: '" + mapping.code + "' to a MessagingError",
         function() {
-          const translatedError = <Errors.MessagingError>(
-            Errors.translate(mapping as any)
-          );
+          const translatedError = <Errors.MessagingError>Errors.translate(mapping as any);
           if (mapping.code === "ECONNRESET") {
             translatedError.name.should.equal("ServiceUnavailableError");
             translatedError.retryable.should.equal(true);

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';
@@ -16,7 +17,6 @@ import { MessagingError } from '@azure/core-amqp';
 import { Receiver } from 'rhea-promise';
 import { ReceiverOptions } from 'rhea-promise';
 import { RetryOptions } from '@azure/core-amqp';
-import { Sender } from 'rhea-promise';
 import { SharedKeyCredential } from '@azure/core-amqp';
 import { TokenCredential } from '@azure/core-amqp';
 import { TokenType } from '@azure/core-amqp';
@@ -279,6 +279,8 @@ export class ReceiveHandler {
     stop(): Promise<void>;
 }
 
+export { RetryOptions }
+
 // @public
 export interface SendOptions {
     abortSignal?: AbortSignalLike;