Improve handling of the client closed error.

device/core/src/device_client.ts

@@ -29,7 +29,7 @@ function safeCallback(callback?: (err?: Error, result?: any) => void, error?: Er
  * to create an IoT Hub device client.
  */
 export class Client extends InternalClient {
-  private _c2dFeature: boolean;
+  private _userRegisteredC2dListener: boolean;
   private _deviceDisconnectHandler: (err?: Error, result?: any) => void;
   private _blobUploadClient: BlobUploadClient;
   private _fileUploadApi: FileUploadInterface;
@@ -46,11 +46,12 @@ export class Client extends InternalClient {
   constructor(transport: DeviceTransport, connStr?: string, blobUploadClient?: BlobUploadClient, fileUploadApi?: FileUploadInterface) {
     super(transport, connStr);
     this._blobUploadClient = blobUploadClient;
-    this._c2dFeature = false;
+    this._userRegisteredC2dListener = false;
     this._fileUploadApi = fileUploadApi;
 
-    this.on('removeListener', (eventName) => {
+    this.on('removeListener', () => {
-      if (eventName === 'message' && this.listeners('message').length === 0) {
+      if (this.listenerCount('message') === 0) {
+        this._userRegisteredC2dListener = false;
         /*Codes_SRS_NODE_DEVICE_CLIENT_16_005: [The client shall stop listening for messages from the service whenever the last listener unsubscribes from the `message` event.]*/
         debug('in removeListener, disabling C2D.');
         this._disableC2D((err) => {
@@ -58,7 +59,6 @@ export class Client extends InternalClient {
             debugErrors('in removeListener, error disabling C2D: ' + err);
             this.emit('error', err);
           } else {
-            this._c2dFeature = false;
             debug('removeListener successfully disabled C2D.');
           }
         });
@@ -67,6 +67,15 @@ export class Client extends InternalClient {
 
     this.on('newListener', (eventName) => {
       if (eventName === 'message') {
+        //
+        // We want to always retain that the we want to have this feature enabled because the API (.on) doesn't really
+        // provide for the capability to say it failed.  It can certainly fail because a network operation is required to
+        // enable.
+        // By saving this off, we are strictly honoring that the feature is enabled.  If it doesn't turn on we signal via
+        // the emitted 'error' that something bad happened.
+        // But if we ever again attain a connected state, this feature will be operational.
+        //
+        this._userRegisteredC2dListener = true;
         /*Codes_SRS_NODE_DEVICE_CLIENT_16_004: [The client shall start listening for messages from the service whenever there is a listener subscribed to the `message` event.]*/
         debug('in newListener, enabling C2D.');
         this._enableC2D((err) => {
@@ -74,7 +83,6 @@ export class Client extends InternalClient {
             debugErrors('in newListener, error enabling C2D: ' + err);
             this.emit('error', err);
           } else {
-            this._c2dFeature = true;
             debug('in newListener, successfully enabled C2D');
           }
         });
@@ -96,7 +104,7 @@ export class Client extends InternalClient {
       if (err && this._retryPolicy.shouldRetry(err)) {
         debugErrors('reconnect policy specifies a reconnect on error');
         /*Codes_SRS_NODE_DEVICE_CLIENT_16_097: [If the transport emits a `disconnect` event while the client is subscribed to c2d messages the retry policy shall be used to reconnect and re-enable the feature using the transport `enableC2D` method.]*/
-        if (this._c2dFeature) {
+        if (this._userRegisteredC2dListener) {
           // turn on C2D
           debug('disconnectHandler re-enabling C2D');
           this._enableC2D((err) => {

device/core/src/internal_client.ts

@@ -58,14 +58,14 @@ export abstract class InternalClient extends EventEmitter {
 
   private _methodCallbackMap: any;
   private _disconnectHandler: (err?: Error, result?: any) => void;
-  private _methodsEnabled: boolean;
+  private _userRegisteredMethodListener: boolean;
 
   constructor(transport: DeviceTransport, connStr?: string) {
     /*Codes_SRS_NODE_INTERNAL_CLIENT_05_001: [The Client constructor shall throw ReferenceError if the transport argument is falsy.]*/
     if (!transport) throw new ReferenceError('transport is \'' + transport + '\'');
 
     super();
-    this._methodsEnabled = false;
+    this._userRegisteredMethodListener = false;
 
     if (connStr) {
       throw new errors.InvalidOperationError('the connectionString parameter of the constructor is not used - users of the SDK should be using the `fromConnectionString` factory method.');
@@ -92,8 +92,7 @@ export abstract class InternalClient extends EventEmitter {
       }
       if (err && this._retryPolicy.shouldRetry(err)) {
         /*Codes_SRS_NODE_INTERNAL_CLIENT_16_098: [If the transport emits a `disconnect` event while the client is subscribed to direct methods the retry policy shall be used to reconnect and re-enable the feature using the transport `enableMethods` method.]*/
-        if (this._methodsEnabled) {
+        if (this._userRegisteredMethodListener) {
-          this._methodsEnabled = false;
           debug('re-enabling Methods link');
           this._enableMethods((err) => {
             if (err) {
@@ -105,7 +104,7 @@ export abstract class InternalClient extends EventEmitter {
         }
 
         /*Codes_SRS_NODE_INTERNAL_CLIENT_16_099: [If the transport emits a `disconnect` event while the client is subscribed to desired properties updates the retry policy shall be used to reconnect and re-enable the feature using the transport `enableTwinDesiredPropertiesUpdates` method.]*/
-        if (this._twin && this._twin.desiredPropertiesUpdatesEnabled) {
+        if (this._twin && this._twin.userRegisteredDesiredPropertiesListener) {
           debug('re-enabling Twin');
           this._twin.enableTwinDesiredPropertiesUpdates((err) => {
             if (err) {
@@ -348,6 +347,16 @@ export abstract class InternalClient extends EventEmitter {
   }
 
   protected _onDeviceMethod(methodName: string, callback: (request: DeviceMethodRequest, response: DeviceMethodResponse) => void): void {
+    //
+    // We want to always retain that the we want to have this feature enabled because the API (.on) doesn't really
+    // provide for the capability to say it failed.  It can certainly fail because a network operation is required to
+    // enable.
+    // By saving this off, we are strictly honoring that the feature is enabled.  If it doesn't turn on we signal via
+    // the emitted 'error' that something bad happened.
+    // But if we ever again attain a connected state, this feature will be operational.
+    //
+    this._userRegisteredMethodListener = true;
+
     // validate input args
     this._validateDeviceMethodInputs(methodName, callback);
 
@@ -426,19 +435,18 @@ export abstract class InternalClient extends EventEmitter {
   }
 
   private _enableMethods(callback: (err?: Error) => void): void {
-    if (!this._methodsEnabled) {
+    debug('enabling methods');
-      const retryOp = new RetryOperation('_enableMethods', this._retryPolicy, this._maxOperationTimeout);
+    const retryOp = new RetryOperation('_enableMethods', this._retryPolicy, this._maxOperationTimeout);
-      retryOp.retry((opCallback) => {
+    retryOp.retry((opCallback) => {
-        this._transport.enableMethods(opCallback);
+      this._transport.enableMethods(opCallback);
-      }, (err) => {
+    }, (err) => {
-        if (!err) {
+      if (!err) {
-          this._methodsEnabled = true;
+        debug('enabled methods');
-        }
+      } else {
-        callback(err);
+        debugErrors('Error while enabling methods: ' + err);
-      });
+      }
-    } else {
+      callback(err);
-      callback();
+    });
-    }
   }
 
   // Currently there is no code making use of this function, because there is no "removeDeviceMethod" corresponding to "onDeviceMethod"

device/core/src/module_client.ts

@@ -30,7 +30,7 @@ function safeCallback(callback?: (err?: Error, result?: any) => void, error?: Er
  * to create an IoT Hub device client.
  */
 export class ModuleClient extends InternalClient {
-  private _inputMessagesEnabled: boolean;
+  private _userRegisteredInputMessageListener: boolean;
   private _moduleDisconnectHandler: (err?: Error, result?: any) => void;
   private _methodClient: MethodClient;
 
@@ -44,7 +44,7 @@ export class ModuleClient extends InternalClient {
    */
   constructor(transport: DeviceTransport, methodClient: MethodClient) {
     super(transport, undefined);
-    this._inputMessagesEnabled = false;
+    this._userRegisteredInputMessageListener = false;
     this._methodClient = methodClient;
 
     /* Codes_SRS_NODE_MODULE_CLIENT_18_012: [ The `inputMessage` event shall be emitted when an inputMessage is received from the IoT Hub service. ]*/
@@ -53,12 +53,17 @@ export class ModuleClient extends InternalClient {
       this.emit('inputMessage', inputName, msg);
     });
 
-    this.on('removeListener', (eventName) => {
+    this.on('removeListener', () => {
-      if (eventName === 'inputMessage' && this.listeners('inputMessage').length === 0) {
+      if (this.listenerCount('inputMessage') === 0) {
+        this._userRegisteredInputMessageListener = false;
         /* Codes_SRS_NODE_MODULE_CLIENT_18_015: [ The client shall stop listening for messages from the service whenever the last listener unsubscribes from the `inputMessage` event. ]*/
+        debug('in removeListener, disabling input messages');
         this._disableInputMessages((err) => {
           if (err) {
+            debugErrors('in removeListener, error disabling input messages: ' + err);
             this.emit('error', err);
+          } else {
+            debug('removeListener successfully disabled input messages.');
           }
         });
       }
@@ -66,11 +71,23 @@ export class ModuleClient extends InternalClient {
 
     this.on('newListener', (eventName) => {
       if (eventName === 'inputMessage') {
+        //
+        // We want to always retain that the we want to have this feature enabled because the API (.on) doesn't really
+        // provide for the capability to say it failed.  It can certainly fail because a network operation is required to
+        // enable.
+        // By saving this off, we are strictly honoring that the feature is enabled.  If it doesn't turn on we signal via
+        // the emitted 'error' that something bad happened.
+        // But if we ever again attain a connected state, this feature will be operational.
+        //
+        this._userRegisteredInputMessageListener = true;
         /* Codes_SRS_NODE_MODULE_CLIENT_18_014: [ The client shall start listening for messages from the service whenever there is a listener subscribed to the `inputMessage` event. ]*/
+        debug('in newListener, enabling input messages');
         this._enableInputMessages((err) => {
           if (err) {
-            /*Codes_SRS_NODE_MODULE_CLIENT_18_017: [The client shall emit an `error` if connecting the transport fails while subscribing to `inputMessage` events.]*/
+            debugErrors('in newListener, error enabling input messages: ' + err);
             this.emit('error', err);
+          } else {
+            debug('in newListener, successfully enabled input messages');
           }
         });
       }
@@ -83,8 +100,7 @@ export class ModuleClient extends InternalClient {
         debug('transport disconnect event:  no error');
       }
       if (err && this._retryPolicy.shouldRetry(err)) {
-        if (this._inputMessagesEnabled) {
+        if (this._userRegisteredInputMessageListener) {
-          this._inputMessagesEnabled = false;
           debug('re-enabling input message link');
           this._enableInputMessages((err) => {
             if (err) {
@@ -251,33 +267,30 @@ export class ModuleClient extends InternalClient {
   }
 
   private _disableInputMessages(callback: (err?: Error) => void): void {
-    if (this._inputMessagesEnabled) {
+    debug('disabling input messages');
-      this._transport.disableInputMessages((err) => {
+    this._transport.disableInputMessages((err) => {
-        if (!err) {
+      if (!err) {
-          this._inputMessagesEnabled = false;
+        debug('disabled input messages');
-        }
+      } else {
-        callback(err);
+        debugErrors('Error while disabling input messages: ' + err);
-      });
+      }
-    } else {
+      callback(err);
-      callback();
+    });
-    }
   }
 
   private _enableInputMessages(callback: (err?: Error) => void): void {
-    if (!this._inputMessagesEnabled) {
+    debug('enabling input messages');
-      const retryOp = new RetryOperation('_enableInputMessages', this._retryPolicy, this._maxOperationTimeout);
+    const retryOp = new RetryOperation('_enableInputMessages', this._retryPolicy, this._maxOperationTimeout);
-      retryOp.retry((opCallback) => {
+    retryOp.retry((opCallback) => {
-        /* Codes_SRS_NODE_MODULE_CLIENT_18_016: [ The client shall connect the transport if needed in order to receive inputMessages. ]*/
+      this._transport.enableInputMessages(opCallback);
-        this._transport.enableInputMessages(opCallback);
+    }, (err) => {
-      }, (err) => {
+      if (!err) {
-        if (!err) {
+        debug('enabled input messages');
-          this._inputMessagesEnabled = true;
+      } else {
-        }
+        debugErrors('Error while enabling input messages: ' + err);
-        callback(err);
+      }
-      });
+      callback(err);
-    } else {
+    });
-      callback();
-    }
   }
 
   /**

device/core/src/twin.ts

@@ -50,7 +50,7 @@ export class Twin extends EventEmitter {
    * The desired and reported properties dictionaries (respectively in `properties.desired` and `properties.reported`).
    */
   properties: TwinProperties;
-  desiredPropertiesUpdatesEnabled: boolean;
+  userRegisteredDesiredPropertiesListener: boolean;
   private _transport: DeviceTransport;
   private _retryPolicy: RetryPolicy;
   private _maxOperationTimeout: number;
@@ -68,7 +68,7 @@ export class Twin extends EventEmitter {
     this._transport = transport;
     this._retryPolicy = retryPolicy;
     this._maxOperationTimeout = maxTimeout;
-    this.desiredPropertiesUpdatesEnabled = false;
+    this.userRegisteredDesiredPropertiesListener = false;
     this.on('newListener', this._handleNewListener.bind(this));
     /*Codes_SRS_NODE_DEVICE_TWIN_16_001: [The `Twin` constructor shall subscribe to the `twinDesiredPropertiesUpdate` event off the `transport` object.]*/
     this._transport.on('twinDesiredPropertiesUpdate', this._onDesiredPropertiesUpdate.bind(this));
@@ -119,13 +119,18 @@ export class Twin extends EventEmitter {
    * @private
    */
   enableTwinDesiredPropertiesUpdates(callback: (err?: Error) => void): void {
+    debug('enabling twin desired properties updates');
     const retryOp = new RetryOperation('enableTwinDesiredPropertiesUpdates', this._retryPolicy, this._maxOperationTimeout);
     retryOp.retry((opCallback) => {
-      this._transport.enableTwinDesiredPropertiesUpdates((err) => {
+      this._transport.enableTwinDesiredPropertiesUpdates(opCallback);
-        this.desiredPropertiesUpdatesEnabled = !err;
+    }, (err) => {
-        opCallback(err);
+      if (!err) {
-      });
+        debug('enabled twin desired properties updates');
-    }, callback);
+      } else {
+        debugErrors('Error while enabling twin desired properties updates: ' + err);
+      }
+      callback(err);
+    });
   }
 
   // Note: Since we currently don't keep track of listeners, so we don't "disable" the twin properties updates when no one is listening.
@@ -212,6 +217,15 @@ export class Twin extends EventEmitter {
           self.emit(eventName, propertyValue);
         });
       }
+      //
+      // We want to always retain that the we want to have this feature enabled because the API (.on) doesn't really
+      // provide for the capability to say it failed.  It can certainly fail because a network operation is required to
+      // enable.
+      // By saving this off, we are strictly honoring that the feature is enabled.  If it doesn't turn on we signal via
+      // the emitted 'error' that something bad happened.
+      // But if we ever again attain a connected state, this feature will be operational.
+      //
+      this.userRegisteredDesiredPropertiesListener = true;
 
       /*Codes_SRS_NODE_DEVICE_TWIN_16_010: [When a listener is added for the first time on an event which name starts with `properties.desired`, the twin shall call the `enableTwinDesiredPropertiesUpdates` method of the `Transport` object.]*/
       this.enableTwinDesiredPropertiesUpdates((err) => {

device/core/test/_module_client_test.js

@@ -615,11 +615,8 @@ describe('ModuleClient', function () {
         sinon.spy(fakeTransport, testConfig.enableFunc);
         var client = new ModuleClient(fakeTransport);
 
-        // Calling 'on' twice to make sure it's called only once on the receiver.
-        // It works because the test will fail if the test callback is called multiple times, and it's called for every time the testConfig.eventName event is subscribed on the receiver.
         client.on(testConfig.eventName, function () {});
-        client.on(testConfig.eventName, function () {});
+        assert.strictEqual(fakeTransport[testConfig.enableFunc].callCount, 1);
-        assert.isTrue(fakeTransport[testConfig.enableFunc].calledOnce);
       });
 
       /*Tests_SRS_NODE_MODULE_CLIENT_18_015: [ The client shall stop listening for messages from the service whenever the last listener unsubscribes from the `inputMessage` event. ]*/

device/samples/javascript/issue1110.js

@@ -1,83 +0,0 @@
-'use strict';
-
-const Protocol = require('azure-iot-device-mqtt').MqttWs;
-const Client = require('azure-iot-device').Client;
-const Message = require('azure-iot-device').Message;
-
-const deviceConnectionString = process.env.IOTHUB_DEVICE_CONNECTION_STRING;
-let sendInterval = 0;
-
-function connectHandler () {
-  console.log('+Client connected');
-  if (!sendInterval) {
-    sendMessage();
-    // const time = 5 * 1000;
-    // sendInterval = setInterval(sendMessage, time);
-  }
-  console.log('-Client connected');
-}
-
-function disconnectHandler () {
-  console.log('+Client disconnected');
-  client.open().catch((err) => {
-    console.error(err.message);
-  });
-  console.log('-Client disconnected');
-}
-
-function messageHandler (msg) {
-  console.log('+Client onMessage');
-  console.log('Id: ' + msg.messageId + ' Body: ' + msg.data);
-  client.complete(msg, printResultFor('completed'));
-  console.log('-Client onMessage');
-}
-
-async function errorHandler (err) {
-  console.log('+Client error');
-  console.error(err.message);
-  console.log('calling DeviceClient.close()');
-  await client.close();
-  console.log('returned from DeviceClient.closed');
-  console.log('-Client error');
-}
-
-function sendMessage() {
-  // SET A BREAKPOINT AT HERE.
-  // ONCE YOU GET BREAK INTO THE DEBUGGER. DISCONNECT NETWORK CABLE AND GO.
-  console.log('Please unplug network cable from your device to simulate disconnection');
-  const message = generateMessage();
-  console.log('Sending message: ' + message.getData());
-  client.sendEvent(message, printResultFor('send'));
-}
-
-function generateMessage () {
-  const windSpeed = 10 + (Math.random() * 4); // range: [10, 14]
-  const temperature = 20 + (Math.random() * 10); // range: [20, 30]
-  const humidity = 60 + (Math.random() * 20); // range: [60, 80]
-  const data = JSON.stringify({ deviceId: 'myFirstDevice', windSpeed: windSpeed, temperature: temperature, humidity: humidity });
-  const message = new Message(data);
-  message.properties.add('temperatureAlert', (temperature > 28) ? 'true' : 'false');
-  return message;
-}
-
-// fromConnectionString must specify a transport constructor, coming from any transport package.
-let client = Client.fromConnectionString(deviceConnectionString, Protocol);
-
-client.on('connect', connectHandler);
-client.on('error', errorHandler);
-client.on('disconnect', disconnectHandler);
-client.on('message', messageHandler);
-
-client.open()
-.catch(err => {
-  console.error('Could not connect: ' + err.message);
-});
-
-
-// Helper function to print results in the console
-function printResultFor(op) {
-  return async function printResult(err, res) {
-    if (err) console.log(op + ' status: ' + err.constructor.name);
-    if (res) console.log(op + ' status: ' + res.constructor.name);
-  };
-}

device/transport/mqtt/src/mqtt.ts

@@ -813,7 +813,7 @@ export class Mqtt extends EventEmitter implements DeviceTransport {
       /*Codes_SRS_NODE_DEVICE_MQTT_16_052: [`enableC2D` shall call its callback with an `Error` if subscribing to the topic fails.]*/
       /*Codes_SRS_NODE_DEVICE_MQTT_16_053: [`enableMethods` shall call its callback with an `Error` if subscribing to the topic fails.]*/
       /*Codes_SRS_NODE_DEVICE_MQTT_18_063: [`enableInputMessages` shall call its callback with an `Error` if subscribing to the topic fails. ]*/
-      callback(err);
+      this._ignoreConnectionClosedInErrorCallback(callback)(err);
     });
   }
 
@@ -831,7 +831,7 @@ export class Mqtt extends EventEmitter implements DeviceTransport {
       /*Codes_SRS_NODE_DEVICE_MQTT_16_043: [`disableC2D` shall call its callback with an `Error` if an error is received while unsubscribing.]*/
       /*Codes_SRS_NODE_DEVICE_MQTT_16_046: [`disableMethods` shall call its callback with an `Error` if an error is received while unsubscribing.]*/
       /*Codes_SRS_NODE_DEVICE_MQTT_18_067: [ `disableInputMessages` shall call its callback with an `Error` if an error is received while unsubscribing. ]*/
-      callback(err);
+      this._ignoreConnectionClosedInErrorCallback(callback)(err);
     });
   }
 
@@ -1020,6 +1020,22 @@ export class Mqtt extends EventEmitter implements DeviceTransport {
       });
     }
   }
+
+  //
+  // We encountered an issue where a closed error was raised which some application handlers would respond to by
+  // calling closed.  This would then deadlock behind the code currently executing in mqttjs's close code.
+  // The best solution that could happen exclusively inside the SDK code would be to drop the close error because
+  // we KNOW that a disconnect would be raised after mqttjs finishes the close 'rundown'.
+  //
+  private _ignoreConnectionClosedInErrorCallback(callback: (err?: Error, ...args: any[]) => void): (err?: Error, ...args: any[]) => void {
+    return (err: Error, ...args: any[]) => {
+      if (err?.name === 'Error' && err?.message === 'Connection closed') {
+        debug('Mqtt subscribe/unsubscribe operation failed due to MQTT.js connection closed error. MqttBase will handle this when MQTT.js emits the close event.');
+        return;
+      }
+      callback(err, ...args);
+    };
+  }
 }
 
 

device/transport/mqtt/src/mqtt_twin_client.ts

@@ -168,7 +168,7 @@ export class MqttTwinClient extends EventEmitter {
       if (err) {
         debugErrors('failed to subscribe to desired properties updates: ' + err);
         /*Codes_SRS_NODE_DEVICE_MQTT_TWIN_CLIENT_16_021: [if subscribing fails with an error the `enableTwinDesiredPropertiesUpdates` shall call its callback with the translated version of this error obtained by using the `translateError` method of the `azure-iot-mqtt-base` package.]*/
-        callback(translateError(err));
+        this._ignoreConnectionClosedInErrorCallback(callback)(err);
       } else {
         debug('suback: ' + JSON.stringify(suback));
         /*Codes_SRS_NODE_DEVICE_MQTT_TWIN_CLIENT_16_020: [The `enableTwinDesiredPropertiesUpdates` shall call its callback with no arguments if the subscription is successful.]*/
@@ -183,7 +183,7 @@ export class MqttTwinClient extends EventEmitter {
       if (err) {
         debugErrors('failed to subscribe to desired properties updates: ' + err);
         /*Codes_SRS_NODE_DEVICE_MQTT_TWIN_CLIENT_16_024: [if unsubscribing fails with an error the `disableTwinDesiredPropertiesUpdates` shall call its callback with the translated version of this error obtained by using the `translateError` method of the `azure-iot-mqtt-base` package.]*/
-        callback(translateError(err));
+        this._ignoreConnectionClosedInErrorCallback(callback)(err);
       } else {
         debug('suback: ' + JSON.stringify(suback));
         /*Codes_SRS_NODE_DEVICE_MQTT_TWIN_CLIENT_16_023: [The `disableTwinDesiredPropertiesUpdates` shall call its callback with no arguments if the unsubscription is successful.]*/
@@ -271,4 +271,20 @@ export class MqttTwinClient extends EventEmitter {
       debugErrors('received a response with a malformed topic property: ' + topic);
     }
   }
+
+  //
+  // We encountered an issue(#1110) where a closed error was raised which some application handlers would respond to by
+  // calling closed.  This would then deadlock behind the code currently executing in mqttjs's close code.
+  // The best solution that could happen exclusively inside the SDK code would be to drop the close error because
+  // we KNOW that a disconnect would be raised after mqttjs finishes the close 'rundown'.
+  //
+  private _ignoreConnectionClosedInErrorCallback(callback: (err?: Error, ...args: any[]) => void): (err?: Error, ...args: any[]) => void {
+    return (err: Error, ...args: any[]) => {
+      if (err?.name === 'Error' && err?.message === 'Connection closed') {
+        debug('Mqtt subscribe/unsubscribe operation failed due to MQTT.js connection closed error. MqttBase will handle this when MQTT.js emits the close event.');
+        return;
+      }
+      callback(translateError(err), ...args);
+    };
+  }
 }

device/transport/mqtt/test/_mqtt_test.js

@@ -1061,6 +1061,20 @@ describe('Mqtt', function () {
         });
       });
 
+      it('Does NOT invoke callback if subscribing fails with closed error', function (testCallback) {
+        const transport = new Mqtt(fakeAuthenticationProvider, fakeMqttBase);
+        const closedError = new Error();
+        closedError.name = 'Error';
+        closedError.message = 'Connection closed';
+        fakeMqttBase.subscribe = sinon.stub().callsArgWith(2, closedError);
+        transport.connect(function () {
+          transport[testConfig.methodName](function (err) {
+            assert.fail('Should not have invoked the callback if the mqtt transport was closed');
+          });
+          testCallback();
+        });
+      });
+
       it('will not subscribe multiple times to the same topic', function (testCallback) {
         const transport = new Mqtt(fakeAuthenticationProvider, fakeMqttBase);
         transport[testConfig.methodName](function (err) {
@@ -1125,6 +1139,22 @@ describe('Mqtt', function () {
         });
       });
 
+      it('Does NOT invoke its callback if err is closed on fail of unsubscribe', function (testCallback) {
+        const transport = new Mqtt(fakeAuthenticationProvider, fakeMqttBase);
+        const closedError = new Error();
+        closedError.name = 'Error';
+        closedError.message = 'Connection closed';
+        fakeMqttBase.unsubscribe = sinon.stub().callsArgWith(1, closedError);
+        transport.connect(function () {
+          transport[testConfig.enableFeatureMethod](function () {
+            transport[testConfig.disableFeatureMethod](function (err) {
+              assert.fail('Should NOT have invoked callback');
+            });
+          });
+          testCallback();
+        });
+      });
+
       /* Tests_SRS_NODE_DEVICE_MQTT_16_054: [`disableC2D` shall call its callback with no arguments when the `UNSUBACK` packet is received.]*/
       /* Tests_SRS_NODE_DEVICE_MQTT_16_055: [`disableMethods` shall call its callback with no arguments when the `UNSUBACK` packet is received.]*/
       /* Tests_SRS_NODE_DEVICE_MQTT_16_064: [`disableTwinDesiredPropertiesUpdates` shall call its callback with no arguments when the `UNSUBACK` packet is received.]*/
@@ -1254,6 +1284,38 @@ describe('Mqtt', function () {
       });
     });
 
+    it('Does NOT invoke its callback if subscribe fails with connection closed', function (testCallback) {
+      const closedError = new Error();
+      closedError.name = 'Error';
+      closedError.message = 'Connection closed';
+      fakeMqttBase.subscribe = sinon.stub().callsArgWith(2, closedError);
+      const transport = new Mqtt(fakeAuthenticationProvider, fakeMqttBase);
+      transport.connect(function () {
+        transport.enableTwinDesiredPropertiesUpdates(function (err) {
+          assert.fail('Should NOT have invoked the callback');
+        })
+      });
+      testCallback();
+    });
+
+    it('Does NOT invoke its callback if unsubscribe fails with connection closed', function (testCallback) {
+      const closedError = new Error();
+      closedError.name = 'Error';
+      closedError.message = 'Connection closed';
+      fakeMqttBase.unsubscribe = sinon.stub().callsArgWith(1, closedError);
+      const transport = new Mqtt(fakeAuthenticationProvider, fakeMqttBase);
+      transport.connect(function () {
+        transport.enableTwinDesiredPropertiesUpdates(function (err) {
+          assert.isUndefined(err, 'Subscribe yielded error');
+          transport.disableTwinDesiredPropertiesUpdates(function (err) {
+            assert.fail('Should NOT have invoked the callback');
+          });
+        });
+      });
+      testCallback();
+    });
+
+
     /* Tests_SRS_NODE_DEVICE_MQTT_16_059: [`enableTwinDesiredPropertiesUpdates` shall call the `enableTwinDesiredPropertiesUpdates` on the `MqttTwinClient` object created by the constructor and pass it its callback.]*/
     it('calls \'enableTwinDesiredPropertiesUpdates\' on the MqttTwinClient and passes its callback', function () {
       const transport = new Mqtt(fakeAuthenticationProvider, fakeMqttBase);

service/package.json

@@ -10,7 +10,7 @@
     "async": "^3.2.3",
     "es5-ext": "0.10.53",
     "@azure/core-http": "1.2.3",
-    "@azure/identity": "1.2.5",
+    "@azure/identity": "2.0.0",
     "@azure/ms-rest-js": "^2.0.5",
     "azure-iot-amqp-base": "2.5.0",
     "azure-iot-common": "1.13.0",