PROSAGA码农传奇-SOA/RPC-gRPC如何在客户端等待StreamingServer的头文件

<div class =“post-text”itemprop =“text”>
  
    标头由服务器在第一条消息之前发送，因此简单的方法是调用
     <code>
 stream.hasNext()
 </code>
    ，这将阻止等待消息。在许多RPC中，第一条消息在响应标头之后很快就会出现，所以这样可以很好地工作。
  
  
    顺便说一句，我注意到你在试验
     <code>
 CallOptions
 </code>
     和
     <code>
 Context
 </code>
    ：
  
   <pre>
 <code>
 callOptions.withOption(someHeader, "true");
Context.current().withValue(Context.key("test"), "testvalue");

</code>
 </pre>
  
    这些行都没有做任何事情，因为两个对象都是不可变的。该
     <code>
 with*
 </code>
     调用创建一个新实例，因此您必须使用该行的返回值来执行任何操作。此外，CallOptions和Context主要以相反的方向传递信息，例如从客户端应用程序到拦截器。为了“反转”方向，应用程序需要设置一个可变的值，如AtomicReference或回调函数，然后拦截器可以与该值进行交互。
  
  
    如果服务器响应标头和第一条消息之间可能存在明显的延迟，那么事情会变得更加复杂。调用sayHello（）的代码：
  
   <pre>
 <code>
 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();

</code>
 </pre>
  
    然后在拦截器中：
  
   <pre>
 <code>
 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);
        }
    };
}

</code>
 </pre>
  
    如果你只需要进行验证，而不是实际延迟，那么你可以避免延迟，如果验证失败就让拦截器失败：
  
   <pre>
 <code>
 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);
 }
}

</code>
 </pre>
</DIV>