Как добавить глобальный перехватчик исключений на сервер gRPC?

Код ниже будет улавливать все исключения времени выполнения. Также см. Ссылку https://github.com/grpc/grpc-java/issues/1552

public class GlobalGrpcExceptionHandler implements ServerInterceptor {

   @Override
   public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
         Metadata requestHeaders, ServerCallHandler<ReqT, RespT> next) {
      ServerCall.Listener<ReqT> delegate = next.startCall(call, requestHeaders);
      return new SimpleForwardingServerCallListener<ReqT>(delegate) {
         @Override
         public void onHalfClose() {
            try {
               super.onHalfClose();
            } catch (Exception e) {
               call.close(Status.INTERNAL
                .withCause (e)
                .withDescription("error message"), new Metadata());
            }
         }
      };
   }
}

TransmitStatceptionRunception / a> очень похоже на то, что вы хотите, за исключением того, что ловит только StatusRuntimeException. Вы можете форкнуть его и заставить ловить все исключения.

Чтобы установить перехватчик для всех служб на сервере, вы можете использовать _ 2_, который был добавлен в gRPC 1.5.0.

В Kotlin вам нужно иначе структурировать ServerInterceptor. Я использовал grpc-kotlin в Micronaut, и исключения никогда не появлялись в SimpleForwardingServerCallListener onHalfClose или других обработчиках.

Если вы хотите перехватывать исключения в всех конечных точках gRPC (включая те, которые обрабатывают клиентские потоки) и перехватчиках, вам, вероятно, понадобится нечто подобное следующему :

import io.grpc.ForwardingServerCallListener;
import io.grpc.Metadata;
import io.grpc.ServerCall;
import io.grpc.ServerCallHandler;
import io.grpc.ServerInterceptor;
import io.grpc.Status;

public class GlobalGrpcExceptionHandler implements ServerInterceptor {

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> serverCall,
                                                                 Metadata requestHeaders,
                                                                 ServerCallHandler<ReqT, RespT> serverCallHandler) {
        try {
            ServerCall.Listener<ReqT> delegate = serverCallHandler.startCall(serverCall, requestHeaders);
            return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(delegate) {
                @Override
                public void onMessage(ReqT message) {
                    try {
                        super.onMessage(message); // Here onNext is called (in case of client streams)
                    } catch (Throwable e) {
                        handleEndpointException(e, serverCall);
                    }
                }

                @Override
                public void onHalfClose() {
                    try {
                        super.onHalfClose(); // Here onCompleted is called (in case of client streams)
                    } catch (Throwable e) {
                        handleEndpointException(e, serverCall);
                    }
                }
            };
        } catch (Throwable t) {
            return handleInterceptorException(t, serverCall);
        }
    }

    private <ReqT, RespT> void handleEndpointException(Throwable t, ServerCall<ReqT, RespT> serverCall) {
        serverCall.close(Status.INTERNAL
                         .withCause(t)
                         .withDescription("An exception occurred in the endpoint implementation"), new Metadata());
    }

    private <ReqT, RespT> ServerCall.Listener<ReqT> handleInterceptorException(Throwable t, ServerCall<ReqT, RespT> serverCall) {
        serverCall.close(Status.INTERNAL
                         .withCause(t)
                         .withDescription("An exception occurred in a **subsequent** interceptor"), new Metadata());

        return new ServerCall.Listener<ReqT>() {
            // no-op
        };
    }
}

ОТКАЗ ОТ ОТВЕТСТВЕННОСТИ: я собрал это, изучив реализацию, я не читал его в документации и не уверен, что он изменится. Для справки я имею в виду io.grpc версию 1.30.

Вы читали grpc java примеры для перехватчика?

Итак, в моем случае мы используем код и сообщение в качестве стандарта для определения того, какую ошибку сервер отправил клиенту.

Примеры: сервер отправляет ответ, например

{
  code: 409,
  message: 'Id xxx aldready exist'
}

Таким образом, в клиенте вы можете настроить клиентский перехватчик для получения этого кода и ответа с помощью Reflection. Fyi, мы используем Lognet Spring Boot starter для grpc в качестве сервера и Spring boot для клиента.

public class GrpcExceptionHandler implements ServerInterceptor {

private final Logger logger = LoggerFactory.getLogger (GrpcExceptionHandler.class);

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall (ServerCall<ReqT, RespT> call,
                                                              Metadata headers,
                                                              ServerCallHandler<ReqT, RespT> next) {
    logger.info ("GRPC call at: {}", Instant.now ());
    ServerCall.Listener<ReqT> listener;

    try {
        listener = next.startCall (call, headers);
    } catch (Throwable ex) {
        logger.error ("Uncaught exception from grpc service");
        call.close (Status.INTERNAL
                .withCause (ex)
                .withDescription ("Uncaught exception from grpc service"), null);
        return new ServerCall.Listener<ReqT>() {};
    }

    return listener;
}

}

Образец перехватчика выше.

Вам, конечно же, нужно загрузить его, прежде чем ожидать от него чего-либо;

serverBuilder.addService (ServerInterceptors.intercept (bindableService, interceptor));

ОБНОВЛЕНИЕ

public interface ServerCallHandler<RequestT, ResponseT> {
  /**
   * Produce a non-{@code null} listener for the incoming call. Implementations are free to call
   * methods on {@code call} before this method has returned.
   *
   * <p>If the implementation throws an exception, {@code call} will be closed with an error.
   * Implementations must not throw an exception if they started processing that may use {@code
   * call} on another thread.
   *
   * @param call object for responding to the remote client.
   * @return listener for processing incoming request messages for {@code call}
   */
  ServerCall.Listener<RequestT> startCall(
      ServerCall<RequestT, ResponseT> call,
      Metadata headers);
}

К сожалению, другой контекст потока означает отсутствие области обработки исключений, поэтому мой ответ - это не то решение, которое вы ищете.

Если вы можете преобразовать приложение сервера gRPC для весенней загрузки с помощью yidongnan / grpc-spring-boot- starter, тогда вы можете написать @GrpcAdvice, аналогично Spring Boot @ControllerAdvice как

    @GrpcAdvice
    public class ExceptionHandler {
    
      @GrpcExceptionHandler(ValidationErrorException.class)
      public StatusRuntimeException handleValidationError(ValidationErrorException cause) {
    
         Status.INVALID_ARGUMENT.withDescription("Invalid Argument")
            .asRuntimeException()
      }
    }

На Kotlin добавление try / catch для методов слушателя не работает, по какой-то причине исключения проглатываются.

По ссылке, опубликованной @markficket, я реализовал решение, создающее реализацию SimpleForwardingServerCall.

class ErrorHandlerServerInterceptor : ServerInterceptor {

    private inner class StatusExceptionHandlingServerCall<ReqT, RespT>(delegate: ServerCall<ReqT, RespT>) 
        : ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(delegate) {

        override fun close(status: Status, trailers: Metadata) {
            status.run {
                when {
                    isOk -> status
                    cause is MyException -> myExceptionHandler(cause as MyException)
                    else -> defaultExceptionHandler(cause)
                }
            }
                .let { super.close(it, trailers) }
        }

        private fun myExceptionHandler(cause: MyException): Status = cause.run { ... }

        private fun defaultExceptionHandler(cause: Throwable?): Status = cause?.run { ... }

    }

    override fun <ReqT : Any, RespT : Any> interceptCall(
        call: ServerCall<ReqT, RespT>,
        metadata: Metadata,
        next: ServerCallHandler<ReqT, RespT>
    ): ServerCall.Listener<ReqT> =
        next.startCall(StatusExceptionHandlingServerCall(call), metadata)

}

Тогда конечно вам нужно добавить перехватчик при создании сервера.

ServerBuilder
  .forPort(port)
  .executor(Dispatchers.IO.asExecutor())
  .addService(service)
  .intercept(ErrorHandlerServerInterceptor())
  .build()

А затем вы можете просто выбросить исключения в свои методы gRPC.

override suspend fun someGrpcCall(request: Request): Response {
  ... code ...
  throw NotFoundMyException("Cannot find entity")
}

Я использовал АОП для решения глобальных ошибок rpc, и мне это удобно. Я использую АОП в guice, и способ его использования весной должен быть аналогичным

1. определить метод-перехватчик

```

public class ServerExceptionInterceptor implements MethodInterceptor {
    final static Logger logger = LoggerFactory.getLogger(ServerExceptionInterceptor.class);

    public Object invoke(MethodInvocation invocation) throws Throwable {
        try {
            return  invocation.proceed();
        } catch (Exception ex) {
            String stackTrace = Throwables.getStackTraceAsString(ex);
            logger.error("##grpc server side error, {}", stackTrace);
            Object[] args = invocation.getArguments();
            StreamObserver<?> responseObserver = (StreamObserver<?>)args[1];
            responseObserver.onError(Status.INTERNAL
                    .withDescription(stackTrace)
                    .withCause(ex)
                    .asRuntimeException());

            return null;
        }
    }

    @Target({ElementType.METHOD, ElementType.TYPE}) @Retention(RUNTIME)
    public @interface WrapError {
        String value() default "";
    }

}

```

2. добавить @WrapError ко всем методам rpc @Override @WrapError public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) { HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build(); logger.info("#rpc server, sayHello, planId: {}", req.getName()); if(true) throw new RuntimeException("testing-rpc-error"); //simulate an exception responseObserver.onNext(reply); responseObserver.onCompleted(); }

   3. bind the interceptor in guice module

ServerExceptionInterceptor interceptor = new ServerExceptionInterceptor(); requestInjection(interceptor); bindInterceptor(Matchers.any(), Matchers.annotatedWith(WrapError.class), interceptor);

4. тестирование