fix: catch exception when server sends invalid/incomplete json in sub…

…scription (Bug849302)

README.md

@@ -117,6 +117,7 @@ Any use of the source code and related documents of this repository in applicati
 * 2023-09-04: 1.9.3  - fix: subscription with ipv6 address that contains a zone index
 * 2024-03-13: 1.9.4  - fix: timeout of connection with Node.js v19. "Read timeout, received no data in undefinedms, assuming connection is dead"
 * 2024-03-19: 1.9.5  - fix: update dependencies in package-lock
+* 2024-05-24: 1.9.6  - fix: catch exception when server sends invalid/incomplete json in subscription (Bug849302)
 ```
 
 ## About

lib/CtrlxDatalayerSubscription.js

@@ -320,7 +320,7 @@ class CtrlxDatalayerSubscription extends EventEmitter {
           }
         },
         agent: new https.Agent({ keepAlive: false })  // create a dedicated agent to have dedicated connection instance. Also disable the agent-keep-alive explicitly.
-                                                      // This is necessary because since node.js 19 the default behaviour was changed. 
+                                                      // This is necessary because since node.js 19 the default behaviour was changed.
                                                       // https://nodejs.org/en/blog/announcements/v19-release-announce#https11-keepalive-by-default
       };
 
@@ -428,24 +428,31 @@ class CtrlxDatalayerSubscription extends EventEmitter {
             debugUpdate(`update(${e.data})`);
           }
 
-          let payload = CtrlxDatalayer._parseData(e.data);
-
-          if (!this.emit('update', payload, e.lastEventId)) {
-            // Listener seems not yet to be attached. Retry on next tick.
-            setTimeout(()=>this.emit('update', payload, e.lastEventId), 0);
+          try {
+            let payload = CtrlxDatalayer._parseData(e.data);
+            if (!this.emit('update', payload, e.lastEventId)) {
+              // Listener seems not yet to be attached. Retry on next tick.
+              setTimeout(()=>this.emit('update', payload, e.lastEventId), 0);
+            }
+          } catch (err) {
+            this.emit('error', new Error(`Error parsing update event: ${err.message}`));
           }
+
         });
 
         this._es.addEventListener('keepalive', (e) => {
           if (debugUpdate.enabled) {
             debugUpdate(`keepalive(${e.data})`);
           }
 
-          let payload = CtrlxDatalayer._parseData(e.data);
-
-          if (!this.emit('keepalive', payload, e.lastEventId)) {
-            // Listener seems not yet to be attached. Retry on next tick.
-            setTimeout(()=>this.emit('keepalive', payload, e.lastEventId), 0);
+          try {
+            let payload = CtrlxDatalayer._parseData(e.data);
+            if (!this.emit('keepalive', payload, e.lastEventId)) {
+              // Listener seems not yet to be attached. Retry on next tick.
+              setTimeout(()=>this.emit('keepalive', payload, e.lastEventId), 0);
+            }
+          } catch (err) {
+            this.emit('error', new Error(`Error parsing keepalive event: ${err.message}`));
           }
         });
 
@@ -476,10 +483,9 @@ class CtrlxDatalayerSubscription extends EventEmitter {
       // @ts-ignore
       this._es.onretrying = undefined;
       // @ts-ignore
-      this._es.onerror = (e) => { 
-        // ignore any pending errors in the pipeline, which might get emitted and result in an uncaught exception 
+      this._es.onerror = (e) => {
+        // ignore any pending errors in the pipeline, which might get emitted and result in an uncaught exception
         debug(`onerror(${e.type})`);
-        console.log("error after");
       };
       // @ts-ignore
       this._es.onmessage = undefined;

package.json

@@ -1,6 +1,6 @@
 {
   "name": "node-red-contrib-ctrlx-automation",
-  "version": "1.9.5",
+  "version": "1.9.6",
   "description": "Node-RED nodes for ctrlX AUTOMATION",
   "repository": {
     "type": "git",

test/ctrlx-datalayer-subscribe.test.js

@@ -464,6 +464,63 @@ describe('ctrlx-datalayer-subscribe', function () {
 
   });
 
+  describe('ctrlx-datalayer-subscribe: Error Handling', function () {
+    it('should handle invalid send json messages', function (done) {
 
+      let flow = [
+        { "id": "f1", "type": "tab", "label": "Test flow"},
+        { "id": "h1", "z":"f1", "type": "helper" },
+        { "id": "n1", "z":"f1", "type": "ctrlx-datalayer-subscribe", "subscription": "s1", "path": "test/invalid/json", "name": "subscribe", "wires": [["h1"]] },
+        { "id": "s1", "z":"f1", "type": "ctrlx-config-subscription", "device": "c1", "name": "sub1", "publishIntervalMs": "1000" },
+        { "id": "c1", "z":"f1", "type": "ctrlx-config", "name": "ctrlx", "hostname": getHostname(), "debug": true },
+      ];
+      let credentials = {
+        c1: {
+          username: getUsername(),
+          password: getPassword()
+        }
+      };
+
+      helper.load([ctrlxConfigNode, ctrlxConfigSubscriptionNode, ctrlxDatalayerSubscribeNode], flow, credentials, () => {
+
+        let s1 = helper.getNode("s1");
+        let h1 = helper.getNode("h1");
+        let n1 = helper.getNode("n1");
+
+        let numErrors = 0;
+
+        // @ts-ignore
+        h1.on("input", (msg) => {
+          try {
+            expect(msg).to.have.property('topic').to.be.a('string').eql('test/invalid/json');
+            expect(msg).to.have.property('timestamp').to.be.a('number');
+            expect(msg).to.have.property('type').to.be.a('string').eql('object');
+            expect(msg).to.have.property('payload').to.be.a('object').eql({
+              valid: "data"
+            });
+            // Make sure we received an error before we reveice the valid data.
+            expect(numErrors).eql(1);
+            s1.subscription.close();
+            done();
+          }
+          catch (err) {
+            s1.subscription.close();
+            done(err);
+          }
+        });
+
+        // We expect to reveive an error message, because the mockup will send an invalid JSON message.
+        n1.on('call:error', call => {
+          numErrors++;
+          try {
+            expect(call.firstArg).eql('Error parsing update event: Unexpected end of JSON input');
+          } catch (err) {
+            done(err);
+          }
+        });
+
+      });
+    });
+  });
 
 });

test/helper/CtrlxMockupV2.js

@@ -523,6 +523,17 @@ class CtrlxMockupV2 {
                 data.value = options;
                 data.type = 'object';
                 break;
+              case 'test/invalid/json':
+                // This is a special node which sends an invalid and malformed json to for example
+                // simulate a broken connection.
+                sseStream.write({
+                  id: id++,
+                  event: 'update',
+                  data: `{"type": "object", "value": { "invalid formed json`
+                });
+                data.value = { "valid": "data" };
+                data.type = 'object';
+                break;
 
               default:
                 data.value = 'error: unknown value';

test/helper/benchmark.js

@@ -27,7 +27,6 @@
 
 const CtrlxCore = require('../../lib/CtrlxCore')
 const { performance, PerformanceObserver } = require('perf_hooks')
-const async = require('async');
 const { assert } = require('console');