mergeFrom

mergeFrom is a helper function that merges the values of Observables or Signals and emits the latest emitted value. It also gives us the possibility to change the emitted value before emitting it using RxJS operators.

It is similar to merge(), but it also takes Signals into consideration.

From ngxtension perspective, mergeFrom is similar to derivedFrom, but it doesn’t emit the combined value, but the latest emitted value by using the merge operator instead of combineLatest.

import { mergeFrom } from 'ngxtension/merge-from';

Usage

mergeFrom accepts an array of Observables or Signals and returns a Signal that emits the latest value of the Observables or Signals. By default, it needs to be called in an injection context, but it can also be called outside of it by passing the Injector in the third argument options object. If your Observable doesn’t emit synchronously, you can use the startWith operator to change the starting value, or pass an initialValue in the third argument options object.

const a = signal(1);
const b$ = new BehaviorSubject(2);

// array type
const merged = mergeFrom([a, b$]);
// both sources are sync, so emits the last emitted value
console.log(merged()); // 2

It can be used in multiple ways:

1. Merge multiple Signals
2. Merge multiple Observables
3. Merge multiple Signals and Observables
4. Using initialValue param
5. Use it outside of an injection context

1. Merge multiple Signals

We can use mergeFrom to merge multiple Signals into one Signal, which will emit last value of the Signals.

const runnerOne = signal({ name: 'Naruto' });
const runnerTwo = signal({ name: 'Goku' });

const merged = mergeFrom([runnerOne, runnerTwo]);

console.log(merged()); // [1, { name: 'Goku' }]

As we said before, mergeFrom allows us to change the last value before emitting it using rxjs operators (applying asynchronous operations).

import { mergeFrom } from 'ngxtension/merge-from';
import { delay, of, pipe, map } from 'rxjs';

const runnerOne = signal({ name: 'Naruto', time: '12:43PM' });
const runnerTwo = signal({ name: 'Goku', time: '12:44PM' });

const lastRunner = mergeFrom(
  [runnerOne, runnerTwo],
  pipe(
    switchMap((runner) => {
      return of('The last runner to arive was:  ' + runner.name).pipe(
        delay(1000),
      );
    }),
  ),
);
effect(() => console.log(lastRunner())); // 👈 will throw an error!! 💥

setTimeout(() => {
  runnerOne.update((runner) => {
    return {
      ...runner,
      time: '12:45PM',
    };
  });
}, 3000);

// You can copy the above example inside an Angular constructor and see the result in the console.

This will throw an error because the operation pipeline will produce an observable that will not have a sync value because they emit their values later on, so the resulting lastRunner signal doesn’t have an initial value, and this causes the error.

You can solve this by using the initialValue param in the third argument options object, to define the starting value of the resulting Signal and prevent throwing an error in case of real async observable.

const lastRunner = mergeFrom(
  [runnerOne, runnerTwo],
  pipe(
    switchMap((runner) => {
      return of('The last runner to arive was:  ' + runner.name).pipe(
        delay(1000),
      );
    }),
  ),
  { initialValue: 'Waiting for someone to arrive...' },
);

This works, and you can copy the above example inside a component constructor and see the result in the console:

  Waiting for someone to arrive...      - // initial value passed as third argument
    The last runner to arive was:  Goku   - // last value after 1 second
    The last runner to arive was:  Naruto - // last value after 3 seconds

Another way to solve this problem is using the startWith rxjs operator in the pipe to force the observable to have a starting value like below.

const lastRunner = mergeFrom(
  [runnerOne, runnerTwo],
  pipe(
    switchMap((runner) => {
      return of('The last runner to arive was:  ' + runner.name).pipe(
        delay(1000),
      );
    }),
    startWith('Waiting for someone to arrive...'),
  ),
);

The console log will be:

  Waiting for someone to arrive...      - // initial value passed as third argument
    The last runner to arive was:  Goku   - // last value after 1 second
    The last runner to arive was:  Naruto - // last value after 3 seconds

2. Combine multiple Observables

We can use mergeFrom to combine multiple Observables into one Signal, which will emit the combined value of the Observables.

const runnerOne$ = new BehaviorSubject({ name: 'Naruto', time: '12:43PM' });
const runnerTwo$ = new BehaviorSubject({ name: 'Goku', time: '12:44PM' });

const lastRunner = mergeFrom([runnerOne, runnerTwo]);

console.log(lastRunner()); // { name: 'Goku', time: '12:44PM' }

This is just a better version of:

const lastRunner = toSignal(merge([runnerOne$, runnerTwo$]));

And it can be used in the same way as in the previous example with rxjs operators.

3. Combine multiple Signals and Observables

We can use it to combine multiple Signals and Observables into one Signal.

const runnerOne = signal({ name: 'Naruto', time: '12:43PM' });
const runnerTwo$ = new BehaviorSubject({ name: 'Goku', time: '12:44PM' });

const lastRunner = mergeFrom([runnerOne, runnerTwo$]);

console.log(lastRunner()); // {name: 'Goku', time: '12:44PM'}

Tricky part

If all Observables don’t emit any initial value, mergeFrom will throw an error forcing you to fix this situation either by passing an initialValue in the third argument, or using startWith operator in one of the observables to force it to have a sync starting value.

const runnerOne$ = new Subject<number>();
const runnerTwo$ = new Subject<number>();

const lastRunner = mergeFrom([runnerOne$, runnerTwo$]);

console.log(lastRunner()); // 👈 will throw an error!! 💥

Using startWith operator aproach to change the initial value.

const lastRunner = mergeFrom([
  runnerOne$.pipe(startWith({ name: 'Naruto', time: '12:43PM' })),
  runnerTwo$,
]);

console.log(lastRunner()); // {name: 'Naruto', time: '12:43PM'}

Or, we can set initialValue as the third argument operator to change the initial value.

const lastRunner = mergeFrom([runnerOne$, runnerTwo$], undefined, {
  initialValue: { name: 'Naruto', time: '12:43PM' },
});

console.log(lastRunner()); // {name: 'Naruto', time: '12:43PM'}

4. Using initialValue param

Or, you can set the initialValue as the third argument in the options object, to define the starting value of the resulting Signal and prevent throwing error in case of observables that emit later.

const lastRunner = mergeFrom([runnerOne$, runnerTwo$], undefined, {
  initialValue: { name: 'Naruto', time: '12:43PM' },
});

console.log(lastRunner()); // {name: 'Naruto', time: '12:43PM'}

5. Use it outside of an injection context

By default, mergeFrom needs to be called in an injection context, but it can also be called outside of it by passing the Injector in the third argument options object.

@Component()
export class MyComponent implements OnInit {
  readonly injector = inject(Injector);
  ngOnInit(): void {
    const runnerTwo$ = new Subject<number>();
    const runnerOne$ = new Subject<number>();

    const lastRunner = mergeFrom([runnerOne$, runnerTwo$], undefined, {
      initialValue: { name: 'Naruto', time: '12:43PM' },
      injector: this.injector,
    });

    console.log(lastRunner()); // {name: 'Naruto', time: '12:43PM'}
  }
}