grpc - 如何将数据从 grpc rpc 调用传递到 java 中的服务器拦截器

Question

在处理完 rpc 服务器调用后，我正在尝试使用来自响应的值设置一些元数据。计划是使用服务器拦截器并覆盖 close 方法。

是这样的:https://github.com/dconnelly/grpc-error-example/blob/master/src/main/java/example/Errors.java#L38

由于元数据值取决于响应，我需要一些方法将数据从 rpc 服务器调用传递到服务器拦截器或访问拦截器的响应

在 Golang 中，元数据可以在处理后的 rpc 调用 grpc.SetTrailer 中轻松设置，但在 java 中，无法在 rpc 调用中设置。所以我正在尝试使用服务器拦截器。

有人可以帮忙吗？

Answer 1

您可以为此使用 grpc-java 的 Context。 在拦截器中，您附加一个 Context 和一个包含可变引用的自定义键。然后在调用中再次访问该 header 并从中提取值。

public static final Context.Key<TrailerHolder> TRAILER_HOLDER_KEY = Context.key("trailerHolder");
    
Context context = Context.current().withValue(TRAILER_HOLDER_KEY, new TrailerHolder());
Context previousContext = context.attach();
[...]
context.detach(previousContext);

您可以像这样访问上下文值:

TrailerHolder trailerHolder = TRAILER_HOLDER_KEY.get();

您可能希望实现类似于此方法的代码: Contexts#interceptCall(Context, ServerCall, Metadata, ServerCallHandler)

编辑:

import io.grpc.Context;
import io.grpc.ForwardingServerCall.SimpleForwardingServerCall;
import io.grpc.ForwardingServerCallListener;
import io.grpc.Metadata;
import io.grpc.ServerCall;
import io.grpc.ServerCall.Listener;
import io.grpc.ServerCallHandler;
import io.grpc.ServerInterceptor;
import io.grpc.Status;

public class TrailerServerInterceptor implements ServerInterceptor {

    public static final Context.Key<Metadata> TRAILER_HOLDER_KEY = Context.key("trailerHolder");

    @Override
    public <ReqT, RespT> Listener<ReqT> interceptCall(final ServerCall<ReqT, RespT> call, final Metadata headers,
            final ServerCallHandler<ReqT, RespT> next) {
        final TrailerCall<ReqT, RespT> call2 = new TrailerCall<>(call);
        final Context context = Context.current().withValue(TRAILER_HOLDER_KEY, new Metadata());
        final Context previousContext = context.attach();
        try {
            return new TrailerListener<>(next.startCall(call2, headers), context);
        } finally {
            context.detach(previousContext);
        }
    }

    private class TrailerCall<ReqT, RespT> extends SimpleForwardingServerCall<ReqT, RespT> {

        public TrailerCall(final ServerCall<ReqT, RespT> delegate) {
            super(delegate);
        }

        @Override
        public void close(final Status status, final Metadata trailers) {
            trailers.merge(TRAILER_HOLDER_KEY.get());
            super.close(status, trailers);
        }

    }

    private class TrailerListener<ReqT> extends ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT> {

        private final Context context;

        public TrailerListener(final ServerCall.Listener<ReqT> delegate, final Context context) {
            super(delegate);
            this.context = context;
        }

        @Override
        public void onMessage(final ReqT message) {
            final Context previous = this.context.attach();
            try {
                super.onMessage(message);
            } finally {
                this.context.detach(previous);
            }
        }

        @Override
        public void onHalfClose() {
            final Context previous = this.context.attach();
            try {
                super.onHalfClose();
            } finally {
                this.context.detach(previous);
            }
        }

        @Override
        public void onCancel() {
            final Context previous = this.context.attach();
            try {
                super.onCancel();
            } finally {
                this.context.detach(previous);
            }
        }

        @Override
        public void onComplete() {
            final Context previous = this.context.attach();
            try {
                super.onComplete();
            } finally {
                this.context.detach(previous);
            }
        }

        @Override
        public void onReady() {
            final Context previous = this.context.attach();
            try {
                super.onReady();
            } finally {
                this.context.detach(previous);
            }
        }

    }

}

在您的 grpc 服务方法中，您可以简单地使用 TRAILER_HOLDER_KEY.get().put(...)