fix(buffer): subscribe to the closing notifier before the source

spec/operators/buffer-spec.ts

@@ -317,11 +317,11 @@ describe('Observable.prototype.buffer', () => {
     });
 
     subject.next(1);
-    expect(results).to.deep.equal([[1]]);
+    expect(results).to.deep.equal([[]]);
     subject.next(2);
-    expect(results).to.deep.equal([[1], [2]]);
+    expect(results).to.deep.equal([[], [1]]);
     subject.complete();
-    expect(results).to.deep.equal([[1], [2], [], 'complete']);
+    expect(results).to.deep.equal([[], [1], [2], 'complete']);
   });
 
   it('should buffer when Promise resolves', (done) => {

src/internal/operators/buffer.ts

@@ -47,19 +47,7 @@ export function buffer<T>(closingNotifier: ObservableInput<any>): OperatorFuncti
     // The current buffered values.
     let currentBuffer: T[] = [];
 
-    // Subscribe to our source.
-    source.subscribe(
-      createOperatorSubscriber(
-        subscriber,
-        (value) => currentBuffer.push(value),
-        () => {
-          subscriber.next(currentBuffer);
-          subscriber.complete();
-        }
-      )
-    );
-
-    // Subscribe to the closing notifier.
+    // Subscribe to the closing notifier first.
     from(closingNotifier).subscribe(
       createOperatorSubscriber(
         subscriber,
@@ -73,6 +61,18 @@ export function buffer<T>(closingNotifier: ObservableInput<any>): OperatorFuncti
       )
     );
 
+    // Subscribe to our source.
+    source.subscribe(
+      createOperatorSubscriber(
+        subscriber,
+        (value) => currentBuffer.push(value),
+        () => {
+          subscriber.next(currentBuffer);
+          subscriber.complete();
+        }
+      )
+    );
+
     return () => {
       // Ensure buffered values are released on finalization.
       currentBuffer = null!;