io.grpc.Contexts#interceptCall ( )源码实例Demo

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(final ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
    Header headerObject;
    try {
        headerObject = headerReader.extract(headers);
    } catch (Exception e) {
        if (logger.isInfoEnabled()) {
            logger.info("Header extract fail cause={}, method={} headers={}, attr={}",
                    e.getMessage(), call.getMethodDescriptor().getFullMethodName(), headers, call.getAttributes(), e);
        }
        call.close(Status.INVALID_ARGUMENT.withDescription(e.getMessage()), new Metadata());
        return new ServerCall.Listener<ReqT>() {
        };
    }

    final Context currentContext = Context.current();
    final Context newContext = currentContext.withValue(contextKey, headerObject);
    if (logger.isDebugEnabled()) {
        logger.debug("headerPropagation method={}, headers={}, attr={}", call.getMethodDescriptor().getFullMethodName(), headers, call.getAttributes());
    }

    ServerCall.Listener<ReqT> contextPropagateInterceptor = Contexts.interceptCall(newContext, call, headers, next);
    return contextPropagateInterceptor;
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(final ServerCall<ReqT, RespT> serverCall, Metadata headers, ServerCallHandler<ReqT, RespT> serverCallHandler) {
    final Attributes attributes = serverCall.getAttributes();

    final TransportMetadata transportMetadata = attributes.get(MetadataServerTransportFilter.TRANSPORT_METADATA_KEY);
    if (transportMetadata == null) {
        if (logger.isInfoEnabled()) {
            logger.info("Close call. cause=transportMetadata is null, headers={}, attributes={}", headers, serverCall.getAttributes());
        }
        serverCall.close(Status.INTERNAL.withDescription("transportMetadata is null"), new Metadata());
        return new ServerCall.Listener<ReqT>() {
        };
    }

    final Context currentContext = Context.current();
    final Context newContext = currentContext.withValue(ServerContext.getTransportMetadataKey(), transportMetadata);
    if (logger.isDebugEnabled()) {
        logger.debug("bind metadata method={}, headers={}, attr={}", serverCall.getMethodDescriptor().getFullMethodName(), headers, serverCall.getAttributes());
    }
    ServerCall.Listener<ReqT> listener = Contexts.interceptCall(newContext, serverCall, headers, serverCallHandler);
    return listener;
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(final ServerCall<ReqT, RespT> call,
                                                             final Metadata headers,
                                                             final ServerCallHandler<ReqT, RespT> next) {
    Context ctx = Context.current();
    final ServerStream stream = ServerStreamHelper.getServerStream(call);
    if (stream != null) {
        ctx = ctx.withValue(STREAM, stream);
    }
    return Contexts.interceptCall(ctx, call, headers, next);
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(final ServerCall<ReqT, RespT> call,
                                                             final Metadata headers,
                                                             final ServerCallHandler<ReqT, RespT> next) {
    final Context ctx = Context.current() //
        .withValue(REMOTE_ADDRESS, call.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR));
    return Contexts.interceptCall(ctx, call, headers, next);
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
    ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
  // Extract the Span Context from the metadata of the gRPC request
  Context extractedContext = textFormat.extract(Context.current(), headers, getter);
  InetSocketAddress clientInfo =
      (InetSocketAddress) call.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
  // Build a span based on the received context
  try (Scope scope = ContextUtils.withScopedContext(extractedContext)) {
    Span span =
        tracer
            .spanBuilder("helloworld.Greeter/SayHello")
            .setSpanKind(Span.Kind.SERVER)
            .startSpan();
    span.setAttribute("component", "grpc");
    span.setAttribute("rpc.service", "Greeter");
    span.setAttribute("net.peer.ip", clientInfo.getHostString());
    span.setAttribute("net.peer.port", clientInfo.getPort());
    // Process the gRPC call normally
    try {
      span.setStatus(Status.OK);
      return Contexts.interceptCall(Context.current(), call, headers, next);
    } finally {
      span.end();
    }
  }
}
 

@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers,
        ServerCallHandler<ReqT, RespT> next) {
    ScopedBeansContainer container = new ScopedBeansContainer();
    Context context = Context.current().withValue(GRPC_REQUEST_KEY, container);
    context.addListener(this, MoreExecutors.directExecutor());
    return Contexts.interceptCall(context, call, headers, next);
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
    String value = headers.get(CONTEXT_KEY);
    SampleContext context;
    if (value != null) {
        context = new SampleContext(value);
    } else {
        context = CONTEXT_UNDEFINED;
    }
    return Contexts.interceptCall(Context.current().withValue(CALLER_ID_CONTEXT_KEY, context), call, headers, next);
}
 

@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(
    ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
  RequestMetadata meta = requestMetadataFromHeaders(headers);
  if (meta == null) {
    meta = RequestMetadata.getDefaultInstance();
  }
  Context ctx = Context.current().withValue(CONTEXT_KEY, meta);
  return Contexts.interceptCall(ctx, call, headers, next);
}
 

@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
    String clientId = headers.get(CLIENT_ID_HEADER_KEY);
    if (clientId == null || !authenticator.authenticate(clientId)) {
        call.close(Status.UNAUTHENTICATED.withDescription("Invalid or unknown client: " + clientId), headers);
        return NOOP_LISTENER;
    }
    Context context = Context.current().withValue(CLIENT_ID_CONTEXT_KEY, clientId);
    return Contexts.interceptCall(context, call, headers, next);
}
 

@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers,
        ServerCallHandler<ReqT, RespT> next) {
    ScopedBeansContainer container = new ScopedBeansContainer();
    Context context = Context.current().withValue(GRPC_REQUEST_KEY, container);
    context.addListener(this, MoreExecutors.directExecutor());
    return Contexts.interceptCall(context, call, headers, next);
}
 

/**
 * @param call: ServerCall
 * @param requestHeaders : Metadata request headers
 * @param next: ServerCallHandler
 * @param <R>: Request
 * @param <S>: Response
 * @return {@link Contexts}
 */
@Override
public <R, S> Listener<R> interceptCall(
    ServerCall<R, S> call, Metadata requestHeaders, ServerCallHandler<R, S> next) {
  LOGGER.trace("Headers : {}", requestHeaders);
  String methodName = call.getMethodDescriptor().getFullMethodName();
  if (!ModelDBConstants.GRPC_HEALTH_CHECK_METHOD_NAME.equalsIgnoreCase(methodName)) {
    LOGGER.info("methodName: {}", methodName);
  }
  Context context =
      Context.current()
          .withValue(METADATA_INFO, requestHeaders)
          .withValue(METHOD_NAME, methodName);
  ServerCall.Listener<R> delegate = Contexts.interceptCall(context, call, requestHeaders, next);
  ACTIVE_REQUEST_COUNT.incrementAndGet();
  LOGGER.trace("Active Request count {}", ACTIVE_REQUEST_COUNT.get());
  return new ForwardingServerCallListener.SimpleForwardingServerCallListener<R>(delegate) {

    @Override
    public void onCancel() {
      ACTIVE_REQUEST_COUNT.decrementAndGet();
      LOGGER.trace("Decrease Request count oon onCancel()");
      LOGGER.trace("Active Request count {}", ACTIVE_REQUEST_COUNT.get());
      super.onCancel();
    }

    @Override
    public void onComplete() {
      ACTIVE_REQUEST_COUNT.decrementAndGet();
      LOGGER.trace("Decrease Request count on onComplete()");
      LOGGER.trace("Active Request count {}", ACTIVE_REQUEST_COUNT.get());
      super.onComplete();
    }
  };
}
 

@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(
    ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
  RequestMetadata meta = requestMetadataFromHeaders(headers);
  if (meta == null) {
    throw io.grpc.Status.INVALID_ARGUMENT
        .withDescription(
            "RequestMetadata not received from the client for "
                + call.getMethodDescriptor().getFullMethodName())
        .asRuntimeException();
  }
  Context ctx = Context.current().withValue(CONTEXT_KEY, meta);
  return Contexts.interceptCall(ctx, call, headers, next);
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> serverCall,
    Metadata metadata, ServerCallHandler<ReqT, RespT> serverCallHandler) {
  String value = metadata.get(Constant.AUTHORIZATION_METADATA_KEY);

  Status status = Status.OK;
  if (value == null) {
    status = Status.UNAUTHENTICATED.withDescription("Authorization token is missing");
  } else if (!value.startsWith(Constant.BEARER_TYPE)) {
    status = Status.UNAUTHENTICATED.withDescription("Unknown authorization type");
  } else {
    Jws<Claims> claims = null;
    // remove authorization type prefix
    String token = value.substring(Constant.BEARER_TYPE.length()).trim();
    try {
      // verify token signature and parse claims
      claims = parser.parseClaimsJws(token);
    } catch (JwtException e) {
      status = Status.UNAUTHENTICATED.withDescription(e.getMessage()).withCause(e);
    }
    if (claims != null) {
      // set client id into current context
      Context ctx = Context.current()
          .withValue(Constant.CLIENT_ID_CONTEXT_KEY, claims.getBody().getSubject());
      return Contexts.interceptCall(ctx, serverCall, metadata, serverCallHandler);
    }
  }

  serverCall.close(status, new Metadata());
  return new ServerCall.Listener<ReqT>() {
    // noop
  };
}
 

@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(
    ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
  Context ctx = Context.current();
  CallMetricRecorder callMetricRecorder = InternalCallMetricRecorder.CONTEXT_KEY.get(ctx);
  if (callMetricRecorder == null) {
    callMetricRecorder = InternalCallMetricRecorder.newCallMetricRecorder();
    ctx = ctx.withValue(InternalCallMetricRecorder.CONTEXT_KEY, callMetricRecorder);
  }
  final CallMetricRecorder finalCallMetricRecorder = callMetricRecorder;
  ServerCall<ReqT, RespT> trailerAttachingCall =
      new SimpleForwardingServerCall<ReqT, RespT>(call) {
        @Override
        public void close(Status status, Metadata trailers) {
          Map<String, Double> metricValues =
              InternalCallMetricRecorder.finalizeAndDump(finalCallMetricRecorder);
          // Only attach a metric report if there are some metric values to be reported.
          if (!metricValues.isEmpty()) {
            OrcaLoadReport report =
                OrcaLoadReport.newBuilder().putAllRequestCost(metricValues).build();
            trailers.put(ORCA_ENDPOINT_LOAD_METRICS_KEY, report);
          }
          super.close(status, trailers);
        }
      };
  return Contexts.interceptCall(
      ctx,
      trailerAttachingCall,
      headers,
      next);
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
    Context ctx = Context.current().withValue(ctxKey, "ServerAcceptsContext");
    return Contexts.interceptCall(ctx, call, headers, next);
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
    Context wrappedContext = Context.current();

    Object debugValue = headers.get(DEBUG_KEY);
    if (debugValue != null) {
        boolean debugEnabled = Boolean.parseBoolean(debugValue.toString());
        if (debugEnabled) {
            wrappedContext = wrappedContext.withValue(DEBUG_CONTEXT_KEY, "true");
        }
    }
    Object compressionValue = headers.get(COMPRESSION_KEY);
    if (compressionValue != null) {
        String compressionType = compressionValue.toString();
        if (ALLOWED_COMPRESSION_TYPES.contains(compressionType)) {
            call.setCompression(compressionType);
            wrappedContext = wrappedContext.withValue(COMPRESSION_CONTEXT_KEY, compressionType);
        }
    }

    wrappedContext = copyIntoContext(wrappedContext, headers, CALLER_ID_KEY, CALLER_ID_CONTEXT_KEY);
    wrappedContext = copyIntoContext(wrappedContext, headers, CALLER_TYPE_KEY, CALLER_TYPE_CONTEXT_KEY);
    wrappedContext = copyIntoContext(wrappedContext, headers, DIRECT_CALLER_ID_KEY, DIRECT_CALLER_ID_CONTEXT_KEY);
    wrappedContext = copyIntoContext(wrappedContext, headers, CALL_REASON_KEY, CALL_REASON_CONTEXT_KEY);
    wrappedContext = copyDirectCallerContextIntoContext(wrappedContext, call);

    Object callMetadataValue = headers.get(CALL_METADATA_KEY);
    if (callMetadataValue != null) {
        try {
            com.netflix.titus.grpc.protogen.CallMetadata grpcCallMetadata = com.netflix.titus.grpc.protogen.CallMetadata.parseFrom((byte[]) callMetadataValue);
            wrappedContext = wrappedContext.withValue(CALL_METADATA_CONTEXT_KEY, CommonRuntimeGrpcModelConverters.toCallMetadata(grpcCallMetadata));
        } catch (Exception e) {
            // Ignore bad header value.
            logger.info("Invalid CallMetadata in a request header", e);
        }
    }

    return wrappedContext == Context.current()
            ? next.startCall(call, headers)
            : Contexts.interceptCall(wrappedContext, call, headers, next);
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {

    Context context = Context.current();

    // The authorization header has the credentials (e.g., username and password for Basic Authentication).
    // The form of the header is: <Method> <Token> (CustomMethod static-token, or Basic XYZ...., for example)
    String credentials = headers.get(Metadata.Key.of(AuthConstants.AUTHORIZATION, Metadata.ASCII_STRING_MARSHALLER));

    if (!Strings.isNullOrEmpty(credentials)) {
        String[] parts = credentials.split("\\s+", 2);
        if (parts.length == 2) {
            String method = parts[0];
            String token = parts[1];
            if (!Strings.isNullOrEmpty(method)) {
                if (method.equals(handler.getHandlerName())) {
                    log.debug("Handler [{}] successfully matched auth method [{}]", handler, method);
                    Principal principal;
                    try {
                        if ((principal = handler.authenticate(token)) == null) {
                            log.warn("Handler for method [{}] returned a null Principal upon authentication for the"
                                    + "given token", method);
                            call.close(Status.fromCode(Status.Code.UNAUTHENTICATED), headers);
                            return null;
                        }
                    } catch (AuthException e) {
                        log.warn("Authentication failed", e);
                        call.close(Status.fromCode(Status.Code.UNAUTHENTICATED), headers);
                        return null;
                    }
                    // Creates a new Context with the given key/value pairs.
                    context = context.withValues(PRINCIPAL_OBJECT_KEY, principal, AUTH_INTERCEPTOR_OBJECT_KEY, this);
                }
            } else {
                log.debug("Credentials are present, but method [{}] is null or empty", method);
            }
        }
    }

    // reaching this point means that the handler wasn't applicable to this request.
    return Contexts.interceptCall(context, call, headers, next);
}
 

@Override
public <REQ, RESP> Listener<REQ> interceptCall(ServerCall<REQ, RESP> call, Metadata headers,
                                               ServerCallHandler<REQ, RESP> next) {
    final Context grpcContext = Context.current().withValue(CONTEXT_KEY, "value");
    return Contexts.interceptCall(grpcContext, call, headers, next);
}
 

/**
 * Intercept incoming and outgoing messages and enforce any necessary controls
 *
 * @param call the request message
 * @param headers the request metadata
 * @param next the next interceptor in the interceptor chain prior to the service implementation
 * @param <I> The message request type (e.g. ReqT)
 * @param <O> The message reply type (e.g. RespT)
 *
 * @return a listener for the incoming call.
 */
@Override
public <I, O> ServerCall.Listener<I> interceptCall(
        final ServerCall<I, O> call,
        final Metadata headers,
        final ServerCallHandler<I, O> next) {

    final Attributes attributes = call.getAttributes();
    final SocketAddress socketAddress = attributes.get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
    final String clientIp = clientIp(socketAddress);
    String foundSubject = DEFAULT_FOUND_SUBJECT;

    // enforce that the DN on the client cert matches the configured pattern
    final SSLSession sslSession = attributes.get(Grpc.TRANSPORT_ATTR_SSL_SESSION);
    if(this.authorizedDNpattern != null && sslSession != null) {
        try {
            final X509Certificate[] certs = sslSession.getPeerCertificateChain();
            if(certs != null && certs.length > 0) {
                for (final X509Certificate cert : certs) {
                    foundSubject = cert.getSubjectDN().getName();
                    if(authorizedDNpattern.matcher(foundSubject).matches()) {
                        break;
                    } else {
                        logger.warn("Rejecting transfer attempt from " + foundSubject + " because the DN is not authorized, host=" + clientIp);
                        call.close(Status.PERMISSION_DENIED.withDescription(DN_UNAUTHORIZED + foundSubject), headers);
                        return IDENTITY_LISTENER;
                    }
                }
            }
        } catch (final SSLPeerUnverifiedException e) {
            logger.debug("skipping DN authorization for request from {}.", new Object[] {clientIp}, e);
        }
    }
    // contextualize the DN and IP for use in the RPC implementation
    final Context context = Context.current()
            .withValue(REMOTE_HOST_KEY, clientIp)
            .withValue(REMOTE_DN_KEY, foundSubject);

    // if we got to this point, there were no errors, call the next interceptor in the chain
    return Contexts.interceptCall(context, call, headers, next);
}