Conditionally apply pipe operators

user_1002229 0
Viewed 3697

I need a timeout operator to throw an error when parameter isTimeoutNeeded is True, so that I could handle the case if the subscribed observable do not emit any value after X seconds.

if (isTimeoutNeeded) {
    this.service.getData()
      .pipe(
          timeout(10000),
          ...
      )
      .subscribe((...) => ...);
} else {
    this.service.getData()
    .pipe( 
       ... 
    )    
    .subscribe((...) => ...);
}

Is it possible to write this in a rxjs way by conditionally adding timeout operator when needed?

I have tried using iif and switchMap but it did not work.

 this.service.getData()
    .pipe(
       mergeMap((response) =>
          iif(() => !isTimeoutNeeded,
          of(response),
          of(response).pipe(timeout(10000)) // did not apply timeout if isTimeoutNeeded = True
       )
    )
    .subscribe((...) => ...);

switchMap

 this.service.getData()
    .pipe(
           switchMap((response) => {
                if (!isTimeoutNeeded) {
                    return of(response);
                }
                return of(response).timeout(10000); // did not apply timeout if isTimeoutNeeded = True
            })
            .....
       )
    )
    .subscribe((...) => ...);
rxjs angular
4 Answers

RXJS works best when the operators easily describe the behavior without detailing the implementation. When you start using switchMap() and conditions to express a behavior the intent gets lost, and the code becomes difficult to maintain.

So just write your own operator.

function timeoutWhen<T>(cond: boolean, value: number): OperatorFunction<T, T> {
   return function(source: Observable<T>): Observable<T> {
       return cond ? source.pipe(timeout(value)) : source;
   }
}

Now when you use that operator the source code is easy to read and understand.

    this.service.getData().pipe(
       timeoutWhen(isTimeoutNeeded, 10000),
       ...
    ).subscribe((...) => ...);

Now you have something that is reusable and makes your observables easier to read and understand.

user_10315690

For this particular case, I did it like this:

.pipe(
  (isTimeoutNeeded? interval(5000) : identity)
)
user_30461320

You were almost right using switchMap:

const subject = new BehaviorSubject(this.service.getData());

subject
  .pipe(
    switchMap(response => {
      // If isTimeoutNeeded, a timetout is applied else wait result forever.
      return isTimeoutNeeded
        ? response.pipe(timeout(10000))
        : response;
    }),
    ...
  )
  .subscribe(... => ...);

In this code the timeout is applied on the getData observable whereas in your example it was a new observable.

Check my Stackblitz example.

user_33714400

You can use the RxJs pipeable operator delayWhen:

Delay emitted values determined by provided function.

Have it return an interval() observable like

.pipe(
  delayWhen(() => isTimeoutNeeded ? interval(5000) ? interval(0)
)
user_45695060
Related