One by one observables RxJS-CodePudding

I have a problem with observables. I've prepared stackblitz to simplify my problem.

I have 2 observables (obs1$, obs2$) and array of numbers. I want to wait for obs1$ to be completed and then loop through array and return observable of each element, run obs2$.

Here is the function code:

oneByOneObservables(): Observable<unknown> {
  const obs1$ = of(1, 2, 3);
  const arr = [4, 5, 6];
  const obs2$ = of(7, 8, 9);

  return obs1$.pipe(
    concat(() => arr.map((item) => of(item))),
    () => obs2$
  );
}

I've got an error :

No overload matches this call. The last overload gave the following error. Argument of type '() => Observable[]' is not assignable to parameter of type 'SchedulerLike | ObservableInput'. Property '[Symbol.iterator]' is missing in type '() => Observable[]' but required in type 'Iterable'.

Thanks for help

CodePudding user response：

So this will do the trick. I added some logs to track it down.

Here is also Stackblitz.

import { Component } from '@angular/core';
import { ChangeDetectionStrategy } from '@angular/core';
import { Observable, of, forkJoin } from 'rxjs';
import { concatMap, tap, last } from 'rxjs/operators';

@Component({
  selector: 'my-app',
  styleUrls: ['./app.component.scss'],
  templateUrl: './app.component.html',
  changeDetection: ChangeDetectionStrategy.OnPush,
})
export class AppComponent {
  oneByOneObservables(): Observable<unknown> {
    const obs1$ = of(1, 2, 3).pipe(tap((v) => console.log('Obs1', v)));
    const arr = [4, 5, 6];
    const obs2$ = of(7, 8, 9);

    return obs1$.pipe(
      last(),
      tap((v) => console.log('Obs1 last value', v)),
      concatMap(() => forkJoin(arr.map((item) => of(item)))),
      tap((v) => console.log('Array of observables value', v)),
      concatMap(() => obs2$),
      tap((v) => console.log('Obs2 value', v))
    );
  }
}

CodePudding user response：

Heres some code:

wait for obs1$ to be completed ✓
then loop through array and return observable of each element ✓ (This is currently a noop wasting a few cpu cycles. Creating an observable doesn't do anything, do you plan to subscribe to these? In order? All at once? Up to you, I guess)
run obs2$ ✓

function oneByOneObservables(): Observable<number> {
  const obs1$ = of(1, 2, 3);
  const arr = [4, 5, 6];
  const obs2$ = of(7, 8, 9);

  return obs1$.pipe(
    concatWith(defer(() => {
      arr.map((item) => of(item))
      return obs2$
    }))
  );
}

oneByOneObservables().subscribe(console.log);

output:

An example where you subscribe to the array of observables one after another.

function oneByOneObservables(): Observable<number> {
  const obs1$ = of(1, 2, 3);
  const arr = [4, 5, 6];
  const obs2$ = of(7, 8, 9);

  return obs1$.pipe(
    concatWith(defer(() => 
      concat(...arr.map((item) => of(item)))
    )),
    concatWith(obs2$)
  );
}

oneByOneObservables().subscribe(console.log)

output:

CodePudding user response：

You can use switchMap? Though I'm not sure what output from the observables you're looking for.

oneByOneObservables(): Observable<unknown> {
  const obs1$ = of(1, 2, 3);
  const arr = [4, 5, 6];
  const obs2$ = of(7, 8, 9);

  return obs1$.pipe(
    concat(() => arr.map((item) => of(item))),
    switchMap(() => obs2$)
  );
}