JavaScriptPlugins
diff --git a/‎src/client.ts
Lines changed: 25 additions & 20 deletions b/‎src/client.ts
Lines changed: 25 additions & 20 deletions
diff --git a/‎src/interface/clientOptions.ts
Lines changed: 7 additions & 0 deletions b/‎src/interface/clientOptions.ts
Lines changed: 7 additions & 0 deletions
diff --git a/‎src/interface/packets.ts
Lines changed: 0 additions & 21 deletions b/‎src/interface/packets.ts
Lines changed: 0 additions & 21 deletions
@@ -10,16 +10,16 @@ import {
 } from 'mqtt-packet';
 import { write } from './write.js';
 import { ConnectOptions } from './interface/connectOptions.js';
-import { Duplex } from 'stream';
+import { Duplex } from 'node:stream';
+import { Socket } from 'node:net';
 import { EventEmitter } from 'node:events';
 import { connectionFactory } from './connectionFactory/index.js';
 import eos from 'end-of-stream';
 import { defaultConnectOptions } from './util/constants.js';
-import { ReasonCodeErrors } from './util/errors.js';
+import { ReasonCodeErrors } from './util/reasonCodes.js';
 import { logger } from './util/logger.js';
 import { defaultClientId } from './util/defaultClientId.js';
 import { PublishPacket } from './interface/packets.js';
-import { NumberAllocator } from 'number-allocator';
 import { Logger } from 'pino';
 import * as sequencer from './sequencer.js';
 
@@ -37,16 +37,15 @@ export class MqttClient extends EventEmitter {
   connected: boolean;
   errored: boolean;
   _eos: Promise<void> | undefined;
-  conn: Duplex;
+  conn: Duplex | Socket;
   _clientLogger: Logger;
   /**
    * Use packet ID as key if there is one (e.g., SUBACK)
    * Use packet type as key if there is no packet ID (e.g., CONNACK)
    */
   // TODO: This should be removed after we remove CONNECT into the sequencer
   _inflightPackets: Map<string | number, (err: Error | null, packet: IPacket) => void>;
-  private _numberAllocator: NumberAllocator;
-  private _packetSequencer = new sequencer.MqttPacketSequencer(this._sendPacketCallback.bind(this));
+  private _packetSequencer = new sequencer.MqttPacketSequencer(write.bind(null, this));
 
   constructor(options: ConnectOptions) {
     super();
@@ -67,8 +66,6 @@ export class MqttClient extends EventEmitter {
 
     this._clientLogger = logger.child({ id: this._options.clientId });
 
-    this._numberAllocator = new NumberAllocator(1, 65535);
-
     this.conn = this._options.customStreamFactory
       ? this._options.customStreamFactory(this._options)
       : connectionFactory(this._options);
@@ -124,15 +121,6 @@ export class MqttClient extends EventEmitter {
     });
   }
 
-  private _sendPacketCallback(packetType: sequencer.PacketType, message: sequencer.Message): void {
-    if (packetType == 'publish') {
-      write(this, message as IPublishPacket);
-    } else {
-      logger.error(`Unexpected packet type: ${packetType}`);
-      throw new Error(`Unexpected packet type: ${packetType}`);
-    }
-  }
-
   async handleIncomingPacket(packet: Packet): Promise<void> {
     this._clientLogger.trace(`handleIncomingPacket packet.cmd=${packet.cmd}`);
     switch (packet.cmd) {
@@ -150,7 +138,7 @@ export class MqttClient extends EventEmitter {
         // (We need the sequencer to do this because it has to send puback messages and it needs to do the whole QOS-2 thing when packets come in.)
         //
         // Also, another random thought, when we get suback back from the broker, it will include granted QOS values and we'll need to return those.
-        this._packetSequencer.handleIncomingPacket('puback', (packet as unknown) as sequencer.Packet);
+        this._packetSequencer.handleIncomingPacket((packet as unknown) as sequencer.Packet);
         break;
       }
     }
@@ -165,7 +153,7 @@ export class MqttClient extends EventEmitter {
   public async connect(): Promise<IConnackPacket> {
     logger.trace('sending connect...');
     this.connecting = true;
-
+    
     const connackPromise = this._awaitConnack();
     const packet: IConnectPacket = {
       cmd: 'connect',
@@ -179,7 +167,8 @@ export class MqttClient extends EventEmitter {
       will: this._options.will,
       properties: this._options.properties,
     };
-    logger.trace(`writing connect...`);
+    this._packetSequencer.runSequence(packet)
+    logger.trace(`running connect sequence...`);
     await write(this, packet);
     logger.trace('waiting for connack...');
     const connack = await connackPromise;
@@ -319,11 +308,27 @@ export class MqttClient extends EventEmitter {
     }
   }
 
+  // TODO: follow up on Aedes to see if there is a better way than breaking the Node Streams contract and accessing _writableState
+  // to make sure that the write callback is cleaned up in case of error.
   onError(err?: Error | null | undefined) {
     this.emit('error', err);
     this.errored = true;
     this.conn.removeAllListeners('error');
     this.conn.on('error', () => {});
+    // hack to clean up the write callbacks in case of error
+    this.hackyCleanupWriteCallback();
     this._destroyClient(true);
   }
+
+  hackyCleanupWriteCallback() {
+    // _writableState is not part of the public API for Duplex or Socket, so we have to do some typecasting here to work with it as the stream state.
+    // See https://github.com/nodejs/node/issues/445 for information on this.
+    const state = (this.conn as any)._writableState;
+    if (typeof state.getBuffer !== 'function') {
+      // See https://github.com/nodejs/node/pull/31165
+      throw new Error('_writableState.buffer is EOL. _writableState should have getBuffer() as a function.');
+    }
+    const list: any[] = state.getBuffer();
+    list.forEach((req) => {req.callback()});
+  }
 }
@@ -0,0 +1,7 @@
+export interface ClientOptions {
+  /**
+   * MQTT protocol version to use. Use 4 for vMQTT 3.1.1, and 5 for MQTT v5.0
+   * Default: 5
+   */
+  protocolVersion?: 4 | 5;
+}