io.grpc.ServerCall#close ( )源码实例Demo

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

    if (SupervisorServiceGrpc.getServiceDescriptor().getMethods().contains(call.getMethodDescriptor())) {
        // Supervisor API calls are not restricted to the active leader.
        return next.startCall(call, headers);
    }

    if (leaderActivator.isLeader()) {
        if (leaderActivator.isActivated()) {
            return next.startCall(call, headers);
        } else {
            call.close(Status.UNAVAILABLE.withDescription("Titus Master is initializing and not yet available."), new Metadata());
            return new ServerCall.Listener<ReqT>() {
            };
        }
    } else {
        call.close(Status.UNAVAILABLE.withDescription("Titus Master is not leader."), new Metadata());
        return new ServerCall.Listener<ReqT>() {
        };
    }
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
    AdmissionControllerResponse result;
    try {
        AdmissionControllerRequest request = AdmissionControllerRequest.newBuilder()
                .withCallerId(callerIdResolver.get())
                .withEndpointName(call.getMethodDescriptor().getFullMethodName())
                .build();
        result = admissionController.apply(request);
    } catch (Exception e) {
        logger.warn("Admission controller error: {}", e.getMessage());
        logger.debug("Stack trace", e);

        return next.startCall(call, headers);
    }

    if (result.isAllowed()) {
        return next.startCall(call, headers);
    }

    call.close(Status.RESOURCE_EXHAUSTED.withDescription(result.getReasonMessage()), new Metadata());
    return (ServerCall.Listener<ReqT>) NO_OP_LISTENER;
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
    if (!fitRequestHandler.isPresent()) {
        return next.startCall(call, headers);
    }

    FitInjection fitInjection = fitRequestHandler.get();
    String injectionPoint = call.getMethodDescriptor().getFullMethodName();

    // Request failure
    try {
        fitInjection.beforeImmediate(injectionPoint);
    } catch (Exception e) {
        call.close(Status.UNAVAILABLE.withDescription("FIT server failure"), new Metadata());
        return new ServerCall.Listener<ReqT>() {
        };
    }

    // Increased latency.
    return fitInjection.findAction(FitLatencyAction.ACTION_ID)
            .map(action -> {
                int latencyMs = ExceptionExt.doTry(() -> Integer.parseInt(action.getProperties().get("latency"))).orElse(100);
                return new LatencyHandler<>(call, headers, next, latencyMs).getLatencyListener();
            })
            .orElse(next.startCall(call, headers));
}
 

@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 ServerCall.Listener<ByteBuf> startCall(
    final ServerCall<ByteBuf, ByteBuf> call, Metadata headers) {
  call.sendHeaders(new Metadata());
  call.request(1);
  return new ServerCall.Listener<ByteBuf>() {
    @Override
    public void onMessage(ByteBuf message) {
      // no-op
      message.release();
      call.request(1);
      call.sendMessage(genericResponse.slice());
    }

    @Override
    public void onHalfClose() {
      call.close(Status.OK, new Metadata());
    }

    @Override
    public void onCancel() {
    }

    @Override
    public void onComplete() {
    }
  };
}
 

@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(
    ServerCall<ReqT, RespT> serverCall,
    Metadata metadata,
    ServerCallHandler<ReqT, RespT> nextHandler) {
  serverCall.close(status, new Metadata());
  return new Listener<ReqT>() {};
}
 

@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> serverCall, Metadata metadata,
                                                  ServerCallHandler<ReqT, RespT> serverCallHandler) {
    String resourceName = serverCall.getMethodDescriptor().getFullMethodName();
    // Remote address: serverCall.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
    Entry entry = null;
    try {
        ContextUtil.enter(resourceName);
        entry = SphU.entry(resourceName, EntryType.IN);
        // Allow access, forward the call.
        return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(
            serverCallHandler.startCall(
                new ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(serverCall) {
                    @Override
                    public void close(Status status, Metadata trailers) {
                        super.close(status, trailers);
                        // Record the exception metrics.
                        if (!status.isOk()) {
                            recordException(status.asRuntimeException());
                        }
                    }
                }, metadata)) {};
    } catch (BlockException e) {
        serverCall.close(FLOW_CONTROL_BLOCK, new Metadata());
        return new ServerCall.Listener<ReqT>() {};
    } finally {
        if (entry != null) {
            entry.exit();
        }
        ContextUtil.exit();
    }
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
        ServerCall<ReqT, RespT> serverCall, Metadata metadata,
        ServerCallHandler<ReqT, RespT> serverCallHandler) {
    accessLogger.info("gRPC [" +
            serverCall.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR) +
            "]: " + serverCall.getMethodDescriptor().getFullMethodName());

    ServerCall.Listener<ReqT> delegate = serverCallHandler.startCall(serverCall, metadata);
    return new SimpleForwardingServerCallListener<ReqT>(delegate) {
        @Override
        public void onHalfClose() {
            try {
                super.onHalfClose();
            } catch (Exception e) {
                logger.error("Caught an unexpected error.", e);
                serverCall.close(Status.INTERNAL
                        .withCause(e)
                        .withDescription(e.toString() + "\n" + e.getMessage()),
                        new Metadata());
            }
        }

        @Override
        public void onMessage(ReqT request) {
            accessLogger.info("Request Data: " + request);
            super.onMessage(request);
        }
    };
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
    ServerCall<ReqT, RespT> serverCall,
    Metadata metadata,
    ServerCallHandler<ReqT, RespT> serverCallHandler) {
  String token = metadata.get(Metadata.Key.of("token", ASCII_STRING_MARSHALLER));
  System.out.println("Token: " + token);

  boolean tokenIsValid = false;

  try {
    FirebaseToken firebaseToken = FirebaseAuth.getInstance().verifyIdToken(token);
    System.err.println("Email for token: " + firebaseToken.getEmail());

    // TODO: properly validate whether user has rights
    //noinspection ConstantConditions
    if (true) {
      tokenIsValid = true;
    }

  } catch (Exception e) {
    e.printStackTrace();
    System.err.println("Was unable to parse token");
  }

  if (!tokenIsValid) {
    serverCall.close(Status.UNAUTHENTICATED, metadata);
    return new ServerCall.Listener<ReqT>() {};
  } else {
    return serverCallHandler.startCall(serverCall, metadata);
  }
}
 

@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
    MethodDescriptor<ReqT, RespT> methodDescriptor = call.getMethodDescriptor();
    if (notProtectedMethods.contains(methodDescriptor)) {
        return next.startCall(call, headers);
    }

    if (leaderActivationStatus.isActivatedLeader()) {
        return next.startCall(call, headers);
    } else {
        call.close(Status.UNAVAILABLE.withDescription("Not a leader or not ready yet."), new Metadata());
        return new ServerCall.Listener<ReqT>() {
        };
    }
}
 

@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);
}
 

/**
 * intercept point of call.
 *
 * @param serverCall        call of server.
 * @param metadata          of call.
 * @param serverCallHandler handler of call.
 * @param <REQUEST>         of call.
 * @param <RESPONSE>        of call.
 * @return lister of call.
 */
@Override
public <REQUEST, RESPONSE> ServerCall.Listener<REQUEST> interceptCall(ServerCall<REQUEST, RESPONSE> serverCall,
    Metadata metadata, ServerCallHandler<REQUEST, RESPONSE> serverCallHandler) {
    String token = metadata.get(AUTH_HEAD_HEADER_NAME);
    if (expectedToken.equals(token)) {
        return serverCallHandler.startCall(serverCall, metadata);
    } else {
        serverCall.close(Status.PERMISSION_DENIED, new Metadata());
        return listener;
    }
}
 

@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 ServerCall.Listener<ByteBuf> startCall(
    final ServerCall<ByteBuf, ByteBuf> call, Metadata headers) {
  call.sendHeaders(new Metadata());
  call.request(1);
  return new ServerCall.Listener<ByteBuf>() {
    @Override
    public void onMessage(ByteBuf message) {
      // no-op
      message.release();
      call.request(1);
      call.sendMessage(genericResponse.slice());
    }

    @Override
    public void onHalfClose() {
      call.close(Status.OK, new Metadata());
    }

    @Override
    public void onCancel() {
    }

    @Override
    public void onComplete() {
    }
  };
}
 

@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);
}
 

/**
 * Close the call with {@link Status#UNAUTHENTICATED}.
 *
 * @param call The call to close.
 * @param aex The exception that was the cause.
 */
protected void closeCallUnauthenticated(final ServerCall<?, ?> call, final AuthenticationException aex) {
    log.debug(UNAUTHENTICATED_DESCRIPTION, aex);
    call.close(Status.UNAUTHENTICATED.withCause(aex).withDescription(UNAUTHENTICATED_DESCRIPTION), new Metadata());
}
 

/**
 * Close the call with {@link Status#PERMISSION_DENIED}.
 *
 * @param call The call to close.
 * @param aex The exception that was the cause.
 */
protected void closeCallAccessDenied(final ServerCall<?, ?> call, final AccessDeniedException aex) {
    log.debug(ACCESS_DENIED_DESCRIPTION, aex);
    call.close(Status.PERMISSION_DENIED.withCause(aex).withDescription(ACCESS_DENIED_DESCRIPTION), new Metadata());
}
 

/**
 * Close the call with {@link Status#UNAUTHENTICATED}.
 *
 * @param call The call to close.
 * @param aex The exception that was the cause.
 */
protected void closeCallUnauthenticated(final ServerCall<?, ?> call, final AuthenticationException aex) {
    log.debug(UNAUTHENTICATED_DESCRIPTION, aex);
    call.close(Status.UNAUTHENTICATED.withCause(aex).withDescription(UNAUTHENTICATED_DESCRIPTION), new Metadata());
}
 

/**
 * Close the call with {@link Status#PERMISSION_DENIED}.
 *
 * @param call The call to close.
 * @param aex The exception that was the cause.
 */
protected void closeCallAccessDenied(final ServerCall<?, ?> call, final AccessDeniedException aex) {
    log.debug(ACCESS_DENIED_DESCRIPTION, aex);
    call.close(Status.PERMISSION_DENIED.withCause(aex).withDescription(ACCESS_DENIED_DESCRIPTION), new Metadata());
}