Java 类io.grpc.Metadata 实例源码

private void copyMetadataToThreadLocal(Metadata headers) {
  String attachments = headers.get(GrpcUtil.GRPC_CONTEXT_ATTACHMENTS);
  String values = headers.get(GrpcUtil.GRPC_CONTEXT_VALUES);
  try {
    if (attachments != null) {
      Map<String, String> attachmentsMap =
          SerializerUtil.fromJson(attachments, new TypeToken<Map<String, String>>() {}.getType());
      RpcContext.getContext().setAttachments(attachmentsMap);
    }
    if (values != null) {
      Map<String, Object> valuesMap =
          SerializerUtil.fromJson(values, new TypeToken<Map<String, Object>>() {}.getType());
      for (Map.Entry<String, Object> entry : valuesMap.entrySet()) {
        RpcContext.getContext().set(entry.getKey(), entry.getValue());
      }
    }
  } catch (Throwable e) {
    log.error(e.getMessage(), e);
  }
}

private void copyThreadLocalToMetadata(Metadata headers) {
  Map<String, String> attachments = RpcContext.getContext().getAttachments();
  Map<String, Object> values = RpcContext.getContext().get();
  try {
    if (!attachments.isEmpty()) {
      headers.put(GrpcUtil.GRPC_CONTEXT_ATTACHMENTS, SerializerUtil.toJson(attachments));
    }
    if (!values.isEmpty()) {
      headers.put(GrpcUtil.GRPC_CONTEXT_VALUES, SerializerUtil.toJson(values));
    }
  } catch (Throwable e) {
    log.error(e.getMessage(), e);
  } finally {
    RpcContext.removeContext();
  }
}

private void statusError(Status status, Metadata trailers) {
  if (enabledRetry) {
    final NameResolverNotify nameResolverNotify = this.createNameResolverNotify();
    boolean retryHaveDone = this.retryHaveDone();
    if (retryHaveDone) {
      completionFuture.setException(status.asRuntimeException(trailers));
    } else {
      nameResolverNotify.refreshChannel();
      scheduleRetryService.execute(this);
      SocketAddress remoteAddress =
          (SocketAddress) callOptions.getOption(GrpcCallOptions.CALLOPTIONS_CUSTOME_KEY)
              .get(GrpcCallOptions.GRPC_CURRENT_ADDR_KEY);
      logger.error(String.format("Retrying failed call. Failure #%d，Failure Server: %s",
          currentRetries.get(), String.valueOf(remoteAddress)));
      currentRetries.getAndIncrement();
    }
  } else {
    completionFuture.setException(status.asRuntimeException(trailers));
  }

}

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
        ServerCall<ReqT, RespT> call,
        Metadata headers,
        ServerCallHandler<ReqT, RespT> next) {
    if (Objects.isNull(SecurityContextHolder.getContext().getAuthentication())) {
        SecurityContextHolder.getContext().setAuthentication(new AnonymousAuthenticationToken(key,
                "anonymousUser", Collections.singletonList(new SimpleGrantedAuthority("ROLE_ANONYMOUS"))));

        log.debug("Populated SecurityContextHolder with anonymous token: {}",
                SecurityContextHolder.getContext().getAuthentication());
    } else {
        log.debug("SecurityContextHolder not populated with anonymous token, as it already contained: {}",
                SecurityContextHolder.getContext().getAuthentication());
    }

    return next.startCall(call, headers);
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
        final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions,
        final Channel next) {
    return new ClientInterceptors.CheckedForwardingClientCall<ReqT, RespT>(
            next.newCall(method, callOptions)) {
        @Override
        protected void checkedStart(Listener<RespT> responseListener, Metadata headers)
                throws StatusException {
            Metadata cachedSaved;
            URI uri = serviceUri(next, method);
            synchronized (this) {
                Map<String, List<String>> latestMetadata = getRequestMetadata(uri);
                if (mLastMetadata == null || mLastMetadata != latestMetadata) {
                    mLastMetadata = latestMetadata;
                    mCached = toHeaders(mLastMetadata);
                }
                cachedSaved = mCached;
            }
            headers.merge(cachedSaved);
            delegate().start(responseListener, headers);
        }
    };
}

@BeforeClass
public static void startServer() throws IOException {
    AfricasTalking.initialize(Fixtures.USERNAME, Fixtures.API_KEY);
    server = new Server(new Authenticator() {
        @Override
        public boolean authenticate(String client) {
            return client.compareToIgnoreCase(TEST_CLIENT_ID) == 0;
        }
    });
    server.addSipCredentials("test", "secret", "sip://at.dev");
    server.start(certFile, privateKeyFile, TEST_PORT);

    ManagedChannel ch = NettyChannelBuilder.forAddress("localhost", TEST_PORT)
        .sslContext(GrpcSslContexts.forClient().trustManager(certFile).build())
        .build();


    client = SdkServerServiceGrpc.newBlockingStub(ch)
        .withCallCredentials(new CallCredentials(){
            @Override
            public void applyRequestMetadata(MethodDescriptor<?, ?> method, Attributes attrs, Executor appExecutor,
                    final MetadataApplier applier) {
                        appExecutor.execute(new Runnable(){
                            @Override
                            public void run() {
                                try {
                                    Metadata headers = new Metadata();
                                    Metadata.Key<String> clientIdKey = Metadata.Key.of("X-Client-Id", Metadata.ASCII_STRING_MARSHALLER);
                                    headers.put(clientIdKey, TEST_CLIENT_ID);
                                    applier.apply(headers);
                                } catch(Throwable ex) {
                                    applier.fail(Status.UNAUTHENTICATED.withCause(ex));
                                }
                            }
                        });

            }
        });
}

/**
 * A metadata marshaller that encodes objects as JSON using the google-gson library.
 *
 * <p>All non-ascii characters are unicode escaped to comply with {@code AsciiMarshaller}'s character range
 * requirements.
 *
 * @param clazz the type to serialize
 * @param <T>
 */
public static final <T> Metadata.AsciiMarshaller<T> JSON_MARSHALLER(Class<T> clazz) {
    return new Metadata.AsciiMarshaller<T>() {
        TypeToken<T> typeToken = TypeToken.of(clazz);
        private Gson gson = new Gson();

        @Override
        public String toAsciiString(T value) {
            try {
                try (StringWriter sw = new StringWriter()) {
                    gson.toJson(value, typeToken.getType(), new UnicodeEscapingAsciiWriter(sw));
                    return sw.toString();
                }
            } catch (IOException e) {
                throw new IllegalArgumentException(e);
            }
        }

        @Override
        public T parseAsciiString(String serialized) {
            return gson.fromJson(serialized, typeToken.getType());
        }
    };
}

/**
 * A metadata marshaller that encodes objects as protobuf according to their proto IDL specification.
 *
 * @param clazz the type to serialize
 * @param <T>
 */
public static <T extends GeneratedMessageV3> Metadata.BinaryMarshaller<T> PROTOBUF_MARSHALLER(Class<T> clazz) {
    try {
        Method defaultInstance = clazz.getMethod("getDefaultInstance");
        GeneratedMessageV3 instance = (GeneratedMessageV3) defaultInstance.invoke(null);

        return new Metadata.BinaryMarshaller<T>() {
            @Override
            public byte[] toBytes(T value) {
                return value.toByteArray();
            }

            @SuppressWarnings("unchecked")
            @Override
            public T parseBytes(byte[] serialized) {
                try {
                    return (T) instance.getParserForType().parseFrom(serialized);
                } catch (InvalidProtocolBufferException ipbe) {
                    throw new IllegalArgumentException(ipbe);
                }
            }
        };
    } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException ex) {
        throw new IllegalStateException(ex);
    }
}

@Test
public void interceptorShouldFreezeContext() {
    TestService svc = new TestService();

    // Plumbing
    serverRule.getServiceRegistry().addService(ServerInterceptors.interceptForward(svc,
            new AmbientContextServerInterceptor("ctx-"),
            new AmbientContextFreezeServerInterceptor()));
    GreeterGrpc.GreeterBlockingStub stub = GreeterGrpc
            .newBlockingStub(serverRule.getChannel())
            .withInterceptors(new AmbientContextClientInterceptor("ctx-"));

    // Test
    Metadata.Key<String> key = Metadata.Key.of("ctx-k", Metadata.ASCII_STRING_MARSHALLER);
    AmbientContext.initialize(Context.current()).run(() -> {
        AmbientContext.current().put(key, "value");
        stub.sayHello(HelloRequest.newBuilder().setName("World").build());
    });

    assertThat(svc.frozen).isTrue();
}

@Test
public void jsonMarshallerRoundtrip() {
    Foo foo = new Foo();
    foo.country = "France";
    List<Bar> bars = new ArrayList<>();
    Bar bar1 = new Bar();
    bar1.cheese = "Brë";
    bar1.age = 2;
    bars.add(bar1);
    Bar bar2 = new Bar();
    bar2.cheese = "Guda<>'";
    bar2.age = 4;
    bars.add(bar2);
    foo.bars = bars;

    Metadata.AsciiMarshaller<Foo> marshaller = MoreMetadata.JSON_MARSHALLER(Foo.class);
    String str = marshaller.toAsciiString(foo);
    assertThat(str).doesNotContain("ë");

    Foo foo2 = marshaller.parseAsciiString(str);
    assertThat(foo2).isEqualTo(foo);
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
        final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions,
        final Channel next) {
    return new ClientInterceptors.CheckedForwardingClientCall<ReqT, RespT>(
            next.newCall(method, callOptions)) {
        @Override
        protected void checkedStart(Listener<RespT> responseListener, Metadata headers)
                throws StatusException {
            Metadata cachedSaved;
            URI uri = serviceUri(next, method);
            synchronized (this) {
                Map<String, List<String>> latestMetadata = getRequestMetadata(uri);
                if (mLastMetadata == null || mLastMetadata != latestMetadata) {
                    mLastMetadata = latestMetadata;
                    mCached = toHeaders(mLastMetadata);
                }
                cachedSaved = mCached;
            }
            headers.merge(cachedSaved);
            delegate().start(responseListener, headers);
        }
    };
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(final MethodDescriptor<ReqT, RespT> method,
                                                           CallOptions callOptions, final Channel next) {
    return new ClientInterceptors.CheckedForwardingClientCall<ReqT, RespT>(
            next.newCall(method, callOptions)) {
        @Override
        protected void checkedStart(Listener<RespT> responseListener, Metadata headers)
                throws StatusException {

            Metadata cachedSaved;
            URI uri = serviceUri(next, method);
            synchronized (GoogleCredentialsInterceptor.this) {
                Map<String, List<String>> latestMetadata = getRequestMetadata(uri);
                if (mLastMetadata == null || mLastMetadata != latestMetadata) {
                    mLastMetadata = latestMetadata;
                    mCached = toHeaders(mLastMetadata);
                }
                cachedSaved = mCached;
            }
            headers.merge(cachedSaved);
            delegate().start(responseListener, headers);
        }
    };
}

private ClientInterceptor metadataInterceptor() {
  ClientInterceptor interceptor = new ClientInterceptor() {
    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
        final io.grpc.MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, final Channel next) {
      return new ClientInterceptors.CheckedForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
        @Override
        protected void checkedStart(Listener<RespT> responseListener, Metadata headers)
            throws StatusException {
          for (ConfigProto.CallMetadataEntry entry : callConfiguration.getMetadataList()) {
            Metadata.Key<String> key = Metadata.Key.of(entry.getName(), Metadata.ASCII_STRING_MARSHALLER);
            headers.put(key, entry.getValue());
          }
          delegate().start(responseListener, headers);
        }
      };
    }
  };

  return interceptor;
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
    MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
  return new SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
    @Override
    public void start(Listener<RespT> responseListener, Metadata headers) {
      getToken(next).ifPresent(t -> headers.put(TOKEN, t));
      super.start(new SimpleForwardingClientCallListener<RespT>(responseListener) {
        @Override
        public void onClose(Status status, Metadata trailers) {
          if (isInvalidTokenError(status)) {
            try {
              refreshToken(next);
            } catch (Exception e) {
              // don't throw any error here.
              // rpc will retry on expired auth token.
            }
          }
          super.onClose(status, trailers);
        }
      }, headers);
    }
  };
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
        final MethodDescriptor<ReqT, RespT> method, CallOptions callOptions,
        final Channel next) {
    return new ClientInterceptors.CheckedForwardingClientCall<ReqT, RespT>(
            next.newCall(method, callOptions)) {
        @Override
        protected void checkedStart(Listener<RespT> responseListener, Metadata headers)
                throws StatusException {
            Metadata cachedSaved;
            URI uri = serviceUri(next, method);
            synchronized (this) {
                Map<String, List<String>> latestMetadata = getRequestMetadata(uri);
                if (mLastMetadata == null || mLastMetadata != latestMetadata) {
                    mLastMetadata = latestMetadata;
                    mCached = toHeaders(mLastMetadata);
                }
                cachedSaved = mCached;
            }
            headers.merge(cachedSaved);
            delegate().start(responseListener, headers);
        }
    };
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT>
    newCall(MethodDescriptor<ReqT, RespT> methodDescriptor, CallOptions callOptions) {
  final Context timerContext = timer.time();
  final AtomicBoolean decremented = new AtomicBoolean(false);
  return new CheckedForwardingClientCall<ReqT, RespT>(delegate.newCall(methodDescriptor, callOptions)) {
    @Override
    protected void checkedStart(ClientCall.Listener<RespT> responseListener, Metadata headers)
        throws Exception {
      ClientCall.Listener<RespT> timingListener = wrap(responseListener, timerContext, decremented);
      getStats().ACTIVE_RPC_COUNTER.inc();
      getStats().RPC_METER.mark();
      delegate().start(timingListener, headers);
    }

    @Override
    public void cancel(String message, Throwable cause) {
      if (!decremented.getAndSet(true)) {
        getStats().ACTIVE_RPC_COUNTER.dec();
      }
      super.cancel(message, cause);
    }
  };
}

private static Exception toException(ErrorReporter errorReport) {
  Metadata trailers = new Metadata();
  trailers.put(errorDetailsKey, errorReport.toErrorDetails());
  switch (errorReport.getGeneralCode()) {
    case FUNCTION:
      return Status.FAILED_PRECONDITION.withCause(errorReport.getCause())
          .withDescription(errorReport.getSpecificErrorMsg()).asException(trailers);
    case UNAVAILABLE:
      return Status.UNAVAILABLE.withCause(errorReport.getCause())
          .withDescription(errorReport.getSpecificErrorMsg()).asRuntimeException(trailers);
    case INTERNAL:
      return Status.INTERNAL.withCause(errorReport.getCause())
          .withDescription(errorReport.getSpecificErrorMsg()).asRuntimeException(trailers);
    default:
      return Status.UNKNOWN.withCause(errorReport.getCause())
          .withDescription(errorReport.getSpecificErrorMsg()).asRuntimeException(trailers);
  }
}

private void logCallEnded(String serviceRpcName, String clientIp, String userAgent, Status status,
    Stopwatch stopwatch, Metadata headers) {
  MDC.put(GRPC_RPC_NAME_KEY, serviceRpcName);
  MDC.put(GRPC_CLIENT_IP_KEY, clientIp);
  MDC.put(GRPC_USER_AGENT_KEY, userAgent);
  String statusString = status.getCode().name();
  MDC.put(GRPC_STATUS_KEY, statusString);
  String elapsedString = Long.toString(stopwatch.elapsed(TimeUnit.MILLISECONDS));
  MDC.put(GRPC_ELAPSED_MS_KEY, elapsedString);
  if (LOG.isTraceEnabled()) {
    String headerString = headers.toString();
    MDC.put(GRPC_HEADERS_KEY, headerString);
    LOG.trace("[{}] [{}] [{}] [{}] [{} ms] [{}]", serviceRpcName, clientIp, userAgent,
        statusString, elapsedString, headerString);
  } else {
    LOG.info("[{}] [{}] [{}] [{}] [{} ms]", serviceRpcName, clientIp, userAgent, statusString,
        elapsedString);
  }
}

@SuppressWarnings("checkstyle:MethodTypeParameterName")
@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(
    final ServerCall<ReqT, RespT> call, final Metadata headers,
    final ServerCallHandler<ReqT, RespT> next) {
  TL.set(call);

  return next.startCall(new SimpleForwardingServerCall<ReqT, RespT>(call) {
    @Override
    public void close(final Status status, final Metadata trailers) {
      super.close(status, trailers);
      TL.remove();
    }

  }, headers);
}

/**
 * Executes a unary call with a response {@link SingleObserver}.
 */
public static <ReqT, RespT> Single<RespT> unaryCall(
    final ClientCall<ReqT, RespT> call,
    final ReqT request) {
  final SingleRequestSender<ReqT> requestSender = new SingleRequestSender<ReqT>(call, request);
  SingleResponseReceiver<RespT> responseReceiver = new SingleResponseReceiver<RespT>(call) {
    @Override
    public void startCall() {
      requestSender.startCall();
      super.startCall();
    }
  };

  call.start(responseReceiver, new Metadata());

  return Single.wrap(responseReceiver.singleSource());
}

/**
 * Executes a server-streaming call with a response {@link Subscriber}.
 */
public static <ReqT, RespT> Flowable<RespT> serverStreamingCall(
    final ClientCall<ReqT, RespT> call,
    ReqT request) {
  final SingleRequestSender<ReqT> requestSender = new SingleRequestSender<ReqT>(call, request);
  StreamingResponseReceiver<RespT> responseReceiver = new StreamingResponseReceiver<RespT>(call) {
    @Override
    public void startCall() {
      requestSender.startCall();
      super.startCall();
    }
  };

  call.start(responseReceiver, new Metadata());

  return Flowable.fromPublisher(responseReceiver.publisher());
}

/**
 * Executes a client-streaming call returning a {@link StreamObserver} for the requestMore messages.
 *
 * @return requestMore stream observer.
 */
public static <ReqT, RespT> Single<RespT> clientStreamingCall(
    ClientCall<ReqT, RespT> call,
    Flowable<ReqT> requests,
    CallOptions options) {
  final StreamRequestSender<ReqT> requestSender = new StreamRequestSender<ReqT>(call, getLowWatermark(options), getHighWatermark(options));
  SingleResponseReceiver<RespT> responseReceiver = new SingleResponseReceiver<RespT>(call) {
    @Override
    public void startCall() {
      requestSender.startCall();
      super.startCall();
    }
  };

  ClientCall.Listener<RespT> delegate = new DelegateClientCallListener<RespT>(requestSender, responseReceiver);
  call.start(delegate, new Metadata());

  requests.subscribe(requestSender.subscriber());

  return Single.wrap(responseReceiver.singleSource());
}

/**
 * Executes a bidi-streaming call.
 *
 * @return requestMore stream observer.
 */
public static <ReqT, RespT> Flowable<RespT> bidiStreamingCall(
    ClientCall<ReqT, RespT> call,
    Flowable<ReqT> requests,
    CallOptions options) {
  final StreamRequestSender<ReqT> requestSender = new StreamRequestSender<ReqT>(call, getLowWatermark(options), getHighWatermark(options));
  StreamingResponseReceiver<RespT> responseReceiver = new StreamingResponseReceiver<RespT>(call) {
    @Override
    public void startCall() {
      requestSender.startCall();
      super.startCall();
    }
  };

  ClientCall.Listener<RespT> delegate = new DelegateClientCallListener<RespT>(requestSender, responseReceiver);
  call.start(delegate, new Metadata());

  requests.subscribe(requestSender.subscriber());

  return Flowable.fromPublisher(responseReceiver.publisher());
}

/**
 * Returns the {@link Error} extracted from the {@link Metadata}.
 *
 * @param metadata the metadata to convert
 * @return the error extracted from the metadata or {@code Optional.absent()}
 *         if there is no error.
 */
public static Optional<Error> toError(Metadata metadata) {
    checkNotNull(metadata);
    final byte[] bytes = metadata.get(KEY);

    if (bytes == null) {
        return Optional.absent();
    }

    try {
        final Error error = Error.parseFrom(bytes);
        return Optional.of(error);
    } catch (InvalidProtocolBufferException e) {
        throw Exceptions.illegalStateWithCauseOf(e);
    }
}

private void retryCall(
    RequestT payload,
    Metadata.Headers requestHeaders,
    Listener<ResponseT> listener) {
  final Call<RequestT, ResponseT> delegate = channel.newCall(method);
  delegate.start(listener, requestHeaders);
  delegate.request(1);
  cancelled.addListener(new Runnable() {
    @Override
    public void run() {
      delegate.cancel();
    }
  }, MoreExecutors.directExecutor());

  delegate.sendPayload(payload);
  delegate.halfClose();
}

@Test
public void callCompletionStatusesAreRecorded() throws InterruptedException {
  CallCompletionStatusInterceptor interceptor = new CallCompletionStatusInterceptor();

  when(channelStub.newCall(BigtableServiceGrpc.CONFIG.mutateRow)).thenReturn(callStub);

  CompletionStatusGatheringCall<MutateRowRequest, Empty> wrappedCall =
      interceptor.interceptCall(BigtableServiceGrpc.CONFIG.mutateRow, channelStub);

  Listener<Empty> statusGatheringListener =
      wrappedCall.createGatheringListener(responseListenerStub);

  statusGatheringListener.onClose(Status.INTERNAL, new Metadata.Trailers());

  CallCompletionStatusInterceptor.CallCompletionStatus expectedStatusEntry =
      new CallCompletionStatusInterceptor.CallCompletionStatus(
          BigtableServiceGrpc.CONFIG.mutateRow, Status.INTERNAL);

  Assert.assertEquals(1, interceptor.getCallCompletionStatuses().count(expectedStatusEntry));
}

@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers,ServerCallHandler<ReqT, RespT> next) {
    logger.info("Call intercepted "+headers.toString());
    String token = headers.get(authKey);
    if (StringUtils.notEmpty(token))
    {
        try
        {
            logger.info("Token "+token);
            ConsumerBean consumer = resourceServer.validateResourceFromToken(token);
            logger.info("Setting call to client "+consumer.getShort_name());
            return new SeldonServerCallListener<ReqT>(next.startCall(call, headers),consumer.getShort_name(),this);
        }
        catch (APIException e)
        {
            logger.warn("API exception on getting token ",e);
            return next.startCall(call, headers);
        }
    }
    else
    {
        logger.warn("Empty token ignoring call");
        return next.startCall(call, headers);
    }
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
    MethodDescriptor<ReqT,RespT> method, CallOptions callOptions, Channel next) {
  LOGGER.info("Intercepted " + method.getFullMethodName());
  ClientCall<ReqT, RespT> call = next.newCall(method, callOptions);

  call = new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(call) {
    @Override
    public void start(Listener<RespT> responseListener, Metadata headers) {
      if (apiKey != null && !apiKey.isEmpty()) {
        LOGGER.info("Attaching API Key: " + apiKey);
        headers.put(API_KEY_HEADER, apiKey);
      }
      super.start(responseListener, headers);
    }
  };
  return call;
}

@Override
public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
    MethodDescriptor<ReqT,RespT> method, CallOptions callOptions, Channel next) {
  LOGGER.info("Intercepted " + method.getFullMethodName());
  ClientCall<ReqT, RespT> call = next.newCall(method, callOptions);

  call = new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(call) {
    @Override
    public void start(Listener<RespT> responseListener, Metadata headers) {
      if (apiKey != null && !apiKey.isEmpty()) {
        LOGGER.info("Attaching API Key: " + apiKey);
        headers.put(API_KEY_HEADER, apiKey);
      }
      if (authToken != null && !authToken.isEmpty()) {
        System.out.println("Attaching auth token");
        headers.put(AUTHORIZATION_HEADER, "Bearer " + authToken);
      }
      super.start(responseListener, headers);
    }
  };
  return call;
}

@Override
public void close(Status status, Metadata unusedGrpcMetadata) {
    checkState(!closeCalled, "call already closed");

    closeCalled = true;

    HttpHeaders trailers = statusToTrailers(status, sendHeadersCalled);
    final HttpObject trailersObj;
    if (sendHeadersCalled && GrpcSerializationFormats.isGrpcWeb(serializationFormat)) {
        // Normal trailers are not supported in grpc-web and must be encoded as a message.
        // Message compression is not supported in grpc-web, so we don't bother using the normal
        // ArmeriaMessageFramer.
        trailersObj = serializeTrailersAsMessage(trailers);
    } else {
        trailersObj = trailers;
    }
    try {
        res.write(trailersObj);
        res.close();
    } finally {
        closeListener(status);
    }
}

@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
    MethodInterceptResult result) throws Throwable {
    GRPCDynamicFields cachedObjects = (GRPCDynamicFields)objInst.getSkyWalkingDynamicField();
    Metadata headers = cachedObjects.getMetadata();
    Map<String, String> headerMap = new HashMap<String, String>();
    for (String key : headers.keys()) {
        if (!key.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
            String value = headers.get(Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER));
            headerMap.put(key, value);
        }
    }

    ContextCarrier contextCarrier = new ContextCarrier();
    CarrierItem next = contextCarrier.items();
    while (next.hasNext()) {
        next = next.next();
        String contextValue = headerMap.get(next.getHeadKey());
        if (!StringUtil.isEmpty(contextValue)) {
            next.setHeadValue(contextValue);
        }
    }

    final AbstractSpan span = ContextManager.createEntrySpan(cachedObjects.getRequestMethodName() + (cachedObjects.getMethodType() != MethodDescriptor.MethodType.UNARY ? STREAM_CALL_OPERATION_NAME_SUFFIX : BLOCK_CALL_OPERATION_NAME_SUFFIX), contextCarrier);
    span.setComponent(ComponentsDefine.GRPC);
}

@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
    MethodInterceptResult result) throws Throwable {
    final Metadata headers = (Metadata)allArguments[1];

    final ContextCarrier contextCarrier = new ContextCarrier();
    ContextManager.inject(contextCarrier);
    CarrierItem contextItem = contextCarrier.items();
    while (contextItem.hasNext()) {
        contextItem = contextItem.next();
        Metadata.Key<String> headerKey = Metadata.Key.of(contextItem.getHeadKey(), Metadata.ASCII_STRING_MARSHALLER);
        headers.put(headerKey, contextItem.getHeadValue());
    }

    GRPCDynamicFields cachedObjects = (GRPCDynamicFields)objInst.getSkyWalkingDynamicField();
    GRPCDynamicFields listenerCachedObject = new GRPCDynamicFields();
    listenerCachedObject.setSnapshot(ContextManager.capture());
    listenerCachedObject.setDescriptor(cachedObjects.getDescriptor());
    ((EnhancedInstance)allArguments[0]).setSkyWalkingDynamicField(listenerCachedObject);
}

@Test
public void testOnReadyWithoutContextCarrier() throws Throwable {
    cachedObjects.setMetadata(new Metadata());
    serverCallOnReadyInterceptor.beforeMethod(enhancedInstance, null, null, null, null);
    serverCallOnMessageInterceptor.beforeMethod(enhancedInstance, null, null, null, null);
    serverCallOnMessageInterceptor.afterMethod(enhancedInstance, null, null, null, null);
    serverCallOnCloseInterceptor.afterMethod(enhancedInstance, null, null, null, null);

    assertThat(segmentStorage.getTraceSegments().size(), is(1));
    TraceSegment segment = segmentStorage.getTraceSegments().get(0);

    assertThat(segment.getRefs() == null, is(true));

    assertThat(SegmentHelper.getSpans(segment).size(), is(2));
    AbstractTracingSpan abstractTracingSpan = SegmentHelper.getSpans(segment).get(0);
    assertThat(abstractTracingSpan.getOperationName(), is("org.skywalking.test.grpc.GreetService.sayHello/ResponseStreamObserver/OnNext"));

    abstractTracingSpan = SegmentHelper.getSpans(segment).get(1);
    assertThat(abstractTracingSpan.getOperationName(), is("org.skywalking.test.grpc.GreetService.sayHello/StreamCall"));
    assertThat(abstractTracingSpan.isEntry(), is(true));
    assertThat(SpanHelper.getTags(abstractTracingSpan).size(), is(1));
    assertThat(SpanHelper.getTags(abstractTracingSpan).get(0).getKey(), is("onNext.count"));
    assertThat(SpanHelper.getTags(abstractTracingSpan).get(0).getValue(), is("1"));
}

@Test
public void convertServerHeaders_sanitizes() {
  Metadata metaData = new Metadata();

  // Intentionally being explicit here rather than relying on any pre-defined lists of headers,
  // since the goal of this test is to validate the correctness of such lists in the first place.
  metaData.put(GrpcUtil.CONTENT_TYPE_KEY, "to-be-removed");
  metaData.put(GrpcUtil.TE_HEADER, "to-be-removed");
  metaData.put(GrpcUtil.USER_AGENT_KEY, "to-be-removed");
  metaData.put(userKey, userValue);

  Http2Headers output = Utils.convertServerHeaders(metaData);
  DefaultHttp2Headers headers = new DefaultHttp2Headers();
  for (Map.Entry<CharSequence, CharSequence> entry : output) {
    headers.add(entry.getKey(), entry.getValue());
  }
  // 2 reserved headers, 1 user header
  assertEquals(2 + 1, headers.size());
  assertEquals(Utils.CONTENT_TYPE_GRPC, headers.get(GrpcUtil.CONTENT_TYPE_KEY.name()));
}

@Test
public void methodNotFound() throws Exception {
  createAndStartServer();
  ServerTransportListener transportListener
      = transportServer.registerNewServerTransport(new SimpleServerTransport());
  transportListener.transportReady(Attributes.EMPTY);
  Metadata requestHeaders = new Metadata();
  StatsTraceContext statsTraceCtx =
      StatsTraceContext.newServerContext(
          streamTracerFactories, "Waiter/nonexist", requestHeaders);
  when(stream.statsTraceContext()).thenReturn(statsTraceCtx);
  transportListener.streamCreated(stream, "Waiter/nonexist", requestHeaders);
  verify(stream).setListener(isA(ServerStreamListener.class));
  verify(stream, atLeast(1)).statsTraceContext();

  assertEquals(1, executor.runDueTasks());
  verify(stream).close(statusCaptor.capture(), any(Metadata.class));
  Status status = statusCaptor.getValue();
  assertEquals(Status.Code.UNIMPLEMENTED, status.getCode());
  assertEquals("Method not found: Waiter/nonexist", status.getDescription());

  verify(streamTracerFactory).newServerStreamTracer(eq("Waiter/nonexist"), same(requestHeaders));
  assertNull(streamTracer.getServerCallInfo());
  assertEquals(Status.Code.UNIMPLEMENTED, statusCaptor.getValue().getCode());
}

@Test
public void inboundHeadersReceived_failsOnUnrecognizedMessageEncoding() {
  AbstractClientStream stream =
      new BaseAbstractClientStream(allocator, statsTraceCtx, transportTracer);
  stream.start(mockListener);
  Metadata headers = new Metadata();
  headers.put(GrpcUtil.MESSAGE_ENCODING_KEY, "not-a-real-compression-method");

  stream.transportState().inboundHeadersReceived(headers);

  verifyNoMoreInteractions(mockListener);
  Throwable t = ((BaseTransportState) stream.transportState()).getDeframeFailedCause();
  assertEquals(Status.INTERNAL.getCode(), Status.fromThrowable(t).getCode());
  assertTrue(
      "unexpected deframe failed description",
      Status.fromThrowable(t).getDescription().startsWith("Can't find decompressor for"));
}

@Test
public void cancelBeforeConnected() throws Exception {
  initTransportAndDelayConnected();
  final String message = "Hello Server";
  MockStreamListener listener = new MockStreamListener();
  OkHttpClientStream stream =
      clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
  stream.start(listener);
  InputStream input = new ByteArrayInputStream(message.getBytes(UTF_8));
  stream.writeMessage(input);
  stream.flush();
  stream.cancel(Status.CANCELLED);
  verifyNoMoreInteractions(frameWriter);

  allowTransportConnected();
  verifyNoMoreInteractions(frameWriter);
  shutdownAndVerify();
}

@Test
public void streamFailed() {
  ArgumentCaptor<BidirectionalStream.Callback> callbackCaptor =
      ArgumentCaptor.forClass(BidirectionalStream.Callback.class);
  verify(factory)
      .newBidirectionalStreamBuilder(
          isA(String.class), callbackCaptor.capture(), isA(Executor.class));
  BidirectionalStream.Callback callback = callbackCaptor.getValue();

  // Nothing happens and stream fails

  CronetException exception = mock(CronetException.class);
  callback.onFailed(cronetStream, null, exception);
  verify(transport).finishStream(eq(clientStream), isA(Status.class));
  // finishStream calls transportReportStatus.
  clientStream.transportState().transportReportStatus(Status.UNAVAILABLE, false, new Metadata());

  ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
  verify(clientListener).closed(statusCaptor.capture(), isA(Metadata.class));
  Status status = statusCaptor.getValue();
  assertEquals(Status.UNAVAILABLE.getCode(), status.getCode());
}

public static Http2Headers convertClientHeaders(Metadata headers,
    AsciiString scheme,
    AsciiString defaultPath,
    AsciiString authority,
    AsciiString method,
    AsciiString userAgent) {
  Preconditions.checkNotNull(defaultPath, "defaultPath");
  Preconditions.checkNotNull(authority, "authority");
  Preconditions.checkNotNull(method, "method");

  // Discard any application supplied duplicates of the reserved headers
  headers.discardAll(CONTENT_TYPE_KEY);
  headers.discardAll(GrpcUtil.TE_HEADER);
  headers.discardAll(GrpcUtil.USER_AGENT_KEY);

  return GrpcHttp2OutboundHeaders.clientRequestHeaders(
      toHttp2Headers(headers),
      authority,
      defaultPath,
      method,
      scheme,
      userAgent);
}