fix(buffer): subscribe to source and closingNotifier in proper order

spec/operators/buffer-spec.ts

@@ -225,6 +225,19 @@ describe('Observable.prototype.buffer', () => {
     expectSubscriptions(b.subscriptions).toBe(bsubs);
   });
 
+  it('should work with filtered source as closingNotifier', () => {
+    const values = {0: 0, 1: 1, 2: 2, 3: 3, 4: 4, 5: 5, 6: 6, 7: 7, 8: 8};
+
+    const source =    hot('-0-1-2-3-4-5-6-7-8-|', values);
+    const expected =      '-a---b---c---d---e-|';
+
+    const expectedValues = {a: [0], b: [1, 2], c: [3, 4], d: [5, 6], e: [7, 8]};
+    const filteredSource = source.filter(x => x % 2 === 0);
+
+    const result = source.buffer(filteredSource);
+    expectObservable(result).toBe(expected, expectedValues);
+  });
+
   it('should emit last buffer if source completes', () => {
     const a =    hot('-a-b-c-d-e-f-g-h-i-|');
     const b =    hot('-----B-----B--------');

src/operator/buffer.ts

@@ -1,5 +1,6 @@
 import { Operator } from '../Operator';
 import { Subscriber } from '../Subscriber';
+import { TeardownLogic } from '../Subscription';
 import { Observable } from '../Observable';
 
 import { OuterSubscriber } from '../OuterSubscriber';
@@ -47,8 +48,11 @@ class BufferOperator<T> implements Operator<T, T[]> {
   constructor(private closingNotifier: Observable<any>) {
   }
 
-  call(subscriber: Subscriber<T[]>, source: any): any {
-    return source.subscribe(new BufferSubscriber(subscriber, this.closingNotifier));
+  call(subscriber: Subscriber<T[]>, source: any): TeardownLogic {
+    const bufferSubscriber = new BufferSubscriber(subscriber);
+    const subscription = source.subscribe(bufferSubscriber);
+    bufferSubscriber.subscribeToClosingNotifier(this.closingNotifier);
+    return subscription;
   }
 }
 
@@ -60,8 +64,11 @@ class BufferOperator<T> implements Operator<T, T[]> {
 class BufferSubscriber<T> extends OuterSubscriber<T, any> {
   private buffer: T[] = [];
 
-  constructor(destination: Subscriber<T[]>, closingNotifier: Observable<any>) {
+  constructor(destination: Subscriber<T[]>) {
     super(destination);
+  }
+
+  subscribeToClosingNotifier(closingNotifier: Observable<any>) {
     this.add(subscribeToResult(this, closingNotifier));
   }