feat(mqtt): Stream map serialization to improve memory usage with lar…

…ge maps

backend/lib/mqtt/handles/MapNodeMqttHandle.js

@@ -8,7 +8,7 @@ const Logger = require("../../Logger");
 const NodeMqttHandle = require("./NodeMqttHandle");
 const path = require("path");
 const PropertyMqttHandle = require("./PropertyMqttHandle");
-const zlib = require("zlib");
+const {stringifyAndDeflate} = require("../../utils/streamHelpers");
 
 class MapNodeMqttHandle extends NodeMqttHandle {
     /**
@@ -157,10 +157,7 @@ class MapNodeMqttHandle extends NodeMqttHandle {
         const robot = this.robot;
 
         const promise = new Promise((resolve, reject) => {
-            zlib.deflate(JSON.stringify(robot.state.map), (err, buf) => {
-                if (err !== null) {
-                    return reject(err);
-                }
+            stringifyAndDeflate(robot.state.map).then(buf => {
                 let payload;
 
                 if (mapHack) {
@@ -189,6 +186,8 @@ class MapNodeMqttHandle extends NodeMqttHandle {
                 }
 
                 resolve(payload);
+            }).catch(err => {
+                reject(err);
             });
         });
 

backend/lib/utils/BufferWriteStream.js

@@ -0,0 +1,20 @@
+const stream = require("stream");
+
+class BufferWriteStream extends stream.Writable {
+    constructor() {
+        super();
+
+        this.chunks = [];
+    }
+
+    _write(chunk, encoding, done) {
+        this.chunks.push(chunk);
+        done();
+    }
+
+    getBuffer() {
+        return Buffer.concat(this.chunks);
+    }
+}
+
+module.exports = BufferWriteStream;

backend/lib/utils/streamHelpers.js

@@ -0,0 +1,37 @@
+const BufferWriteStream = require("./BufferWriteStream");
+const stream = require("stream");
+const zlib = require("zlib");
+const {JsonStream} = require("jetsons");
+
+function stringifyAndCompress(data, algorithm) {
+    return new Promise((resolve, reject) => {
+        const bufferStream = new BufferWriteStream();
+
+        stream.pipeline(
+            new JsonStream(data),
+            algorithm,
+            bufferStream,
+
+            (err) => {
+                if (err !== null && err !== undefined) {
+                    reject(err);
+                } else {
+                    resolve(bufferStream.getBuffer());
+                }
+            }
+        );
+    });
+}
+
+function stringifyAndDeflate(data) {
+    return stringifyAndCompress(data, zlib.createDeflate());
+}
+
+function stringifyAndGZip(data) {
+    return stringifyAndCompress(data, zlib.createGzip());
+}
+
+module.exports = {
+    stringifyAndDeflate: stringifyAndDeflate,
+    stringifyAndGZip: stringifyAndGZip
+};

backend/package.json

@@ -41,6 +41,7 @@
     "express-list-endpoints": "^5.0.0",
     "express-rate-limit": "^5.2.6",
     "expresse": "git+https://npm@github.com/Hypfer/expresse",
+    "jetsons": "^1.2.2",
     "mqtt": "^4.0.0",
     "multer": "^1.4.1",
     "uuid": "^8.3.0",