[Event Hubs] Perform sendRequest() operation handling from SDK

sdk/core/core-amqp/changelog.md

@@ -1,3 +1,7 @@
+## 1.0.0-preview.2.0 - 16th July, 2019
+
+- `sendRequest()` function in the `RequestResponseLink` now excludes default retries and leaves it up to the users to implement it as necessary.
+
 ## 1.0.0-preview.1.0 - 28th June, 2019
 
 This library is based off of the [@azure/amqp-common](https://www.npmjs.com/package/@azure/amqp-common)

sdk/core/core-amqp/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@azure/core-amqp",
   "sdk-type": "client",
-  "version": "1.0.0-preview.1",
+  "version": "1.0.0-preview.2",
   "description": "Common library for amqp based azure sdks like @azure/event-hubs.",
   "author": "Microsoft Corporation",
   "license": "MIT",

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

@@ -3,7 +3,6 @@
 
 import { AbortSignalLike, AbortError } from "@azure/abort-controller";
 import * as Constants from "./util/constants";
-import { retry, RetryConfig, RetryOperationType } from "./retry";
 import {
   Session,
   Connection,
@@ -32,19 +31,9 @@ export interface SendRequestOptions {
   abortSignal?: AbortSignalLike;
   /**
    * @property {number} [timeoutInSeconds] Max time to wait for the operation to complete.
-   * Default: `10 seconds`.
+   * Default: `60 seconds`.
    */
   timeoutInSeconds?: number;
-  /**
-   * @property {number} [maxRetries] Number of times the operation needs to be retried in case
-   * of error. Default: 3.
-   */
-  maxRetries?: number;
-  /**
-   * @property {number} [delayInSeconds] Amount of time to wait in seconds before making the
-   * next attempt. Default: 15.
-   */
-  delayInSeconds?: number;
   /**
    * @property {string} [requestName] Name of the request being performed.
    */
@@ -86,9 +75,7 @@ export class RequestResponseLink implements ReqResLink {
 
   /**
    * Sends the given request message and returns the received response. If the operation is not
-   * completed in the provided timeout in seconds `default: 10`, then the request will be retried
-   * linearly for the provided number of times `default: 3` with the provided delay in seconds
-   * `default: 15` between each attempt.
+   * completed in the provided timeout in seconds `default: 60`, then `OperationTimeoutError` is thrown.
    *
    * @param {Message} request The AMQP (request) message.
    * @param {SendRequestOptions} [options] Options that can be provided while sending a request.
@@ -98,65 +85,38 @@ export class RequestResponseLink implements ReqResLink {
     if (!options) options = {};
 
     if (!options.timeoutInSeconds) {
-      options.timeoutInSeconds = 10;
+      options.timeoutInSeconds = Constants.defaultOperationTimeoutInSeconds;
     }
 
-    let count: number = 0;
     const aborter: AbortSignalLike | undefined = options && options.abortSignal;
 
     const sendRequestPromise = () =>
       new Promise<AmqpMessage>((resolve: any, reject: any) => {
-        let waitTimer: any;
         let timeOver: boolean = false;
         type NormalizedInfo = {
           statusCode: number;
           statusDescription: string;
           errorCondition: string;
         };
 
-        count++;
-        if (count !== 1) {
-          // Generate a new message_id every time after the first attempt
-          request.message_id = generate_uuid();
-        } else if (!request.message_id) {
-          // Set the message_id in the first attempt only if it is not set
-          request.message_id = generate_uuid();
-        }
-
-        const rejectOnAbort = () => {
-          const address = this.receiver.address || "address";
-          const requestName = options!.requestName;
-          const desc: string =
-            `[${this.connection.id}] The request "${requestName}" ` +
-            `to "${address}" has been cancelled by the user.`;
-          log.error(desc);
-          const error = new AbortError(
-            `The ${requestName ? requestName + " " : ""}operation has been cancelled by the user.`
-          );
-
-          reject(error);
-        };
-
-        const onAbort = () => {
-          // remove the event listener as this will be registered next time someone makes a request.
+        const actionAfterTimeout = () => {
+          timeOver = true;
           this.receiver.removeListener(ReceiverEvents.message, messageCallback);
-          // safe to clear the timeout if it hasn't already occurred.
-          if (!timeOver) {
-            clearTimeout(waitTimer);
+          if (aborter) {
+            aborter.removeEventListener("abort", onAbort);
           }
-          aborter!.removeEventListener("abort", onAbort);
-
-          rejectOnAbort();
+          const address = this.receiver.address || "address";
+          const desc: string =
+            `The request with message_id "${request.message_id}" to "${address}" ` +
+            `endpoint timed out. Please try again later.`;
+          const e: Error = {
+            name: "OperationTimeoutError",
+            message: desc
+          };
+          return reject(translate(e));
         };
 
-        if (aborter) {
-          // the aborter may have been triggered between request attempts
-          // so check if it was triggered and reject if needed.
-          if (aborter.aborted) {
-            return rejectOnAbort();
-          }
-          aborter.addEventListener("abort", onAbort);
-        }
+        const waitTimer = setTimeout(actionAfterTimeout, options!.timeoutInSeconds! * 1000);
 
         // Handle different variations of property names in responses emitted by EventHubs and ServiceBus.
         const getCodeDescriptionAndError = (props: any): NormalizedInfo => {
@@ -222,25 +182,48 @@ export class RequestResponseLink implements ReqResLink {
           }
         };
 
-        const actionAfterTimeout = () => {
-          timeOver = true;
-          this.receiver.removeListener(ReceiverEvents.message, messageCallback);
-          if (aborter) {
-            aborter.removeEventListener("abort", onAbort);
-          }
+        const rejectOnAbort = () => {
           const address = this.receiver.address || "address";
+          const requestName = options!.requestName;
           const desc: string =
-            `The request with message_id "${request.message_id}" to "${address}" ` +
-            `endpoint timed out. Please try again later.`;
-          const e: AmqpError = {
-            condition: ConditionStatusMapper[408],
-            description: desc
-          };
-          return reject(translate(e));
+            `[${this.connection.id}] The request "${requestName}" ` +
+            `to "${address}" has been cancelled by the user.`;
+          log.error(desc);
+          const error = new AbortError(
+            `The ${requestName ? requestName + " " : ""}operation has been cancelled by the user.`
+          );
+
+          reject(error);
         };
 
+        const onAbort = () => {
+          // remove the event listener as this will be registered next time someone makes a request.
+          this.receiver.removeListener(ReceiverEvents.message, messageCallback);
+          // safe to clear the timeout if it hasn't already occurred.
+          if (!timeOver) {
+            clearTimeout(waitTimer);
+          }
+          aborter!.removeEventListener("abort", onAbort);
+
+          rejectOnAbort();
+        };
+
+        if (aborter) {
+          // the aborter may have been triggered between request attempts
+          // so check if it was triggered and reject if needed.
+          if (aborter.aborted) {
+            return rejectOnAbort();
+          }
+          aborter.addEventListener("abort", onAbort);
+        }
+
+        if (!request.message_id) {
+          // Set the message_id only if it is not set
+          request.message_id = generate_uuid();
+        }
+
         this.receiver.on(ReceiverEvents.message, messageCallback);
-        waitTimer = setTimeout(actionAfterTimeout, options!.timeoutInSeconds! * 1000);
+
         log.reqres(
           "[%s] %s request sent: %O",
           this.connection.id,
@@ -249,17 +232,8 @@ export class RequestResponseLink implements ReqResLink {
         );
         this.sender.send(request);
       });
-    const config: RetryConfig<AmqpMessage> = {
-      operation: sendRequestPromise,
-      connectionId: this.connection.id,
-      operationType:
-        request.to && request.to === Constants.cbsEndpoint
-          ? RetryOperationType.cbsAuth
-          : RetryOperationType.management,
-      delayInSeconds: options.delayInSeconds,
-      maxRetries: options.maxRetries
-    };
-    return retry<AmqpMessage>(config);
+
+    return sendRequestPromise();
   }
 
   /**

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

@@ -114,7 +114,6 @@ describe("RequestResponseLink", function() {
       });
     }, 4000);
     const response = await link.sendRequest(request, {
-      delayInSeconds: 1,
       timeoutInSeconds: 5
     });
     assert.equal(response.correlation_id, messageId);
@@ -240,7 +239,6 @@ describe("RequestResponseLink", function() {
       const signal = controller.signal;
       setTimeout(controller.abort.bind(controller), 100);
       await link.sendRequest(request, {
-        delayInSeconds: 1,
         timeoutInSeconds: 5,
         abortSignal: signal // cancel between request attempts
       });

sdk/eventhub/event-hubs/package.json

@@ -63,7 +63,7 @@
   },
   "dependencies": {
     "@azure/abort-controller": "1.0.0-preview.1",
-    "@azure/core-amqp": "^1.0.0-preview.1",
+    "@azure/core-amqp": "^1.0.0-preview.2",
     "@azure/core-asynciterator-polyfill": "^1.0.0-preview.1",
     "async-lock": "^1.1.3",
     "debug": "^3.1.0",

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

@@ -7,7 +7,11 @@ import {
   defaultLock,
   translate,
   Constants,
-  SendRequestOptions
+  SendRequestOptions,
+  retry,
+  RetryConfig,
+  RetryOperationType,
+  randomNumberFromInterval
 } from "@azure/core-amqp";
 import {
   Message,
@@ -306,31 +310,86 @@ export class ManagementClient extends LinkEntity {
     options?: { retryOptions?: RetryOptions; abortSignal?: AbortSignalLike; requestName?: string }
   ): Promise<any> {
     try {
-      log.mgmt(
-        "[%s] Acquiring lock to get the management req res link.",
-        this._context.connectionId
-      );
-      await defaultLock.acquire(this.managementLock, () => {
-        return this._init();
-      });
-
       if (!options) {
         options = {};
       }
 
-      const sendRequestOptions: SendRequestOptions = {
-        maxRetries: options.retryOptions && options.retryOptions.maxRetries,
-        abortSignal: options.abortSignal,
-        requestName: options.requestName,
-        timeoutInSeconds: getRetryAttemptTimeoutInMs(options.retryOptions) / 1000,
-        delayInSeconds:
-          options.retryOptions &&
-          options.retryOptions.retryInterval &&
-          options.retryOptions.retryInterval >= 0
-            ? options.retryOptions.retryInterval / 1000
-            : undefined
+      const sendOperationPromise = () =>
+        new Promise<Message>(async (resolve, reject) => {
+          try {
+            const actionAfterTimeout = () => {
+              const address = this._mgmtReqResLink!.receiver.address || "address";
+              const desc: string =
+                `The request with message_id "${request.message_id}" to "${address}" ` +
+                `endpoint timed out. Please try again later.`;
+              const e: Error = {
+                name: "OperationTimeoutError",
+                message: desc
+              };
+              return reject(translate(e));
+            };
+
+            const waitTimer = setTimeout(
+              actionAfterTimeout,
+              getRetryAttemptTimeoutInMs(options!.retryOptions)
+            );
+
+            const initOperationStartTime = Date.now();
+            log.mgmt(
+              "[%s] Acquiring lock to get the management req res link.",
+              this._context.connectionId
+            );
+            await defaultLock.acquire(this.managementLock, () => {
+              return this._init().catch((err: Error) => {
+                clearTimeout(waitTimer);
+                reject(translate(err));
+              });
+            });
+            const initOperationEndTime = Date.now();
+
+            if (!options) {
+              options = {};
+            }
+
+            const operationTimeoutInMs = initOperationEndTime - initOperationStartTime;
+            const sendRequestOptions: SendRequestOptions = {
+              abortSignal: options.abortSignal,
+              requestName: options.requestName,
+              timeoutInSeconds: operationTimeoutInMs / 1000
+            };
+
+            const result = await this._mgmtReqResLink!.sendRequest(request, sendRequestOptions);
+            resolve(result);
+          } catch (err) {
+            err = translate(err);
+            const address = this._mgmtReqResLink!.sender.address || "address";
+            log.error(
+              "[%s] An error occurred during send on management request-response link '%s' with address " +
+                "'%s': %O",
+              this._context.connectionId,
+              this._mgmtReqResLink!.name,
+              address,
+              err
+            );
+            reject(err);
+          }
+        });
+
+      const maxRetries = options.retryOptions && options.retryOptions.maxRetries;
+      const delayInSeconds =
+        options.retryOptions &&
+        options.retryOptions.retryInterval &&
+        options.retryOptions.retryInterval >= 0
+          ? options.retryOptions.retryInterval / 1000
+          : Constants.defaultDelayBetweenOperationRetriesInSeconds;
+      const config: RetryConfig<void> = {
+        operation: sendOperationPromise,
+        connectionId: this._context.connectionId,
+        operationType: RetryOperationType.management,
+        maxRetries: maxRetries,
+        delayInSeconds: delayInSeconds
       };
-      return (await this._mgmtReqResLink!.sendRequest(request, sendRequestOptions)).body;
+      return (await retry<Message>(config)).body;
     } catch (err) {
       err = translate(err);
       log.error("An error occurred while making the request to $management endpoint: %O", err);