fixed unsubscribe, added test

src/services/__tests__/mqtt-client-service.test.js

+/**
+ * @jest-environment jsdom
+*/
+import '@testing-library/jest-dom';
+
+import MqttClientService from '../mqtt-client-service';
+
+let subscribeTopicAndValidate = (topic, callback) => {
+  let token = MqttClientService.instance.subscribeToTopic(topic, callback);
+  expect(token).toBeDefined();
+  expect(token.topic).toBe(topic);
+  expect(token.callback).toBe(callback);
+  expect(MqttClientService.instance.subTokensMap.get(topic).includes(token)).toBeTruthy();
+
+  return token;
+};
+
+let unsubscribeAndValidate = (token) => {
+  MqttClientService.instance.unsubscribe(token);
+  expect(MqttClientService.instance.subTokensMap.get(token.topic).includes(token)).toBeFalsy();
+};
+
+test('sub/unsub', async () => {
+  let topicA = 'topic/A';
+  let topicB = 'topic/B';
+
+  let sub1Callback = jest.fn();
+  let sub1Token = subscribeTopicAndValidate(topicA, sub1Callback);
+  let sub2Callback = jest.fn();
+  let sub2Token = subscribeTopicAndValidate(topicA, sub2Callback);
+  let sub3Callback = jest.fn();
+  let sub3Token = subscribeTopicAndValidate(topicB, sub3Callback);
+
+  expect(MqttClientService.instance.subTokensMap.get(topicA).length).toBe(2);
+  expect(MqttClientService.instance.subTokensMap.get(topicB).length).toBe(1);
+
+  MqttClientService.instance.onMessage(topicA, {});
+  MqttClientService.instance.onMessage(topicB, {});
+  expect(sub1Token.callback).toHaveBeenCalledTimes(1);
+  expect(sub2Token.callback).toHaveBeenCalledTimes(1);
+  expect(sub3Token.callback).toHaveBeenCalledTimes(1);
+
+  unsubscribeAndValidate(sub1Token);
+  expect(MqttClientService.instance.subTokensMap.get(topicA).length).toBe(1);
+  expect(MqttClientService.instance.subTokensMap.get(topicB).length).toBe(1);
+
+  MqttClientService.instance.onMessage(topicA, {});
+  MqttClientService.instance.onMessage(topicB, {});
+  expect(sub1Token.callback).toHaveBeenCalledTimes(1);
+  expect(sub2Token.callback).toHaveBeenCalledTimes(2);
+  expect(sub3Token.callback).toHaveBeenCalledTimes(2);
+
+  unsubscribeAndValidate(sub2Token);
+  expect(MqttClientService.instance.subTokensMap.get(topicA).length).toBe(0);
+  expect(MqttClientService.instance.subTokensMap.get(topicB).length).toBe(1);
+
+  MqttClientService.instance.onMessage(topicA, {});
+  MqttClientService.instance.onMessage(topicB, {});
+  expect(sub1Token.callback).toHaveBeenCalledTimes(1);
+  expect(sub2Token.callback).toHaveBeenCalledTimes(2);
+  expect(sub3Token.callback).toHaveBeenCalledTimes(3);
+
+  unsubscribeAndValidate(sub3Token);
+  expect(MqttClientService.instance.subTokensMap.get(topicA).length).toBe(0);
+  expect(MqttClientService.instance.subTokensMap.get(topicB).length).toBe(0);
+
+  MqttClientService.instance.onMessage(topicA, {});
+  MqttClientService.instance.onMessage(topicB, {});
+  expect(sub1Token.callback).toHaveBeenCalledTimes(1);
+  expect(sub2Token.callback).toHaveBeenCalledTimes(2);
+  expect(sub3Token.callback).toHaveBeenCalledTimes(3);
+});

src/services/mqtt-client-service.js

@@ -47,15 +47,18 @@ export default class MqttClientService extends EventEmitter {
   }
 
   onMessage(topic, payload, packet) {
+    if (typeof payload === 'undefined') {
+      return;
+    }
+
     //console.info('MQTT message: [topic, payload, packet]');
     //console.info([topic, payload, packet]);
     //Now we see which callbacks have been assigned for a topic
-    if (typeof this.subTokensMap.get(topic) !== 'undefined') {
-      for (var token in this.subTokensMap.get(topic)){
-        if (typeof token.callback === 'function' && payload !== 'undefined') {
-          //Deserializatin of Data must happen here
-          token.callback(payload);
-        }
+    let subTokens = this.subTokensMap.get(topic);
+    if (typeof subTokens !== 'undefined') {
+      for (var token of subTokens) {
+        //Deserializatin of Data must happen here
+        token.callback(payload);
       };
     };
 
@@ -95,22 +98,25 @@ export default class MqttClientService extends EventEmitter {
         [token]
       );
     }
-    console.info('You have been subscribed to topic ' + topic);
-    console.info(this.subTokensMap);
+    //console.info('You have been subscribed to topic ' + topic);
+    //console.info(this.subTokensMap);
     return token;
   }
 
   unsubscribe(unsubToken) {
     if (this.subTokensMap.has(unsubToken.topic)){
-      this.subTokensMap.get(unsubToken.topic).forEach(token => {
-        if (token === unsubToken){
-          this.subTokensMap.get(unsubToken.topic).pop(token);
-          console.info('You have been unsubscribed from topic ' + unsubToken.topic);
-        };
-      });
+      let tokens = this.subTokensMap.get(unsubToken.topic);
+      let index = tokens.indexOf(unsubToken);
+      if (index !== -1) {
+        tokens.splice(index, 1);
+        //console.info('You have been unsubscribed from topic ' + unsubToken.topic);
+      }
+      else {
+        console.warn('Your provided token could not be found in the subscription list');
+      }
     }
     else{
-      console.info('The topic ' + unsubToken.topic + ' was not found');
+      console.warn('The topic ' + unsubToken.topic + ' was not found');
     }
   }