node stream backpressure test

test.js

const streams = require('stream');
const StreamLimiter = require('throttle');
const uuid = require('uuid');

class TestDuplex extends streams.Duplex {
  nextWrite = null

  constructor() {
    super({
      highWaterMark: 128
    });
  }

  _read(size) {
    if (this.nextWrite) {
      this.nextWrite();
      this.nextWrite = null;
    }
  }

  _write(chunk, encoding, callback) {
    const ret = this.push(chunk);
    if (ret) {
      callback();
    } else {
      console.log('internal buffer is full');
      this.nextWrite = () => {
        callback();
      }
    }
  }
}


const testDuplex = new TestDuplex();

const deque = [];

const doit = () => {
  const a = Buffer.from(uuid.v4());
  deque.push(...a);
  testDuplex.write(a, () => {
    setImmediate(() => doit());
  });
}
doit();

testDuplex
  .pipe(new StreamLimiter(128 * 8))
  .on('data', (data) => {
    for (let i=0; i<data.length; i++) {
      const a = data[i];
      const b = deque.shift();
      if (a !== b) {
        throw new Error(`WRONG DATA ${a} !== ${b}`);
      }
    }
  });

WARNING

DO NOT WRITE AGAIN BEFORE WRITE IS DONE.

BELOW IS NOT WORKING. (this.nextWrite is overwritten.)

const doit = () => {
  let count = 0;

  for (let i=0; i<2; i++) {
    const a = Buffer.from(uuid.v4());
    deque.push(...a);
    testDuplex.write(a, () => {
      count++;
      if (count === 2) {
        setImmediate(() => doit());
      }
    });
  }
}
setImmediate(() => doit());