feat: 🎸 improve publish parsing

Author: streamich

benchmarks/micro/mqtt-codec-publish.js

@@ -0,0 +1,18 @@
+const {MqttDecoder} = require('../../es6/MqttDecoder');
+const {connectShort, publishSample} = require('../packets');
+
+const parser = new MqttDecoder();
+const max = 10e6;
+let i;
+const start = Date.now() / 1000;
+
+for (i = 0; i < max; i++) {
+  parser.push(Buffer.from(publishSample));
+  while (parser.parse()) {}
+}
+
+const time = Date.now() / 1000 - start;
+console.log('Total packets', max);
+console.log('Total time', Math.round(time * 100) / 100);
+console.log('Packets/s', (Math.round(max / time / 1e4) / 1e2) + 'M');
+console.log('Memory used', (process.memoryUsage().heapUsed / 1024 / 1024).toFixed(2) + ' Mb');

src/MqttDecoder.ts

@@ -2,7 +2,7 @@ import {BufferList} from './BufferList';
 import {ERROR, PACKET_TYPE} from './enums';
 import {PacketConnack, parseConnack} from './packets/connack';
 import {PacketConnect, parseConnect} from './packets/connect';
-import {PacketPublish, parsePublish} from './packets/publish';
+import {PacketPublish} from './packets/publish';
 import {PacketPuback, parsePuback} from './packets/puback';
 import {PacketPubrec, parsePubrec} from './packets/pubrec';
 import {PacketPubrel, parsePubrel} from './packets/pubrel';
@@ -15,17 +15,14 @@ import {PacketPingreq} from './packets/pingreq';
 import {PacketPingresp} from './packets/pingresp';
 import {PacketDisconnect, parseDisconnect} from './packets/disconnect';
 import {PacketAuth, parseAuth} from './packets/auth';
-import { parseBinary, parseProps } from './util/parse';
-import { Properties } from './types';
+import {parseBinary, parseProps} from './util/parse';
+import {Properties} from './types';
 
 const enum DECODER_STATE {
   HEADER = 0,
   DATA = 1,
 }
 
-const parsers = new Map();
-parsers.set(PACKET_TYPE.PUBLISH, parsePublish);
-
 export class MqttDecoder {
   /** Keeps track of which part message framing are we in. */
   private state: DECODER_STATE = DECODER_STATE.HEADER;
@@ -126,15 +123,14 @@ export class MqttDecoder {
 
       if (this.state !== DECODER_STATE.DATA) return;
 
-      const {l: length} = this;
+      const {b, l} = this;
       let offset = this.offset;
-      const end = offset + length;
-      if (list.length < end) return;
+      const packetEndOffset = offset + l;
+      if (list.length < packetEndOffset) return;
 
       this.state = DECODER_STATE.HEADER;
       this.offset = 0;
 
-      const {b, l} = this;
       const type: PACKET_TYPE = (b >> 4) as PACKET_TYPE;
       switch (type) {
         case PACKET_TYPE.PUBLISH: {
@@ -151,80 +147,80 @@ export class MqttDecoder {
             p = props;
             offset += size;
           }
-          const d = list.slice(offset, list.length);
+          const d = list.slice(offset, packetEndOffset);
           const t = topic.toString('utf8');
-          list.consume(end);
+          list.consume(packetEndOffset);
           return new PacketPublish(b, l, t, i, p, d);
         }
         case PACKET_TYPE.CONNECT: {
           const packet = parseConnect(b, l, list, offset);
           this.version = packet.v;
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.CONNACK: {
           const packet = parseConnack(b, l, list, this.version, offset);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.PUBACK: {
           const packet = parsePuback(b, l, list, this.version, offset);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.PUBREC: {
           const packet = parsePubrec(b, l, list, this.version, offset);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.PUBREL: {
           const packet = parsePubrel(b, l, list, this.version, offset);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.PUBCOMP: {
           const packet = parsePubcomp(b, l, list, this.version, offset);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.SUBSCRIBE: {
           const packet = parseSubscribe(b, l, list, this.version, offset);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.SUBACK: {
           const packet = parseSuback(b, l, list, this.version, offset);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.UNSUBSCRIBE: {
           const packet = parseUnsubscribe(b, l, list, this.version, offset);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.UNSUBACK: {
           const packet = parseUnsuback(b, l, list, this.version, offset);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.PINGREQ: {
           const packet = new PacketPingreq(b, l);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.PINGRESP: {
           const packet = new PacketPingresp(b, l);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.DISCONNECT: {
           const packet = parseDisconnect(b, l, list, this.version, offset);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         case PACKET_TYPE.AUTH: {
           const packet = parseAuth(b, l, list, this.version, offset);
-          list.consume(end);
+          list.consume(packetEndOffset);
           return packet;
         }
         default: throw ERROR.MALFORMED_PACKET;

src/packets/publish.ts

@@ -1,7 +1,5 @@
-import {BufferList} from '../BufferList';
 import {Packet, PacketHeaderData} from '../packet';
 import {Properties} from '../types';
-import {parseBinary, parseProps} from '../util/parse';
 
 export interface PacketPublishData extends PacketHeaderData {
   /** Topic Name. */
@@ -26,22 +24,3 @@ export class PacketPublish extends Packet implements PacketPublishData {
     super(b, l);
   }
 }
-
-export const parsePublish = (b: number, l: number, data: BufferList, version: number, offset: number): PacketPublish => {
-  const topic = parseBinary(data, offset);
-  offset += 2 + topic.byteLength;
-  let i: number = 0;
-  if (((b >> 1) & 0b11) > 0) {
-    i = data.readUInt16BE(offset);
-    offset += 2;
-  }
-  let p: Properties = {};
-  if (version === 5) {
-    const [props, size] = parseProps(data, offset);
-    p = props;
-    offset += size;
-  }
-  const d = data.slice(offset, data.length);
-  const t = topic.toString('utf8');
-  return new PacketPublish(b, l, t, i, p, d);
-};