diff --git a/benchmarks/message-io/incoming-message-stream.js b/benchmarks/message-io/incoming-message-stream.js
new file mode 100644
index 000000000..2d3aba868
--- /dev/null
+++ b/benchmarks/message-io/incoming-message-stream.js
@@ -0,0 +1,41 @@
+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();
+
+  (async function() {
+    let total = 0;
+
+    for await (const message of incoming) {
+      for await (const chunk of message) {
+        total += chunk.length;
+      }
+    }
+
+    bench.end(n);
+  })();
+}
diff --git a/benchmarks/message-io/outgoing-message-stream.js b/benchmarks/message-io/outgoing-message-stream.js
new file mode 100644
index 000000000..c8c5aecbf
--- /dev/null
+++ b/benchmarks/message-io/outgoing-message-stream.js
@@ -0,0 +1,70 @@
+const { createBenchmark } = require('../common');
+const { Duplex } = require('stream');
+
+const Debug = require('tedious/lib/debug');
+const OutgoingMessageStream = require('tedious/lib/outgoing-message-stream');
+const Message = require('tedious/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();
+
+  function writeNextMessage(i) {
+    if (i === n) {
+      out.end();
+      out.once('finish', () => {
+        bench.end(n);
+      });
+      return;
+    }
+
+    const message = new Message({ type: 2, resetConnection: false });
+    out.write(message);
+
+    for (const chunk of payload) {
+      message.write(chunk);
+    }
+
+    message.end();
+
+    if (out.needsDrain) {
+      out.once('drain', () => {
+        writeNextMessage(i + 1);
+      });
+    } else {
+      process.nextTick(() => {
+        writeNextMessage(i + 1);
+      });
+    }
+  }
+
+  writeNextMessage(0);
+}
diff --git a/benchmarks/message-io/read-message.js b/benchmarks/message-io/read-message.js
new file mode 100644
index 000000000..0bd70320b
--- /dev/null
+++ b/benchmarks/message-io/read-message.js
@@ -0,0 +1,38 @@
+const { createBenchmark } = require('../common');
+const { Readable } = require('stream');
+
+const Debug = require('tedious/lib/debug');
+const { readMessage } = 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();
+
+    let total = 0;
+
+    for (let i = 0; i < n; i++) {
+      for await (const chunk of readMessage(stream, { debug })) {
+        total += chunk.length;
+      }
+    }
+
+    bench.end(n);
+  })();
+}
diff --git a/benchmarks/message-io/write-message.js b/benchmarks/message-io/write-message.js
new file mode 100644
index 000000000..8580980f0
--- /dev/null
+++ b/benchmarks/message-io/write-message.js
@@ -0,0 +1,41 @@
+const { createBenchmark } = require('../common');
+const { Duplex } = require('stream');
+
+const Debug = require('tedious/lib/debug');
+const { writeMessage } = 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();
+
+    for (let i = 0; i < n; i++) {
+      await writeMessage(stream, 8 + 1024, 2, payload, { debug });
+    }
+
+    bench.end(n);
+  })();
+}
diff --git a/src/connection.ts b/src/connection.ts
index 0ff2f90e3..5154bcb0d 100644
--- a/src/connection.ts
+++ b/src/connection.ts
@@ -29,7 +29,7 @@ import NTLMResponsePayload from './ntlm-payload';
 import Request from './request';
 import RpcRequestPayload from './rpcrequest-payload';
 import SqlBatchPayload from './sqlbatch-payload';
-import MessageIO from './message-io';
+import MessageIO, { readMessage, writeMessage } from './message-io';
 import { Parser as TokenStreamParser } from './token/token-stream-parser';
 import { Transaction, ISOLATION_LEVEL, assertValidIsolationLevel } from './transaction';
 import { ConnectionError, RequestError } from './errors';
@@ -2092,17 +2092,32 @@ class Connection extends EventEmitter {
 
           socket.setKeepAlive(true, KEEP_ALIVE_INITIAL_DELAY);
 
-          this.messageIo = new MessageIO(socket, this.config.options.packetSize, this.debug);
-          this.messageIo.on('secure', (cleartext) => { this.emit('secure', cleartext); });
-
           this.socket = socket;
 
           this.debug.log('connected to ' + this.config.server + ':' + this.config.options.port);
 
-          this.sendPreLogin();
+          try {
+            await this.sendPreLogin(socket, signal);
+          } catch (err: any) {
+            signal.throwIfAborted();
+
+            throw this.wrapSocketError(err);
+          }
 
           this.transitionTo(this.STATE.SENT_PRELOGIN);
-          const preloginResponse = await this.readPreloginResponse(signal);
+
+          let preloginResponse;
+          try {
+            preloginResponse = await this.readPreloginResponse(socket, signal);
+          } catch (err: any) {
+            signal.throwIfAborted();
+
+            throw this.wrapSocketError(err);
+          }
+
+          this.messageIo = new MessageIO(socket, this.config.options.packetSize, this.debug);
+          this.messageIo.on('secure', (cleartext) => { this.emit('secure', cleartext); });
+
           await this.performTlsNegotiation(preloginResponse, signal);
 
           this.sendLogin7Packet();
@@ -2456,7 +2471,7 @@ class Connection extends EventEmitter {
   /**
    * @private
    */
-  sendPreLogin() {
+  async sendPreLogin(socket: net.Socket, signal: AbortSignal) {
     const [, major, minor, build] = /^(\d+)\.(\d+)\.(\d+)/.exec(version) ?? ['0.0.0', '0', '0', '0'];
     const payload = new PreloginPayload({
       // If encrypt setting is set to 'strict', then we should have already done the encryption before calling
@@ -2466,7 +2481,7 @@ class Connection extends EventEmitter {
       version: { major: Number(major), minor: Number(minor), build: Number(build), subbuild: 0 }
     });
 
-    this.messageIo.sendMessage(TYPE.PRELOGIN, payload.data);
+    await writeMessage(socket, this.config.options.packetSize, TYPE.PRELOGIN, [payload.data], { debug: this.debug, signal });
     this.debug.payload(function() {
       return payload.toString('  ');
     });
@@ -3365,37 +3380,12 @@ class Connection extends EventEmitter {
     });
   }
 
-  async readPreloginResponse(signal: AbortSignal): Promise<PreloginPayload> {
+  async readPreloginResponse(socket: net.Socket, signal: AbortSignal): Promise<PreloginPayload> {
     let messageBuffer = Buffer.alloc(0);
 
-    await withAbortRace(signal, async (signalAborted) => {
-      const message = await Promise.race([
-        this.messageIo.readMessage().catch((err) => {
-          throw this.wrapSocketError(err);
-        }),
-        signalAborted
-      ]);
-
-      const iterator = message[Symbol.asyncIterator]();
-      try {
-        while (true) {
-          const { done, value } = await Promise.race([
-            iterator.next(),
-            signalAborted
-          ]);
-
-          if (done) {
-            break;
-          }
-
-          messageBuffer = Buffer.concat([messageBuffer, value]);
-        }
-      } finally {
-        if (iterator.return) {
-          await iterator.return();
-        }
-      }
-    });
+    for await (const data of readMessage(socket, { debug: this.debug, signal })) {
+      messageBuffer = Buffer.concat([messageBuffer, data]);
+    }
 
     const preloginPayload = new PreloginPayload(messageBuffer);
     this.debug.payload(function() {
diff --git a/src/message-io.ts b/src/message-io.ts
index d068d73a0..98f28759e 100644
--- a/src/message-io.ts
+++ b/src/message-io.ts
@@ -1,6 +1,7 @@
 import DuplexPair from 'native-duplexpair';
 
-import { Duplex } from 'stream';
+import BufferList from 'bl';
+import { Duplex, type Readable, type Writable } from 'stream';
 import * as tls from 'tls';
 import { isIP, Socket } from 'net';
 import { EventEmitter } from 'events';
@@ -8,7 +9,8 @@ import { EventEmitter } from 'events';
 import Debug from './debug';
 
 import Message from './message';
-import { TYPE } from './packet';
+import { HEADER_LENGTH, Packet, TYPE } from './packet';
+import { ConnectionError } from './errors';
 
 import IncomingMessageStream from './incoming-message-stream';
 import OutgoingMessageStream from './outgoing-message-stream';
@@ -190,5 +192,387 @@ class MessageIO extends EventEmitter {
   }
 }
 
+/**
+ * Write a message with the given `type` and `payload` to the given `stream`,
+ * wrapping it into TDS packets of the given `packetSize`.
+ *
+ * Respects backpressure from the stream, waiting for it to drain before
+ * writing more data.
+ *
+ * If iterating the `payload` throws, the message is terminated with a final
+ * packet that has the `IGNORE` flag set (telling the server to disregard the
+ * message) and the error is re-thrown. Errors from the stream itself are
+ * thrown as-is.
+ *
+ * If the given `cancelSignal` is aborted, the remaining payload is discarded
+ * and the message is terminated with the `IGNORE` flag set, telling the
+ * server to disregard it. This is a normal protocol outcome: the returned
+ * promise resolves, the TDS stream stays aligned, and the server will send a
+ * (short) response to the ignored message. Callers can check
+ * `cancelSignal.aborted` to distinguish this from a fully sent message.
+ *
+ * If the given `signal` is aborted, writing stops and the signal's abort
+ * reason is thrown. Note that this can leave a partially written message on
+ * the stream, so it must only be used when the connection is being torn down
+ * anyway. When both signals are aborted, `signal` wins.
+ *
+ * @param stream The stream to write the message to.
+ * @param packetSize The maximum packet size to use.
+ * @param type The type of the message to write.
+ * @param payload The payload to write.
+ * @param options.debug A debug instance to log packets to.
+ * @param options.resetConnection Whether the server should reset the connection when processing the message.
+ * @param options.cancelSignal An abort signal to cancel the message while keeping the connection usable.
+ * @param options.signal An abort signal to stop writing the message during connection teardown.
+ */
+export async function writeMessage(stream: Writable, packetSize: number, type: number, payload: AsyncIterable<Buffer> | Iterable<Buffer>, options: { debug?: Debug, resetConnection?: boolean, cancelSignal?: AbortSignal, signal?: AbortSignal } = {}): Promise<void> {
+  const { debug, resetConnection = false, cancelSignal, signal } = options;
+
+  signal?.throwIfAborted();
+
+  if (!stream.writable) {
+    throw new Error('Premature close');
+  }
+
+  let drain: PromiseWithResolvers<void> | null = null;
+
+  const onDrain = () => {
+    if (drain) {
+      const { resolve } = drain;
+      drain = null;
+      resolve();
+    }
+  };
+
+  const onError = (err: Error) => {
+    if (drain) {
+      const { reject } = drain;
+      drain = null;
+      reject(err);
+    }
+  };
+
+  let abortPromise: Promise<never> | null = null;
+  let onAbort: (() => void) | null = null;
+
+  if (signal) {
+    const { promise, reject } = Promise.withResolvers<never>();
+
+    // Prevent unhandled rejections if the signal is aborted while
+    // nothing is currently racing against `abortPromise`.
+    promise.catch(() => {});
+
+    abortPromise = promise;
+    onAbort = () => { reject(signal.reason); };
+    signal.addEventListener('abort', onAbort, { once: true });
+  }
+
+  const CANCEL = Symbol('cancel');
+
+  let canceled = cancelSignal?.aborted ?? false;
+  let cancelPromise: Promise<typeof CANCEL> | null = null;
+  let onCancel: (() => void) | null = null;
+
+  if (cancelSignal && !canceled) {
+    const { promise, resolve } = Promise.withResolvers<typeof CANCEL>();
+
+    cancelPromise = promise;
+    onCancel = () => {
+      canceled = true;
+      resolve(CANCEL);
+    };
+    cancelSignal.addEventListener('abort', onCancel, { once: true });
+  }
+
+  const waitForDrain = () => {
+    drain = Promise.withResolvers();
+
+    if (abortPromise || cancelPromise) {
+      const contenders: Promise<unknown>[] = [drain.promise];
+      if (abortPromise) {
+        contenders.push(abortPromise);
+      }
+      if (cancelPromise) {
+        contenders.push(cancelPromise);
+      }
+      return Promise.race(contenders);
+    }
+
+    return drain.promise;
+  };
+
+  stream.on('drain', onDrain);
+  stream.on('close', onDrain);
+  stream.on('error', onError);
+
+  try {
+    const bl = new BufferList();
+    const length = packetSize - HEADER_LENGTH;
+    let packetNumber = 0;
+
+    const writePacket = async (packet: Packet) => {
+      debug?.packet('Sent', packet);
+      debug?.data(packet);
+
+      if (stream.write(packet.buffer) === false) {
+        await waitForDrain();
+      }
+    };
+
+    let iterator;
+    if ((payload as AsyncIterable<Buffer>)[Symbol.asyncIterator]) {
+      iterator = (payload as AsyncIterable<Buffer>)[Symbol.asyncIterator]();
+    } else {
+      iterator = (payload as Iterable<Buffer>)[Symbol.iterator]();
+    }
+
+    while (!canceled) {
+      let value, done;
+      try {
+        const result = iterator.next();
+
+        let raceResult: IteratorResult<Buffer> | typeof CANCEL;
+        if (abortPromise || cancelPromise) {
+          const contenders: Promise<IteratorResult<Buffer> | typeof CANCEL>[] = [Promise.resolve(result)];
+          if (abortPromise) {
+            contenders.push(abortPromise);
+          }
+          if (cancelPromise) {
+            contenders.push(cancelPromise);
+          }
+          raceResult = await Promise.race(contenders);
+        } else {
+          raceResult = await result;
+        }
+
+        if (raceResult === CANCEL) {
+          break;
+        }
+
+        ({ value, done } = raceResult);
+      } catch (err) {
+        // The payload errored while being iterated. If the stream is still
+        // writable, terminate the message with the `IGNORE` flag set so the
+        // server disregards everything sent so far. If the signal was
+        // aborted instead, the connection is being torn down and the
+        // message is left unterminated.
+        if (stream.writable && !signal?.aborted) {
+          const packet = new Packet(type);
+          packet.packetId(packetNumber += 1);
+          packet.resetConnection(resetConnection);
+          packet.last(true);
+          packet.ignore(true);
+
+          await writePacket(packet);
+        }
+
+        throw err;
+      }
+
+      if (done) {
+        break;
+      }
+
+      bl.append(value);
+
+      while (!canceled && bl.length > length) {
+        const data = bl.slice(0, length);
+        bl.consume(length);
+
+        const packet = new Packet(type);
+        packet.packetId(packetNumber += 1);
+        packet.resetConnection(resetConnection);
+        packet.addData(data);
+
+        await writePacket(packet);
+      }
+    }
+
+    // On cancellation, any buffered payload data is discarded and the final
+    // packet is flagged so the server ignores the whole message.
+    const data = canceled ? Buffer.alloc(0) : bl.slice();
+    bl.consume(bl.length);
+
+    const packet = new Packet(type);
+    packet.packetId(packetNumber += 1);
+    packet.resetConnection(resetConnection);
+    packet.last(true);
+    packet.ignore(canceled);
+    packet.addData(data);
+
+    await writePacket(packet);
+  } finally {
+    stream.removeListener('drain', onDrain);
+    stream.removeListener('close', onDrain);
+    stream.removeListener('error', onError);
+
+    if (signal && onAbort) {
+      signal.removeEventListener('abort', onAbort);
+    }
+
+    if (cancelSignal && onCancel) {
+      cancelSignal.removeEventListener('abort', onCancel);
+    }
+  }
+}
+
+/**
+ * Read the next TDS message from the given `stream`.
+ *
+ * Returns an async generator that yields the data of the message's packets
+ * as they arrive. The generator throws if the stream emits an error or is
+ * closed before the message was fully read.
+ *
+ * Any bytes following the message's last packet (e.g. the start of the next
+ * message) are pushed back onto the stream, to be consumed by the next read.
+ *
+ * If the given `signal` is aborted, reading stops and the signal's abort
+ * reason is thrown. This also interrupts waiting for more data on a quiet
+ * stream, which an external `.return()` or `.throw()` call can not do (it
+ * would be queued behind the pending read).
+ *
+ * @param stream The stream to read the message from.
+ * @param options.debug A debug instance to log packets to.
+ * @param options.signal An abort signal to stop reading the message.
+ */
+export async function* readMessage(stream: Readable, options: { debug?: Debug, signal?: AbortSignal } = {}): AsyncGenerator<Buffer, void, undefined> {
+  const { debug, signal } = options;
+
+  signal?.throwIfAborted();
+
+  if (!stream.readable) {
+    throw new Error('Premature close');
+  }
+
+  const bl = new BufferList();
+
+  let error: Error | null = null;
+  let closed = false;
+  let waiting: PromiseWithResolvers<void> | null = null;
+
+  const onReadable = () => {
+    if (waiting) {
+      const { resolve } = waiting;
+      waiting = null;
+      resolve();
+    }
+  };
+
+  const onError = (err: Error) => {
+    error = err;
+
+    if (waiting) {
+      const { reject } = waiting;
+      waiting = null;
+      reject(err);
+    }
+  };
+
+  const onClose = () => {
+    closed = true;
+
+    if (waiting) {
+      const { reject } = waiting;
+      waiting = null;
+      reject(new Error('Premature close'));
+    }
+  };
+
+  let onAbort: (() => void) | null = null;
+  if (signal) {
+    onAbort = () => {
+      error ??= signal.reason;
+
+      if (waiting) {
+        const { reject } = waiting;
+        waiting = null;
+        reject(signal.reason);
+      }
+    };
+    signal.addEventListener('abort', onAbort, { once: true });
+  }
+
+  stream.on('readable', onReadable);
+  stream.on('error', onError);
+  stream.on('close', onClose);
+
+  try {
+    while (true) {
+      if (error) {
+        throw error;
+      }
+
+      if (closed) {
+        throw new Error('Premature close');
+      }
+
+      let chunk: Buffer;
+      while ((chunk = stream.read()) !== null) {
+        bl.append(chunk);
+
+        // The packet header is always 8 bytes of length.
+        while (bl.length >= HEADER_LENGTH) {
+          // Get the full packet length
+          const length = bl.readUInt16BE(2);
+          if (length < HEADER_LENGTH) {
+            throw new ConnectionError('Unable to process incoming packet');
+          }
+
+          if (bl.length < length) {
+            break;
+          }
+
+          const data = bl.slice(0, length);
+          bl.consume(length);
+
+          const packet = new Packet(data);
+          debug?.packet('Received', packet);
+          debug?.data(packet);
+
+          yield packet.data();
+
+          // Did the stream error out or close while we yielded? The events
+          // have already fired, so the wait below would never settle.
+          if (error) {
+            throw error;
+          }
+
+          if (closed) {
+            throw new Error('Premature close');
+          }
+
+          if (packet.isLast()) {
+            // This was the last packet of the message. Any data left in the
+            // buffer belongs to the next message (e.g. the response to an
+            // `ATTENTION` message sent by the client while reading an
+            // incoming response), so push it back onto the stream.
+            if (bl.length) {
+              stream.unshift(bl.slice());
+            }
+
+            return;
+          }
+        }
+      }
+
+      // Wait for the stream to become readable again (or error out or close).
+      waiting = Promise.withResolvers();
+      await waiting.promise;
+    }
+  } finally {
+    stream.removeListener('readable', onReadable);
+    stream.removeListener('error', onError);
+    stream.removeListener('close', onClose);
+
+    if (signal && onAbort) {
+      signal.removeEventListener('abort', onAbort);
+    }
+  }
+}
+
 export default MessageIO;
 module.exports = MessageIO;
+// The `module.exports` assignment above replaces the named exports on the
+// CommonJS side, so re-attach them.
+module.exports.writeMessage = writeMessage;
+module.exports.readMessage = readMessage;
diff --git a/test/unit/message-io-test.ts b/test/unit/message-io-test.ts
index ab1376dfe..cbbe97eac 100644
--- a/test/unit/message-io-test.ts
+++ b/test/unit/message-io-test.ts
@@ -1,22 +1,794 @@
 import { type AddressInfo, createConnection, createServer, Server, Socket } from 'net';
-import { once } from 'events';
+import { getEventListeners, once } from 'events';
 import { assert } from 'chai';
 import { promisify } from 'util';
 import DuplexPair from 'native-duplexpair';
 import { checkServerIdentity, type PeerCertificate, TLSSocket } from 'tls';
 import { readFileSync } from 'fs';
-import { Duplex } from 'stream';
+import { Duplex, Readable } from 'stream';
+import BufferListStream from 'bl';
 
 import Debug from '../../src/debug';
-import MessageIO from '../../src/message-io';
+import MessageIO, { readMessage, writeMessage } from '../../src/message-io';
 import Message from '../../src/message';
 import { Packet, TYPE } from '../../src/packet';
+import { ConnectionError } from '../../src/errors';
 
 const packetType = 2;
 const packetSize = 8 + 4;
 
 const delay = promisify(setTimeout);
 
+function assertNoDanglingEventListeners(stream: Duplex) {
+  assert.strictEqual(stream.listenerCount('error'), 0);
+  assert.strictEqual(stream.listenerCount('drain'), 0);
+  assert.strictEqual(stream.listenerCount('readable'), 0);
+}
+
+function splitPackets(data: Buffer): Packet[] {
+  const packets = [];
+
+  let offset = 0;
+  while (offset < data.length) {
+    const length = data.readUInt16BE(offset + 2);
+    packets.push(new Packet(data.subarray(offset, offset + length)));
+    offset += length;
+  }
+
+  return packets;
+}
+
+describe('writeMessage', function() {
+  it('wraps the given payload into a TDS packet and writes it to the given stream', async function() {
+    const payload = Buffer.from([1, 2, 3]);
+    const stream = new BufferListStream();
+
+    await writeMessage(stream, packetSize, packetType, [payload]);
+
+    const buf = stream.read();
+    assert.instanceOf(buf, Buffer);
+
+    const packet = new Packet(buf);
+    assert.strictEqual(packet.type(), packetType);
+    assert.strictEqual(packet.length(), payload.length + 8);
+    assert.isTrue(packet.isLast());
+    assert.deepEqual(packet.data(), payload);
+
+    assert.isNull(stream.read());
+    assertNoDanglingEventListeners(stream);
+  });
+
+  it('splits payloads that are larger than the packet size across multiple packets', async function() {
+    const payload = Buffer.from([1, 2, 3, 4, 5, 6]);
+    const stream = new BufferListStream();
+
+    // `packetSize` allows for 4 bytes of data per packet
+    await writeMessage(stream, packetSize, packetType, [payload]);
+
+    const [firstPacket, secondPacket, ...rest] = splitPackets(stream.read());
+    assert.lengthOf(rest, 0);
+
+    assert.strictEqual(firstPacket.packetId(), 1);
+    assert.isFalse(firstPacket.isLast());
+    assert.deepEqual(firstPacket.data(), payload.subarray(0, 4));
+
+    assert.strictEqual(secondPacket.packetId(), 2);
+    assert.isTrue(secondPacket.isLast());
+    assert.deepEqual(secondPacket.data(), payload.subarray(4));
+  });
+
+  it('writes an empty final packet for an empty payload', async function() {
+    const stream = new BufferListStream();
+
+    await writeMessage(stream, packetSize, packetType, []);
+
+    const packet = new Packet(stream.read());
+    assert.isTrue(packet.isLast());
+    assert.deepEqual(packet.data(), Buffer.alloc(0));
+
+    assert.isNull(stream.read());
+  });
+
+  it('terminates the message with the ignore flag set when iterating the payload errors', async function() {
+    const payload = Buffer.from([1, 2, 3, 4, 5, 6]);
+    const stream = new BufferListStream();
+
+    let hadError = false;
+    try {
+      await writeMessage(stream, packetSize, packetType, (async function*() {
+        yield payload;
+        throw new Error('iteration error');
+      })());
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'iteration error');
+    }
+
+    assert(hadError);
+
+    const [firstPacket, lastPacket, ...rest] = splitPackets(stream.read());
+    assert.lengthOf(rest, 0);
+
+    // The part of the payload that filled a whole packet was sent before the
+    // error occurred, the rest is discarded.
+    assert.isFalse(firstPacket.isLast());
+    assert.deepEqual(firstPacket.data(), payload.subarray(0, 4));
+
+    assert.isTrue(lastPacket.isLast());
+    assert.include(lastPacket.statusAsString(), 'IGNORE');
+    assert.deepEqual(lastPacket.data(), Buffer.alloc(0));
+
+    assertNoDanglingEventListeners(stream);
+  });
+
+  it('handles errors from the payload while the stream is waiting for drain', async function() {
+    const payload = Buffer.from([1, 2, 3, 4]);
+
+    const callbacks: Array<() => void> = [];
+    const stream = new Duplex({
+      write(chunk, encoding, callback) {
+        // Collect all callbacks so that we can simulate draining the stream later
+        callbacks.push(callback);
+      },
+      read() {},
+
+      // instantly return false on write requests to indicate that the stream needs to drain
+      highWaterMark: 1
+    });
+
+    let hadError = false;
+    try {
+      await writeMessage(stream, packetSize, packetType, (async function*() {
+        yield payload;
+
+        // Simulate draining the stream after the exception was thrown
+        setTimeout(() => {
+          let cb;
+          while (cb = callbacks.shift()) {
+            cb();
+          }
+        }, 20);
+
+        throw new Error('iteration error');
+      })());
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'iteration error');
+    }
+
+    assert(hadError);
+    assertNoDanglingEventListeners(stream);
+  });
+
+  it('handles errors on the stream during writing', async function() {
+    const payload = Buffer.from([1, 2, 3]);
+    const stream = new Duplex({
+      write(chunk, encoding, callback) {
+        callback(new Error('write error'));
+      },
+      read() {}
+    });
+
+    let hadError = false;
+    try {
+      await writeMessage(stream, packetSize, packetType, [payload]);
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'write error');
+    }
+
+    assert(hadError);
+    assertNoDanglingEventListeners(stream);
+  });
+
+  it('handles errors on the stream while waiting for the stream to drain', async function() {
+    const payload = Buffer.from([1, 2, 3]);
+    const stream = new Duplex({
+      write(chunk, encoding, callback) {
+        // never call callback so that the stream never drains
+      },
+      read() {},
+
+      // instantly return false on write requests to indicate that the stream needs to drain
+      highWaterMark: 1
+    });
+
+    setTimeout(() => {
+      assert(stream.writableNeedDrain);
+      stream.destroy(new Error('write error'));
+    }, 20);
+
+    let hadError = false;
+    try {
+      await writeMessage(stream, packetSize, packetType, [payload, payload, payload]);
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'write error');
+    }
+
+    assert(hadError);
+    assertNoDanglingEventListeners(stream);
+  });
+
+  it('handles errors on the stream while waiting for more payload data', async function() {
+    const payload = Buffer.from([1, 2, 3]);
+    const stream = new Duplex({
+      write(chunk, encoding, callback) {
+        // never call callback so that the stream never drains
+      },
+      read() {},
+
+      // instantly return false on write requests to indicate that the stream needs to drain
+      highWaterMark: 1
+    });
+
+    setTimeout(() => {
+      assert(stream.writableNeedDrain);
+      stream.destroy(new Error('write error'));
+    }, 20);
+
+    let hadError = false;
+    try {
+      await writeMessage(stream, packetSize, packetType, (async function*() {
+        yield payload;
+        yield payload;
+        yield payload;
+      })());
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'write error');
+    }
+
+    assert(hadError);
+    assertNoDanglingEventListeners(stream);
+  });
+
+  it('throws when the given stream is not writable', async function() {
+    const stream = new BufferListStream();
+    stream.end();
+
+    let hadError = false;
+    try {
+      await writeMessage(stream, packetSize, packetType, []);
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'Premature close');
+    }
+
+    assert(hadError);
+  });
+
+  it('terminates the message with the ignore flag set when the given cancel signal is aborted while writing', async function() {
+    const payload = Buffer.from([1, 2, 3, 4, 5, 6]);
+    const stream = new BufferListStream();
+
+    const controller = new AbortController();
+    setTimeout(() => {
+      controller.abort();
+    }, 20);
+
+    await writeMessage(stream, packetSize, packetType, (async function*() {
+      yield payload;
+
+      // Stall forever, waiting for more data that never arrives.
+      await new Promise(() => {});
+    })(), { cancelSignal: controller.signal });
+
+    const [firstPacket, lastPacket, ...rest] = splitPackets(stream.read());
+    assert.lengthOf(rest, 0);
+
+    // The part of the payload that filled a whole packet was sent before the
+    // cancellation, the rest is discarded and the message is terminated with
+    // the `IGNORE` flag set.
+    assert.isFalse(firstPacket.isLast());
+    assert.deepEqual(firstPacket.data(), payload.subarray(0, 4));
+
+    assert.isTrue(lastPacket.isLast());
+    assert.include(lastPacket.statusAsString(), 'IGNORE');
+    assert.deepEqual(lastPacket.data(), Buffer.alloc(0));
+
+    assertNoDanglingEventListeners(stream);
+    assert.lengthOf(getEventListeners(controller.signal, 'abort'), 0);
+  });
+
+  it('writes an ignored message without consuming the payload when the given cancel signal is already aborted', async function() {
+    const stream = new BufferListStream();
+    const cancelSignal = AbortSignal.abort();
+
+    let payloadConsumed = false;
+
+    await writeMessage(stream, packetSize, packetType, (async function*() {
+      payloadConsumed = true;
+      yield Buffer.from([1, 2, 3]);
+    })(), { cancelSignal });
+
+    assert.isFalse(payloadConsumed);
+
+    const [packet, ...rest] = splitPackets(stream.read());
+    assert.lengthOf(rest, 0);
+
+    assert.isTrue(packet.isLast());
+    assert.include(packet.statusAsString(), 'IGNORE');
+    assert.deepEqual(packet.data(), Buffer.alloc(0));
+
+    assertNoDanglingEventListeners(stream);
+  });
+
+  it('stops waiting for the stream to drain when the given cancel signal is aborted', async function() {
+    const payload = Buffer.from([1, 2, 3]);
+    const stream = new Duplex({
+      write(chunk, encoding, callback) {
+        // never call callback so that the stream never drains
+      },
+      read() {},
+
+      // instantly return false on write requests to indicate that the stream needs to drain
+      highWaterMark: 1
+    });
+
+    const controller = new AbortController();
+    setTimeout(() => {
+      assert(stream.writableNeedDrain);
+      controller.abort();
+    }, 20);
+
+    // Resolves normally: the remaining payload is discarded and the
+    // `IGNORE`-flagged terminator is written without further drain waits.
+    await writeMessage(stream, packetSize, packetType, [payload, payload, payload], { cancelSignal: controller.signal });
+
+    assertNoDanglingEventListeners(stream);
+    assert.lengthOf(getEventListeners(controller.signal, 'abort'), 0);
+  });
+
+  it('prefers teardown over cancellation when both signals are aborted', async function() {
+    const stream = new BufferListStream();
+    const signal = AbortSignal.abort(new Error('teardown'));
+    const cancelSignal = AbortSignal.abort();
+
+    let hadError = false;
+    try {
+      await writeMessage(stream, packetSize, packetType, [Buffer.from([1, 2, 3])], { cancelSignal, signal });
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'teardown');
+    }
+
+    assert(hadError);
+
+    // Nothing was written.
+    assert.isNull(stream.read());
+    assertNoDanglingEventListeners(stream);
+  });
+
+  it('throws when the given signal is already aborted', async function() {
+    const stream = new BufferListStream();
+    const signal = AbortSignal.abort(new Error('canceled'));
+
+    let hadError = false;
+    try {
+      await writeMessage(stream, packetSize, packetType, [Buffer.from([1, 2, 3])], { signal });
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'canceled');
+    }
+
+    assert(hadError);
+
+    // Nothing was written.
+    assert.isNull(stream.read());
+    assertNoDanglingEventListeners(stream);
+  });
+
+  it('stops writing when the given signal is aborted while waiting for the stream to drain', async function() {
+    const payload = Buffer.from([1, 2, 3]);
+    const stream = new Duplex({
+      write(chunk, encoding, callback) {
+        // never call callback so that the stream never drains
+      },
+      read() {},
+
+      // instantly return false on write requests to indicate that the stream needs to drain
+      highWaterMark: 1
+    });
+
+    const controller = new AbortController();
+    setTimeout(() => {
+      assert(stream.writableNeedDrain);
+      controller.abort(new Error('canceled'));
+    }, 20);
+
+    let hadError = false;
+    try {
+      await writeMessage(stream, packetSize, packetType, [payload, payload, payload], { signal: controller.signal });
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'canceled');
+    }
+
+    assert(hadError);
+    assertNoDanglingEventListeners(stream);
+    assert.lengthOf(getEventListeners(controller.signal, 'abort'), 0);
+  });
+
+  it('stops writing when the given signal is aborted while waiting for more payload data', async function() {
+    const payload = Buffer.from([1, 2, 3, 4, 5, 6]);
+    const stream = new BufferListStream();
+
+    const controller = new AbortController();
+    setTimeout(() => {
+      controller.abort(new Error('canceled'));
+    }, 20);
+
+    let hadError = false;
+    try {
+      await writeMessage(stream, packetSize, packetType, (async function*() {
+        yield payload;
+
+        // Stall forever, waiting for more data that never arrives.
+        await new Promise(() => {});
+      })(), { signal: controller.signal });
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'canceled');
+    }
+
+    assert(hadError);
+
+    // The part of the payload that filled a whole packet was written, but
+    // no `IGNORE` terminator: the message is intentionally left unterminated
+    // as the connection is being torn down.
+    const packets = splitPackets(stream.read());
+    assert.lengthOf(packets, 1);
+    assert.isFalse(packets[0].isLast());
+
+    assertNoDanglingEventListeners(stream);
+    assert.lengthOf(getEventListeners(controller.signal, 'abort'), 0);
+  });
+});
+
+describe('readMessage', function() {
+  it('reads a message consisting of a single TDS packet from the given stream', async function() {
+    const payload = Buffer.from([1, 2, 3]);
+    const packet = new Packet(packetType);
+    packet.last(true);
+    packet.addData(payload);
+
+    const stream = new BufferListStream();
+    stream.write(packet.buffer);
+
+    const chunks = [];
+    for await (const chunk of readMessage(stream)) {
+      chunks.push(chunk);
+    }
+
+    assert.deepEqual(chunks, [payload]);
+    assertNoDanglingEventListeners(stream);
+  });
+
+  it('reads a message that spans multiple TDS packets', async function() {
+    const payload = Buffer.from([1, 2, 3]);
+
+    const firstPacket = new Packet(packetType);
+    firstPacket.addData(payload.subarray(0, 2));
+
+    const lastPacket = new Packet(packetType);
+    lastPacket.last(true);
+    lastPacket.addData(payload.subarray(2));
+
+    const stream = new BufferListStream();
+    stream.write(firstPacket.buffer);
+    stream.write(lastPacket.buffer);
+
+    const chunks = [];
+    for await (const chunk of readMessage(stream)) {
+      chunks.push(chunk);
+    }
+
+    assert.deepEqual(Buffer.concat(chunks), payload);
+  });
+
+  it('reads packets that arrive in chunks that do not align with packet boundaries', async function() {
+    const payload = Buffer.from([1, 2, 3]);
+
+    const firstPacket = new Packet(packetType);
+    firstPacket.addData(payload.subarray(0, 2));
+
+    const lastPacket = new Packet(packetType);
+    lastPacket.last(true);
+    lastPacket.addData(payload.subarray(2));
+
+    const data = Buffer.concat([firstPacket.buffer, lastPacket.buffer]);
+
+    const stream = new Readable({ read() {} });
+
+    // Deliver the data byte by byte.
+    for (let i = 0; i < data.length; i++) {
+      setTimeout(() => {
+        stream.push(data.subarray(i, i + 1));
+      }, i);
+    }
+
+    const chunks = [];
+    for await (const chunk of readMessage(stream)) {
+      chunks.push(chunk);
+    }
+
+    assert.deepEqual(Buffer.concat(chunks), payload);
+  });
+
+  it('pushes bytes that follow the last packet of a message back onto the stream', async function() {
+    const payload = Buffer.from([1, 2, 3]);
+    const packet = new Packet(packetType);
+    packet.last(true);
+    packet.addData(payload);
+
+    const trailingBytes = Buffer.from([9, 9, 9, 9]);
+
+    const stream = new BufferListStream();
+    stream.write(Buffer.concat([packet.buffer, trailingBytes]));
+
+    const chunks = [];
+    for await (const chunk of readMessage(stream)) {
+      chunks.push(chunk);
+    }
+
+    assert.deepEqual(chunks, [payload]);
+    assert.deepEqual(stream.read(), trailingBytes);
+  });
+
+  it('handles errors while reading from the stream', async function() {
+    const stream = Readable.from((async function*(): AsyncGenerator<Buffer> {
+      throw new Error('read error');
+    })());
+
+    let hadError = false;
+    try {
+      const message = readMessage(stream);
+      while (!(await message.next()).done) {
+        // Discard the message contents.
+      }
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'read error');
+    }
+
+    assert(hadError);
+  });
+
+  it('throws when the stream is closed before the message was fully read', async function() {
+    const packet = new Packet(packetType);
+    packet.addData(Buffer.from([1, 2, 3]));
+
+    const stream = new Readable({ read() {} });
+    stream.push(packet.buffer);
+
+    setTimeout(() => {
+      stream.destroy();
+    }, 20);
+
+    let hadError = false;
+    try {
+      const message = readMessage(stream);
+      while (!(await message.next()).done) {
+        // Discard the message contents.
+      }
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'Premature close');
+    }
+
+    assert(hadError);
+  });
+
+  it('throws when the stream is closed while a yielded chunk is being processed', async function() {
+    const payload = Buffer.from([1, 2, 3]);
+    const packet = new Packet(packetType);
+    packet.addData(payload);
+
+    const stream = new Readable({ read() {} });
+    stream.push(packet.buffer);
+
+    const message = readMessage(stream);
+
+    const { value } = await message.next();
+    assert.deepEqual(value, payload);
+
+    // The generator is now suspended at `yield`. Close the stream without
+    // an error while the consumer is processing the chunk.
+    stream.destroy();
+    await once(stream, 'close');
+
+    let hadError = false;
+    try {
+      await message.next();
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'Premature close');
+    }
+
+    assert(hadError);
+  });
+
+  it('throws when the given stream is not readable', async function() {
+    const stream = new Readable({ read() {} });
+    stream.destroy();
+
+    let hadError = false;
+    try {
+      const message = readMessage(stream);
+      while (!(await message.next()).done) {
+        // Discard the message contents.
+      }
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'Premature close');
+    }
+
+    assert(hadError);
+  });
+
+  it('throws when the given signal is already aborted', async function() {
+    const stream = new BufferListStream();
+    const signal = AbortSignal.abort(new Error('canceled'));
+
+    let hadError = false;
+    try {
+      const message = readMessage(stream, { signal });
+      while (!(await message.next()).done) {
+        // Discard the message contents.
+      }
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'canceled');
+    }
+
+    assert(hadError);
+    assertNoDanglingEventListeners(stream);
+  });
+
+  it('stops reading when the given signal is aborted while waiting for data on a quiet stream', async function() {
+    // A stream that stays open but never produces data, like a connection
+    // to a server that accepted the connection but never responds.
+    const stream = new Readable({ read() {} });
+
+    const controller = new AbortController();
+    setTimeout(() => {
+      controller.abort(new Error('canceled'));
+    }, 20);
+
+    let hadError = false;
+    try {
+      const message = readMessage(stream, { signal: controller.signal });
+      while (!(await message.next()).done) {
+        // Discard the message contents.
+      }
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'canceled');
+    }
+
+    assert(hadError);
+    assert.lengthOf(getEventListeners(controller.signal, 'abort'), 0);
+  });
+
+  it('stops reading when the given signal is aborted after a partial message was read', async function() {
+    const firstPacket = new Packet(packetType);
+    firstPacket.addData(Buffer.from([1, 2, 3]));
+
+    const stream = new Readable({ read() {} });
+    stream.push(firstPacket.buffer);
+
+    const controller = new AbortController();
+    setTimeout(() => {
+      controller.abort(new Error('canceled'));
+    }, 20);
+
+    const chunks = [];
+    let hadError = false;
+    try {
+      for await (const chunk of readMessage(stream, { signal: controller.signal })) {
+        chunks.push(chunk);
+      }
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'canceled');
+    }
+
+    assert(hadError);
+    assert.deepEqual(chunks, [Buffer.from([1, 2, 3])]);
+    assert.lengthOf(getEventListeners(controller.signal, 'abort'), 0);
+  });
+
+  it('settles pending reads when the given signal is aborted, instead of deadlocking on `.return()`', async function() {
+    // Regression test: calling `.return()` on the generator while it is
+    // suspended waiting for stream data is queued behind the pending
+    // `.next()` call and would deadlock. Aborting the signal settles the
+    // pending read, after which `.return()` resolves normally.
+    const stream = new Readable({ read() {} });
+
+    const controller = new AbortController();
+    const message = readMessage(stream, { signal: controller.signal });
+
+    const pendingRead = message.next();
+
+    controller.abort(new Error('canceled'));
+
+    let hadError = false;
+    try {
+      await pendingRead;
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, Error);
+      assert.strictEqual(err.message, 'canceled');
+    }
+
+    assert(hadError);
+
+    // The generator has completed; `.return()` settles immediately.
+    assert.deepEqual(await message.return(), { value: undefined, done: true });
+    assert.lengthOf(getEventListeners(controller.signal, 'abort'), 0);
+  });
+
+  it('throws when receiving a packet with an invalid length', async function() {
+    const invalidPacket = Buffer.alloc(8);
+    invalidPacket.writeUInt16BE(4, 2);
+
+    const stream = new BufferListStream();
+    stream.write(invalidPacket);
+
+    let hadError = false;
+    try {
+      const message = readMessage(stream);
+      while (!(await message.next()).done) {
+        // Discard the message contents.
+      }
+    } catch (err: any) {
+      hadError = true;
+
+      assert.instanceOf(err, ConnectionError);
+      assert.strictEqual(err.message, 'Unable to process incoming packet');
+    }
+
+    assert(hadError);
+    assertNoDanglingEventListeners(stream);
+  });
+});
+
 describe('MessageIO', function() {
   let server: Server;
   let serverConnection: Socket;