How to "break" out of current pipe?

[nandin-borjigin]

What is the most idiomatic way to do this break operator?

const source$ = new Subject<string | number>()

source$.pipe(
  break((x: string | number): x is string => typeof x === "string"),
  map(x => x + 1), // x should be inferred as of `number` type.
  map(x => x * 2)
).subscribe((x) => console.log(x));


source$.next("this should bypass") // prints "this should bypass"
source$.next(10) // prints 22

I don't want to introduce any "nesting" of operators as I would if I use switchMap or something like that.

[voliva]

You really can't.... pipe is just taking one observable and passing it to the next operator. So these three ways are equivalent:

source$.pipe(
  map(x => x + 1),
  map(x => x * 2)
)

source$.pipe(
  map(x => x + 1)
).pipe(
  map(x => x * 2)
)

const tmp$ = source$.pipe(
  map(x => x + 1)
)
tmp$.pipe(
  map(x => x * 2)
)

How can you tell an operator that the value should be skipped over the next few operators? Those operators are just functions that take in an observable and return a new one.

You have to imagine each observable as a standalone entity. You could just:

const isString = (x: string | number): x is string => typeof x === "string";
const isNumber = (x: string | number): x is number => typeof x === "number";
merge(
  source$.pipe(filter(isString)),
  source$.pipe(
    filter(isNumber),
    map(x => x + 1),
    map(x => x * 2)
  )
)

And if you don't like nesting, then break it down into smaller observables:

const isString = (x: string | number): x is string => typeof x === "string";
const isNumber = (x: string | number): x is number => typeof x === "number";

const stringSource$ = source$.pipe(filter(isString));
const numberSource$ = source$.pipe(
  filter(isNumber),
  map(x => x + 1),
  map(x => x * 2)
);

merge(stringSource$, numberSource$)
  .subscribe(x => console.log(x));

Anything else you would need to write a custom pipe that does that.... but that's not "idiomatic".

[nandin-borjigin]

So these three ways are equivalent:
[code]

This example explains the impossibility of a solution very clearly. Thanks!

[BribeFromTheHive]

I wouldn't say a solution to this is impossible at all.

Here are a couple of options:

const source$ = new Subject<string | number>();

source$.pipe(
  filter((x): x is number => typeof x === 'number'),
  map(x => x + 1),
  map(x => x * 2)
).subscribe(x => console.log(x));

source$.next("this should bypass");  // Nothing happens
source$.next(10);  // prints 22

But my understanding is that you want to actually log 'this should bypass' and not just emit nothing in that condition. So to do that, you'd use a switchMap or mergeMap with an inner observable:

const source$ = new Subject<string | number>();

source$.pipe(
  mergeMap((x: string | number) => {
    if (typeof x === 'string') {
      return of(x);  // Bypass the following operations and emit the string as-is
    }
    return of(x).pipe(
      map(x => x + 1),
      map(x => x * 2)
    );
  })
).subscribe(x => console.log(x));

source$.next("this should bypass");  // prints "this should bypass"
source$.next(10);  // prints 22

[voliva]

@BribeFromTheHive OP asked how would you do an operator that makes some values "skip" the next operators towards the end of the pipe, and it still gets emitted, and that's impossible.

You can of course work around it by designing your observable differently, but it's literally impossible to build an operator that does that.

OP also said

I don't want to introduce any "nesting" of operators as I would if I use switchMap or something like that.

So OP already was aware of the mergeMap/switchMap solution

[BribeFromTheHive]

Thanks for that catch, I didn't see the part about the not wanting to nest stuff. There is a roundabout way to avoid callback hell, if it's that important:

const source$ = new Subject<string | number>();

source$.pipe(
  mergeMap((x: string | number) => {
    if (typeof x === 'string') {
      return throwError(x);  // Throw an error to be caught later
    }
    return of(x);
  }),
  map(x => x + 1),
  map(x => x * 2),
  catchError((x: string) => of(x))
).subscribe(
  x => console.log(x)
);

source$.next("this should bypass");  // prints "this should bypass"
source$.next(10);  // prints 22