@Override
public void watch(HealthCheckRequest request,
    StreamObserver<HealthCheckResponse> responseObserver) {
  final ServerSideCall call = new ServerSideCall(request, responseObserver);
  Context.current().addListener(
      new CancellationListener() {
        @Override
        public void cancelled(Context ctx) {
          call.cancelled = true;
        }
      }, MoreExecutors.directExecutor());
  calls.add(call);
}
 

@Override
public void streamCoreMetrics(
    OrcaLoadReportRequest request, StreamObserver<OrcaLoadReport> responseObserver) {
  final ServerSideCall call = new ServerSideCall(request, responseObserver);
  Context.current()
      .addListener(
          new CancellationListener() {
            @Override
            public void cancelled(Context ctx) {
              call.cancelled = true;
            }
          },
          MoreExecutors.directExecutor());
  calls.add(call);
}
 

@Override
public void watch(HealthCheckRequest request,
    StreamObserver<HealthCheckResponse> responseObserver) {
  final ServerSideCall call = new ServerSideCall(request, responseObserver);
  Context.current().addListener(
      new CancellationListener() {
        @Override
        public void cancelled(Context ctx) {
          call.cancelled = true;
        }
      }, MoreExecutors.directExecutor());
  calls.add(call);
}
 

@SuppressWarnings("unchecked")
@Before
public void setUp() throws Exception {
  MockitoAnnotations.initMocks(this);
  mockLoadReportingService = mock(LoadReportingServiceGrpc.LoadReportingServiceImplBase.class,
      delegatesTo(
          new LoadReportingServiceGrpc.LoadReportingServiceImplBase() {
            @Override
            public StreamObserver<LoadStatsRequest> streamLoadStats(
                final StreamObserver<LoadStatsResponse> responseObserver) {
              assertThat(callEnded.get()).isTrue();  // ensure previous call was ended
              callEnded.set(false);
              Context.current().addListener(
                  new CancellationListener() {
                    @Override
                    public void cancelled(Context context) {
                      callEnded.set(true);
                    }
                  }, MoreExecutors.directExecutor());
              StreamObserver<LoadStatsRequest> requestObserver =
                  mock(StreamObserver.class);
              lrsRequestObservers.add(requestObserver);
              return requestObserver;
            }
          }
      ));
  cleanupRule.register(InProcessServerBuilder.forName("fakeLoadReportingServer").directExecutor()
      .addService(mockLoadReportingService).build().start());
  channel = cleanupRule.register(
      InProcessChannelBuilder.forName("fakeLoadReportingServer").directExecutor().build());
  when(backoffPolicyProvider.get()).thenReturn(backoffPolicy1, backoffPolicy2);
  when(backoffPolicy1.nextBackoffNanos())
      .thenReturn(TimeUnit.SECONDS.toNanos(1L), TimeUnit.SECONDS.toNanos(10L));
  when(backoffPolicy2.nextBackoffNanos())
      .thenReturn(TimeUnit.SECONDS.toNanos(1L), TimeUnit.SECONDS.toNanos(10L));
  lrsClient =
      new LoadReportClient(
          logId,
          TARGET_NAME,
          channel,
          NODE,
          syncContext,
          fakeClock.getScheduledExecutorService(),
          backoffPolicyProvider,
          fakeClock.getStopwatchSupplier());
  lrsClient.startLoadReporting(callback);
}