gRPC 如何在客户端等待来自 StreamingServer 的标头

Question

对于一个由 proto 文件定义的简单 gRPC 服务：

service HelloService {
    rpc sayHello() returns (stream string)
}

在我开始处理来自服务器的响应之前，如何在客户端先等待标头到达？我尝试使用 ClientInterceptor 并覆盖 onHeaders() 方法，但是在对 sayHello() 的调用已经完成后会调用它。如何验证 sayHello 客户端中的特定标头并根据标头的有效性继续调用？

public class SomeHeaderInterceptor implements ClientInterceptor {
private static final String FULL_METHOD_NAME = "sayHello";
public static CallOptions.Key<String> someHeader = CallOptions.Key.of("some_header_active", "false");

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> methodDescriptor,
                                                           CallOptions callOptions, Channel channel) {
    return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(channel.newCall(methodDescriptor, callOptions)) {

        @Override
        public void start(Listener<RespT> responseListener, Metadata headers) {
            super.start(new ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(responseListener) {
                @Override
                public void onHeaders(Metadata headers) {
                    Metadata.Key<String> SAYHELLO_ACTIVE_HEADER = Metadata.Key.of("some_header_active",
                            Metadata.ASCII_STRING_MARSHALLER);
                    if (methodDescriptor.getFullMethodName().equals(FULL_METHOD_NAME)) {
                        if (!headers.containsKey(SAYHELLO_ACTIVE_HEADER)) {
                            LOGGER.logError("some_header activation missing from header: " + headers);
                        } else {
                            callOptions.withOption(someHeader, "true");
                            Context.current().withValue(Context.key("test"), "testvalue");
                        }
                    }

                    super.onHeaders(headers);
                }
            }, headers);
        }
    };
}

sayHello 的代码如下：

public Iterator<String> sayHello() {
Iterator<String> stream = blockingStub.sayHello();

// wait for the sayhello active header 
boolean isActive = Boolean.parseBoolean(blockingStub.getCallOptions().getOption(SomeHeaderInterceptor. someHeader));
System.out.println("the some_header header value is: " + isActive);
System.out.println("the context key : " + Context.key("test").get(Context.current()));

return stream;
}

在上面的 sayHello() 代码中，它不等待标头到达并在 onHeaders() 中设置上下文。我怎样才能做到这一点？在我验证服务器发送的流中存在 some_header 后，我只想将流返回给客户端调用者。

Answer 1

标头由服务器在第一条消息之前发送，因此最简单的方法是调用stream.hasNext()，这将阻止等待消息。在许多 RPC 中，第一条消息在响应头之后很快就出现了，所以这会很好地工作。

顺便说一句，我注意到您正在尝试使用 CallOptions 和 Context：

callOptions.withOption(someHeader, "true");
Context.current().withValue(Context.key("test"), "testvalue");

这两行都没有真正做任何事情，因为这两个对象都是不可变的。 with* 调用创建一个新实例，因此您必须使用该行的返回值来执行任何操作。此外，CallOptions 和 Context 主要以相反的方向传递信息，例如从客户端应用程序到拦截器。要“反转”方向，应用程序需要设置一个可变的值，例如 AtomicReference 或回调函数，然后拦截器可以与该值进行交互。

如果服务器响应标头和第一条消息之间可能存在明显延迟，那么事情会变得更加复杂。调用sayHello()的代码：

CompletableFuture<Boolean> future = new CompletableFuture<>();
Iterator<String> stream = blockingStub
    .withOption(SomeHeaderInterceptor.SOME_HEADER, future)
    .sayHello();
// wait for the sayhello active header 
boolean isActive = future.get();

然后在拦截器中：

private static final String FULL_METHOD_NAME =
    //"helloworld.Greeter/SayHello";
    GreeterGrpc.getSayHelloMethod().getFullMethodName();
public static final CallOptions.Key<CompletableFuture<Boolean>> SOME_HEADER =
    CallOptions.Key.create("some_header_active", CompletableFuture.completedFuture(false));
private static final Metadata.Key<String> SAYHELLO_ACTIVE_HEADER =
    Metadata.Key.of("some_header_active", Metadata.ASCII_STRING_MARSHALLER);

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> methodDescriptor,
                                                           CallOptions callOptions, Channel channel) {
    CompletableFuture<Boolean> future = callOptions.getOption(SOME_HEADER);
    if (!methodDescriptor.getFullMethodName().equals(FULL_METHOD_NAME)) {
        future.complete(false);
        return channel.newCall(methodDescriptor, callOptions);
    }
    return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(channel.newCall(methodDescriptor, callOptions)) {

        @Override
        public void start(Listener<RespT> responseListener, Metadata headers) {
            super.start(new ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(responseListener) {
                @Override
                public void onHeaders(Metadata headers) {
                    if (!headers.containsKey(SAYHELLO_ACTIVE_HEADER)) {
                        // Could also be future.complete(false)
                        future.completeExceptionally(new Exception("some_header activation missing from header: " + headers));
                    } else {
                        future.complete(true);
                    }
                    super.onHeaders(headers);
                }

                @Override
                public void onClose(Status status, Metadata trailers) {
                    // onHeaders() might not have been called, especially if there was an error
                    if (!future.isDone()) {
                        future.completeExceptionally(status.asRuntimeException(trailers));
                    }
                    super.onClose(status, trailers);
                }
            }, headers);
        }
    };
}

如果您只需要进行验证，而不是实际延迟，那么您可以避免延迟，如果验证失败，则让拦截器调用失败：

private static final String FULL_METHOD_NAME =
    //"helloworld.Greeter/SayHello";
    GreeterGrpc.getSayHelloMethod().getFullMethodName();
private static final Metadata.Key<String> SAYHELLO_ACTIVE_HEADER =
    Metadata.Key.of("some_header_active", Metadata.ASCII_STRING_MARSHALLER);

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> methodDescriptor,
                                                           CallOptions callOptions, Channel channel) {
    if (!methodDescriptor.getFullMethodName().equals(FULL_METHOD_NAME)) {
        return channel.newCall(methodDescriptor, callOptions);
    }
    // We use context to cancel since it is thread-safe, whereas ClientCall.cancel is not
    CancellableContext context = Context.current().withCancellation();
    class ValidatingListener extends ForwardingClientCallListener<RespT> {
        private Listener<RespT> responseListener;

        public ValidatingListener(Listener<RespT> responseListener) {
            this.responseListener = responseListener;
        }

        @Override
        protected Listener<RespT> delegate() {
            return responseListener;
        }

        @Override
        public void onHeaders(Metadata headers) {
            if (!headers.containsKey(SAYHELLO_ACTIVE_HEADER)) {
                Listener<RespT> saved = responseListener;
                responseListener = new Listener<RespT>() {}; // noop listener; throw away future events
                Status status = Status.UNKNOWN.withDescription(
                        "some_header activation missing from header: " + headers);
                context.cancel(status.asRuntimeException());
                saved.onClose(status, new Metadata());
                return;
            }
            // Validation successful
            super.onHeaders(headers);
        }

        @Override
        public void onClose(Status status, Metadata trailers) {
            context.close();
        }
    }

    Context toRestore = context.attach();
    try {
        return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(channel.newCall(methodDescriptor, callOptions)) {

            @Override
            public void start(Listener<RespT> responseListener, Metadata headers) {
                super.start(new ValidatingListener(responseListener), headers);
            }
        };
    } finally {
        context.detach(toRestore);
    }
}