refactor: move message handling internals to use async/await

benchmarks/message-io/incoming-message-stream.js

@@ -0,0 +1,43 @@
+const { createBenchmark } = require('../common');
+const { Readable } = require('stream');
+
+const Debug = require('tedious/lib/debug');
+const IncomingMessageStream = require('tedious/lib/incoming-message-stream');
+const { Packet } = require('tedious/lib/packet');
+
+const bench = createBenchmark(main, {
+  n: [100, 1000, 10000, 100000]
+});
+
+function main({ n }) {
+  const debug = new Debug();
+
+  const stream = Readable.from((async function*() {
+    for (let i = 0; i < n; i++) {
+      const packet = new Packet(2);
+      packet.last(true);
+      packet.addData(Buffer.from([1, 2, 3, 4, 5, 6, 7, 8, 9]));
+
+      yield packet.buffer;
+    }
+  })());
+
+  const incoming = new IncomingMessageStream(debug);
+  stream.pipe(incoming);
+
+  bench.start();
+  console.profile('incoming-message-stream');
+
+  (async function() {
+    let total = 0;
+
+    for await (m of incoming) {
+      for await (const buf of m) {
+        total += buf.length;
+      }
+    }
+
+    console.profileEnd('incoming-message-stream');
+    bench.end(n);
+  })();
+}

benchmarks/message-io/outgoing-message-stream.js

@@ -0,0 +1,72 @@
+const { createBenchmark } = require('../common');
+const { Duplex } = require('stream');
+
+const Debug = require('../../lib/debug');
+const OutgoingMessageStream = require('../../lib/outgoing-message-stream');
+const Message = require('../../lib/message');
+
+const bench = createBenchmark(main, {
+  n: [100, 1000, 10000, 100000]
+});
+
+function main({ n }) {
+  const debug = new Debug();
+
+  const stream = new Duplex({
+    read() {},
+    write(chunk, encoding, callback) {
+      // Just consume the data
+      callback();
+    }
+  });
+
+  const payload = [
+    Buffer.alloc(1024),
+    Buffer.alloc(1024),
+    Buffer.alloc(1024),
+    Buffer.alloc(256),
+    Buffer.alloc(256),
+    Buffer.alloc(256),
+    Buffer.alloc(256),
+  ];
+
+  const out = new OutgoingMessageStream(debug, {
+    packetSize: 8 + 1024
+  });
+  out.pipe(stream);
+
+  bench.start();
+  console.profile('write-message');
+
+  function writeNextMessage(i) {
+    if (i == n) {
+      out.end();
+      out.once('finish', () => {
+        console.profileEnd('write-message');
+        bench.end(n);
+      });
+      return;
+    }
+
+    const m = new Message({ type: 2, resetConnection: false });
+    out.write(m);
+
+    for (const buf of payload) {
+      m.write(buf);
+    }
+
+    m.end();
+
+    if (out.needsDrain) {
+      out.once('drain', () => {
+        writeNextMessage(i + 1);
+      });
+    } else {
+      process.nextTick(() => {
+        writeNextMessage(i + 1);
+      });
+    }
+  }
+
+  writeNextMessage(0);
+}

benchmarks/message-io/read-message.js

@@ -0,0 +1,39 @@
+const { createBenchmark } = require('../common');
+const { Readable } = require('stream');
+
+const Debug = require('tedious/lib/debug');
+const MessageIO = require('tedious/lib/message-io');
+const { Packet } = require('tedious/lib/packet');
+
+const bench = createBenchmark(main, {
+  n: [100, 1000, 10000, 100000]
+});
+
+function main({ n }) {
+  const debug = new Debug();
+
+  const stream = Readable.from((async function*() {
+    for (let i = 0; i < n; i++) {
+      const packet = new Packet(2);
+      packet.last(true);
+      packet.addData(Buffer.from([1, 2, 3, 4, 5, 6, 7, 8, 9]));
+
+      yield packet.buffer;
+    }
+  })());
+
+  (async function() {
+    bench.start();
+    console.profile('read-message');
+
+    let total = 0;
+    for (let i = 0; i < n; i++) {
+      for await (const chunk of MessageIO.readMessage(stream, debug)) {
+        total += chunk.length;
+      }
+    }
+
+    console.profileEnd('read-message');
+    bench.end(n);
+  })();
+}

benchmarks/message-io/write-message.js

@@ -0,0 +1,43 @@
+const { createBenchmark, createConnection } = require('../common');
+const { Duplex } = require('stream');
+
+const Debug = require('tedious/lib/debug');
+const MessageIO = require('tedious/lib/message-io');
+
+const bench = createBenchmark(main, {
+  n: [100, 1000, 10000, 100000]
+});
+
+function main({ n }) {
+  const debug = new Debug();
+
+  const stream = new Duplex({
+    read() {},
+    write(chunk, encoding, callback) {
+      // Just consume the data
+      callback();
+    }
+  });
+
+  const payload = [
+    Buffer.alloc(1024),
+    Buffer.alloc(1024),
+    Buffer.alloc(1024),
+    Buffer.alloc(256),
+    Buffer.alloc(256),
+    Buffer.alloc(256),
+    Buffer.alloc(256),
+  ];
+
+  (async function() {
+    bench.start();
+    console.profile('write-message');
+
+    for (let i = 0; i <= n; i++) {
+      await MessageIO.writeMessage(stream, debug, 8 + 1024, 2, payload);
+    }
+
+    console.profileEnd('write-message');
+    bench.end(n);
+  })();
+}

package.json

@@ -80,7 +80,8 @@
     "semantic-release": "^22.0.12",
     "sinon": "^15.2.0",
     "typedoc": "^0.26.6",
-    "typescript": "^5.5.4"
+    "typescript": "^5.5.4",
+    "wtfnode": "^0.9.3"
   },
   "scripts": {
     "docs": "typedoc",
@@ -124,6 +125,7 @@
     ]
   },
   "mocha": {
+    "nodeOption": "require=wtfnode",
     "require": "test/setup.js",
     "timeout": 5000,
     "extension": [