Get current Item in catchError

[victor-perez]

In short I get a event where the data contains a ID and each ID needs to be process by some internal code that sometimes can throw an error. In the catchError I want to start a new observable. To handle the error correctly, but for that I need to the know what was the item that gave the error. I can't find anywhere how to solve this.

I'm using RxJS 7.x

/// ---> 1 ---- 2 --- 1283 --- 8 --- etc
fromEvent(eventEmitter, 'data')
.pipe(
  mergeMap(async (id) => await someInternalCode (id)),
  catchError((e) => {
    if (e instanceof Error) {
      console.log(e)
      ///which ID threw a error
      return of(id) // 1 --- 2 ---  1283 --- 8 --- etc
        .pipe(
          // code to handle error
        )
    }
    throw e
  })
)
.subscribe()

[demensky]

The error can only occur once per subscription. Maybe you should use catchError inside mergeMap instead of outside?

[demensky]

fromEvent(eventEmitter, 'data')
  .pipe(
    mergeMap((id) =>
      from(someInternalCode(id)).pipe(
        catchError((e) => {
          if (e instanceof Error) {
            console.log(e);
            return of(id).pipe();
          }
          throw e;
        }),
      ),
    ),
  )
  .subscribe();

[victor-perez]

Yes I was also already thinking of that, just was wondering of I can get it maybe out the cause stream.
I also thought of using the Error.cause

mergeMap(async (id) => {
  return await someInternalCode(id)
    .cache((error) => {
      throw Error('some internal code threw an error', {
        cause: {
          error,
          value: id,
        },
      })
    })
}),
catchError((error) => {
  if (error?.cause?.error instanceof ErrorX) {
    console.log(error?.cause.value)
  }
  throw e
})

Multiple ways to do this anyway thanks @demensky