fix(buffer): closingNotifier completion does not complete resulting o…

…bservable

Resolves an issue where the resulting observable would complete when the closingNotifier completed. Notifier completion should not complete the result, only source completion should do that.

BREAKING CHANGE: closingNotifier no longer closes the result of `buffer`. If that is truly a desired behavior, then you should use `takeUntil`. Something like: `source$.pipe(buffer(notifier$), takeUntil(notifier$.pipe(ignoreElements(), endWith(true))))`, where `notifier$` is multicast, although there are many ways to compose this behavior.

spec/operators/buffer-spec.ts

@@ -49,7 +49,7 @@ describe('Observable.prototype.buffer', () => {
     testScheduler.run(({ hot, expectObservable }) => {
       const a = hot('--1--2--^--3--4--5---6----7--8--9---0---|');
       const b = EMPTY;
-      const expected = '|';
+      const expected = '     --------------------------------|';
       expectObservable(a.pipe(buffer(b))).toBe(expected);
     });
   });
@@ -67,7 +67,7 @@ describe('Observable.prototype.buffer', () => {
     testScheduler.run(({ expectObservable }) => {
       const a = NEVER;
       const b = EMPTY;
-      const expected = '|';
+      const expected = '-';
       expectObservable(a.pipe(buffer(b))).toBe(expected);
     });
   });
@@ -139,9 +139,9 @@ describe('Observable.prototype.buffer', () => {
     // Buffshoulder Boundaries onCompletedBoundaries (RxJS 4)
     testScheduler.run(({ hot, expectObservable, expectSubscriptions }) => {
       const a = hot('--1--2--^--3--4--5---6----7--8--9---0---|');
-      const subs = '         ^----------------!               ';
+      const subs = '         ^-------------------------------!';
       const b = hot('--------^--a-------b---cd|               ');
-      const expected = '     ---a-------b---cd|               ';
+      const expected = '     ---a-------b---cd---------------|';
       const expectedValues = {
         a: ['3'],
         b: ['4', '5'],

src/internal/operators/buffer.ts

@@ -1,6 +1,7 @@
 import { Observable } from '../Observable';
 import { OperatorFunction } from '../types';
 import { operate } from '../util/lift';
+import { noop } from '../util/noop';
 import { OperatorSubscriber } from './OperatorSubscriber';
 
 /**
@@ -50,12 +51,19 @@ export function buffer<T>(closingNotifier: Observable<any>): OperatorFunction<T,
 
     // Subscribe to the closing notifier.
     closingNotifier.subscribe(
-      new OperatorSubscriber(subscriber, () => {
-        // Start a new buffer and emit the previous one.
-        const b = currentBuffer;
-        currentBuffer = [];
-        subscriber.next(b);
-      })
+      new OperatorSubscriber(
+        subscriber,
+        () => {
+          // Start a new buffer and emit the previous one.
+          const b = currentBuffer;
+          currentBuffer = [];
+          subscriber.next(b);
+        },
+        // Pass all errors to the consumer.
+        undefined,
+        // Closing notifier should not complete the resulting observable.
+        noop
+      )
     );
 
     return () => {