Observable.bufferWhen emits data before trigger if source observable completes

[Jason3S]

RxJS version:
5.1.0

Code to reproduce:

import * as Rx from 'rxjs/Rx';
import {ISuiteCallbackContext} from 'mocha';

describe('Experiment with Observable buffer', function (this: ISuiteCallbackContext) {
    this.timeout(10000);

    it('Demonstrates a bug with bufferWhen', () => {
        const src = Rx.Observable.interval(5).take(10);
        const interval = Rx.Observable.interval(1000);
        const trigger = new Rx.Subject<number>();

        interval
            .do(n => console.log('Trigger'))
            .subscribe(trigger);

        return src
            .bufferWhen(() => trigger)
            .do(v => console.log(v))
            .toArray()
            .toPromise()
            .then(values => {
                console.log(values);
                console.log('done.');
            });
        // Actual output:
        // [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ]
        // [ [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ] ]
        // done.

        // Expected output:
        // Trigger
        // [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ]
        // [ [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ] ]
        // done.
    });
});

Expected behavior:

src         ---1---2---3--|
trigger     --------------------s-------------->
bufferWhen
            --------------------[1,2,3]-|

I would expect that bufferWhen should only emit when the closingSelect (trigger) emits a value.

Actual behavior:

src         ---1---2---3--|
trigger     ----------------------------s------->
bufferWhen
            ---------------[1,2,3]-|

bufferWhen emits as soon as src is complete. That is unexpected and undesired.

Additional information:
This is an attempt at working around #2154. The issue there is that it doesn't emit at all.

[JaminFarr]

Thought I should put the workaround in this issue too where it's more relevant.

Zip with the trigger after the buffer gives the expected behaviour.

import * as Rx from 'rxjs/Rx';
import {ISuiteCallbackContext} from 'mocha';

describe('Experiment with Observable buffer', function (this: ISuiteCallbackContext) {
    this.timeout(10000);

    it('Demonstrates a bug with bufferWhen', () => {
        const src = Rx.Observable.interval(5).take(10);
        const interval = Rx.Observable.interval(1000);
        const trigger = new Rx.Subject<number>();

        interval
            .do(n => console.log('Trigger'))
            .subscribe(trigger);

        return src
            .bufferWhen(() => trigger)
            .zip(trigger, src => src)
            .do(v => console.log(v))
            .toArray()
            .toPromise()
            .then(values => {
                console.log(values);
                console.log('done.');
            });
        // Actual output:
        // Trigger
        // [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ]
        // [ [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ] ]
        // done.

        // Expected output:
        // Trigger
        // [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ]
        // [ [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ] ]
        // done.
    });
});

[Jason3S]

Nice workaround. Cool idea with the zip.

[benlesh]

Closing as stale