feat: 🎸 add unsuback

Author: streamich

src/MqttDecoder.ts

@@ -1,5 +1,5 @@
 import BufferList from 'bl';
-import {DECODER_STATE, PACKET_TYPE} from './enums';
+import {DECODER_STATE, ERROR, PACKET_TYPE} from './enums';
 import {PacketConnack, parseConnack} from './packets/connack';
 import {PacketConnect, parseConnect} from './packets/connect';
 import {PacketPublish, parsePublish} from './packets/publish';
@@ -10,6 +10,7 @@ import {PacketPubcomp, parsePubcomp} from './packets/pubcomp';
 import {PacketSubscribe, parseSubscribe} from './packets/subscribe';
 import {PacketSuback, parseSuback} from './packets/suback';
 import {PacketUnsubscribe, parseUnsubscribe} from './packets/unsubscribe';
+import {PacketUnsuback, parseUnsuback} from './packets/unsuback';
 
 export class MqttDecoder {
   public state: DECODER_STATE = DECODER_STATE.HEADER;
@@ -37,7 +38,19 @@ export class MqttDecoder {
     this.list = new BufferList();
   }
 
-  public parse(): null | PacketConnect | PacketConnack | PacketPublish | PacketPuback | PacketPubrec | PacketPubrel | PacketPubcomp | PacketSubscribe | PacketSuback | PacketUnsubscribe {
+  public parse():
+  | null
+  | PacketConnect
+  | PacketConnack
+  | PacketPublish
+  | PacketPuback
+  | PacketPubrec
+  | PacketPubrel
+  | PacketPubcomp
+  | PacketSubscribe
+  | PacketSuback
+  | PacketUnsubscribe
+  | PacketUnsuback {
     this.parseFixedHeader();
     const data = this.parseVariableData();
     if (!data) return null;
@@ -84,8 +97,12 @@ export class MqttDecoder {
         const packet = parseUnsubscribe(b, l, data, this.version);
         return packet;
       }
+      case PACKET_TYPE.UNSUBACK: {
+        const packet = parseUnsuback(b, l, data, this.version);
+        return packet;
+      }
       default: {
-        return null;
+        throw ERROR.MALFORMED_PACKET;
       }
     }
   }

src/__tests__/MqttDecoder.spec.ts

@@ -9,6 +9,7 @@ import {PacketPubrec} from '../packets/pubrec';
 import {PacketSubscribe} from '../packets/subscribe';
 import { PacketSuback } from '../packets/suback';
 import { PacketUnsubscribe } from '../packets/unsubscribe';
+import { PacketUnsuback } from '../packets/unsuback';
 
 it('can instantiate', () => {
   const decoder = new MqttDecoder();
@@ -572,3 +573,29 @@ describe('UNSUBSCRIBE', () => {
     ]);
   });
 });
+
+describe('UNSUBACK', () => {
+  it('parses MQTT 5.0 packet', () => {
+    const decoder = new MqttDecoder();
+    decoder.version = 5;
+    decoder.push(Buffer.from([
+      176, 25, // Header
+      0, 8, // Message ID
+      20, // properties length
+      31, 0, 4, 116, 101, 115, 116, // reasonString
+      38, 0, 4, 116, 101, 115, 116, 0, 4, 116, 101, 115, 116, // userProperties
+      0, 128 // success and error
+    ]));
+    const packet: PacketUnsuback = decoder.parse() as PacketUnsuback;
+    expect(packet.b).toBe(176);
+    expect(packet.l).toBe(25);
+    expect(packet.i).toBe(8);
+    expect(packet.p).toEqual({
+      [PROPERTY.ReasonString]: 'test',
+      [PROPERTY.UserProperty]: [
+        ['test', 'test'],
+      ],
+    });
+    expect(packet.s).toEqual([0, 128]);
+  });
+});

src/packets/unsuback.ts

@@ -0,0 +1,42 @@
+import BufferList from 'bl';
+import {Packet, PacketHeaderData} from '../packet';
+import {Properties} from '../types';
+import {parseProps} from '../util/parse';
+
+export interface PacketUnsubackData extends PacketHeaderData {
+  /** Packet Identifier. */
+  i: number;
+  /** Properties. */
+  p: Properties;
+  /** SUBACK Payload. */
+  s: number[];
+}
+
+export class PacketUnsuback extends Packet implements PacketUnsubackData {
+  constructor(
+    b: number,
+    l: number,
+    public i: number,
+    public p: Properties,
+    public s: number[],
+  ) {
+    super(b, l);
+  }
+}
+
+export const parseUnsuback = (b: number, l: number, data: BufferList, version: number): PacketUnsuback => {
+  const i = data.readUInt16BE(0);
+  let offset = 2;
+  let p: Properties = {};
+  if (version === 5) {
+    const [props, size] = parseProps(data, offset);
+      p = props;
+      offset += size;
+  }
+  const len = data.length;
+  const s: number[] = [];
+  while (offset < len) {
+    s.push(data.readUInt8(offset++));
+  }
+  return new PacketUnsuback(b, l, i, p, s);
+};