diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ff5e8d..37c527b 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/COVERAGE.md b/COVERAGE.md index 4ee8128..4199a5c 100644 --- a/COVERAGE.md +++ b/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%. diff --git a/README.md b/README.md index ec017e8..257a310 100644 --- a/README.md +++ b/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' }); ``` diff --git a/index.d.ts b/index.d.ts index d0faf3f..9bc879f 100644 --- a/index.d.ts +++ b/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 /// @@ -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; } diff --git a/lib/msgpack.js b/lib/msgpack.js index 81fae7f..8686024 100644 --- a/lib/msgpack.js +++ b/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); diff --git a/package-lock.json b/package-lock.json index bc2946c..389cc91 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "msgpack", - "version": "3.2.0", + "version": "3.3.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "msgpack", - "version": "3.2.0", + "version": "3.3.0", "license": "BSD-3-Clause", "dependencies": { "nan": "^2.23.1" diff --git a/package.json b/package.json index 88a2a1e..7fa82b1 100644 --- a/package.json +++ b/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 ", "contributors": [ diff --git a/test/msgpack.test.js b/test/msgpack.test.js index 5c98aa4..8ea7576 100644 --- a/test/msgpack.test.js +++ b/test/msgpack.test.js @@ -126,6 +126,25 @@ describe('msgpack pack/unpack', () => { }); }); +function mockWritable(returnSeq) { + const s = new EventEmitter(); + const seq = returnSeq ? returnSeq.slice() : []; + s.writes = []; + s.write = function (chunk) { + s.writes.push({ + chunk: chunk, + extra: Array.prototype.slice.call(arguments, 1), + }); + const last = arguments[arguments.length - 1]; + const ret = seq.length > 0 ? seq.shift() : true; + if (typeof last === 'function') { + last(); + } + return ret; + }; + return s; +} + describe('msgpack.Stream', () => { it('sends a packed message through write', () => { const s = new EventEmitter(); @@ -150,6 +169,206 @@ describe('msgpack.Stream', () => { assert.deepEqual(Array.prototype.slice.call(s.write.args, 1), [1, 2, 3]); }); + it('returns true when write returns true and does not queue', () => { + const s = mockWritable([true]); + const ms = new msgpack.Stream(s); + assert.equal(ms.send('hello'), true); + assert.equal(s.writes.length, 1); + assert.deepEqual(msgpack.unpack(s.writes[0].chunk), 'hello'); + assert.equal(ms.send('again'), true); + assert.equal(s.writes.length, 2); + }); + + it('queues send after write returns false and flushes FIFO on drain', () => { + const s = mockWritable([false, true]); + const ms = new msgpack.Stream(s); + const drained = []; + ms.on('drain', () => drained.push(true)); + + assert.equal(ms.send('one'), false); + assert.equal(s.writes.length, 1); + assert.equal(ms.send('two'), false); + assert.equal(ms.send('three'), false); + assert.equal(s.writes.length, 1); + + s.emit('drain'); + + assert.equal(s.writes.length, 3); + assert.deepEqual( + s.writes.map((w) => msgpack.unpack(w.chunk)), + ['one', 'two', 'three'] + ); + assert.equal(drained.length, 1); + }); + + it('re-emits drain when the underlying writable drains with an empty queue', () => { + const s = mockWritable([false]); + const ms = new msgpack.Stream(s); + let got = 0; + ms.on('drain', () => { + got += 1; + }); + assert.equal(ms.send('x'), false); + s.emit('drain'); + assert.equal(got, 1); + assert.equal(s.writes.length, 1); + }); + + it('throws when more than 1024 messages are queued', () => { + const s = mockWritable(); + s.writes = []; + s.write = function (chunk) { + s.writes.push({ chunk: chunk, extra: [] }); + return false; + }; + const ms = new msgpack.Stream(s); + assert.equal(ms.send('head'), false); + for (let i = 0; i < 1024; i++) { + assert.equal(ms.send(i), false); + } + assert.equal(s.writes.length, 1); + assert.throws(() => ms.send('overflow'), /backpressure|queue full/); + }); + + it('runs the extra callback argument on send', () => { + const s = mockWritable([true]); + const ms = new msgpack.Stream(s); + let n = 0; + assert.equal( + ms.send('hello', () => { + n += 1; + }), + true + ); + assert.equal(n, 1); + }); + + it('runs the callback of a queued send after that buffer is written', () => { + const s = mockWritable([false, true]); + const ms = new msgpack.Stream(s); + let n = 0; + assert.equal(ms.send('one'), false); + assert.equal( + ms.send('two', () => { + n += 1; + }), + false + ); + assert.equal(n, 0); + s.emit('drain'); + assert.equal(n, 1); + assert.deepEqual( + s.writes.map((w) => msgpack.unpack(w.chunk)), + ['one', 'two'] + ); + assert.equal(s.writes[1].extra.length, 1); + assert.equal(typeof s.writes[1].extra[0], 'function'); + }); + + it('does not pass encoding to a queued flush write', () => { + const s = mockWritable([false, true]); + const ms = new msgpack.Stream(s); + ms.send('one'); + ms.send('two', 'utf8'); + s.emit('drain'); + assert.equal(s.writes[1].extra.length, 0); + assert.deepEqual(msgpack.unpack(s.writes[1].chunk), 'two'); + }); + + it('stops a flush when write returns false again', () => { + const s = mockWritable([false, false, true]); + const ms = new msgpack.Stream(s); + ms.send('a'); + ms.send('b'); + ms.send('c'); + s.emit('drain'); + assert.equal(s.writes.length, 2); + s.emit('drain'); + assert.equal(s.writes.length, 3); + assert.deepEqual( + s.writes.map((w) => msgpack.unpack(w.chunk)), + ['a', 'b', 'c'] + ); + }); + + it('does not spin if write emits drain synchronously during flush', () => { + const s = new EventEmitter(); + let n = 0; + s.write = function () { + n += 1; + if (n === 1) { + return false; + } + s.emit('drain'); + return true; + }; + const ms = new msgpack.Stream(s); + ms.send('a'); + ms.send('b'); + ms.send('c'); + s.emit('drain'); + assert.equal(n, 3); + }); + + it('emits error and drops the queue on close with pending sends', () => { + const s = mockWritable([false]); + const ms = new msgpack.Stream(s); + const errors = []; + ms.on('error', (e) => errors.push(e)); + ms.send('one'); + ms.send('two'); + s.emit('close'); + assert.equal(errors.length, 1); + assert.match(errors[0].message, /unsent|backpressure/); + s.emit('drain'); + assert.equal(s.writes.length, 1); + }); + + it('does not emit error on close when the queue is empty', () => { + const s = mockWritable([true]); + const ms = new msgpack.Stream(s); + const errors = []; + ms.on('error', (e) => errors.push(e)); + ms.send('one'); + s.emit('close'); + s.emit('end'); + s.emit('error', new Error('socket')); + assert.equal(errors.length, 0); + }); + + it('drops the queue on underlying error and end', () => { + const s = mockWritable([false]); + const ms = new msgpack.Stream(s); + const errors = []; + ms.on('error', (e) => errors.push(e)); + ms.send('a'); + ms.send('b'); + s.emit('error', new Error('socket')); + assert.equal(errors.length, 1); + + const s2 = mockWritable([false]); + const ms2 = new msgpack.Stream(s2); + const errors2 = []; + ms2.on('error', (e) => errors2.push(e)); + ms2.send('a'); + ms2.send('b'); + s2.emit('end'); + assert.equal(errors2.length, 1); + }); + + it('invokes queued callbacks when the queue is dropped', (t, done) => { + const s = mockWritable([false]); + const ms = new msgpack.Stream(s); + ms.on('error', () => {}); + ms.send('one'); + ms.send('two', (err) => { + assert.ok(err); + assert.match(err.message, /unsent|backpressure/); + done(); + }); + s.emit('close'); + }); + it('emits msg for a complete packet', () => { const s = new EventEmitter(); const ms = new msgpack.Stream(s);