CircuitBreakingConnectionStateListener implements ConnectionStateListener { @Override public synchronized void stateChanged(CuratorFramework client, ConnectionState newState) { if ( circuitBreaker.isOpen() ) { handleOpenStateChange(newState); } else { handleClosedStateChange(newState); } } CircuitBreakingConnectionStateListener(CuratorFramework client, ConnectionStateListener listener, RetryPolicy retryPolicy);  CircuitBreakingConnectionStateListener(CuratorFramework client, ConnectionStateListener listener, RetryPolicy retryPolicy, ScheduledExecutorService service); @Override synchronized void stateChanged(CuratorFramework client, ConnectionState newState); synchronized boolean isOpen();  }
@Test public void testBasic() throws Exception { RecordingListener recordingListener = new RecordingListener(); TestRetryPolicy retryPolicy = new TestRetryPolicy(); CircuitBreakingConnectionStateListener listener = new CircuitBreakingConnectionStateListener(dummyClient, recordingListener, retryPolicy, service); listener.stateChanged(dummyClient, ConnectionState.RECONNECTED); Assert.assertEquals(timing.takeFromQueue(recordingListener.stateChanges), ConnectionState.RECONNECTED); listener.stateChanged(dummyClient, ConnectionState.SUSPENDED); Assert.assertEquals(timing.takeFromQueue(recordingListener.stateChanges), ConnectionState.SUSPENDED); listener.stateChanged(dummyClient, ConnectionState.SUSPENDED); Assert.assertTrue(recordingListener.stateChanges.isEmpty()); listener.stateChanged(dummyClient, ConnectionState.LOST); Assert.assertEquals(timing.takeFromQueue(recordingListener.stateChanges), ConnectionState.LOST); synchronized(listener) { listener.stateChanged(dummyClient, ConnectionState.READ_ONLY); listener.stateChanged(dummyClient, ConnectionState.RECONNECTED); listener.stateChanged(dummyClient, ConnectionState.SUSPENDED); listener.stateChanged(dummyClient, ConnectionState.LOST); listener.stateChanged(dummyClient, ConnectionState.SUSPENDED); } retryTiming.multiple(2).sleep(); Assert.assertTrue(recordingListener.stateChanges.isEmpty()); retryPolicy.isRetrying = false; Assert.assertEquals(timing.takeFromQueue(recordingListener.stateChanges), ConnectionState.SUSPENDED); }
@Test public void testResetsAfterReconnect() throws Exception { RecordingListener recordingListener = new RecordingListener(); TestRetryPolicy retryPolicy = new TestRetryPolicy(); CircuitBreakingConnectionStateListener listener = new CircuitBreakingConnectionStateListener(dummyClient, recordingListener, retryPolicy, service); synchronized(listener) { listener.stateChanged(dummyClient, ConnectionState.LOST); listener.stateChanged(dummyClient, ConnectionState.LOST); } Assert.assertEquals(timing.takeFromQueue(recordingListener.stateChanges), ConnectionState.LOST); Assert.assertTrue(recordingListener.stateChanges.isEmpty()); listener.stateChanged(dummyClient, ConnectionState.RECONNECTED); Assert.assertEquals(timing.takeFromQueue(recordingListener.stateChanges), ConnectionState.RECONNECTED); }