Define how "push" interacts with highWaterMark on readable side (suspending of push)

[lightsofapollo]

Node's internal use of highWaterMark is mostly in conjunction with the ability to call an explicit method to read with an explicit number number of bytes triggered by a request for data.

To have sane piping I believe we need the ability to halt the flow of data on the readable side (if the writer is blocked we can attempt to pause the reader). This is where having some enforced convention seems useful:

My theoretical example of how this might work:

const stream = new ReadableStream({ highWaterMark: 1024 }, (push, done, error) => {
  var timerId;  
  function spam() {
    var defer = push(str2ab("why can't you come up with a better example"));

    // lets say your a good person
    if (defer) return promise.then(spam);

    timerId = setTimeout(spam);
  }

  spam();
});

In the case where the promise is ignored (we should also set a flag?) then report it somehow?

---

[domenic]

The existing API already provides for this (although I admit I don't have an example of using it yet). If push returns false, you should pause the underlying source of data until onRead is called again. In other words, it uses onRead being called as the signaling mechanism, instead of a new out-of-band promise. This gives much more flexibility to the system to schedule calls to onRead when appropriate, perhaps multiple times, instead of only the zero or one times that a promise allows. In essence, this is the difference between pull streams and push streams: you are creating a push stream, which pushes data at the consumer and opts-in to information about how it is being consumed; the existing API is for a pull stream, where the primary mode of interaction is through notification that you are being pulled from.

In other words, I do not understand what this proposal offers over the API in the readme. I also don't understand if you are abandoning the idea of onRead being called multiple times, or if you're giving it a different meaning, or what.

[lightsofapollo]

Re-reading the readme I am mistaken about the constructor... As I understand it from reading it again the function passed to the constructor toRead is invoked for each call to .read.

After reading the various long threads about promises the first thing I think is this thing will only be called once.

---

To be clear then toRead functions much the same way _read and doing something like this would work?

// (but with ES6 and extending ReadableStream)
function MyReader() {
  ReadableStream.call(this);
}

MyReader.prototype.onRead = function(push, done, error) {}

Assuming the above is true why do we need the function on the constructor?

[domenic]

onRead is not a method of streams; it is a function you pass to a stream constructor to tell it how it that stream reacts to read calls. A sample implementation might be something like:

class ReadableStream {
  constructor(options, onRead) {
    this._options = options;
    this._onRead = onRead;
    this._internalBuffer = new ArrayBuffer(options.highWaterMark);
  }

  read() {
    this._onRead(data => {
      pushInto(this._internalBuffer, data);
      return !hasRoomLeft(this._internalBuffer, this._options);
    }, ...);
    process.nextTick(() => this._internalBuffer = new ArrayBuffer(options.highWaterMark);

    if (hasData(this._internalBuffer)) {
      return Promise.resolve(this._internalBuffer);
    }

    return waitForData(this._internalBuffer).then(() => this._internalBuffer);
  }
}

[domenic]

The new BaseReadableStream makes it somewhat clearer how this works, especially the onRead-type interaction we were discussing. But I will leave this open until I finish writing ReadableStream and showing how a configurable backpressure strategy with a given high water mark works.