@Override
  public void listenToCluster(final String clusterName,
                              final LoadBalancerStateListenerCallback callback)
  {
    trace(_log, "listenToCluster: ", clusterName);

    // wrap the callback since we need to wait for both uri and cluster listeners to
    // onInit before letting the callback know that we're done.
    final LoadBalancerStateListenerCallback wrappedCallback =
        new LoadBalancerStateListenerCallback()
        {
          private final AtomicInteger _count = new AtomicInteger(2);

          @Override
          public void done(int type, String name)
          {
            if (_count.decrementAndGet() <= 0)
            {
              callback.done(type, clusterName);
            }
          }
        };

    _clusterSubscriber.ensureListening(clusterName, wrappedCallback);
    _uriSubscriber.ensureListening(clusterName, wrappedCallback);
  }
@Test(groups = { "small", "back-end" })
  public void testListenToCluster() throws URISyntaxException,
      InterruptedException
  {
    reset();

    List<String> schemes = new ArrayList<>();

    schemes.add("http");

    assertFalse(_state.isListeningToCluster("cluster-1"));
    assertNull(_state.getClusterProperties("cluster-1"));

    final CountDownLatch latch = new CountDownLatch(1);
    LoadBalancerStateListenerCallback callback = new LoadBalancerStateListenerCallback()
    {
      @Override
      public void done(int type, String name)
      {
        latch.countDown();
      }
    };

    _state.listenToCluster("cluster-1", callback);

    if (!latch.await(5, TimeUnit.SECONDS))
    {
      fail("didn't get callback when listenToCluster was called");
    }

    assertTrue(_state.isListeningToCluster("cluster-1"));
    assertNotNull(_state.getClusterProperties("cluster-1"));
    assertNull(_state.getClusterProperties("cluster-1").getProperty());

    ClusterProperties property = new ClusterProperties("cluster-1", schemes);

    _clusterRegistry.put("cluster-1", property);

    assertTrue(_state.isListeningToCluster("cluster-1"));
    assertNotNull(_state.getClusterProperties("cluster-1"));
    assertEquals(_state.getClusterProperties("cluster-1").getProperty(), property);
  }