@Test public void userInterceptor_throwsOnOnHalfClose() throws IOException {
  init(new ServerInterceptor() {
    @Override public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
        ServerCall<ReqT, RespT> call, Metadata metadata, ServerCallHandler<ReqT, RespT> next) {
      return new SimpleForwardingServerCallListener<ReqT>(next.startCall(call, metadata)) {
        @Override public void onHalfClose() {
          throw new IllegalStateException("I'm a bad interceptor.");
        }
      };
    }
  });

  assertThatThrownBy(() -> GreeterGrpc.newBlockingStub(client).sayHello(HELLO_REQUEST))
      .isInstanceOf(StatusRuntimeException.class);
  testSpanHandler.takeRemoteSpanWithErrorMessage(Span.Kind.SERVER, "I'm a bad interceptor.");
}
 

@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> call,
        Metadata headers,
        ServerCallHandler<ReqT, RespT> next) {
    try {
        ServerCall.Listener<ReqT> delegate = next.startCall(call, headers);
        return new SimpleForwardingServerCallListener<ReqT>(delegate) {
            @Override
            public void onComplete() {
                try {
                    super.onComplete();
                } finally {
                    SecurityContextHolder.clearContext();
                    log.debug("SecurityContextHolder now cleared, as request processing completed");
                }
            }

            @Override
            public void onCancel() {
                try {
                    super.onCancel();
                } finally {
                    SecurityContextHolder.clearContext();
                    log.debug("SecurityContextHolder now cleared, as request processing was canceled");
                }
            }
        };
    } catch (Throwable t) {
        SecurityContextHolder.clearContext();
        throw t;
    }
}
 

@Before
public void setUp() {
  MockitoAnnotations.initMocks(this);
  forwarder = new SimpleForwardingServerCallListener<Integer>(serverCallListener) {};
}
 

@Before
public void setUp() {
  forwarder = new SimpleForwardingServerCallListener<Integer>(serverCallListener) {};
}
 

/**
 * This shows that a {@link ServerInterceptor} can see the server server span when processing the
 * request and response.
 */
@Test public void bodyTaggingExample() throws IOException {
  SpanCustomizer customizer = CurrentSpanCustomizer.create(tracing);
  AtomicInteger sends = new AtomicInteger();
  AtomicInteger recvs = new AtomicInteger();

  init(new ServerInterceptor() {
    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
        ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
      call = new SimpleForwardingServerCall<ReqT, RespT>(call) {
        @Override public void sendMessage(RespT message) {
          delegate().sendMessage(message);
          customizer.tag("grpc.message_send." + sends.getAndIncrement(), message.toString());
        }
      };
      return new SimpleForwardingServerCallListener<ReqT>(next.startCall(call, headers)) {
        @Override public void onMessage(ReqT message) {
          customizer.tag("grpc.message_recv." + recvs.getAndIncrement(), message.toString());
          delegate().onMessage(message);
        }
      };
    }
  });

  GreeterGrpc.newBlockingStub(client).sayHello(HELLO_REQUEST);

  assertThat(testSpanHandler.takeRemoteSpan(Span.Kind.SERVER).tags()).containsKeys(
      "grpc.message_recv.0", "grpc.message_send.0"
  );

  Iterator<HelloReply> replies = GreeterGrpc.newBlockingStub(client)
      .sayHelloWithManyReplies(HELLO_REQUEST);
  assertThat(replies).toIterable().hasSize(10);

  // Intentionally verbose here to show that only one recv and 10 replies
  assertThat(testSpanHandler.takeRemoteSpan(Span.Kind.SERVER).tags()).containsKeys(
      "grpc.message_recv.1",
      "grpc.message_send.1",
      "grpc.message_send.2",
      "grpc.message_send.3",
      "grpc.message_send.4",
      "grpc.message_send.5",
      "grpc.message_send.6",
      "grpc.message_send.7",
      "grpc.message_send.8",
      "grpc.message_send.9",
      "grpc.message_send.10"
  );
}