feat: 🎸 add encoding for PUBLISH packets

Author: streamich

src/packets/__tests__/publish.spec.ts

@@ -0,0 +1,68 @@
+import {PACKET_TYPE, PROPERTY} from '../../enums';
+import {PacketPublish} from '../publish';
+import {MqttDecoder} from '../../MqttDecoder';
+
+test('can create a packet', () => {
+  const packet = PacketPublish.create('topic', 0, {}, Buffer.from([1, 2, 3]));
+  expect(packet.b >> 4).toBe(PACKET_TYPE.PUBLISH);
+  expect(packet.t).toBe('topic');
+  expect(packet.i).toBe(0);
+  expect(packet.p).toEqual({});
+  expect(packet.d).toEqual(Buffer.from([1, 2, 3]));
+});
+
+test('can serialize a basic packet', () => {
+  const packet1 = PacketPublish.create('topic', 0, {}, Buffer.from([1, 2, 3]));
+  const buf = packet1.toBuffer(5);
+  const decoder = new MqttDecoder();
+  decoder.version = 5;
+  decoder.push(buf);
+  const packet2 = decoder.parse()! as PacketPublish;
+  expect(packet2.b).toBe(packet1.b);
+  expect(packet2.t).toBe('topic');
+  expect(packet2.p).toEqual({});
+  expect(packet2.d).toEqual(Buffer.from([1, 2, 3]));
+  expect(packet2).toEqual(packet1);
+});
+
+test('can encode packet with QoS = 1', () => {
+  const packet1 = PacketPublish.create('topic', 1, {}, Buffer.from([1, 2, 3]));
+  packet1.setQualityOfService(1);
+  const buf = packet1.toBuffer(5);
+  const decoder = new MqttDecoder();
+  decoder.version = 5;
+  decoder.push(buf);
+  const packet2 = decoder.parse()! as PacketPublish;
+  expect(packet2.b).toBe(packet1.b);
+  expect(packet2.qualityOfService()).toBe(1);
+  expect(packet2.t).toBe('topic');
+  expect(packet2.p).toEqual({});
+  expect(packet2.d).toEqual(Buffer.from([1, 2, 3]));
+  expect(packet2).toEqual(packet1);
+});
+
+test('can encode properties', () => {
+  const packet1 = PacketPublish.create('topic', 1, {
+    [PROPERTY.AssignedClientIdentifier]: 'test',
+    [PROPERTY.UserProperty]: [
+      ['test', 'test'],
+    ],
+  }, Buffer.from([1, 2, 3]));
+  packet1.setQualityOfService(1);
+  const buf = packet1.toBuffer(5);
+  const decoder = new MqttDecoder();
+  decoder.version = 5;
+  decoder.push(buf);
+  const packet2 = decoder.parse()! as PacketPublish;
+  expect(packet2.b).toBe(packet1.b);
+  expect(packet2.qualityOfService()).toBe(1);
+  expect(packet2.t).toBe('topic');
+  expect(packet2.p).toEqual({
+    [PROPERTY.AssignedClientIdentifier]: 'test',
+    [PROPERTY.UserProperty]: [
+      ['test', 'test'],
+    ],
+  });
+  expect(packet2.d).toEqual(Buffer.from([1, 2, 3]));
+  expect(packet2).toEqual(packet1);
+});

src/packets/publish.ts

@@ -1,5 +1,7 @@
+import { PACKET_TYPE } from '../enums';
 import {Packet, PacketHeaderData} from '../packet';
 import {Properties} from '../types';
+import { genProps } from '../util/genProps/v7';
 
 export interface PacketPublishData extends PacketHeaderData {
   /** Topic Name. */
@@ -13,6 +15,20 @@ export interface PacketPublishData extends PacketHeaderData {
 }
 
 export class PacketPublish extends Packet implements PacketPublishData {
+  /**
+   * @param t Topic
+   * @param i Packet ID
+   * @param p Properties
+   * @param d Payload
+   */
+  static create(
+    t: string,
+    i: number,
+    p: Properties,
+    d: Buffer,): PacketPublish {
+    return new PacketPublish(PACKET_TYPE.PUBLISH << 4, 0, t, i, p, d);
+  }
+
   constructor(
     b: number,
     l: number,
@@ -23,4 +39,67 @@ export class PacketPublish extends Packet implements PacketPublishData {
   ) {
     super(b, l);
   }
+
+  public toBuffer(version: number): Buffer {
+    const payload = this.d;
+    const lenTopic = Buffer.byteLength(this.t);
+    const isQosHigh = this.qualityOfService() > 0;
+    const emitProps = version === 5;
+    const props = emitProps ? genProps(this.p) : null;
+    const propsLength = emitProps ? props!.length : 0;
+    const remainingLength: number =
+      2 + lenTopic +            // topic length
+      (isQosHigh ? 2 : 0) +     // packet ID
+      propsLength +             // properties
+      payload.length;           // payload length
+    const remainingLengthSize = remainingLength < 128 ? 1 : remainingLength < 16_384 ? 2 : remainingLength < 2_097_152 ? 3 : 4;
+    const bufferLength = 1 + remainingLengthSize + remainingLength;
+    const buf = Buffer.allocUnsafe(bufferLength);
+    this.l = remainingLength;
+    
+    buf.writeUInt8(this.b);
+
+    let offset = 1;
+
+    switch (remainingLengthSize) {
+      case 1:
+        buf.writeUInt8(remainingLength, 1);
+        offset = 2;
+        break;
+      case 2:
+        buf.writeUInt16LE(((remainingLength & 0b011111110000000) << 1) | (0b10000000 | (remainingLength & 0b01111111)), 1);
+        offset = 3;
+        break;
+      case 3:
+        buf.writeUInt16LE(((0b100000000000000 | (remainingLength & 0b011111110000000)) << 1) | (0b10000000 | (remainingLength & 0b01111111)), 1);
+        buf.writeUInt8((remainingLength >> 14) & 0b01111111, 3);
+        offset = 4;
+        break;
+      case 4:
+        buf.writeUInt32LE((((((remainingLength >> 21) & 0b01111111) << 8) | (0b10000000 | ((remainingLength >> 14) & 0b01111111))) << 16) |
+          ((0b100000000000000 | (remainingLength & 0b011111110000000)) << 1) | (0b10000000 | (remainingLength & 0b01111111)), 1);
+        offset = 5;
+        break;
+    }
+
+    buf.writeUInt16BE(lenTopic, offset);
+    offset += 2;
+    buf.write(this.t, offset);
+    offset += lenTopic;
+
+    if (isQosHigh) {
+      buf.writeUInt16BE(this.i, offset);
+      offset += 2;
+    }
+
+    if (emitProps) {
+      props!.copy(buf, offset);
+      offset += propsLength;
+    }
+
+    payload.copy(buf, offset);
+    // offset += payload.length;
+
+    return buf;
+  }
 }