[service-bus] Reliability improvements and testing updates

sdk/servicebus/service-bus/CHANGELOG.md

@@ -1,6 +1,6 @@
 # Release History
 
-## 7.1.0-beta.1 (Unreleased)
+## 7.0.6 (Unreleased)
 
 ### New Features
 
@@ -16,6 +16,12 @@
 - Settling messages now use the `retryOptions` passed to `ServiceBusClient`, making it more resilient against network failures.
   [PR#14867](https://github.com/Azure/azure-sdk-for-js/pull/14867/files)
 
+## 7.0.5 (2021-04-06)
+
+### Bug fixes
+
+- Some of the queue properties such as "forwardTo" and "autoDeleteOnIdle" were not being set as requested through the `ServiceBusAdministrationClient.createQueue` method because of a bug with regards to the ordering of XML properties. The issue has been fixed in [#14692](https://github.com/Azure/azure-sdk-for-js/pull/14692).
+
 ## 7.0.4 (2021-03-31)
 
 ### Bug fixes

sdk/servicebus/service-bus/package.json

@@ -2,7 +2,7 @@
   "name": "@azure/service-bus",
   "sdk-type": "client",
   "author": "Microsoft Corporation",
-  "version": "7.1.0-beta.1",
+  "version": "7.0.6",
   "license": "MIT",
   "description": "Azure Service Bus SDK for JavaScript",
   "homepage": "https://github.com/Azure/azure-sdk-for-js/tree/master/sdk/servicebus/service-bus/",
@@ -73,7 +73,7 @@
     "test:node": "npm run clean && npm run build:test:node && npm run integration-test:node",
     "test": "npm run test:node && npm run test:browser",
     "unit-test:browser": "echo skipped",
-    "unit-test:node": "mocha -r esm --require source-map-support/register --reporter ../../../common/tools/mocha-multi-reporter.js --timeout 120000 --full-trace \"dist-esm/test/internal/unit/*.spec.js\" \"dist-esm/test/internal/node/*.spec.js\"",
+    "unit-test:node": "mocha -r esm -r ts-node/register --require source-map-support/register --reporter ../../../common/tools/mocha-multi-reporter.js --timeout 120000 --full-trace \"test/internal/unit/*.spec.ts\" \"test/internal/node/*.spec.ts\"",
     "unit-test": "npm run unit-test:node && npm run unit-test:browser",
     "docs": "typedoc --excludePrivate --excludeNotExported --excludeExternals --stripInternal --mode file --out ./dist/docs ./src"
   },

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

@@ -118,16 +118,6 @@ export abstract class MessageReceiver extends LinkEntity<Receiver> {
     number,
     DeferredPromiseAndTimer
   >();
-  /**
-   * The message handler provided by the user that will be wrapped
-   * inside _onAmqpMessage.
-   */
-  protected _onMessage!: OnMessage;
-  /**
-   * The error handler provided by the user that will be wrapped
-   * inside _onAmqpError.
-   */
-  protected _onError?: OnError;
 
   /**
    * A lock renewer that handles message lock auto-renewal. This is undefined unless the user

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

@@ -1,8 +1,10 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
+import { AbortError } from "@azure/abort-controller";
 import { Receiver, ReceiverEvents } from "rhea-promise";
 import { receiverLogger as logger } from "../log";
+import { ServiceBusError } from "../serviceBusError";
 
 /**
  * Wraps the receiver with some higher level operations for managing state
@@ -11,37 +13,61 @@ import { receiverLogger as logger } from "../log";
  * @internal
  */
 export class ReceiverHelper {
-  private _isSuspended: boolean = false;
+  private _isSuspended: boolean = true;
 
   constructor(
     private _getCurrentReceiver: () => { receiver: Receiver | undefined; logPrefix: string }
   ) {}
 
+  private _getCurrentReceiverOrError():
+    | "is undefined"
+    | "is not open"
+    | "is suspended"
+    | { receiver: Receiver | undefined; logPrefix: string } {
+    const currentReceiverData = this._getCurrentReceiver();
+
+    if (currentReceiverData.receiver == null) {
+      return "is undefined";
+    }
+
+    if (!currentReceiverData.receiver.isOpen()) {
+      return "is not open";
+    }
+
+    if (this._isSuspended) {
+      return "is suspended";
+    }
+
+    return currentReceiverData;
+  }
+
   /**
    * Adds credits to the receiver, respecting any state that
    * indicates the receiver is closed or should not continue
    * to receive more messages.
    *
    * @param credits - Number of credits to add.
-   * @returns true if credits were added, false if there is no current receiver instance
    * or `stopReceivingMessages` has been called.
    */
-  addCredit(credits: number): boolean {
-    const { receiver, logPrefix } = this._getCurrentReceiver();
+  addCredit(credits: number): void {
+    const currentReceiverOrError = this._getCurrentReceiverOrError();
 
-    if (!this.canReceiveMessages()) {
-      logger.verbose(
-        `${logPrefix} Asked to add ${credits} credits but the receiver is not able to receive messages`
-      );
-      return false;
-    }
+    if (typeof currentReceiverOrError === "string") {
+      const errorMessage = `Cannot request messages on the receiver since it ${currentReceiverOrError}.`;
 
-    if (receiver != null) {
-      logger.verbose(`${logPrefix} Adding ${credits} credits`);
-      receiver.addCredit(credits);
+      if (currentReceiverOrError === "is suspended") {
+        // if a user has suspended the receiver we should consider this a non-retryable
+        // error since it absolutely requires user intervention.
+        throw new AbortError(errorMessage);
+      }
+
+      throw new ServiceBusError(errorMessage, "GeneralError");
     }
 
-    return true;
+    if (currentReceiverOrError.receiver != null) {
+      logger.verbose(`${currentReceiverOrError.logPrefix} Adding ${credits} credits`);
+      currentReceiverOrError.receiver.addCredit(credits);
+    }
   }
 
   /**
@@ -60,36 +86,38 @@ export class ReceiverHelper {
     logger.verbose(
       `${logPrefix} User has requested to stop receiving new messages, attempting to drain.`
     );
+
     return this.drain();
   }
 
   /**
-   * Resets tracking so `addCredit` works again.
+   * Resets tracking so `addCredit` works again by toggling the `_isSuspended` flag.
    */
   resume(): void {
     this._isSuspended = false;
   }
 
-  /**
-   * Whether the receiver can receive messages.
-   *
-   * This checks if the the caller has decided to disable adding
-   * credits via 'suspend' as well as whether the receiver itself is
-   * still open.
-   */
-  canReceiveMessages(): boolean {
-    const { receiver } = this._getCurrentReceiver();
-    return !this._isSuspended && this._isValidReceiver(receiver);
+  isSuspended(): boolean {
+    return this._isSuspended;
   }
 
   /**
    * Initiates a drain for the current receiver and resolves when
    * the drain has completed.
+   *
+   * NOTE: This method returns immediately if the receiver is not valid or if there
+   * are no pending credits on the receiver (ie: `receiver.credit === 0`).
    */
   async drain(): Promise<void> {
     const { receiver, logPrefix } = this._getCurrentReceiver();
 
     if (!this._isValidReceiver(receiver)) {
+      // TODO: should we throw?
+      return;
+    }
+
+    if (receiver.credit === 0) {
+      // nothing to drain
       return;
     }