@Override
  public final Scoped onStateChange(Consumer<NodeState> listener) {
    // Wrap listeners in a reference of our own to guarantee uniqueness for listener references.
    AtomicReference<Consumer<NodeState>> listenerRef = new AtomicReference<>(listener);
    synchronized (mListeners) {
      Preconditions.checkState(mListeners.add(listenerRef), "listener already exists");
    }
    return () -> {
      synchronized (mListeners) {
        mListeners.remove(listenerRef);
      }
    };
  }
@Test(timeout = TIMEOUT)
  public void onStateChange() {
    AtomicInteger primaryCounter = new AtomicInteger(0);
    AtomicInteger standbyCounter = new AtomicInteger(0);
    Scoped listener = mSelector.onStateChange(state -> {
      if (state.equals(NodeState.PRIMARY)) {
        primaryCounter.incrementAndGet();
      } else {
        standbyCounter.incrementAndGet();
      }
    });
    for (int i = 0; i < 10; i++) {
      mSelector.setState(NodeState.PRIMARY);
      mSelector.setState(NodeState.STANDBY);
    }
    assertEquals(10, primaryCounter.get());
    assertEquals(10, standbyCounter.get());
    listener.close();
    mSelector.setState(NodeState.PRIMARY);
    mSelector.setState(NodeState.STANDBY);
    assertEquals(10, primaryCounter.get());
    assertEquals(10, standbyCounter.get());
  }