fix(forkJoin): now finalizes sources before emitting (#6546)

fixes #4914

Author: benlesh

spec/observables/forkJoin-spec.ts

@@ -1,6 +1,6 @@
 /** @prettier */
 import { expect } from 'chai';
-import { forkJoin, of } from 'rxjs';
+import { finalize, forkJoin, map, of, timer } from 'rxjs';
 import { lowerCaseO } from '../helpers/test-helper';
 import { TestScheduler } from 'rxjs/testing';
 import { observableMatcher } from '../helpers/observableMatcher';
@@ -303,6 +303,22 @@ describe('forkJoin', () => {
     });
   });
 
+  it('should finalize in the proper order', () => {
+    const results: any[] = [];
+    const source = forkJoin(
+      [1, 2, 3, 4].map((n) =>
+        timer(100, rxTestScheduler).pipe(
+          map(() => n),
+          finalize(() => results.push(`finalized ${n}`))
+        )
+      )
+    );
+
+    source.subscribe((value) => results.push(value));
+    rxTestScheduler.flush();
+    expect(results).to.deep.equal(['finalized 1', 'finalized 2', 'finalized 3', 'finalized 4', [1, 2, 3, 4]]);
+  });
+
   describe('forkJoin({ foo, bar, baz })', () => {
     it('should join the last values of the provided observables into an array', () => {
       rxTestScheduler.run(({ hot, expectObservable }) => {

src/internal/observable/forkJoin.ts

@@ -166,8 +166,10 @@ export function forkJoin(...args: any[]): Observable<any> {
             }
             values[sourceIndex] = value;
           },
+          () => remainingCompletions--,
+          undefined,
           () => {
-            if (!--remainingCompletions || !hasValue) {
+            if (!remainingCompletions || !hasValue) {
               if (!remainingEmissions) {
                 subscriber.next(keys ? createObject(keys, values) : values);
               }