takeUntil

[ccorcos]

Hey, I thought I'd add some of the functions you've been helping me to this library.

Some of the tests fail with testling, but they seem to be outside the code I wrote. There are also some linting errors, but also outside the code I wrote. Anyways, let me know what you think.

[ccorcos]

Also, Ive never actually used Travis CI. Would you mind giving me your 2 cents? I seems it just runs tests on each pull request?

[vqvu]

Thanks for this @ccorcos. Travis CI will run tests on your branch whenever you submit a pull request or push updates to your PR.

Anyway, I've changed npm test to also run eslint and fixed the browser error. Can you do a rebase on top of master?

[vqvu]

Don't call resume on either stream. You shouldn't need to. Just move the c.toArray above the write calls.

[ccorcos]

then to array should call twice right? the first time it will be be [1] and the second will be [1,2] right?

[vqvu]

Nope. The callback will only be called once. toArray will wait until the source stream (c in this case) ends before emitting everything it saw.

[vqvu]

Also, can you add

1. A setTimeout-based test. Use sinon. See the ratelimit for an example of how to set that up. We want to check that it works with a regular generator too.
2. Add a noValueOnError test. Like this. This will check that the transform doesn't forward errors as values.

[ccorcos]

hmm. this isn't working...

    'async generator': function (test) {
        function delay(push, ms, x) {
            setTimeout(function () {
                push(null, x);
            }, ms);
        }
        var source = _(function (push, next) {
            delay(push, 10, 1);
            delay(push, 20, 2);
            delay(push, 30, 3);
            // should be stopped
            delay(push, 40, 4);
            delay(push, 50, 5);
            delay(push, 60, _.nil);
        })
        var stopStream = _(function (push, next) {
            delay(push, 35, 1);
            delay(push, 45, _.nil);
        })
        var results = [];
        source.takeUntil(stopStream).each(function (x) {
            results.push(x);
        });
        this.clock.tick(10);
        test.same(results, [1]);
        this.clock.tick(10);
        test.same(results, [1, 2]);
        this.clock.tick(10);
        test.same(results, [1, 2, 3]);
        this.clock.tick(10);
        test.same(results, [1, 2, 3]);
        this.clock.tick(20);
        test.same(results, [1, 2, 3]);
        test.done();
    },

[ccorcos]

it worked when I tried with 20...

        var stopStream = _(function (push, next) {
            delay(push, 20, 1);
            delay(push, 45, _.nil);
        })

[ccorcos]

ahh, probably because stop isn't started until you get the first value from the source...

that make sense

[ccorcos]

Oy. That rebase in in the wrong spot... Any ideas how to fix that?

[ccorcos]

P.S. If you want to check out what I've been working on:

https://github.com/ccorcos/meteor-react-highland

I threw you in the acknowledgements at the bottom. Thanks for all the help!

And some discussion here:

https://crater.io/posts/JBsraJN8rMpf47ahu

[vqvu]

That rebase in in the wrong spot... Any ideas how to fix that?

Try

$ git reset --hard c33dd7e1ac898346cdd618539d3040bff4141975
$ git push -f

ahh, probably because stop isn't started until you get the first value from the source...

This is a good point, and I can see this tripping people up. Is this reasonable behavior? For instance, a usecase for this transform may be as a timeout: "wait 30 seconds for stream to emit something and stop it if it takes too long." In this case, you can't start consuming from the selector (i.e., the stopping stream) when the first value gets emitted, so the current behavior doesn't seem very reasonable.

There are two ways to do this:

1. Consume from the selector immediately, when takeUntil is called.
2. Consume from the selector when downstream first request for data.

I believe Rx's takeUntil does the latter, so we should probably do the same. It also fits with our laziness model.

Another question: what happens if we pause after a resume? Do we continue consuming from the source and buffer the output? Or do we pause the source? Buffering isn't idea, but if we pause the source, then are we OK with the following situation?

var src = _(function (push, next) {
    delay(push, 1, 10); // push 1 after 20 ms.
    delay(push, 2, 20);
    delay(push, _.nil, 30);
});

var selector = _(function (push, next) {
    delay(push, _.nil, 25);
});

src.takeUntil(selector).toArray(_.log);
// => [1, 2]

src.takeUntil(selector).consume(function (err, x, push, next) {
    push(err, x);
    if (x !== _.nil) {
        // This effectively pauses the source for 20 ms.
        setTimeout(next, 20);
    }
}).toArray(_.log);
// => [1]

We'd get different results because the src never got the chance to emit 2 because selector already finished in the time that we were paused. A buffering scheme would result in the same values in both cases. On the other hand, this this is at its core a time-dependent transform, does it really matter that much that attaching subsequent time-dependent transforms changes the result? Plus, it wouldn't be very lazy.

This is more of a problem than I thought...

@ccorcos, what do you think?

Highland collaborator folks (if you see this), any thoughts?

[svozza]

Instinctively I'd lean towards a solution that gives a guaranteed result in all cases but it's a good point that this compromises our laziness and with a particularly fast stream that buffer could become quite large quite quickly. As long as we detail the behaviour clearly in the docs, I don't think it's much of a problem.

[vqvu]

Makes sense. Plus, you can always retroactively buffer, but you can't take back laziness.

@ccorcos Do you think you can make the appropriate changes? I can take over if you don't have the time.

[ccorcos]

Hey, I fixed the rebase... Not sure why that worked but it did!

[ccorcos]

I think takeUntil makes sense as is. In terms of buffering, I'm not sure how you'd do that but it makes sense as well