Proposal Stream.buffer(highWaterMark)

[stephen-dahl]

I propose Stream.buffer as a new method that mimics the functionality of the node stream writev function

const deleteKeys = new Writable({
	highWaterMark: 50,
	write: (key, encoding, callback) => {
		//handle 1 item in stream
	},
	writev: (keys, callback) => {
		//keys contains all keys that have been buffered since the last iteration started
	},
});

This fills a similar role to batch. the difference is that batch waits for the provided count to be pushed to it before it pushes the array downstream.
buffer would collect items until downstream requests another batch up to a limit

_(source)
  .buffer(50)
  .tap(batch => {
    //batch contains all items that have flown downstream while tap was processing its last iteration (limit 50)
  });

I am willing to write this assuming it is something the project is willing to accept.

[vqvu]

This sounds reasonable, though I'd like call it something like batchUpTo(...) or batchMax(...).

buffer sounds more like collecting up to the highWaterMark, but still emitting one at a time to me. Essentially highWaterMark+write instead of highWaterMark+writev.

[stephen-dahl]

I can go with batchUpTo.

This whole thing depends on my correct understanding that for this pipeline

_(source)
  .batchUpTo(50)
  .map(batch => _(somethingAsync))
  .mergeWithLimit(1)

map would only request the next value from batchUpTo once mergeWithLimit finishes with the previous substream.

[vqvu]

Yes. That should be the case. BTW, I suggest you use flatMap(...) instead of .map(...).mergeWithLimit(1). They are equivalent, but flatMap is clearer, and potentially more efficient.

…

On Thu, Oct 18, 2018 at 9:32 AM Stephen Dahl ***@***.***> wrote: I can go with batchUpTo. This whole thing depends on my correct understanding that for this pipeline _(source) .batchUpTo(50) .tap(batch => _(somethingAsync)) .mergeWithLimit(1) tap would only request the next value from batchUpTo once mergeWithLimit finishes with the previous substream. — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#655 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AGIyZWv-nLNzKBM0Qm-j9xX8wIUJVuZWks5umJDqgaJpZM4XpLMH> .

[stephen-dahl]

Thanks for the tip. this library has too many useful things and it is easy to lose track of what tools are available.
I was looking through the source code to see if I could get started but it looks like I can't use consume as a base since I need to define my own pull function to pass downstream. can you recommend another function to base batchUpTo on or should I duplicate code and reimplement consume inside my method? If there is not another method to base my work on it seems that abstracting consume to a lower level method that consume and my function calls would be best but that is a bit invasive of a code change for my first expedition into this repo.

[vqvu]

The trick is to use consume along with create. See the implementation for latest, which is probably the closest to what you're trying to do.

The difference is that with batchUpTo, you want to apply backpressure if the buffer is filled up, whereas limit fully decouples the backpressure for child stream from the source stream. One way to do this kind of backpressure coordination is like this

var batched = [];
var resumeSource;
var resumeDest;
var pushDest;
var waitingForBatch = false;

var source = this.consume(function (err, x, ignore, next) {
  resumeSource = next;

  if (err) {
    // pushDest the error
    // resumeDest and continue this stream
  } else if (x === nil) {
    // pushDest batched if not empty.
    // pushDest nil
  } else {
    batched.push(x);
    if (waitingForBatch) {
      waitingForBatch = false;
      // pushDest batch and clear it.
      // resumeDest
    }

    // Continue this stream if batched is not full.
    // This stream will implicitly pause if the batch *is* full
});

resumeSource = source.resume.bind(source);

return source.createChild(function (push, next) {
  resumeSource();
  pushDest = push;
  resumeDest = next;

  if (buffer.length > 0) {
    // push batch and clear it.
    // continue this stream.
  } else {
    // This part signifies that the source should push the next element as soon as it has it.
    waitingForBuffer = true;
  }
});

You can also use pull and createChild. It'll probably look very similar, except instead of calling resumeSource, you call pull.

[stephen-dahl]

finally got around to working on this. I decided that this fit best as part of .batch with the signature .batch(min, max) and default max to min so that you can still get the same behavior if you call it with 1 param. I have my implementation here https://github.com/stephen-dahl/highland/commit/f2fe2fffb85f03a8efb0892469874f70609e7f2d but I am getting an infinite loop on tests and my manual test quits after the first batch. Latest didn't seem to call next in its consume so I didn't it mine but that may be what is stopping it.

I was also having trouble seeing where to extend the unit tests. I noticed that nodeunit is not being maintained anymore. is there a plan to switch to mocha or jest for the new version? I have worked with both and jest seems better to me.

[stephen-dahl]

have it working now in my own test but still getting an infinite loop and stack overflow when running grunt test

let _ = require('./lib/index');

let stream = _();
let promise = stream
    .tap(b => console.log(b))
    .batchWithTimeOrCount(1000, 5, 7)
    .tap(b => console.log(b))
    .collect()
    .tap(b => console.log(b))
    .toPromise(Promise)

async function waitRandom() {
    return new Promise((res) => {
        let deley = Math.ceil(Math.random() * Math.random() * 1000);
        setTimeout(res, deley);
    })
}

(async () => {
    let i = 0
    while(true){
        await waitRandom();
        stream.write(i++);
        if(i > 50){
            stream.end();
            break;
        }
    }

    await promise;
})()

[vqvu]

I'm not sure why you see an infinite loop, but I left some comments in your commit.

The other problem is that you can't add arguments to transforms. They get exported as a top-level transform that is curried, and adding additional arguments breaks that API. That is, you can run code like _.batchWithTimeOrCount(1000)(2)(stream).

I think merging it into batchWithTimeOrCount make sense, but we need to come up with another name. Maybe batchWithTimeOrCountRange? Not great, I know, but this is how things are.