@Before
public void setUp() throws Exception {
    String serverName = InProcessServerBuilder.generateName();

    grpcCleanup.register(InProcessServerBuilder
            .forName(serverName)
            .directExecutor()
            .addService(cacheService)
            .build()
            .start());

    channel = grpcCleanup.register(InProcessChannelBuilder
            .forName(serverName)
            .directExecutor()
            .build());

    when(managedChannelFactory.getInstance()).thenReturn(channel);
}
 

@Before
public void setUp() throws Exception {
  reflectionService = ProtoReflectionService.newInstance();
  Server server =
      InProcessServerBuilder.forName("proto-reflection-test")
          .directExecutor()
          .addService(reflectionService)
          .addService(new ReflectableServiceGrpc.ReflectableServiceImplBase() {})
          .fallbackHandlerRegistry(handlerRegistry)
          .build()
          .start();
  grpcCleanupRule.register(server);
  ManagedChannel channel =
      grpcCleanupRule.register(
          InProcessChannelBuilder.forName("proto-reflection-test").directExecutor().build());
  stub = ServerReflectionGrpc.newStub(channel);
}
 

@Before
public void setUp() throws Exception {
    // Generate a unique in-process server name.
    String serverName = InProcessServerBuilder.generateName();

    // Create a server, add service, start, and register for automatic graceful shutdown.
    grpcCleanup.register(InProcessServerBuilder
            .forName(serverName).directExecutor().addService(serviceImpl).build().start());

    // Create a client channel and register for automatic graceful shutdown.
    ManagedChannel channel = grpcCleanup.register(
            InProcessChannelBuilder.forName(serverName).directExecutor().build());

    // Create a client using the in-process channel;
    client = new GoalStateProvisionerClient(channel);
}
 

@Before
public void setup() throws Exception {
  this.grpcRule = new GrpcCleanupRule();
  // setup fake serving service
  String serverName = InProcessServerBuilder.generateName();
  this.grpcRule.register(
      InProcessServerBuilder.forName(serverName)
          .directExecutor()
          .addService(this.servingMock)
          .build()
          .start());

  // setup test feast client target
  ManagedChannel channel =
      this.grpcRule.register(
          InProcessChannelBuilder.forName(serverName).directExecutor().build());
  this.client = new FeastClient(channel);
}
 

@Before
public void setUp() throws Exception {
  GreeterImplBase greeterImplBase =
      new GreeterImplBase() {
        @Override
        public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
          responseObserver.onNext(HelloReply.getDefaultInstance());
          responseObserver.onCompleted();
        }
      };
  // Generate a unique in-process server name.
  String serverName = InProcessServerBuilder.generateName();
  // Create a server, add service, start, and register for automatic graceful shutdown.
  grpcCleanup.register(InProcessServerBuilder.forName(serverName).directExecutor()
      .addService(ServerInterceptors.intercept(greeterImplBase, new HeaderServerInterceptor()))
      .build().start());
  // Create a client channel and register for automatic graceful shutdown.
  channel =
      grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build());
}
 

@Before
public final void setupChannels() throws IOException {
    if(gRpcServerProperties.isEnabled()) {
        ManagedChannelBuilder<?> channelBuilder = ManagedChannelBuilder.forAddress("localhost", getPort());
        Resource certChain = Optional.ofNullable(gRpcServerProperties.getSecurity())
                .map(GRpcServerProperties.SecurityProperties::getCertChain)
                .orElse(null);
        if(null!= certChain){
            ((NettyChannelBuilder)channelBuilder)
                    .useTransportSecurity()
                    .sslContext(GrpcSslContexts.forClient().trustManager(certChain.getInputStream()).build());
        }else{
            channelBuilder.usePlaintext();
        }


        channel = onChannelBuild(channelBuilder).build();
    }
    if(StringUtils.hasText(gRpcServerProperties.getInProcessServerName())){
        inProcChannel = onChannelBuild(
                            InProcessChannelBuilder.forName(gRpcServerProperties.getInProcessServerName())
                            .usePlaintext()
                        ).build();

    }
}
 

private GrpcRemoteDownloader newDownloader(RemoteCacheClient cacheClient) throws IOException {
  final RemoteOptions remoteOptions = Options.getDefaults(RemoteOptions.class);
  final RemoteRetrier retrier =
      TestUtils.newRemoteRetrier(
          () -> new ExponentialBackoff(remoteOptions),
          RemoteRetrier.RETRIABLE_GRPC_ERRORS,
          retryService);
  final ReferenceCountedChannel channel =
      new ReferenceCountedChannel(
          InProcessChannelBuilder.forName(fakeServerName).directExecutor().build());
  return new GrpcRemoteDownloader(
      channel.retain(),
      Optional.<CallCredentials>empty(),
      retrier,
      withEmptyMetadata,
      cacheClient,
      remoteOptions);
}
 

@Setup
public void setup() throws IOException {
    System.out.println("---------- SETUP ONCE -------------");
    ScheduledExecutorService scheduledExecutorService =
        Executors.newScheduledThreadPool(Runtime.getRuntime()
                                                .availableProcessors());
    reactiveServer =
        InProcessServerBuilder.forName("benchmark-reactiveServer")
                              .scheduledExecutorService(scheduledExecutorService)
                              .addService(new BenchmarkReactorServerServiceImpl(100000))
                              .build()
                              .start();

    reactiveChannel = InProcessChannelBuilder.forName("benchmark-reactiveServer")
                                             .build();
    reactiveClient = ReactorBenchmarkServiceGrpc.newReactorStub(reactiveChannel);
}
 

@Test
public void testRetryOnUnavailable() throws IOException {
  String uniqueName = InProcessServerBuilder.generateName();
  ExecutionImpl service = new ExecutionImpl(Status.UNAVAILABLE, 0);
  InProcessServerBuilder.forName(uniqueName).addService(service).build().start();
  CallCounter beforeRetry = new CallCounter();
  ManagedChannel channel =
      InProcessChannelBuilder.forName(uniqueName)
          .intercept(
              new RetryClientInterceptor(
                  RetryPolicy.builder().setMaxRetries(2).setBeforeRetry(beforeRetry).build()))
          .build();
  ExecutionBlockingStub stub = ExecutionGrpc.newBlockingStub(channel);
  try {
    stub.execute(ExecuteRequest.newBuilder().build()).forEachRemaining(resp -> {});
    Assert.fail("Final retry should cause an exception");
  } catch (StatusRuntimeException ex) {
    Assert.assertEquals(Status.Code.UNAVAILABLE, ex.getStatus().getCode());
  }

  Assert.assertEquals(3, service.calls);
  Assert.assertEquals(2, beforeRetry.calls);
}
 

public TestRemoteExecutionClients(List<BindableService> services) throws IOException {
  eventBus = new DefaultBuckEventBus(new DefaultClock(), new BuildId("dontcare"));
  String serverName = "uniquish-" + new Random().nextLong();

  InProcessServerBuilder serverBuilder =
      InProcessServerBuilder.forName(serverName).directExecutor();
  for (BindableService service : services) {
    serverBuilder.addService(service);
  }

  server = serverBuilder.build().start();
  ManagedChannel channel = InProcessChannelBuilder.forName(serverName).directExecutor().build();

  clients =
      new GrpcRemoteExecutionClients(
          "buck",
          channel,
          channel,
          100,
          MetadataProviderFactory.emptyMetadataProvider(),
          eventBus,
          FakeBuckConfig.builder()
              .build()
              .getView(RemoteExecutionConfig.class)
              .getStrategyConfig());
}
 

@Before
public final void setUp() throws Exception {
  // Use a mutable service registry for later registering the service impl for each test case.
  fakeServer =
      InProcessServerBuilder.forName(fakeServerName)
          .fallbackHandlerRegistry(serviceRegistry)
          .directExecutor()
          .build()
          .start();
  logStream = Mockito.mock(AsynchronousFileOutputStream.class);
  clock = new ManualClock();
  interceptor = new LoggingInterceptor(logStream, clock);
  loggedChannel =
      ClientInterceptors.intercept(
          InProcessChannelBuilder.forName(fakeServerName).directExecutor().build(), interceptor);
}
 

@Before
public void setUp() throws Exception {
  // Generate a unique in-process server name.
  String serverName = InProcessServerBuilder.generateName();

  // Create a server, add service, start, and register for automatic graceful shutdown.
  grpcCleanup.register(InProcessServerBuilder.forName(serverName).
      fallbackHandlerRegistry(serviceRegistry).directExecutor().build().start());

  // Create a client channel and register for automatic graceful shutdown.
  ManagedChannel channel = grpcCleanup.register(
      InProcessChannelBuilder.forName(serverName).directExecutor().build());

  // Create a TccEventServiceStub using the in-process channel;
  service = new GrpcTccClientMessageSender(serviceConfig, channel, address, handler, null);
}
 

@Test
public void AbstractStubFeaturesShouldPropagate() throws Exception {
    com.google.common.base.Preconditions.checkArgument(true);
    Channel channel = InProcessChannelBuilder.forName("ignore").build();
    com.salesforce.jprotoc.GreeterGrpc8.GreeterCompletableFutureStub stub = com.salesforce.jprotoc.GreeterGrpc8
                    .newCompletableFutureStub(channel)
                    .withCompression("bz2")
                    .withMaxInboundMessageSize(42);

    Field innerStubField = com.salesforce.jprotoc.GreeterGrpc8.GreeterCompletableFutureStub.class.getDeclaredField("innerStub");
    innerStubField.setAccessible(true);
    com.salesforce.jprotoc.GreeterGrpc.GreeterFutureStub innerStub = (com.salesforce.jprotoc.GreeterGrpc.GreeterFutureStub) innerStubField.get(stub);

    assertEquals("bz2", stub.getCallOptions().getCompressor());
    assertEquals(new Integer(42), stub.getCallOptions().getMaxInboundMessageSize());

    assertEquals("bz2", innerStub.getCallOptions().getCompressor());
    assertEquals(new Integer(42), innerStub.getCallOptions().getMaxInboundMessageSize());

    assertEquals(stub.getCallOptions().toString(), innerStub.getCallOptions().toString());
}
 

@Before
public void setup() throws IOException {
    final MockJobManagerService mockJobManagerService = new MockJobManagerService();

    testServer = InProcessServerBuilder
            .forName("testServer")
            .directExecutor()
            .addService(mockJobManagerService)
            .build()
            .start();

    final ManagedChannel channel = InProcessChannelBuilder
            .forName("testServer")
            .directExecutor()
            .usePlaintext(true)
            .build();
    final JobManagementServiceStub jobManagementServiceStub = JobManagementServiceGrpc.newStub(channel);
    final JobManagementServiceFutureStub jobManagementServiceFutureStub = JobManagementServiceGrpc.newFutureStub(channel);
    titusClient = new TitusClientImpl(jobManagementServiceStub, jobManagementServiceFutureStub, new DefaultRegistry());
}
 

/**
 * Before the test has started, create the server and channel.
 */
@Override
protected void before() throws Throwable {
  serverName = UUID.randomUUID().toString();

  serviceRegistry = new MutableHandlerRegistry();

  InProcessServerBuilder serverBuilder = InProcessServerBuilder.forName(serverName)
      .fallbackHandlerRegistry(serviceRegistry);

  if (useDirectExecutor) {
    serverBuilder.directExecutor();
  }

  server = serverBuilder.build().start();

  InProcessChannelBuilder channelBuilder = InProcessChannelBuilder.forName(serverName);

  if (useDirectExecutor) {
    channelBuilder.directExecutor();
  }

  channel = channelBuilder.build();
}
 

@BeforeClass
public static void setUp() throws IOException {
  final String name = InProcessServerBuilder.generateName();
  server = InProcessServerBuilder.forName(name)
    .directExecutor()
    .addService(new JobsServiceAdapter(p(LocalJobsService.class)))
    .addService(new Chronicle(p(LocalJobsService.class)))
    .build();
  server.start();

  channel = InProcessChannelBuilder.forName(name)
    .directExecutor()
    .build();

  asyncStub = JobsServiceGrpc.newStub(channel);
  chronicleStub = ChronicleGrpc.newBlockingStub(channel);
}
 

@BeforeClass
public static void setUp() throws Exception {
  final String name = InProcessServerBuilder.generateName();
  server = InProcessServerBuilder.forName(name)
    .directExecutor()
    .addService(new JobsServiceAdapter(p(LocalJobsService.class)))
    .addService(new Chronicle(p(LocalJobsService.class)))
    .build();
  server.start();

  channel = InProcessChannelBuilder.forName(name)
    .directExecutor()
    .build();
  asyncStub = JobsServiceGrpc.newStub(channel);

  query = getFile("tpch_quoted.sql");
}
 

@Before
public void setUp() throws Exception {
  // Generate a unique in-process server name.
  String serverName = InProcessServerBuilder.generateName();

  // Create a server, add service, start, and register for automatic graceful shutdown.
  grpcCleanup.register(InProcessServerBuilder
      .forName(serverName).directExecutor().addService(serviceImpl).build().start());

  // Create a client channel and register for automatic graceful shutdown.
  ManagedChannel channel = grpcCleanup.register(
      InProcessChannelBuilder.forName(serverName).directExecutor().build());

  // Create a HelloWorldClient using the in-process channel;
  client = new HelloWorldClient(channel);
}
 

@Test
public void healthException() throws Exception {
    // Generate a unique in-process server name.
    String serverName = InProcessServerBuilder.generateName();
    HealthCheckAggregator hca = mock(HealthCheckAggregator.class);
    CompletableFuture<HealthCheckStatus> hcsf = mock(CompletableFuture.class);
    when(hcsf.get()).thenThrow(InterruptedException.class);
    when(hca.check()).thenReturn(hcsf);
    HealthServiceImpl healthyService = new HealthServiceImpl(hca);

    addService(serverName, healthyService);
    HealthGrpc.HealthBlockingStub blockingStub = HealthGrpc.newBlockingStub(
            // Create a client channel and register for automatic graceful shutdown.
            grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build()));

    thrown.expect(StatusRuntimeException.class);
    thrown.expect(hasProperty("status", is(Status.INTERNAL)));
    blockingStub.check(HealthCheckRequest.newBuilder().build());

}
 

@Test
public void putAddsExpiration() throws IOException, InterruptedException {
  ByteString uploadContent = ByteString.copyFromUtf8("uploaded");
  Digest digest = DIGEST_UTIL.compute(uploadContent);
  String instanceName = "test";
  ListMultimap<Digest, Runnable> onExpirations =
      MultimapBuilder.hashKeys().arrayListValues().build();
  Channel channel = InProcessChannelBuilder.forName(fakeServerName).directExecutor().build();
  ByteStreamUploader uploader = mock(ByteStreamUploader.class);
  GrpcCAS cas = new GrpcCAS(instanceName, channel, uploader, onExpirations);
  Runnable onExpiration = mock(Runnable.class);
  cas.put(new Blob(uploadContent, digest), onExpiration);
  verify(uploader, times(1))
      .uploadBlob(eq(HashCode.fromString(digest.getHash())), any(Chunker.class));
  assertThat(onExpirations.get(digest)).containsExactly(onExpiration);
  verifyZeroInteractions(onExpiration);
}
 

@Test
public void missingWriteQueryIsNotFound() throws IOException {
  ByteString helloWorld = ByteString.copyFromUtf8("Hello, World!");
  Digest digest = DIGEST_UTIL.compute(helloWorld);
  String uuid = UUID.randomUUID().toString();
  String resourceName = createBlobUploadResourceName(uuid, digest);

  Channel channel = InProcessChannelBuilder.forName(fakeServerName).directExecutor().build();
  ByteStreamBlockingStub service = ByteStreamGrpc.newBlockingStub(channel);

  StatusRuntimeException notFoundException = null;
  try {
    service.queryWriteStatus(
        QueryWriteStatusRequest.newBuilder().setResourceName(resourceName).build());
  } catch (StatusRuntimeException e) {
    assertThat(Status.fromThrowable(e).getCode()).isEqualTo(Code.NOT_FOUND);
    notFoundException = e;
  }
  assertThat(notFoundException).isNotNull();
}
 

@Test
public void completedWriteQueryIsFound() throws IOException, InterruptedException {
  ByteString helloWorld = ByteString.copyFromUtf8("Hello, World!");
  Digest digest = DIGEST_UTIL.compute(helloWorld);
  String uuid = UUID.randomUUID().toString();
  String resourceName = createBlobUploadResourceName(uuid, digest);

  when(simpleBlobStore.containsKey(digest.getHash())).thenReturn(true);

  Channel channel = InProcessChannelBuilder.forName(fakeServerName).directExecutor().build();
  ByteStreamBlockingStub service = ByteStreamGrpc.newBlockingStub(channel);
  QueryWriteStatusResponse response =
      service.queryWriteStatus(
          QueryWriteStatusRequest.newBuilder().setResourceName(resourceName).build());
  assertThat(response)
      .isEqualTo(
          QueryWriteStatusResponse.newBuilder()
              .setCommittedSize(digest.getSizeBytes())
              .setComplete(true)
              .build());
  verify(simpleBlobStore, times(1)).containsKey(eq(digest.getHash()));
}
 

@Test
public void missingBlobReadIsNotFound() {
  ByteString helloWorld = ByteString.copyFromUtf8("Hello, World!");
  Digest digest = DIGEST_UTIL.compute(helloWorld);

  Channel channel = InProcessChannelBuilder.forName(fakeServerName).directExecutor().build();
  ByteStreamBlockingStub service = ByteStreamGrpc.newBlockingStub(channel);

  when(simpleBlobStore.get(eq(digest.getHash()), any(OutputStream.class)))
      .thenReturn(immediateFuture(false));
  ReadRequest request =
      ReadRequest.newBuilder().setResourceName(createBlobDownloadResourceName(digest)).build();
  StatusRuntimeException notFoundException = null;
  try {
    if (service.read(request).hasNext()) {
      fail("no responses should be available");
    }
  } catch (StatusRuntimeException e) {
    assertThat(Status.fromThrowable(e).getCode()).isEqualTo(Code.NOT_FOUND);
    notFoundException = e;
  }
  assertThat(notFoundException).isNotNull();
}
 

@Test
public void healthServing() throws Exception {
    // Generate a unique in-process server name.
    String serverName = InProcessServerBuilder.generateName();
    HealthCheckAggregator hca = mock(HealthCheckAggregator.class);
    CompletableFuture<HealthCheckStatus> hcsf = mock(CompletableFuture.class);
    HealthCheckStatus hcs = mock(HealthCheckStatus.class);
    when(hcs.isHealthy()).thenReturn(true);
    when(hcsf.get()).thenReturn(hcs);
    when(hca.check()).thenReturn(hcsf);
    HealthServiceImpl healthyService = new HealthServiceImpl(hca);

    addService(serverName, healthyService);
    HealthGrpc.HealthBlockingStub blockingStub = HealthGrpc.newBlockingStub(
            // Create a client channel and register for automatic graceful shutdown.
            grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build()));


    HealthCheckResponse reply = blockingStub.check(HealthCheckRequest.newBuilder().build());

    assertEquals(HealthCheckResponse.ServingStatus.SERVING, reply.getStatus());
}
 

@Test
public void healthNotServing() throws Exception {
    // Generate a unique in-process server name.
    String serverName = InProcessServerBuilder.generateName();
    HealthCheckAggregator hca = mock(HealthCheckAggregator.class);
    CompletableFuture<HealthCheckStatus> hcsf = mock(CompletableFuture.class);
    HealthCheckStatus hcs = mock(HealthCheckStatus.class);
    when(hcs.isHealthy()).thenReturn(false);
    when(hcsf.get()).thenReturn(hcs);
    when(hca.check()).thenReturn(hcsf);
    HealthServiceImpl healthyService = new HealthServiceImpl(hca);

    addService(serverName, healthyService);
    HealthGrpc.HealthBlockingStub blockingStub = HealthGrpc.newBlockingStub(
            // Create a client channel and register for automatic graceful shutdown.
            grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build()));


    HealthCheckResponse reply = blockingStub.check(HealthCheckRequest.newBuilder().build());

    assertEquals(HealthCheckResponse.ServingStatus.NOT_SERVING, reply.getStatus());
}
 

private RegistryGrpc.RegistryBlockingStub getSchemaRegistryStub(MetaStore metaStore)
    throws IOException {
  String serverName = InProcessServerBuilder.generateName();
  grpcCleanup.register(
      InProcessServerBuilder.forName(serverName)
          .directExecutor()
          .addService(new RegistryService(metaStore))
          .build()
          .start());
  return RegistryGrpc.newBlockingStub(
      grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build()));
}
 

@Before
public void setUp() throws Exception {
  // Generate a unique in-process server name.
  String serverName = InProcessServerBuilder.generateName();
  features = new ArrayList<>();
  // Use directExecutor for both InProcessServerBuilder and InProcessChannelBuilder can reduce the
  // usage timeouts and latches in test. But we still add timeout and latches where they would be
  // needed if no directExecutor were used, just for demo purpose.
  server = new RouteGuideServer(
      InProcessServerBuilder.forName(serverName).directExecutor(), 0, features);
  server.start();
  // Create a client channel and register for automatic graceful shutdown.
  inProcessChannel = grpcCleanup.register(
      InProcessChannelBuilder.forName(serverName).directExecutor().build());
}
 

@Before
public void setUp() throws Exception {
  // Generate a unique in-process server name.
  String serverName = InProcessServerBuilder.generateName();
  // Use a mutable service registry for later registering the service impl for each test case.
  grpcCleanup.register(InProcessServerBuilder.forName(serverName)
      .fallbackHandlerRegistry(serviceRegistry).directExecutor().build().start());
  client =
      new RouteGuideClient(InProcessChannelBuilder.forName(serverName).directExecutor());
  client.setTestHelper(testHelper);
}
 

@Test
public void clientHeaderDeliveredToServer() throws Exception {
  // Generate a unique in-process server name.
  String serverName = InProcessServerBuilder.generateName();
  // Create a server, add service, start, and register for automatic graceful shutdown.
  grpcCleanup.register(InProcessServerBuilder.forName(serverName).directExecutor()
      .addService(ServerInterceptors.intercept(new GreeterImplBase() {}, mockServerInterceptor))
      .build().start());
  // Create a client channel and register for automatic graceful shutdown.
  ManagedChannel channel = grpcCleanup.register(
      InProcessChannelBuilder.forName(serverName).directExecutor().build());
  GreeterBlockingStub blockingStub = GreeterGrpc.newBlockingStub(
      ClientInterceptors.intercept(channel, new HeaderClientInterceptor()));
  ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);

  try {
    blockingStub.sayHello(HelloRequest.getDefaultInstance());
    fail();
  } catch (StatusRuntimeException expected) {
    // expected because the method is not implemented at server side
  }

  verify(mockServerInterceptor).interceptCall(
      Matchers.<ServerCall<HelloRequest, HelloReply>>any(),
      metadataCaptor.capture(),
      Matchers.<ServerCallHandler<HelloRequest, HelloReply>>any());
  assertEquals(
      "customRequestValue",
      metadataCaptor.getValue().get(HeaderClientInterceptor.CUSTOM_HEADER_KEY));
}
 

@Test
public void blockingServerStreamingCall_interruptedWaitsForOnClose() throws Exception {
  Integer req = 2;

  class NoopServerStreamingMethod implements ServerStreamingMethod<Integer, Integer> {
    ServerCallStreamObserver<Integer> observer;

    @Override public void invoke(Integer request, StreamObserver<Integer> responseObserver) {
      observer = (ServerCallStreamObserver<Integer>) responseObserver;
    }
  }

  NoopServerStreamingMethod methodImpl = new NoopServerStreamingMethod();
  server = InProcessServerBuilder.forName("noop").directExecutor()
      .addService(ServerServiceDefinition.builder("some")
          .addMethod(SERVER_STREAMING_METHOD, ServerCalls.asyncServerStreamingCall(methodImpl))
          .build())
      .build().start();

  InterruptInterceptor interceptor = new InterruptInterceptor();
  channel = InProcessChannelBuilder.forName("noop")
      .directExecutor()
      .intercept(interceptor)
      .build();
  Iterator<Integer> iter = ClientCalls.blockingServerStreamingCall(
      channel.newCall(SERVER_STREAMING_METHOD, CallOptions.DEFAULT), req);
  try {
    iter.next();
    fail();
  } catch (StatusRuntimeException ex) {
    assertTrue(Thread.interrupted());
    assertTrue("interrupted", ex.getCause() instanceof InterruptedException);
  }
  assertTrue("onCloseCalled", interceptor.onCloseCalled);
  assertTrue("context not cancelled", methodImpl.observer.isCancelled());
}