How to pipe series of sync. and async. tasks using RxJs Operators in an Angular service

I dont think you can return Promise<T[]>. I guess you have to return Observable<Promise<T[]>>. I wrote some code, maybe that is what you want:

public requestCanceled$: Subject<boolean>;

  public stepOneFinished$: Subject<boolean>;
  public stepTwoFinished$: Subject<boolean>;

  handleDataRequest(actionSteps: ActionStep[]): Observable<Promise<T[]>> {
    const startSTepTwo$ = merge(
      this.requestCanceled$,
      this.stepOneFinished$
    ).pipe(take(1));
    const startStepThree$ = merge(
      this.requestCanceled$,
      this.stepTwoFinished$
    ).pipe(take(1));

const actionSteps$ = from(actionSteps);

const ObservableOne = actionSteps$.pipe(
  filter((step: ActionStep) => step.taskGroup === 1), // 1 = Prep.Work
  takeUntil(this.requestCanceled$),
  map((step: ActionStep) => {
    this.prepWork(step);
    this.stepOneFinished$.next();
    this.stepOneFinished$.complete();
    return true;
  })
);

const ObservableTwo = startSTepTwo$.pipe(
  mergeMap(x =>
    actionSteps$.pipe(
      filter((step: ActionStep) => step.taskGroup === 2), // 2 = Data
      takeUntil(this.requestCanceled$),
      map((step: ActionStep) => {
        return this.getData(step);
      }),
      tap(d => {
        this.stepTwoFinished$.next();
        this.stepTwoFinished$.complete();
      })
    )
  )
);

const ObservableThree = startStepThree$.pipe(
  mergeMap(b =>
    actionSteps$.pipe(
      filter(step => step.taskGroup === 3), // 3 = Cleanup
      map((step: ActionStep) => {
        this.cleanupWork(step);
        return true;
      })
    )
  )
);

return combineLatest(ObservableOne, ObservableTwo, ObservableThree).pipe(
  map(d => d[1])
);
}

Here is one version of the solution combining Observable and Promise tools, for those who may be interested. Comments and adjustments welcome.

To test, paste the class to your Angular test project, create an instance, e.g.

const aDataRequestDemo = new DataRequestDemo();
                     // run like this
aDataRequestDemo.dataRequestTest();

Here is the code:

import { Subject, from, merge } from 'rxjs';
import { filter, take, takeUntil, tap, delay } from 'rxjs/operators';


export interface ActionStep {
   id: string,
   taskGroup: number  // 1 = Prep.Work, 2 = Data, 3 = Cleanup
   // etc.
}

const actionSteps: ActionStep[] = [
  { id: 'PrepWork1', taskGroup: 11 },
  { id: 'PrepWork2', taskGroup: 11 },
  { id: 'PrepWork3', taskGroup: 11  },
  { id: 'Data',      taskGroup: 21  },
  { id: 'Cleanup1',  taskGroup: 31   },
  { id: 'Cleanup2',  taskGroup: 31   },
]



export class DataRequestDemo {

  requestCanceled$: Subject<any> = new Subject<any>();
  requestCanceled: boolean;
  
                    /**
                     * Test mentod for a data request 
                     * from and async source 
                     * - returning a Promise<T[]> data 
                     */
  async dataRequestTest() {
    const data: string[] = 
      await this.dataRequest();
    console.log('-- RETREIVED DATA: ', JSON.stringify(data));
  }
    
                    /**
                     * Test mentod for a data request 
                     * from and async source 
                     * - returning a Promise<T[]> data 
                     */
  async dataRequest(): Promise<string[]> {
                    // run preparation tasks
    await this.prepTasks();
                    // get data from source/db
    const data = await this.getData('x');
                    // Run cleanup tasks, later, 
                    // 1st retur data to requestor.
                    // NOTE: Cleanup always runs,
                    // even if request was canceled
    this.cleanup(1000);
                    // if request got canceled, 
                    // return undefined
    if (this.requestCanceled) 
      return undefined;
                    // if request NOT canceled, 
                    // return data
    return data;
  }

  
                    /**
                     * Performs all preparation tasks
                     * outlined in the actionSteps array
                     * - is cancelable
                     */
  async prepTasks(): Promise<boolean> {
    if (this.requestCanceled)
      return undefined;
    const max = actionSteps.filter(step => step.taskGroup === 11).length;
    from(actionSteps).pipe(
                    // only select tasks of taskGroup = 11 = Prep.Work
      filter(step => step.taskGroup === 11),
                    // will unsubscribe after all tasks done
      take(max),
                    // will unsubscribe if user cancles
      takeUntil(
        merge(this.requestCanceled$)
      ),
      tap((step: ActionStep) => this.prepWork(step)),
    ).subscribe();
    return true
  }
                    /**
                     * Performs all cleanup tasks
                     * outlined in the actionSteps array.
                     * Runs delayed - after prep tasks and
                     * after retreived data sent to requestor; 
                     * - not cancelable, always runs
                     */
  private cleanup(ms: number) {
    const max = actionSteps.filter(step => step.taskGroup === 31).length;
    console.log('--cleanup max: ', max);
    from(actionSteps).pipe(
                    // delay start of the tasks by x millis
                    // (not each task is delayed, only the start)
      delay(ms),
                    // only select tasks of taskGroup = 31 / Cleanup
      filter(step => step.taskGroup === 31),
                    // will unsubscribe after all tasks done
      take(max),
                    // run the taks fn:
      tap((step: ActionStep) => this.cleanupWork(step))
    ).subscribe(v => console.log('--CLEANUP: ', v));
  }

  

                    // simulates prep. tasks
  private prepWork(param): boolean  {
    console.log('-- prepWork, param: ', param);
    return true;
  }

                    // simulates data retreival from source / db / ...
  private async getData(param): Promise<string[]> {

    if (this.requestCanceled) 
      return undefined;

    console.log('-- getData, param: ', param);

    let data: string[] = [
      'test-data-1', 
      'test-data-2'
    ];
    return data;
  }

                    // simulates cleanup tasks
  private cleanupWork(param) {
    console.log('-- cleanupWork, param: ', param);
  }

                    /**
                     * User can use this fn. to cancel
                     * the dataRequestTest() method.
                     * It is set for cancelation of both steps,
                     * sync and async 
                     */
  cancel() {
    this.requestCanceled = true;
    this.requestCanceled$.next();
    this.requestCanceled$.complete();
  }

}