[service-bus] remove unnecessary wrapper in streaming receiver (Azure#12604)

Fixes Azure#12260 
The code that's removed is no longer necessary since the `retry` method is being called in a loop and the loop only calls the user's processError handler after the retries in a single `retry` call are exhausted.

Author: chradek

sdk/servicebus/service-bus/src/core/streamingReceiver.ts

@@ -360,26 +360,6 @@ export class StreamingReceiver extends MessageReceiver {
     await this._receiverHelper.suspend();
   }
 
-  /**
-   * Returns a wrapped function that makes any thrown errors retryable (_except_  for AbortError) by
-   * wrapping them in a StreamingReceiverError .
-   */
-  private static wrapRetryOperation(fn: () => Promise<void>) {
-    return async () => {
-      try {
-        await fn();
-      } catch (err) {
-        if (err.name === "AbortError") {
-          throw err;
-        }
-
-        // once we're at this point where we can spin up a connection we're past the point
-        // of fatal errors like the connection string just outright being malformed, for instance.
-        throw new StreamingReceiverError(err);
-      }
-    };
-  }
-
   /**
    * Initializes the link. This method will retry infinitely until a connection is established.
    *
@@ -404,7 +384,7 @@ export class StreamingReceiver extends MessageReceiver {
       ++numRetryCycles;
 
       const config: RetryConfig<void> = {
-        operation: StreamingReceiver.wrapRetryOperation(() => this._initOnce(args)),
+        operation: () => this._initOnce(args),
         connectionId: args.connectionId,
         operationType: RetryOperationType.receiverLink,
         // even though we're going to loop infinitely we allow them to control the pattern we use on each
@@ -417,11 +397,6 @@ export class StreamingReceiver extends MessageReceiver {
         await this._retry<void>(config);
         break;
       } catch (err) {
-        if (StreamingReceiverError.isStreamingReceiverError(err)) {
-          // report the original error to the user
-          err = err.originalError;
-        }
-
         // we only report the error here - this avoids spamming the user with too many
         // redundant reports of errors while still providing them incremental status on failures.
         args.onError({
@@ -544,28 +519,3 @@ export class StreamingReceiver extends MessageReceiver {
     }
   }
 }
-
-/**
- * Wraps an error thrown from the operation for retry<>.
- *
- * Used so we don't have to worry about modifying the errors that
- * also might have originated from the user's processMessage handler,
- * which they rightfully own.
- *
- * @internal
- * @ignore
- */
-export class StreamingReceiverError extends Error {
-  constructor(public originalError: Error | MessagingError) {
-    super(originalError.message);
-    this.name = "StreamingReceiverError";
-  }
-
-  retryable: boolean = true;
-
-  static isStreamingReceiverError(
-    err: Error | StreamingReceiverError
-  ): err is StreamingReceiverError {
-    return err.name === "StreamingReceiverError";
-  }
-}

sdk/servicebus/service-bus/test/internal/streamingReceiver.spec.ts

@@ -6,7 +6,7 @@ import chaiAsPromised from "chai-as-promised";
 import { createConnectionContextForTests } from "./unittestUtils";
 import { ProcessErrorArgs } from "../../src";
 import { ReceiveMode } from "../../src/models";
-import { StreamingReceiver, StreamingReceiverError } from "../../src/core/streamingReceiver";
+import { StreamingReceiver } from "../../src/core/streamingReceiver";
 import sinon from "sinon";
 import { EventContext } from "rhea-promise";
 import { Constants, MessagingError, RetryConfig, RetryMode } from "@azure/core-amqp";
@@ -322,13 +322,7 @@ describe("StreamingReceiver unit tests", () => {
         );
 
         ++numTimesRetryFnCalled;
-        if (numTimesRetryFnCalled === 0) {
-          // add in a little variety - throwing this wrapper error should also
-          // work just fine and it'll be properly extracted when it's reported to the user's onError handler.
-          throw new StreamingReceiverError(
-            new Error(`Error in retry cycle ${numTimesRetryFnCalled}`)
-          );
-        } else if (numTimesRetryFnCalled < 4) {
+        if (numTimesRetryFnCalled < 4) {
           throw new Error(`Error in retry cycle ${numTimesRetryFnCalled}`);
         } else {
           // go ahead and let the retry loop succeed (and it should properly terminate!)
@@ -422,48 +416,6 @@ describe("StreamingReceiver unit tests", () => {
         retryable: undefined
       });
     });
-
-    describe("wrapped operation", () => {
-      it("wrapped operation reports via onError for exceptions.", async () => {
-        const wrappedOperation = StreamingReceiver["wrapRetryOperation"](async () => {
-          throw new Error("Normal errors are logged and rethrown.");
-        });
-
-        try {
-          await wrappedOperation();
-          assert.fail("Should have thrown");
-        } catch (err) {
-          assertError(err, {
-            name: "StreamingReceiverError",
-            message: "Normal errors are logged and rethrown.",
-            // they're also marked as retryable.
-            retryable: true
-          });
-        }
-      });
-
-      it("wrapped operation does throw if it gets an abortError.", async () => {
-        const wrappedOperation = StreamingReceiver["wrapRetryOperation"](async () => {
-          //. the user is obviously welcome to pass in and abort their passed in abortSignal.
-          // Another way this can happen is that LinkEntity.initLink will also throw an AbortError if the link has
-          // been closed by the user.
-          throw new AbortError(
-            "AbortError's should propagate or the link has had .close() called on it"
-          );
-        });
-
-        try {
-          await wrappedOperation();
-          assert.fail("AbortError should have been thrown");
-        } catch (err) {
-          assertError(err, {
-            name: "AbortError",
-            message: "AbortError's should propagate or the link has had .close() called on it",
-            retryable: undefined
-          });
-        }
-      });
-    });
   });
 
   it("onDetach calls through to init", async () => {