RXJS Operator that checks a condition and directly emits value if true

Question

I am searching for an RXJS Operator that takes a stream of values and checks each value for a certain condition.
If this condition is true, all following operators should be ignored and the value should just be emitted as is.
If the condition is false, all following operators should be used normally.

So for example:

  const source$ = of(1, 2, 3, 4, 5);
  source$
     .pipe(
        theOperatorISearchFor((v) => v <= 4),
        map((v) => v + 100)
     )
     .subscribe((v) => console.log(v));

In this case my output would be: 1, 2, 3, 4, 105

The first 4 times my condition is true, so all following (the map operator) are ignored and the values are just emitted.
The last value does not fullfil the condition, so 105 is outputted.

Any idea?

Answer 1

There is no such operator. However, you could merge 2 different streams.

const source$ = of(1, 2, 3, 4, 5);

const [small$, high$] = partition(source$, (v) => v <= 4);

merge(small$, high$.pipe(map((v) => v + 100))).subscribe((v) => console.log(v));

Stackblitz: https://stackblitz.com/edit/rxjs-sqmj2m?file=index.ts

Answer 2

There is no such operator because it'd lack flexibility. That said, you can either write the code to do that directly yourself or build a reusable operator if you prefer.

The reason I say that it'd lack flexibility is because this operator could handle back pressure in one way only. In your example, you've got something really simple as after your custom operator, you only have a map which is a simple observable transformation. But what if you had an operator that was dealing with another observable. If the source emits faster than the rest of the chain can proceed, should your custom operator stop the following stream? Concat? Skip values in between? That's why you have to chose between switchMap, concatMap, exhaustMap, mergeMap and it has to be nested, it cannot be following your custom operator.

Here's how it'd look like:

const source$ = of(1, 2, 3, 4, 5);
source$
  .pipe(
    switchMap((v) => {
      return v <= 4 ? of(v) : of(v).pipe(map((v) => v + 100));
    })
  )
  .subscribe((v) => console.log(v));

This way, you have all the flexibility you need to deal with back pressure.