You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@curator.apache.org by ca...@apache.org on 2016/08/31 01:01:54 UTC
[3/3] curator git commit: Merge branch 'master' into CURATOR-3.0
Merge branch 'master' into CURATOR-3.0
Conflicts:
curator-recipes/src/main/java/org/apache/curator/framework/recipes/shared/SharedValue.java
curator-recipes/src/test/java/org/apache/curator/framework/recipes/shared/TestSharedCount.java
pom.xml
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/9e400bc8
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/9e400bc8
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/9e400bc8
Branch: refs/heads/CURATOR-3.0
Commit: 9e400bc8620430c245db86d08bafafe2ecb8534f
Parents: 9612c5f 022de39
Author: Cam McKenzie <ca...@apache.org>
Authored: Wed Aug 31 11:01:23 2016 +1000
Committer: Cam McKenzie <ca...@apache.org>
Committed: Wed Aug 31 11:01:23 2016 +1000
----------------------------------------------------------------------
.../framework/recipes/shared/SharedValue.java | 24 ++++-
.../recipes/shared/TestSharedCount.java | 106 +++++++++++++++++++
2 files changed, 127 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/9e400bc8/curator-recipes/src/main/java/org/apache/curator/framework/recipes/shared/SharedValue.java
----------------------------------------------------------------------
diff --cc curator-recipes/src/main/java/org/apache/curator/framework/recipes/shared/SharedValue.java
index 7c2febd,1f9df37..1a3d889
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/shared/SharedValue.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/shared/SharedValue.java
@@@ -22,7 -22,8 +22,9 @@@ package org.apache.curator.framework.re
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.WatcherRemoveCuratorFramework;
+ import org.apache.curator.framework.api.BackgroundCallback;
+ import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.listen.ListenerContainer;
import org.apache.curator.framework.state.ConnectionState;
http://git-wip-us.apache.org/repos/asf/curator/blob/9e400bc8/curator-recipes/src/test/java/org/apache/curator/framework/recipes/shared/TestSharedCount.java
----------------------------------------------------------------------
diff --cc curator-recipes/src/test/java/org/apache/curator/framework/recipes/shared/TestSharedCount.java
index 28df3f9,7939f6e..0690d6a
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/shared/TestSharedCount.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/shared/TestSharedCount.java
@@@ -23,8 -23,11 +23,12 @@@ import com.google.common.collect.Lists
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.imps.TestCleanState;
+ import org.apache.curator.framework.api.BackgroundCallback;
+ import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.state.ConnectionState;
+ import org.apache.curator.framework.state.ConnectionStateListener;
+ import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.test.BaseClassForTests;
import org.apache.curator.test.Timing;
@@@ -321,4 -288,105 +326,105 @@@ public class TestSharedCount extends Ba
CloseableUtils.closeQuietly(client1);
}
}
+
+
+ @Test
+ public void testDisconnectEventOnWatcherDoesNotRetry() throws Exception
+ {
+ final CountDownLatch gotSuspendEvent = new CountDownLatch(1);
+
+ CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryNTimes(10, 1000));
+ curatorFramework.start();
+ curatorFramework.blockUntilConnected();
+
+ SharedCount sharedCount = new SharedCount(curatorFramework, "/count", 10);
+ sharedCount.start();
+
+ curatorFramework.getConnectionStateListenable().addListener(new ConnectionStateListener() {
+ @Override
+ public void stateChanged(CuratorFramework client, ConnectionState newState) {
+ if (newState == ConnectionState.SUSPENDED) {
+ gotSuspendEvent.countDown();
+ }
+ }
+ });
+
+ try
+ {
+ server.stop();
+ // if watcher goes into 10second retry loop we won't get timely notification
+ Assert.assertTrue(gotSuspendEvent.await(5, TimeUnit.SECONDS));
+ }
+ finally
+ {
+ CloseableUtils.closeQuietly(sharedCount);
- CloseableUtils.closeQuietly(curatorFramework);
++ TestCleanState.closeAndTestClean(curatorFramework);
+ }
+ }
+
+ @Test
+ public void testDisconnectReconnectEventDoesNotFireValueWatcher() throws Exception
+ {
+ final CountDownLatch gotSuspendEvent = new CountDownLatch(1);
+ final CountDownLatch gotChangeEvent = new CountDownLatch(1);
+ final CountDownLatch getReconnectEvent = new CountDownLatch(1);
+
+ final AtomicInteger numChangeEvents = new AtomicInteger(0);
+
+
+ CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryNTimes(10, 500));
+ curatorFramework.start();
+ curatorFramework.blockUntilConnected();
+
+ SharedCount sharedCount = new SharedCount(curatorFramework, "/count", 10);
+
+ sharedCount.addListener(new SharedCountListener() {
+ @Override
+ public void countHasChanged(SharedCountReader sharedCount, int newCount) throws Exception {
+ numChangeEvents.incrementAndGet();
+ gotChangeEvent.countDown();
+ }
+
+ @Override
+ public void stateChanged(CuratorFramework client, ConnectionState newState) {
+ if (newState == ConnectionState.SUSPENDED) {
+ gotSuspendEvent.countDown();
+ } else if (newState == ConnectionState.RECONNECTED) {
+ getReconnectEvent.countDown();
+ }
+ }
+ });
+ sharedCount.start();
+
+ try
+ {
+ sharedCount.setCount(11);
+ Assert.assertTrue(gotChangeEvent.await(2, TimeUnit.SECONDS));
+
+ server.stop();
+ Assert.assertTrue(gotSuspendEvent.await(2, TimeUnit.SECONDS));
+
+ server.restart();
+ Assert.assertTrue(getReconnectEvent.await(2, TimeUnit.SECONDS));
+
+ sharedCount.trySetCount(sharedCount.getVersionedValue(), 12);
+
+ // flush background task queue
+ final CountDownLatch flushDone = new CountDownLatch(1);
+ curatorFramework.getData().inBackground(new BackgroundCallback() {
+ @Override
+ public void processResult(CuratorFramework client, CuratorEvent event) throws Exception {
+ flushDone.countDown();
+ }
+ }).forPath("/count");
+ flushDone.await(5, TimeUnit.SECONDS);
+
+ Assert.assertEquals(2, numChangeEvents.get());
+ }
+ finally
+ {
+ CloseableUtils.closeQuietly(sharedCount);
- CloseableUtils.closeQuietly(curatorFramework);
++ TestCleanState.closeAndTestClean(curatorFramework);
+ }
+ }
}