feat: Stream backpressure and drain (#43)

CHANGELOG.md

@@ -7,6 +7,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [3.3.0] - 2026-09-19
+
+`Stream.send` queues packed messages when the underlying writable returns
+`false`, re-emits `drain`, and refuses more than 1024 pending messages.
+See `#43`.
+
+### Added
+
+- `Stream` re-emits `drain` from the underlying writable so callers can
+  listen on the msgpack Stream, not only on the raw socket.
+- After `write()` returns `false`, further `send()` calls queue the already
+  packed Buffer and flush FIFO on `drain`. `send()` stays synchronous and
+  returns the boolean from `write()`, or `false` if the message was queued.
+- Extra arguments (encoding, callback) are still forwarded to `write` on an
+  immediate write. Queued flushes call `write(buf)` without inventing an
+  encoding; a callback supplied on a queued `send` runs after that buffer is
+  written, or with an error if the queue is dropped.
+- The pending-send queue is capped at **1024** messages. A further `send()`
+  throws a catchable `Error` whose message mentions backpressure / queue
+  full.
+- If the underlying stream emits `error`, `close`, or `end` with messages
+  still queued, the queue is dropped and Stream emits `error`. An empty
+  queue does not emit that extra error. Handlers do not throw.
+
 ## [3.2.0] - 2026-09-19
 
 Optional second-argument unpack option `{ lazy: true }` wraps maps and arrays
@@ -128,7 +152,8 @@ GitHub Actions tests Node 18/20/22 on Ubuntu, macOS, and Windows 2022.
 - Pack throw paths free or return pooled sbuffers on every exit.
 - msgpack-c c-7.0.2 includes unpacker buffer-expansion overflow checks.
 
-[Unreleased]: https://github.com/msgpack/msgpack-node/compare/v3.2.0...HEAD
+[Unreleased]: https://github.com/msgpack/msgpack-node/compare/v3.3.0...HEAD
+[3.3.0]: https://github.com/msgpack/msgpack-node/compare/v3.2.0...v3.3.0
 [3.2.0]: https://github.com/msgpack/msgpack-node/compare/v3.1.0...v3.2.0
 [3.1.0]: https://github.com/msgpack/msgpack-node/compare/v3.0.0...v3.1.0
 [3.0.0]: https://github.com/msgpack/msgpack-node/compare/e04c9b55f98d64512174d6e859b8294b729659a2...HEAD

COVERAGE.md

@@ -1,4 +1,4 @@
-# Coverage — msgpack 3.2.0
+# Coverage — msgpack 3.3.0
 
 `npm run coverage` runs both halves and fails the build under 95%.
 

README.md

@@ -2,10 +2,11 @@
 and de-serializes JavaScript values with [MessagePack](https://msgpack.org).
 Packed output is a `Buffer` and is typically much smaller than JSON.
 
-Version 3.2 requires **Node.js 18+**, vendors **msgpack-c c-7.0.2**, unpacks
+Version 3.3 requires **Node.js 18+**, vendors **msgpack-c c-7.0.2**, unpacks
 64-bit integers outside `Number.MAX_SAFE_INTEGER` as `bigint`, accepts
-optional pack type/family hints, and can unpack maps and arrays lazily
-(`unpack(buf, { lazy: true })`). See [`SECURITY.md`](SECURITY.md).
+optional pack type/family hints, can unpack maps and arrays lazily
+(`unpack(buf, { lazy: true })`), and applies write backpressure on
+`Stream.send`. See [`SECURITY.md`](SECURITY.md).
 
 ### Usage
 
@@ -26,7 +27,14 @@ and returns a JavaScript value, or `null` if the buffer is a truncated
 (incomplete) MessagePack object. Oversized array/map/string bombs throw.
 
 A streaming helper wraps a readable socket and emits `msg`, plus `error` when
-a packet cannot be unpacked (the offending buffer is dropped):
+a packet cannot be unpacked (the offending buffer is dropped). `send()` packs
+and writes; it returns the boolean from the underlying `write()`, or `false`
+if the message was queued because a previous write returned `false` and
+`drain` has not fired yet. `drain` is re-emitted from the underlying
+writable onto the Stream. At most **1024** messages may wait in that queue;
+a further `send()` throws. Extra `write` arguments (encoding, callback) are
+forwarded on an immediate write. On underlying `error` / `close` / `end`,
+queued messages are dropped and Stream emits `error` if any were unsent:
 
 ```javascript
 const msgpack = require('msgpack');
@@ -37,6 +45,9 @@ ms.on('msg', (m) => {
 ms.on('error', (e) => {
   console.error('bad packet', e.message);
 });
+ms.on('drain', () => {
+  /* underlying writable is ready for more send() calls */
+});
 ms.send({ hello: 'world' });
 ```
 

index.d.ts

@@ -1,4 +1,4 @@
-// Type definitions for msgpack 3.2.0
+// Type definitions for msgpack 3.3.0
 // Project: https://github.com/msgpack/msgpack-node
 
 /// <reference types="node" />
@@ -96,8 +96,11 @@ export namespace unpack {
 /**
  * Frames MessagePack messages over a stream.
  *
- * Emits `'msg'` with each decoded value, and `'error'` if a packet cannot be
- * decoded (the buffered data is then dropped).
+ * Emits `'msg'` with each decoded value, `'drain'` when the underlying
+ * writable is ready for more data and the send queue is empty, and
+ * `'error'` if a packet cannot be decoded (the buffered data is then
+ * dropped) or if queued sends are discarded because the underlying stream
+ * emitted `error`/`close`/`end`.
  */
 export class Stream extends EventEmitter {
     constructor(s: NodeJS.ReadWriteStream);
@@ -107,7 +110,13 @@ export class Stream extends EventEmitter {
 
     /**
      * Pack `m` and write it to the underlying stream. Extra arguments are
-     * forwarded to `stream.write()` (encoding, callback).
+     * forwarded to `stream.write()` (encoding, callback) on an immediate
+     * write. Returns the boolean from `write()`, or `false` if the message
+     * was queued because a previous write returned false and `drain` has
+     * not fired yet. At most 1024 messages may wait in that queue; further
+     * `send()` throws. Queued flushes call `write(buf)` without inventing
+     * an encoding; a supplied callback runs after that buffer is written
+     * or if the queue is dropped.
      */
     send(m: any, ...args: any[]): boolean;
 }

lib/msgpack.js

@@ -26,19 +26,92 @@ function unpack(buf, opts) {
 
 unpack.bytes_remaining = 0;
 
+const SEND_QUEUE_CAP = 1024;
+
 function Stream(s) {
   const self = this;
   events.EventEmitter.call(self);
   self.buf = null;
 
+  const queue = [];
+  let waitingForDrain = false;
+  let flushing = false;
+
+  function abandonQueue() {
+    const pending = queue.splice(0, queue.length);
+    waitingForDrain = false;
+    if (pending.length === 0) {
+      return;
+    }
+    const err = new Error(
+      'msgpack Stream dropped ' + pending.length +
+        ' unsent message(s) after backpressure'
+    );
+    for (let i = 0; i < pending.length; i++) {
+      const cb = pending[i].cb;
+      if (cb) {
+        process.nextTick(cb, err);
+      }
+    }
+    self.emit('error', err);
+  }
+
+  function onWritableDrain() {
+    if (flushing) {
+      return;
+    }
+    flushing = true;
+    try {
+      while (queue.length > 0) {
+        const item = queue.shift();
+        const ok = item.cb ? s.write(item.buf, item.cb) : s.write(item.buf);
+        if (ok === false) {
+          waitingForDrain = true;
+          return;
+        }
+      }
+      waitingForDrain = false;
+      self.emit('drain');
+    } finally {
+      flushing = false;
+    }
+  }
+
   self.send = function (m) {
-    const args = [pack(m)];
+    const packed = pack(m);
+    if (waitingForDrain) {
+      if (queue.length >= SEND_QUEUE_CAP) {
+        throw new Error(
+          'msgpack Stream backpressure queue full (' +
+            SEND_QUEUE_CAP +
+            ' pending messages)'
+        );
+      }
+      let cb;
+      if (arguments.length > 1 &&
+          typeof arguments[arguments.length - 1] === 'function') {
+        cb = arguments[arguments.length - 1];
+      }
+      queue.push({ buf: packed, cb: cb });
+      return false;
+    }
+
+    const args = [packed];
     for (let i = 1; i < arguments.length; i++) {
       args.push(arguments[i]);
     }
-    return s.write.apply(s, args);
+    const ok = s.write.apply(s, args);
+    if (ok === false) {
+      waitingForDrain = true;
+    }
+    return ok;
   };
 
+  s.addListener('drain', onWritableDrain);
+  s.addListener('error', abandonQueue);
+  s.addListener('close', abandonQueue);
+  s.addListener('end', abandonQueue);
+
   s.addListener('data', function (d) {
     if (self.buf) {
       const b = buffer.Buffer.allocUnsafe(self.buf.length + d.length);

package.json

@@ -1,7 +1,7 @@
 {
   "name": "msgpack",
   "description": "A space-efficient object serialization library for Node.js",
-  "version": "3.2.0",
+  "version": "3.3.0",
   "homepage": "https://github.com/msgpack/msgpack-node",
   "author": "Peter Griess <pg@std.in>",
   "contributors": [