fix(mergeWith): works correctly with an Array (#7281)

* fix(mergeWith): works correctly with an Array

* test: just use a synchronous test to ensure it doesn't emit more or less than it's supposed to (including emitting undefineds).

* test: fix `mergeWith` test

---------

Co-authored-by: Ben Lesh <ben@benlesh.com>

Author: josepot

spec/operators/mergeWith-spec.ts

@@ -46,6 +46,24 @@ describe('merge operator', () => {
     });
   });
 
+  it('should merge a source with a second, when the second is just a plain array', (done) => {
+    const a = of(1, 2, 3);
+    const b = [4, 5, 6, 7, 8];
+    const r = [1, 2, 3, 4, 5, 6, 7, 8];
+
+    a.pipe(mergeWith(b)).subscribe({
+      next: (val) => {
+        expect(val).to.equal(r.shift());
+      },
+      error: () => {
+        done(new Error('should not be called'));
+      },
+      complete: () => {
+        done();
+      },
+    });
+  });
+
   it('should merge cold and cold', () => {
     rxTestScheduler.run(({ cold, expectObservable, expectSubscriptions }) => {
       const e1 = cold(' ---a-----b-----c----|');

src/internal/operators/merge.ts

@@ -1,6 +1,5 @@
 import { ObservableInput, ObservableInputTuple, OperatorFunction, SchedulerLike } from '../types';
 import { operate } from '../util/lift';
-import { argsOrArgArray } from '../util/argsOrArgArray';
 import { mergeAll } from './mergeAll';
 import { popNumber, popScheduler } from '../util/args';
 import { from } from '../observable/from';
@@ -23,7 +22,6 @@ export function merge<T, A extends readonly unknown[]>(
 export function merge<T>(...args: unknown[]): OperatorFunction<T, unknown> {
   const scheduler = popScheduler(args);
   const concurrent = popNumber(args, Infinity);
-  args = argsOrArgArray(args);
 
   return operate((source, subscriber) => {
     mergeAll(concurrent)(from([source, ...(args as ObservableInput<T>[])], scheduler)).subscribe(subscriber);