Retry operation when network connection is down.

lib/cbs.ts

@@ -176,6 +176,25 @@ export class CbsClient {
     }
   }
 
+  /**
+   * Removes the AMQP cbs session to the EventHub/ServiceBus for this client.
+   * @return {Promise<void>}
+   */
+  async remove(): Promise<void> {
+    try {
+      if (this._cbsSenderReceiverLink) {
+        const cbsLink = this._cbsSenderReceiverLink;
+        this._cbsSenderReceiverLink = undefined;
+        await cbsLink!.remove();
+        log.cbs("[%s] Successfully removed the cbs session.", this.connection.id);
+      }
+    } catch (err) {
+      const msg = `An error occurred while removing the cbs link: ${err.stack || JSON.stringify(err)}.`;
+      log.error("[%s] %s", this.connection.id, msg);
+      throw new Error(msg);
+    }
+  }
+
   /**
    * Closes the AMQP cbs session to the EventHub/ServiceBus for this client,
    * returning a promise that will be resolved when disconnection is completed.

lib/requestResponseLink.ts

@@ -175,6 +175,16 @@ export class RequestResponseLink implements ReqResLink {
     return retry<AmqpMessage>(config);
   }
 
+  /**
+   * Removes the sender, receiver link and the underlying session.
+   * @returns {Promise<void>} Promise<void>
+   */
+  async remove(): Promise<void> {
+    await this.sender.remove();
+    await this.receiver.remove();
+    await this.session.remove();
+  }
+
   /**
    * Closes the sender, receiver link and the underlying session.
    * @returns {Promise<void>} Promise<void>

lib/retry.ts

@@ -5,6 +5,7 @@ import { translate, MessagingError } from "./errors";
 import { delay } from ".";
 import * as log from "./log";
 import { defaultRetryAttempts, defaultDelayBetweenRetriesInSeconds } from "./util/constants";
+import { resolve } from "dns";
 
 /**
  * Determines whether the object is a Delivery object.
@@ -62,6 +63,11 @@ export interface RetryConfig<T> {
    * next attempt. Default: 15.
    */
   delayInSeconds?: number;
+  /**
+   * @property {string} connectionHost The host "<yournamespace>.servicebus.windows.net".
+   * Used to check network connectivity.
+   */
+  connectionHost?: string;
 }
 
 /**
@@ -91,6 +97,20 @@ function validateRetryConfig<T>(config: RetryConfig<T>): void {
   }
 }
 
+
+async function checkNetworkConnection(host: string): Promise<boolean> {
+  return new Promise((res) => {
+    resolve(host, function (err: any): void {
+      if (err && err.code === "ECONNREFUSED") {
+        res(false);
+      } else {
+        res(true);
+      }
+    });
+  });
+}
+
+
 /**
  * It will attempt to linearly retry an operation specified number of times with a specified
  * delay in between each retry. The retries will only happen if the error is retryable.
@@ -122,6 +142,14 @@ export async function retry<T>(config: RetryConfig<T>): Promise<T> {
       if (!err.translated) {
         err = translate(err);
       }
+
+      if (!err.retryable && err.name === "ServiceCommunicationError" && config.connectionHost) {
+        const isConnected = await checkNetworkConnection(config.connectionHost);
+        if (!isConnected) {
+          err.name = "ConnectionLostError";
+          err.retryable = true;
+        }
+      }
       lastError = err;
       log.error("[%s] Error occured for '%s' in attempt number %d: %O", config.connectionId,
         config.operationType, j, err);