perf: ⚡️ inline connect parsing

Author: streamich

README.md

@@ -2,6 +2,7 @@
 
 - MQTT packet encoder/decoder for Node.js.
 - `mqtt-codec` is 4-10x faster than `mqtt-packet`, see benchmarks below and uses less memory.
+- Zero dependencies.
 
 ## Benchmarks
 

package.json

@@ -41,7 +41,6 @@
     "@semantic-release/changelog": "^5.0.1",
     "@semantic-release/git": "^9.0.0",
     "@semantic-release/npm": "^7.0.6",
-    "@types/bl": "^2.1.0",
     "@types/jest": "^26.0.14",
     "benchmark": "^2.1.4",
     "husky": "^4.3.0",

src/MqttDecoder.ts

@@ -1,7 +1,7 @@
 import {BufferList} from './BufferList';
 import {ERROR, PACKET_TYPE} from './enums';
 import {PacketConnack, parseConnack} from './packets/connack';
-import {PacketConnect, parseConnect} from './packets/connect';
+import {PacketConnect} from './packets/connect';
 import {PacketPublish} from './packets/publish';
 import {PacketPuback, parsePuback} from './packets/puback';
 import {PacketPubrec, parsePubrec} from './packets/pubrec';
@@ -153,9 +153,52 @@ export class MqttDecoder {
           return new PacketPublish(b, l, t, i, p, d);
         }
         case PACKET_TYPE.CONNECT: {
-          const packet = parseConnect(b, l, list, offset);
-          this.version = packet.v;
+          const buf = list.slice(offset, offset + l);
           list.consume(packetEndOffset);
+          offset = 2 + buf.readUInt16BE(0); // Skip "MQTT" or "MQIsdp" protocol name.
+          const v = buf.readUInt8(offset++);
+          const f = buf.readUInt8(offset++);
+          const k = buf.readUInt16BE(offset);
+          offset += 2;
+          // const ui32 = buf.readUInt32BE(offset);
+          // const v = (ui32 & 0xFF000000) >> 24;
+          // const f = (ui32 & 0xFF0000) >> 16;
+          // const k = (ui32 & 0xFFFF);
+          // offset += 4;
+          let p: Properties = {};
+          if (v === 5) {
+            const [props, propsSize] = parseProps(buf, offset);
+            p = props;
+            offset += propsSize;
+          }
+          const clientId = parseBinary(buf, offset);
+          const id = clientId.toString('utf8');
+          offset += 2 + clientId.byteLength;
+          const packet = new PacketConnect(b, l, v, f, k, p, id);
+          if (packet.willFlag()) {
+            if (v === 5) {
+              const [props, propsSize] = parseProps(buf, offset);
+              packet.wp = props;
+              offset += propsSize;
+            } else packet.wp = {};
+            const willTopic = parseBinary(buf, offset);
+            packet.wt = willTopic.toString('utf8');
+            offset += 2 + willTopic.byteLength;
+            const willPayload = parseBinary(buf, offset);
+            packet.w = willPayload;
+            offset += 2 + willPayload.byteLength;
+          }
+          if (packet.userNameFlag()) {
+            const userName = parseBinary(buf, offset);
+            packet.usr = userName.toString('utf8');
+            offset += 2 + userName.byteLength;
+          }
+          if (packet.passwordFlag()) {
+            const password = parseBinary(buf, offset);
+            packet.pwd = password;
+            offset += 2 + password.byteLength;
+          }
+          this.version = packet.v;
           return packet;
         }
         case PACKET_TYPE.CONNACK: {

src/packets/connect.ts

@@ -69,45 +69,3 @@ export class PacketConnect extends Packet implements PacketConnectData {
     return !!(this.f & 0b00000010);
   }
 }
-
-export const parseConnect = (b: number, l: number, data: BufferList, offset: number): PacketConnect => {
-  offset += 2 + data.readUInt16BE(offset); // Skip "MQTT" or "MQIsdp" protocol name.
-  const v = data.readUInt8(offset++);
-  const f = data.readUInt8(offset++);
-  const k = data.readUInt16BE(offset);
-  offset += 2;
-  let p: Properties = {};
-  if (v === 5) {
-    const [props, propsSize] = parseProps(data, offset);
-    p = props;
-    offset += propsSize;
-  }
-  const clientId = parseBinary(data, offset);
-  const id = clientId.toString('utf8');
-  offset += 2 + clientId.byteLength;
-  const packet = new PacketConnect(b, l, v, f, k, p, id);
-  if (packet.willFlag()) {
-    if (v === 5) {
-      const [props, propsSize] = parseProps(data, offset);
-      packet.wp = props;
-      offset += propsSize;
-    } else packet.wp = {};
-    const willTopic = parseBinary(data, offset);
-    packet.wt = willTopic.toString('utf8');
-    offset += 2 + willTopic.byteLength;
-    const willPayload = parseBinary(data, offset);
-    packet.w = willPayload;
-    offset += 2 + willPayload.byteLength;
-  }
-  if (packet.userNameFlag()) {
-    const userName = parseBinary(data, offset);
-    packet.usr = userName.toString('utf8');
-    offset += 2 + userName.byteLength;
-  }
-  if (packet.passwordFlag()) {
-    const password = parseBinary(data, offset);
-    packet.pwd = password;
-    offset += 2 + password.byteLength;
-  }
-  return packet;
-};

src/types.ts

@@ -31,3 +31,10 @@ export interface Properties {
   [PROPERTY.SubscriptionIdentifierAvailable]?: number;
   [PROPERTY.SharedSubscriptionAvailable]?: number;
 }
+
+export interface BufferLike {
+  readUInt8(offset: number): number;
+  readUInt16BE(offset: number): number;
+  readUInt32BE(offset: number): number;
+  slice(start: number, end: number): Buffer;
+}

src/util/parse.ts

@@ -1,8 +1,7 @@
-import {BufferList} from '../BufferList';
 import {ERROR, PROPERTY} from '../enums';
-import {Properties} from '../types';
+import {BufferLike, Properties} from '../types';
 
-export const parseVarInt = (data: BufferList, offset: number): [int: number, size: number] => {
+export const parseVarInt = (data: BufferLike, offset: number): [int: number, size: number] => {
   const b1 = data.readUInt8(offset);
   if (!(b1 & 0b10000000)) return [b1 & 0b01111111, 1];
   const b2 = data.readUInt8(offset + 1);
@@ -13,37 +12,19 @@ export const parseVarInt = (data: BufferList, offset: number): [int: number, siz
   return [((b4 & 0b01111111) << 21) + ((b3 & 0b01111111) << 14) + ((b2 & 0b01111111) << 7) + (b1 & 0b01111111), 4];
 };
 
-export const parseVarIntBuf = (buf: Buffer, offset: number): [int: number, size: number] => {
-  const b1 = buf.readUInt8(offset);
-  if (!(b1 & 0b10000000)) return [b1 & 0b01111111, 1];
-  const b2 = buf.readUInt8(offset + 1);
-  if (!(b2 & 0b10000000)) return [((b2 & 0b01111111) << 7) + (b1 & 0b01111111), 2];
-  const b3 = buf.readUInt8(offset + 2);
-  if (!(b3 & 0b10000000)) return [((b3 & 0b01111111) << 14) + ((b2 & 0b01111111) << 7) + (b1 & 0b01111111), 3];
-  const b4 = buf.readUInt8(offset + 3);
-  return [((b4 & 0b01111111) << 21) + ((b3 & 0b01111111) << 14) + ((b2 & 0b01111111) << 7) + (b1 & 0b01111111), 4];
-};
-
-export const parseBinary = (list: BufferList, offset: number): Buffer => {
+export const parseBinary = (list: BufferLike, offset: number): Buffer => {
   const stringOffset = offset + 2;
   const dataLength = list.readUInt16BE(offset);
   return list.slice(stringOffset, stringOffset + dataLength);
 };
 
-export const parseBinaryBuf = (buf: Buffer, offset: number): Buffer => {
-  const stringOffset = offset + 2;
-  const dataLength = buf.readUInt16BE(offset);
-  return buf.slice(stringOffset, stringOffset + dataLength);
-};
-
-export const parseProps = (data: BufferList, offset: number): [props: Properties, size: number] => {
+export const parseProps = (data: BufferLike, offset: number): [props: Properties, size: number] => {
   const [int, size] = parseVarInt(data, offset);
   offset += size;
-  const buf = data.slice(offset, offset + int);
+  const end = offset + int;
   const props: Properties = {};
-  offset = 0;
-  while (offset < int) {
-    const byte = buf.readUInt8(offset++);
+  while (offset < end) {
+    const byte = data.readUInt8(offset++);
     switch (byte) {
       case PROPERTY.PayloadFormatIndicator:
       case PROPERTY.RequestProblemInformation:
@@ -53,35 +34,35 @@ export const parseProps = (data: BufferList, offset: number): [props: Properties
       case PROPERTY.WildcardSubscriptionAvailable:
       case PROPERTY.SubscriptionIdentifierAvailable:
       case PROPERTY.SharedSubscriptionAvailable: {
-        props[byte] = buf.readUInt8(offset);
+        props[byte] = data.readUInt8(offset);
         offset += 1;
         break;
       }
       case PROPERTY.ServerKeepAlive:
       case PROPERTY.ReceiveMaximum:
       case PROPERTY.TopicAliasMaximum:
       case PROPERTY.TopicAlias: {
-        props[byte] = buf.readUInt16BE(offset);
+        props[byte] = data.readUInt16BE(offset);
         offset += 2;
         break;
       }
       case PROPERTY.MessageExpiryInterval:
       case PROPERTY.WillDelayInterval:
       case PROPERTY.SessionExpiryInterval:
       case PROPERTY.MaximumPacketSize: {
-        props[byte] = buf.readUInt32BE(offset);
+        props[byte] = data.readUInt32BE(offset);
         offset += 4;
         break;
       }
       case PROPERTY.SubscriptionIdentifier: {
-        const [value, size] = parseVarIntBuf(buf, offset);
+        const [value, size] = parseVarInt(data, offset);
         props[byte] = value;
         offset += size;
         break;
       }
       case PROPERTY.CorrelationData:
       case PROPERTY.AuthenticationData: {
-        const {byteLength} = (props[byte] = parseBinaryBuf(buf, offset));
+        const {byteLength} = (props[byte] = parseBinary(data, offset));
         offset += 2 + byteLength;
         break;
       }
@@ -92,16 +73,16 @@ export const parseProps = (data: BufferList, offset: number): [props: Properties
       case PROPERTY.ResponseInformation:
       case PROPERTY.ServerReference:
       case PROPERTY.ReasonString: {
-        const binary = parseBinaryBuf(buf, offset);
+        const binary = parseBinary(data, offset);
         props[byte] = binary.toString('utf8');
         offset += 2 + binary.byteLength;
         break;
       }
       case PROPERTY.UserProperty: {
         if (!props[PROPERTY.UserProperty]) props[PROPERTY.UserProperty] = [];
-        const key = parseBinaryBuf(buf, offset);
+        const key = parseBinary(data, offset);
         offset += 2 + key.byteLength;
-        const value = parseBinaryBuf(buf, offset);
+        const value = parseBinary(data, offset);
         offset += 2 + value.byteLength;
         props[PROPERTY.UserProperty]!.push([key.toString('utf8'), value.toString('utf8')]);
         break;

yarn.lock

@@ -753,13 +753,6 @@
   dependencies:
     "@babel/types" "^7.3.0"
 
-"@types/bl@^2.1.0":
-  version "2.1.0"
-  resolved "https://registry.yarnpkg.com/@types/bl/-/bl-2.1.0.tgz#45c881c97feae1223d63bbc5b83166153fcb2a15"
-  integrity sha512-1TdA9IXOy4sdqn8vgieQ6GZAiHiPNrOiO1s2GJjuYPw4QVY7gYoVjkW049avj33Ez7IcIvu43hQsMsoUFbCn2g==
-  dependencies:
-    "@types/node" "*"
-
 "@types/color-name@^1.1.1":
   version "1.1.1"
   resolved "https://registry.yarnpkg.com/@types/color-name/-/color-name-1.1.1.tgz#1c1261bbeaa10a8055bbc5d8ab84b7b2afc846a0"