fix(buffer): emit last buffer if source completes

[JaminFarr]

Description:
This fixes an issue with the bufffer operator.
If the source emits values and completes before the notifier emits, those values are lost.

source  = --a--b--c--d--e--|
notifer = ----x-----x-------
source.buffer(notifer)

results
[a]
[b, c]

d & e are never emitted

Related issue: #2154

Fix buffer operator not emitting the last buffer when the source
completes. This is closer to rxjs v4* behaviour and matches the
behaviour of the other buffer operators. The _complete method is very
similar to thoses in bufferCount, bufferTime and bufferWhen.

*rxjs v4 will always emit the buffer if the source completes even if the
buffer is empty. This fix only emits if the buffer is non empty.

BREAKING CHANGE:

extra value may be emitted when source completes

[coveralls]

Coverage increased (+0.002%) to 97.688% when pulling 25307d0 on JaminFarr:fix-buffer-last-array-missing into 89b506d on ReactiveX:master.

[mpodlasin]

@JaminFarr does in this branch last buffer is emited also when notifier observable completes? (for context see: #1754)

[benlesh]

We'll need to verify that this is how Rx4 behaved. As it stands right now this would be more of a breaking change than a "fix", I think. But if it behaved this way in Rx4, there's an argument to be made that this is a fix.

[benlesh]

@JaminFarr confirmed that Rx4 behaves this way and Rx5 does not... therefor this is a bug fix. 👍

[benlesh]

This LGTM... Will merge after conferring with others. Pretty sure this should land in 5.0.2 as a bug fix though.

[JaminFarr]

@Podlas29 No, This PR only changes the source complete behavior.

Examples from #1754

Observable.never().buffer(Observable.empty());

// Result: Same, emits nothing. Source never completes so it is not affected by this fix. 

const source = Observable.from([1,2,3,4,5,6]);
source.buffer(source.take(2));

// Result:
[] // Same
[] // Same
[1, 2, 3, 4, 5, 6] // new from this change
// The notifier emits twice and completes before source is subscribed to, producing
// the first two empty array results. Then the source completes and the array of the
// remaining values is emitted.

@Blesh Great.

[jayphelps]

Merged into next branch, as we decided in our core team meeting today this was indeed a breaking change that has to go into a major version bump.

[jayphelps]

Thanks @JaminFarr !!

[Jason3S]

@JaminFarr What is the behavior of this fix? Does it emit immediately or wait until the closingNotifier. Please see #2361

[JaminFarr]

@Jason3S This fix emits immediately when the source completes with the remaining values. This matches RxJS4 but looks like the opposite behavior to what you're after. I don't think there is a version of BufferWhen in RxJS4 to compare against.

As a workaround to delay the last buffer value to emit in time with the trigger, you can zip the result with the trigger.

const Observable = Rx.Observable;

const source$ = Observable.interval(200).take(25)
const trigger$ = Observable.interval(2000).share()

// I'm using window and mergeMap as a workaround until this PR is merged.
// This is the same as .buffer(trigger$) post fix.
source$
  .window(trigger$)
  .mergeMap(x => x.toArray())

// Use zip to delay the last value and a selector to remove the trigger value.
  .zip(trigger$, (sourceValue /*, triggerValue*/) => sourceValue)

  .subscribe(x => console.log(x))

/*
2s: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
4s: [10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
6s: [20, 21, 22, 23, 24] -- without the zip this will emit at ~5s.
*/

[Jason3S]

Nice workaround.

[lock]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.