concatMap observables, go to next on error, stop when one completes

[keithlayne]

I was surprised that it didn't seem very easy to do what I was trying to accomplish. I was able to get close enough in a very ugly way (please don't ask me to share 😳).

onErrorResumeNext is pretty close in spirit to what I want to do, just complete early if one of the inner observables completes, while emitting all the values along the way.

I looked at the onErrorResumeNext source for inspiration, but that didn't really help, at least the v7 version. I also saw some of the discussion around implementing operators which kinda seemed relevant.

Is there some easy way to do what I want? I know this is prolly pretty weird, but mumble mumble.

[demensky]

Did you want to get something like this?

// the function is just for example
export function ofError<T>(...values: T[]): Observable<T> {
  return concat(
    from(values),
    throwError(() => new Error())
  );
}

of(
  ofError('a', 'b', 'c'), // log a b c
  ofError('e', 'f', 'g'), // log e f g
  of('h', 'i', 'g'), // log h i g & complete
  ofError('k', 'j', 'm'), // ignore
)
  .pipe(
    concatMap((item$) => item$.pipe(materialize())),
    takeWhile(({ kind }) => kind !== 'C'),
    filter(({ kind }) => kind !== 'E'),
    dematerialize()
  )
  .subscribe((value) => {
    console.log(value);
  });

You can wrap it in something like:

function concatMapError<I, O>(
  project: (value: I) => Observable<O>
): OperatorFunction<I, O> {
  return pipe(
    concatMap((item) => project(item).pipe(materialize())),
    takeWhile(({ kind }) => kind !== 'C'),
    filter(({ kind }) => kind !== 'E'),
    dematerialize()
  );
}

or

function concatMapError<T, O extends ObservableInput<unknown>>(
  project: (value: T) => O
): OperatorFunction<T, ObservedValueOf<O>> {
  return pipe(
    concatMap((item) => from(project(item)).pipe(materialize())),
    takeWhile(({ kind }) => kind !== 'C'),
    filter(({ kind }) => kind !== 'E'),
    dematerialize()
  );
}

[keithlayne]

That's what I did, felt pretty gross. I figured there had to be a better way.

Also this approach seems to have a shortcoming: if you never get a completion from an inner observable, the outer will complete anyway. I think.

[demensky]

I can also suggest this solution:

const END = Symbol();

function concatMapError<I, O>(
  project: (value: I) => Observable<O>
): OperatorFunction<I, O> {
  return pipe(
    concatMap((value) =>
      project(value).pipe(
        endWith(END),
        catchError(() => EMPTY)
      )
    ),
    takeWhile((value): value is O => value !== END)
  );
}

But in my opinion materialize and dematerialize are much clearer.

---

That's what I did, felt pretty gross. I figured there had to be a better way.

You can always make a completely custom operator that will fully satisfy your specific requirements.

---

Also this approach seems to have a shortcoming: if you never get a completion from an inner observable, the outer will complete anyway. I think.

dematerialize() also emits complete if source emits complete.

of() // or EMPTY
  .pipe(dematerialize())
  .subscribe({
    complete: () => {
      console.log('dematerialize complete'); // will be displayed in the console
    },
  });

The difference is that if dematerialize() receives { kind: 'C'} it will unsubscribe from source and then emit complete.

[keithlayne]

Awesome, thanks for all your explanation. I guess I was completely unfamiliar with (de)materialize, and I guess I felt like I was abusing them somehow.

My original intent was to write a custome operator (this is a one-off use case, so meh), but that kinda seemed more complicated than I needed. I actually like the idea of ending with a sentinel, I didn't think of that. I will keep that in my back pocket :)