EventInterator error is not caught by the onError interceptor

[viniciusbitt]

I'm putting an interceptor in RPCHandler to show the errors in console.log, but it's not working if it's a route with EventIterator... I don't know if it's expected

The current solution would be try catch block in the handler of each route, but doing this for many routes is complicated

[unnoq]

I believe we should keep current behavior of onError, if you want catch event iterator error you can do somthing like this (clientInterceptors or middlewares)

import { AsyncIteratorClass, isAsyncIteratorObject } from '@orpc/shared'

const handler = new RPCHandler(router, {
  clientInterceptors: [
    async (options) => {
      try {
        const output = await options.next()

        if (isAsyncIteratorObject(output)) {
          return new AsyncIteratorClass(async () => {
            try {
              return await output.next()
            }
            catch (e) {
              console.log(e) // event iterator error

              throw e
            }
          }, async () => {
            await output.return?.()
          })
        }

        return output
      }
      catch (e) {
        console.log(e) // normal error

        throw e
      }
    },
  ],
})

WARNING THIS JUST DRAFT, CAN BE BUG

[n9]

@unnoq I get:

Module '"@orpc/shared"' has no exported member 'EventIteratorClass'.

Has the API changed since then? Since orpc/shared is an internal package, could there be an official helper function for this?

[unnoq]

Has the API changed since then? Since orpc/shared is an internal package, could there be an official helper function for this?

My mistake it should be AsyncIteratorClass

[n9]

@unnoq What about to have a function like this:

import { Context, Meta, ProcedureClientInterceptorOptions } from "@orpc/server";
import { AsyncIteratorClass, Interceptor, isAsyncIteratorObject } from "@orpc/shared";

export function createErrorClientInterceptor<TContext extends Context, TResult>(options: {
    onError: (e: unknown) => TResult | never
    onAsyncIteratorError?: (e: unknown) => IteratorResult<TResult> | never
}): Interceptor<ProcedureClientInterceptorOptions<TContext, Record<never, never>, Meta>, Promise<unknown>> {
    const { onError, onAsyncIteratorError } = options 
    return async (options) => {
        try {
            const output = await options.next()
            if (onAsyncIteratorError && isAsyncIteratorObject(output)) {
                return new AsyncIteratorClass(async () => {
                    try {
                        return await output.next()
                    }
                    catch (e) {
                        return onAsyncIteratorError(e);
                    }
                }, async () => {
                    await output.return?.()
                })
            } else {
                return output
            }
        } catch (e) {
            return onError(e);
        }
    }
}

And usage example:

const handler = new RPCHandler(rpcRouter, {
    clientInterceptors: [createErrorClientInterceptor({
        onError: e => {
            console.error(e)
            throw e
        },
        onAsyncIteratorError: e => {
            console.error(e)
            throw e
        },
    })]
})

PS: Not sure about typing...

[unnoq]

Related: https://github.com/unnoq/orpc/pull/1191

[mutewinter]

Updated based on the Pino PR, which avoids aborted errors:

import { mapEventIterator } from "@orpc/client";
import {
  type Context,
  type Meta,
  type ProcedureClientInterceptorOptions,
} from "@orpc/server";
import {
  type Interceptor,
  isAsyncIteratorObject,
  overlayProxy,
} from "@orpc/shared";

export function createErrorClientInterceptor<
  TContext extends Context,
  TResult,
>(options: {
  onAsyncIteratorError?: (e: unknown) => IteratorResult<TResult> | never;
  onError: (e: unknown) => never | TResult;
}): Interceptor<
  ProcedureClientInterceptorOptions<TContext, Record<never, never>, Meta>,
  Promise<unknown>
> {
  const { onAsyncIteratorError, onError } = options;
  return async (o) => {
    try {
      const output = await o.next();
      if (isAsyncIteratorObject(output)) {
        return overlayProxy(
          output,
          mapEventIterator(output, {
            // eslint-disable-next-line @typescript-eslint/require-await
            error: async (error) => {
              /**
               * DON'T treat aborted signal as error if happen during business logic,
               * Because this is streaming response, so the error can't catch by `interceptors` above.
               */
              if (o.signal?.aborted && o.signal.reason === error) {
                // Signal aborted - pass through
              } else if (onAsyncIteratorError) {
                return onAsyncIteratorError(error);
              }
              return error;
            },
            // eslint-disable-next-line @typescript-eslint/no-unsafe-return
            value: (v) => v,
          }),
        );
      }
      return output;
    } catch (error) {
      return onError(error);
    }
  };
}