fix(buffer): emit last buffer if source completes

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 maybe emitted when source completes

spec/operators/buffer-spec.ts

@@ -219,4 +219,17 @@ describe('Observable.prototype.buffer', () => {
     expectObservable(a.buffer(b).take(1)).toBe(expected, expectedValues);
     expectSubscriptions(b.subscriptions).toBe(bsubs);
   });
+
+
+  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--------');
+    const expected = '-----x-----y-------(z|)';
+    const expectedValues = {
+      x: ['a', 'b', 'c'],
+      y: ['d', 'e', 'f'],
+      z: ['g', 'h', 'i']
+    };
+    expectObservable(a.buffer(b)).toBe(expected, expectedValues);
+  });
 });

src/operator/buffer.ts

@@ -69,6 +69,14 @@ class BufferSubscriber<T> extends OuterSubscriber<T, any> {
     this.buffer.push(value);
   }
 
+  protected _complete() {
+    const buffer = this.buffer;
+    if (buffer.length > 0) {
+      this.destination.next(buffer);
+    }
+    super._complete();
+  }
+
   notifyNext(outerValue: T, innerValue: any,
              outerIndex: number, innerIndex: number,
              innerSub: InnerSubscriber<T, any>): void {