@Override
    public void watch(HealthCheckRequest request, StreamObserver<HealthCheckResponse> responseObserver) {
        final String service = request.getService();
        synchronized (watchLock) {
            HealthCheckResponse.ServingStatus status = statusMap.get(service);
            responseObserver.onNext(getResponseForWatch(status));
            IdentityHashMap<StreamObserver<HealthCheckResponse>, Boolean> serviceWatchers = watchers.get(service);
            if (serviceWatchers == null) {
                serviceWatchers = new IdentityHashMap<>();
                watchers.put(service, serviceWatchers);
            }
            serviceWatchers.put(responseObserver, Boolean.TRUE);
        }
        RpcContext.getCancellationContext().addListener(context -> {
            synchronized (watchLock) {
                IdentityHashMap<StreamObserver<HealthCheckResponse>, Boolean> serviceWatchers = watchers.get(service);
                if (serviceWatchers != null) {
                    serviceWatchers.remove(responseObserver);
                    if (serviceWatchers.isEmpty()) {
                        watchers.remove(service);
                    }
                }
            }
        });
    }
@Test
    void testWatch() throws Exception {
        TriHealthImpl triHealth = new TriHealthImpl();

        HealthCheckRequest request =
                HealthCheckRequest.newBuilder().setService("testWatch").build();
        triHealth.setStatus(request.getService(), ServingStatus.SERVING);
        StreamObserver<HealthCheckResponse> streamObserver = new MockStreamObserver();

        RpcContext.removeCancellationContext();
        // test watch
        triHealth.watch(request, streamObserver);
        Assertions.assertNotNull(RpcContext.getCancellationContext().getListeners());
        HashMap<String, IdentityHashMap<StreamObserver<HealthCheckResponse>, Boolean>> watches = getWatches(triHealth);
        Assertions.assertTrue(watches.containsKey(request.getService()));
        Assertions.assertTrue(watches.get(request.getService()).containsKey(streamObserver));
        Assertions.assertTrue(watches.get(request.getService()).get(streamObserver));
        MockStreamObserver mockStreamObserver = (MockStreamObserver) streamObserver;
        Assertions.assertEquals(mockStreamObserver.getCount(), 1);
        Assertions.assertEquals(
                mockStreamObserver.getResponse().getStatus(), HealthCheckResponse.ServingStatus.SERVING);

        // test setStatus
        triHealth.setStatus(request.getService(), HealthCheckResponse.ServingStatus.SERVICE_UNKNOWN);
        Assertions.assertEquals(mockStreamObserver.getCount(), 2);
        Assertions.assertEquals(
                mockStreamObserver.getResponse().getStatus(), HealthCheckResponse.ServingStatus.SERVICE_UNKNOWN);

        triHealth.enterTerminalState();
        Assertions.assertEquals(mockStreamObserver.getCount(), 3);
        Assertions.assertEquals(
                mockStreamObserver.getResponse().getStatus(), HealthCheckResponse.ServingStatus.NOT_SERVING);

        // test clearStatus
        turnOffTerminal(triHealth);
        triHealth.clearStatus(request.getService());
        Assertions.assertEquals(mockStreamObserver.getCount(), 4);
        Assertions.assertEquals(
                mockStreamObserver.getResponse().getStatus(), HealthCheckResponse.ServingStatus.SERVICE_UNKNOWN);

        // test listener
        RpcContext.getCancellationContext().close();
        Assertions.assertTrue(watches.isEmpty());
    }