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

sdk/core/core-amqp/changelog.md

@@ -1,3 +1,7 @@
+## 1.0.0-preview.2.0 - Coming soon...
+
+- `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/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,
@@ -15,8 +14,7 @@ import {
   SenderOptions,
   ReceiverOptions,
   ReceiverEvents,
-  ReqResLink,
-  generate_uuid
+  ReqResLink
 } from "rhea-promise";
 import { translate, ConditionStatusMapper } from "./errors";
 import * as log from "./log";
@@ -32,19 +30,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 +74,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,168 +84,145 @@ 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();
-        }
+    return new Promise<AmqpMessage>((resolve: any, reject: any) => {
+      let waitTimer: any;
+      let timeOver: boolean = false;
+      type NormalizedInfo = {
+        statusCode: number;
+        statusDescription: string;
+        errorCondition: string;
+      };
 
-        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.`
-          );
+      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);
-        };
+        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);
+      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();
-        };
+        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 (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);
+      }
 
-        // Handle different variations of property names in responses emitted by EventHubs and ServiceBus.
-        const getCodeDescriptionAndError = (props: any): NormalizedInfo => {
-          if (!props) props = {};
-          return {
-            statusCode: (props[Constants.statusCode] || props.statusCode) as number,
-            statusDescription: (props[Constants.statusDescription] ||
-              props.statusDescription) as string,
-            errorCondition: (props[Constants.errorCondition] || props.errorCondition) as string
-          };
+      // Handle different variations of property names in responses emitted by EventHubs and ServiceBus.
+      const getCodeDescriptionAndError = (props: any): NormalizedInfo => {
+        if (!props) props = {};
+        return {
+          statusCode: (props[Constants.statusCode] || props.statusCode) as number,
+          statusDescription: (props[Constants.statusDescription] ||
+            props.statusDescription) as string,
+          errorCondition: (props[Constants.errorCondition] || props.errorCondition) as string
         };
+      };
 
-        const messageCallback = (context: EventContext) => {
-          // remove the event listeners as they will be registered next time when someone makes a request.
-          this.receiver.removeListener(ReceiverEvents.message, messageCallback);
-          if (aborter) {
-            aborter.removeEventListener("abort", onAbort);
-          }
-          const info = getCodeDescriptionAndError(context.message!.application_properties);
-          const responseCorrelationId = context.message!.correlation_id;
-          log.reqres(
-            "[%s] %s response: ",
-            this.connection.id,
-            request.to || "$management",
-            context.message
-          );
-          if (info.statusCode > 199 && info.statusCode < 300) {
-            if (
-              request.message_id === responseCorrelationId ||
-              request.correlation_id === responseCorrelationId
-            ) {
-              if (!timeOver) {
-                clearTimeout(waitTimer);
-              }
-              log.reqres(
-                "[%s] request-messageId | '%s' == '%s' | response-correlationId.",
-                this.connection.id,
-                request.message_id,
-                responseCorrelationId
-              );
-              return resolve(context.message);
-            } else {
-              log.error(
-                "[%s] request-messageId | '%s' != '%s' | response-correlationId. " +
-                  "Hence dropping this response and waiting for the next one.",
-                this.connection.id,
-                request.message_id,
-                responseCorrelationId
-              );
+      const messageCallback = (context: EventContext) => {
+        // remove the event listeners as they will be registered next time when someone makes a request.
+        this.receiver.removeListener(ReceiverEvents.message, messageCallback);
+        if (aborter) {
+          aborter.removeEventListener("abort", onAbort);
+        }
+        const info = getCodeDescriptionAndError(context.message!.application_properties);
+        const responseCorrelationId = context.message!.correlation_id;
+        log.reqres(
+          "[%s] %s response: ",
+          this.connection.id,
+          request.to || "$management",
+          context.message
+        );
+        if (info.statusCode > 199 && info.statusCode < 300) {
+          if (
+            request.message_id === responseCorrelationId ||
+            request.correlation_id === responseCorrelationId
+          ) {
+            if (!timeOver) {
+              clearTimeout(waitTimer);
             }
+            log.reqres(
+              "[%s] request-messageId | '%s' == '%s' | response-correlationId.",
+              this.connection.id,
+              request.message_id,
+              responseCorrelationId
+            );
+            return resolve(context.message);
           } else {
-            const condition =
-              info.errorCondition ||
-              ConditionStatusMapper[info.statusCode] ||
-              "amqp:internal-error";
-            const e: AmqpError = {
-              condition: condition,
-              description: info.statusDescription
-            };
-            const error = translate(e);
-            log.error(error);
-            return reject(error);
-          }
-        };
-
-        const actionAfterTimeout = () => {
-          timeOver = true;
-          this.receiver.removeListener(ReceiverEvents.message, messageCallback);
-          if (aborter) {
-            aborter.removeEventListener("abort", onAbort);
+            log.error(
+              "[%s] request-messageId | '%s' != '%s' | response-correlationId. " +
+                "Hence dropping this response and waiting for the next one.",
+              this.connection.id,
+              request.message_id,
+              responseCorrelationId
+            );
           }
-          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.`;
+        } else {
+          const condition =
+            info.errorCondition || ConditionStatusMapper[info.statusCode] || "amqp:internal-error";
           const e: AmqpError = {
-            condition: ConditionStatusMapper[408],
-            description: desc
+            condition: condition,
+            description: info.statusDescription
           };
-          return reject(translate(e));
+          const error = translate(e);
+          log.error(error);
+          return reject(error);
+        }
+      };
+
+      const actionAfterTimeout = () => {
+        timeOver = true;
+        this.receiver.removeListener(ReceiverEvents.message, messageCallback);
+        if (aborter) {
+          aborter.removeEventListener("abort", onAbort);
+        }
+        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));
+      };
 
-        this.receiver.on(ReceiverEvents.message, messageCallback);
-        waitTimer = setTimeout(actionAfterTimeout, options!.timeoutInSeconds! * 1000);
-        log.reqres(
-          "[%s] %s request sent: %O",
-          this.connection.id,
-          request.to || "$managment",
-          request
-        );
-        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);
+      waitTimer = setTimeout(actionAfterTimeout, options!.timeoutInSeconds! * 1000);
+      this.receiver.on(ReceiverEvents.message, messageCallback);
+
+      log.reqres(
+        "[%s] %s request sent: %O",
+        this.connection.id,
+        request.to || "$managment",
+        request
+      );
+      this.sender.send(request);
+    });
   }
 
   /**