Code within RxJS pipe() not Being Executed

Question

For the following code:

  public getExecution(modelId: string, reportId: string, params: ParameterSet | null): Observable<ExecutionToken> {

    return this.getNewExecution(modelId, reportId, params, null)
      .pipe(
        mergeMap((token: ExecutionToken) => {
          return this.reportExecutionHubService.assignExecutionToClient(token).then(resolve => token);
        })
      )
      .pipe(
        tap((token: ExecutionToken) => this.updateExecutionTokenState(token))
      );
  }

I put a breakpoint on the line with the mergeMap(); but the breakpoint was never hit.

The breakpoint on the tap(...) line was hit; but this.updateExecutionTokenState(token)) was not called - I put a breakpoint on the first line of private updateExecutionTokenState(token: ExecutionToken): void { and that breakpoint was never hit.

  private updateExecutionTokenState(token: ExecutionToken): void {
    this.store.dispatch(new executionTokenActions.UpdateExecutionToken({ // breakpoint on this line
      reportId: token.reportId,
      executionSessionId: token.executionSessionId,
      executionRequestId: token.executionRequestId,
      modelRevisionId: token.modelRevisionId,
      dateReceived: token.dateReceived
    }));
  }

Likewise, for the following code:

  public get(payload: ReportExecutionRequest): Observable<ExecutionToken> {
    return this.apiService.postWithReturn<ReportExecutionRequestResponse>('reportExecution/', payload)
      .pipe(
        catchError((response: HttpErrorResponse): Observable<any> => {
          if (response.error?.code === this.SESSION_TIMEOUT_ERROR_CODE) {
            // this.store.dispatch(new modelActions.SetPageErrored(true));
            this.store.dispatch(new executionTokenActions.ExpireCurrentReportExecutionToken());

            // swallow the error
            return of();
          }

          return throwError(response);
        }),
        map<ReportExecutionRequestResponse, ExecutionToken>((response) => {
          if (payload.executionSessionId !== response.executionSessionId) {
            this.toastrService.info('toast_info_preparing_execution_session');
          }

          const token: ExecutionToken = {
            reportId: payload.targetSet,
            dateReceived: new Date(),
            executionSessionId: response.executionSessionId,
            executionRequestId: response.executionRequestId,
            modelRevisionId: payload.modelRevisionId
          };

          // this.reportExecutionLocalStorageService.add(token);
          this.store.dispatch(new executionTokenActions.AddExecutionToken(token));

          return token;
        })
      );
  }

I placed a breakpoint on the line:

if (payload.executionSessionId !== response.executionSessionId) {

which is just inside the map(), but again, the breakpoint was never hit.

How I am subscribing to the observable:

    this.reportExecutionFacadeService.getExecution(this.modelId, this.reportId, null).subscribe((token) => {
      console.log(token);
    });

I am using Chrome developer tools debugger if that helps - please let me know what I am not understanding here. Thanks.

Answer 1

If b breakpoint on the mergeMap is never hit, that means that your function this.getNewExecution in turn, is not emitting any value.

Make sure that the function this.getNewExecution returns any Observable type (Subject, BehaviorSubject, etc.) and that the observable emits a value.

Only when the Observable returned by this.getNewExecution emits, only then mergeMap inside the pipe would run.

BTW, you can write your code like so (no need to chain pipe twice):

public getExecution(modelId: string, reportId: string, params: ParameterSet | null): Observable<ExecutionToken> {

    return this.getNewExecution(modelId, reportId, params, null)
        .pipe(
            mergeMap((token: ExecutionToken) => {
                return this
                    .reportExecutionHubService
                    .assignExecutionToClient(token)
                    .then(resolve => token);
            }),

            tap((token: ExecutionToken) => this.updateExecutionTokenState(token))
        )
}

BTW2: It looks like assignExecutionToClient returns a promise. If you're looking to take the result resolved by the promise and merge it with the getNewExecution stream, its better to turn it into an observable like so (using from from 'rxjs'):

public getExecution(modelId: string, reportId: string, params: ParameterSet | null): Observable<ExecutionToken> {

    return this.getNewExecution(modelId, reportId, params, null)
        .pipe(
            mergeMap((token: ExecutionToken) => {
                return from(this
                    .reportExecutionHubService
                    .assignExecutionToClient(token)
                );
            }),

            tap((token: ExecutionToken) => this.updateExecutionTokenState(token))
        )
}

And of course, for the whole thing to actually run, you must subscribe to the returned observable. That is, in the calling function, you must write:

    someCallingFunction() {
        this.getExecution(modelId, reportId, params)
            .subscribe(token => console.log(token));
    }

Observables pipe will not run until they are subscribed to. So not subscribing to getExecution will not trigger the whole thing.