fix(buffer): do not emit empty buffer when completes

- do not emit empty buffer when observable completes
- update test spec to use marble test scheduler

spec/operators/buffer-spec.js

@@ -1,19 +1,12 @@
-/* globals describe, it, expect */
+/* globals describe, it, expect, expectObservable, hot */
 var Rx = require('../../dist/cjs/Rx');
-var Observable = Rx.Observable;
 
 describe('Observable.prototype.buffer', function () {
-  it('should emit buffers that close and reopen', function (done) {
-    var expected = [
-      [0, 1, 2],
-      [3, 4, 5],
-      [6, 7, 8]
-    ];
-    Observable.interval(100)
-      .buffer(Observable.interval(320))
-      .take(3)
-      .subscribe(function (w) {
-        expect(w).toEqual(expected.shift())
-      }, null, done);
-  }, 2000);
+  it('should emit buffers that close and reopen', function () {
+    var e1 =       hot('-a-b-c-d-e-f-g-h-i-|');
+    var expected =     '-----x-----y-----z-|';
+    var interval = hot('-----1-----2-----3-|');
+
+    expectObservable(e1.buffer(interval)).toBe(expected, {x: ['a','b','c'], y: ['d','e','f'], z: ['g','h','i']});
+  });
 });

src/operators/buffer.ts

@@ -53,7 +53,10 @@ class BufferSubscriber<T> extends Subscriber<T> {
   flushBuffer() {
     const buffer = this.buffer;
     this.buffer = [];
-    this.destination.next(buffer);
+
+    if (buffer.length > 0) {
+      this.destination.next(buffer);
+    }
   }
 }