fix(withlatestfrom): allow synchronous source (#5828)

spec/operators/withLatestFrom-spec.ts

@@ -260,4 +260,16 @@ describe('withLatestFrom operator', () => {
         expect(x).to.deep.equal([1, 4, 6]);
       });
   });
+
+  it('should work with synchronous observables', () => {
+    const result: Array<Array<number>> = [];
+    of(1, 2, 3).pipe(withLatestFrom(of(4, 5))).subscribe((x) => {
+      result.push(x);
+    });
+
+    expect(result.length).to.equal(3);
+    expect(result[0]).to.deep.equal([1,5]);
+    expect(result[1]).to.deep.equal([2,5]);
+    expect(result[2]).to.deep.equal([3,5]);
+  })
 });

src/internal/operators/withLatestFrom.ts

@@ -159,17 +159,6 @@ export function withLatestFrom<T, R>(...inputs: any[]): OperatorFunction<T, R |
     // we are ready to start emitting values.
     let ready = false;
 
-    // Source subscription
-    source.subscribe(
-      new OperatorSubscriber(subscriber, (value) => {
-        if (ready) {
-          // We have at least one value from the other sources. Go ahead and emit.
-          const values = [value, ...otherValues];
-          subscriber.next(project ? project(...values) : values);
-        }
-      })
-    );
-
     // Other sources. Note that here we are not checking `subscriber.closed`,
     // this causes all inputs to be subscribed to, even if nothing can be emitted
     // from them. This is an important distinction because subscription constitutes
@@ -197,5 +186,16 @@ export function withLatestFrom<T, R>(...inputs: any[]): OperatorFunction<T, R |
         )
       );
     }
+
+    // Source subscription
+    source.subscribe(
+      new OperatorSubscriber(subscriber, (value) => {
+        if (ready) {
+          // We have at least one value from the other sources. Go ahead and emit.
+          const values = [value, ...otherValues];
+          subscriber.next(project ? project(...values) : values);
+        }
+      })
+    );
   });
 }