import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@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);
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@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);
	}
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@Override
protected AbstractServerImplBuilder<?> getServerBuilder() {
  return NettyServerBuilder.forPort(0)
      .maxMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
      .compressorRegistry(compressors)
      .decompressorRegistry(decompressors)
      .intercept(new ServerInterceptor() {
          @Override
          public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call,
              Metadata headers, ServerCallHandler<ReqT, RespT> next) {
            Listener<ReqT> listener = next.startCall(call, headers);
            // TODO(carl-mastrangelo): check that encoding was set.
            call.setMessageCompression(true);
            return listener;
          }
        });
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@Test
public void callNextTwice() {
  ServerInterceptor interceptor = new ServerInterceptor() {
    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
        ServerCall<ReqT, RespT> call,
        Metadata headers,
        ServerCallHandler<ReqT, RespT> next) {
      // Calling next twice is permitted, although should only rarely be useful.
      assertSame(listener, next.startCall(call, headers));
      return next.startCall(call, headers);
    }
  };
  ServerServiceDefinition intercepted = ServerInterceptors.intercept(serviceDefinition,
      interceptor);
  assertSame(listener,
      getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers));
  verify(handler, times(2)).startCall(same(call), same(headers));
  verifyNoMoreInteractions(handler);
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@Test
public void argumentsPassed() {
  @SuppressWarnings("unchecked")
  final ServerCall<String, Integer> call2 = new NoopServerCall<String, Integer>();
  @SuppressWarnings("unchecked")
  final ServerCall.Listener<String> listener2 = mock(ServerCall.Listener.class);

  ServerInterceptor interceptor = new ServerInterceptor() {
      @SuppressWarnings("unchecked") // Lot's of casting for no benefit.  Not intended use.
      @Override
      public <R1, R2> ServerCall.Listener<R1> interceptCall(
          ServerCall<R1, R2> call,
          Metadata headers,
          ServerCallHandler<R1, R2> next) {
        assertSame(call, ServerInterceptorsTest.this.call);
        assertSame(listener,
            next.startCall((ServerCall<R1, R2>)call2, headers));
        return (ServerCall.Listener<R1>) listener2;
      }
    };
  ServerServiceDefinition intercepted = ServerInterceptors.intercept(
      serviceDefinition, Arrays.asList(interceptor));
  assertSame(listener2,
      getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers));
  verify(handler).startCall(call2, headers);
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@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);
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(
    ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
  RequestMetadata meta = headers.get(METADATA_KEY);
  if (meta == null) {
    throw new IllegalStateException("RequestMetadata not received from the client.");
  }
  Context ctx = Context.current().withValue(CONTEXT_KEY, meta);
  return Contexts.interceptCall(ctx, call, headers, next);
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
/**
 * Set up the registry.
 */
@Setup(Level.Trial)
public void setup() throws Exception {
  registry = new MutableHandlerRegistry();
  fullMethodNames = new ArrayList<String>(serviceCount * methodCountPerService);
  for (int serviceIndex = 0; serviceIndex < serviceCount; ++serviceIndex) {
    String serviceName = randomString();
    ServerServiceDefinition.Builder serviceBuilder = ServerServiceDefinition.builder(serviceName);
    for (int methodIndex = 0; methodIndex < methodCountPerService; ++methodIndex) {
      String methodName = randomString();

      MethodDescriptor<Void, Void> methodDescriptor = MethodDescriptor.<Void, Void>newBuilder()
          .setType(MethodDescriptor.MethodType.UNKNOWN)
          .setFullMethodName(MethodDescriptor.generateFullMethodName(serviceName, methodName))
          .setRequestMarshaller(TestMethodDescriptors.voidMarshaller())
          .setResponseMarshaller(TestMethodDescriptors.voidMarshaller())
          .build();
      serviceBuilder.addMethod(methodDescriptor,
          new ServerCallHandler<Void, Void>() {
            @Override
            public Listener<Void> startCall(ServerCall<Void, Void> call,
                Metadata headers) {
              return null;
            }
          });
      fullMethodNames.add(methodDescriptor.getFullMethodName());
    }
    registry.addService(serviceBuilder.build());
  }
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@Override
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
    ServerCall<ReqT, RespT> call,
    Metadata headers,
    ServerCallHandler<ReqT, RespT> next) {
  return next.startCall(call, headers);
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
public MessagePrincipalListener(ServerCall.Listener<R> delegate,String principal,SeldonGrpcServer server) {
    this.delegate = delegate;
    this.server = server;
    this.principal = principal;
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@Override
protected Listener<R> delegate() {
    return delegate;
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers,
    ServerCallHandler<ReqT, RespT> next) {
  calledInterceptors.add(2);
  return next.startCall(call, headers);
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
public ContextualizedServerCallListener(ServerCall.Listener<ReqT> delegate, Context context) {
  super(delegate);
  this.context = context;
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
public SeldonServerCallListener(ServerCall.Listener<R> delegate,String client,ExternalRpcServer server) {
	this.delegate = delegate;
	this.server = server;
	this.client = client;
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@Override
protected Listener<R> delegate() {
	return delegate;
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@Test(timeout = 10000)
public void shutdownShouldCancelOngoingUploads() throws Exception {
  withEmptyMetadata.attach();
  RemoteRetrier retrier =
      new RemoteRetrier(() -> new FixedBackoff(1, 10), (e) -> true, Retrier.ALLOW_ALL_CALLS);
  ByteStreamUploader uploader =
      new ByteStreamUploader(INSTANCE_NAME, channel, null, 3, retrier, retryService);

  CountDownLatch cancellations = new CountDownLatch(2);

  ServerServiceDefinition service =
      ServerServiceDefinition.builder(ByteStreamGrpc.SERVICE_NAME)
      .addMethod(ByteStreamGrpc.METHOD_WRITE,
          new ServerCallHandler<WriteRequest, WriteResponse>() {
            @Override
            public Listener<WriteRequest> startCall(ServerCall<WriteRequest, WriteResponse> call,
                Metadata headers) {
              // Don't request() any messages from the client, so that the client will be blocked
              // on flow control and thus the call will sit there idle long enough to receive the
              // cancellation.
              return new Listener<WriteRequest>() {
                @Override
                public void onCancel() {
                  cancellations.countDown();
                }
              };
            }
          })
      .build();

  serviceRegistry.addService(service);

  byte[] blob1 = new byte[CHUNK_SIZE];
  Chunker chunker1 = new Chunker(blob1, CHUNK_SIZE, DIGEST_UTIL);

  byte[] blob2 = new byte[CHUNK_SIZE + 1];
  Chunker chunker2 = new Chunker(blob2, CHUNK_SIZE, DIGEST_UTIL);

  ListenableFuture<Void> f1 = uploader.uploadBlobAsync(chunker1);
  ListenableFuture<Void> f2 = uploader.uploadBlobAsync(chunker2);

  assertThat(uploader.uploadsInProgress()).isTrue();

  uploader.shutdown();

  cancellations.await();

  assertThat(f1.isCancelled()).isTrue();
  assertThat(f2.isCancelled()).isTrue();

  blockUntilInternalStateConsistent(uploader);
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
@Override
public <ReqT, RespT> Listener<ReqT> interceptCall(
    ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
  return next.startCall(call, headers);
}
 

import io.grpc.ServerCall.Listener; //导入依赖的package包/类
/**
 * Handle AuthToken Exception.
 * <p>
 * 1. illegal token, need to ask client side to relogin
 * </p>
 * <p>
 * 2. refresh token is expired, need to ask client side to relogin
 * </p>
 * 
 * @param call
 * @param userPrincipal
 * @param tokenAuthenticator
 * @param tokenStatus
 * @return Listener<ReqT>
 */
private <ReqT, RespT> Listener<ReqT> handleAuthTokenException(ServerCall<ReqT, RespT> call, TokenStatus tokenStatus) {
  Metadata trailers = new Metadata();
  // illegal rpc access or the refresh token is expired
  if (TokenStatus.INVALID.getCode().equals(tokenStatus.getCode())
      || TokenStatus.EXPIRED.getCode().equals(tokenStatus.getCode())) {
    trailers.put(errorDetailsKey, ErrorReporter.UNAUTHENTICATED
        .withSpecificErrorMsg(MessageErrorCode.ERROR_UNAUTHENTICATED_KEY, true).toErrorDetails());
    call.close(Status.UNAUTHENTICATED, trailers);
  }
  return new ServerCall.Listener<ReqT>() {};
}