You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@curator.apache.org by ra...@apache.org on 2018/06/24 15:51:38 UTC
[1/4] curator git commit: Add new service cache listener to let users
know what instances have changed, while preserving backward compatibility.
Repository: curator
Updated Branches:
refs/heads/CURATOR-470 [created] c0c0ecad5
Add new service cache listener to let users know what instances have changed, while preserving backward compatibility.
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/88e8d9a1
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/88e8d9a1
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/88e8d9a1
Branch: refs/heads/CURATOR-470
Commit: 88e8d9a15d4f1690a6df1c1cabb355936f327aaa
Parents: 9383aa3
Author: Qin Meijie <me...@eques.com.cn>
Authored: Fri Jun 8 13:06:15 2018 +0800
Committer: Qin Meijie <me...@eques.com.cn>
Committed: Fri Jun 8 13:06:15 2018 +0800
----------------------------------------------------------------------
.../details/ServiceCacheEventListener.java | 50 +++++++++
.../x/discovery/details/ServiceCacheImpl.java | 109 +++++++++++++------
.../curator/x/discovery/TestServiceCache.java | 80 ++++++++++++++
3 files changed, 207 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/88e8d9a1/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java
new file mode 100644
index 0000000..1f783f3
--- /dev/null
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java
@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.curator.x.discovery.details;
+
+import org.apache.curator.x.discovery.ServiceInstance;
+
+/**
+ * Listener for events (addition/update/deletion) that happen to a service cache
+ */
+public interface ServiceCacheEventListener<T> extends ServiceCacheListener
+{
+
+ /**
+ * Called when a new cache is added.
+ *
+ * @param added instance added
+ */
+ public void cacheAdded(ServiceInstance<T> added);
+
+ /**
+ * Called when a cache is deleted.
+ *
+ * @param deleted instance deleted
+ */
+ public void cacheDeleted(ServiceInstance<T> deleted);
+
+ /**
+ * Called when a cache is updated.
+ *
+ * @param old old instance
+ * @param updated updated instance
+ */
+ public void cacheUpdated(ServiceInstance<T> old, ServiceInstance<T> updated);
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/88e8d9a1/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
index b8f39d5..df6696a 100644
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
@@ -93,7 +93,7 @@ public class ServiceCacheImpl<T> implements ServiceCache<T>, PathChildrenCacheLi
cache.start(true);
for ( ChildData childData : cache.getCurrentData() )
{
- addInstance(childData, true);
+ addInstanceOnlyIfAbsent(childData);
}
discovery.cacheOpened(this);
}
@@ -146,40 +146,72 @@ public class ServiceCacheImpl<T> implements ServiceCache<T>, PathChildrenCacheLi
@Override
public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception
{
- boolean notifyListeners = false;
- switch ( event.getType() )
+ final Tuple<T> tuple;
+ switch ( event.getType() )
{
case CHILD_ADDED:
+ tuple = addOrUpdateInstance(event.getData());
+ listenerContainer.forEach(
+ new Function<ServiceCacheListener, Void>()
+ {
+ @Override
+ public Void apply(ServiceCacheListener listener)
+ {
+ listener.cacheChanged();
+
+ if(listener instanceof ServiceCacheEventListener) {
+ ((ServiceCacheEventListener) listener).cacheAdded(tuple.newInstance);
+ }
+
+ return null;
+ }
+ }
+ );
+ break;
case CHILD_UPDATED:
{
- addInstance(event.getData(), false);
- notifyListeners = true;
+ tuple = addOrUpdateInstance(event.getData());
+ listenerContainer.forEach(
+ new Function<ServiceCacheListener, Void>()
+ {
+ @Override
+ public Void apply(ServiceCacheListener listener)
+ {
+ listener.cacheChanged();
+
+ if(listener instanceof ServiceCacheEventListener) {
+ ((ServiceCacheEventListener) listener).cacheUpdated(tuple.oldInstance, tuple.newInstance);
+ }
+
+ return null;
+ }
+ }
+ );
break;
}
case CHILD_REMOVED:
{
- instances.remove(instanceIdFromData(event.getData()));
- notifyListeners = true;
+ final ServiceInstance<T> serviceInstance = instances.remove(instanceIdFromData(event.getData()));
+ listenerContainer.forEach(
+ new Function<ServiceCacheListener, Void>()
+ {
+ @Override
+ public Void apply(ServiceCacheListener listener)
+ {
+ listener.cacheChanged();
+
+ if(listener instanceof ServiceCacheEventListener) {
+ ((ServiceCacheEventListener) listener).cacheDeleted(serviceInstance);
+ }
+
+ return null;
+ }
+ }
+ );
break;
}
}
-
- if ( notifyListeners )
- {
- listenerContainer.forEach
- (
- new Function<ServiceCacheListener, Void>()
- {
- @Override
- public Void apply(ServiceCacheListener listener)
- {
- listener.cacheChanged();
- return null;
- }
- }
- );
- }
}
private String instanceIdFromData(ChildData childData)
@@ -187,18 +219,31 @@ public class ServiceCacheImpl<T> implements ServiceCache<T>, PathChildrenCacheLi
return ZKPaths.getNodeFromPath(childData.getPath());
}
- private void addInstance(ChildData childData, boolean onlyIfAbsent) throws Exception
+ private void addInstanceOnlyIfAbsent(ChildData childData) throws Exception
{
String instanceId = instanceIdFromData(childData);
ServiceInstance<T> serviceInstance = discovery.getSerializer().deserialize(childData.getData());
- if ( onlyIfAbsent )
- {
- instances.putIfAbsent(instanceId, serviceInstance);
- }
- else
- {
- instances.put(instanceId, serviceInstance);
- }
+
+ instances.putIfAbsent(instanceId, serviceInstance);
+ cache.clearDataBytes(childData.getPath(), childData.getStat().getVersion());
+ }
+
+ private Tuple<T> addOrUpdateInstance(ChildData childData) throws Exception
+ {
+ String instanceId = instanceIdFromData(childData);
+ ServiceInstance<T> serviceInstance = discovery.getSerializer().deserialize(childData.getData());
+ final Tuple<T> result = new Tuple<>(instances.put(instanceId, serviceInstance), serviceInstance);
cache.clearDataBytes(childData.getPath(), childData.getStat().getVersion());
+ return result;
+ }
+
+ private static class Tuple<T> {
+ public final ServiceInstance<T> oldInstance;
+ public final ServiceInstance<T> newInstance;
+
+ private Tuple(final ServiceInstance<T> oldInstance, final ServiceInstance<T> newInstance) {
+ this.oldInstance = oldInstance;
+ this.newInstance = newInstance;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/curator/blob/88e8d9a1/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java b/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java
index fda5c26..65b7cb9 100644
--- a/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java
+++ b/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java
@@ -28,6 +28,7 @@ import org.apache.curator.test.BaseClassForTests;
import org.apache.curator.test.ExecuteCalledWatchingExecutorService;
import org.apache.curator.test.Timing;
import org.apache.curator.utils.CloseableUtils;
+import org.apache.curator.x.discovery.details.ServiceCacheEventListener;
import org.apache.curator.x.discovery.details.ServiceCacheListener;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -39,6 +40,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
public class TestServiceCache extends BaseClassForTests
{
@@ -310,4 +312,82 @@ public class TestServiceCache extends BaseClassForTests
}
}
}
+
+ @Test
+ public void testServiceCacheEventListener() throws Exception
+ {
+ List<Closeable> closeables = Lists.newArrayList();
+ try
+ {
+ CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+ closeables.add(client);
+ client.start();
+
+ ServiceDiscovery<String> discovery = ServiceDiscoveryBuilder.builder(String.class).basePath("/discovery").client(client).build();
+ closeables.add(discovery);
+ discovery.start();
+
+ ServiceCache<String> cache = discovery.serviceCacheBuilder().name("test").build();
+ closeables.add(cache);
+
+ final CountDownLatch latch = new CountDownLatch(6);
+
+ final AtomicBoolean notifyError = new AtomicBoolean(false);
+ ServiceCacheListener listener = new ServiceCacheEventListener<String>()
+ {
+ @Override
+ public void cacheAdded(final ServiceInstance<String> added) {
+ latch.countDown();
+
+ notifyError.compareAndSet(false,added == null);
+ }
+
+ @Override
+ public void cacheDeleted(final ServiceInstance<String> deleted) {
+ latch.countDown();
+
+ notifyError.compareAndSet(false,deleted == null);
+ }
+
+ @Override
+ public void cacheUpdated(final ServiceInstance<String> before, final ServiceInstance<String> after) {
+ latch.countDown();
+
+ notifyError.compareAndSet(false, !"before".equals(before.getPayload()));
+ notifyError.compareAndSet(false, !"after".equals(after.getPayload()));
+ }
+
+ @Override
+ public void cacheChanged()
+ {
+ latch.countDown();
+ }
+
+ @Override
+ public void stateChanged(CuratorFramework client, ConnectionState newState)
+ {
+
+ }
+ };
+ cache.addListener(listener);
+ cache.start();
+
+ ServiceInstance<String> instance = ServiceInstance.<String>builder().payload("before").name("test").port(10064).build();
+ discovery.registerService(instance);
+ instance = ServiceInstance.<String>builder().id(instance.getId()).payload("after").name("test").port(10064).build();
+ discovery.updateService(instance);
+ discovery.unregisterService(instance);
+ Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+ Assert.assertFalse(notifyError.get());
+ }
+ finally
+ {
+ Collections.reverse(closeables);
+ for ( Closeable c : closeables )
+ {
+ CloseableUtils.closeQuietly(c);
+ }
+ }
+ }
}
[2/4] curator git commit: Merge branch 'CURATOR-470' of
github.com:JerryChin/curator into CURATOR-470
Posted by ra...@apache.org.
Merge branch 'CURATOR-470' of github.com:JerryChin/curator into CURATOR-470
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/6b1522c5
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/6b1522c5
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/6b1522c5
Branch: refs/heads/CURATOR-470
Commit: 6b1522c5f2892451be45df89e6537799f146ab51
Parents: ba8ade0 88e8d9a
Author: randgalt <ra...@apache.org>
Authored: Sun Jun 24 09:26:35 2018 -0500
Committer: randgalt <ra...@apache.org>
Committed: Sun Jun 24 09:26:35 2018 -0500
----------------------------------------------------------------------
.../details/ServiceCacheEventListener.java | 50 ++++++++
.../x/discovery/details/ServiceCacheImpl.java | 115 +++++++++++++------
.../curator/x/discovery/TestServiceCache.java | 80 +++++++++++++
3 files changed, 213 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/6b1522c5/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
----------------------------------------------------------------------
diff --cc curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
index d1a31ad,df6696a..ffaf1a4
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
@@@ -97,23 -91,9 +97,23 @@@ public class ServiceCacheImpl<T> implem
Preconditions.checkState(state.compareAndSet(State.LATENT, State.STARTED), "Cannot be started more than once");
cache.start(true);
+ if ( debugStartLatch != null )
+ {
+ debugStartLatch.countDown();
+ debugStartLatch = null;
+ }
+ if ( debugStartWaitLatch != null )
+ {
+ debugStartWaitLatch.await();
+ debugStartWaitLatch = null;
+ }
+
for ( ChildData childData : cache.getCurrentData() )
{
- addInstanceOnlyIfAbsent(childData);
+ if ( childData.getData() != null ) // else already processed by the cache listener
+ {
- addInstance(childData, true);
++ addInstanceOnlyIfAbsent(childData);
+ }
}
discovery.cacheOpened(this);
}
@@@ -166,21 -146,69 +166,75 @@@
@Override
public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception
{
- boolean notifyListeners = false;
- switch ( event.getType() )
+ final Tuple<T> tuple;
+ switch ( event.getType() )
{
case CHILD_ADDED:
+ tuple = addOrUpdateInstance(event.getData());
+ listenerContainer.forEach(
+ new Function<ServiceCacheListener, Void>()
+ {
+ @Override
+ public Void apply(ServiceCacheListener listener)
+ {
+ listener.cacheChanged();
+
- if(listener instanceof ServiceCacheEventListener) {
++ if ( listener instanceof ServiceCacheEventListener )
++ {
++ //noinspection unchecked
+ ((ServiceCacheEventListener) listener).cacheAdded(tuple.newInstance);
+ }
+
+ return null;
+ }
+ }
+ );
+ break;
case CHILD_UPDATED:
{
- addInstance(event.getData(), false);
- notifyListeners = true;
+ tuple = addOrUpdateInstance(event.getData());
+ listenerContainer.forEach(
+ new Function<ServiceCacheListener, Void>()
+ {
+ @Override
+ public Void apply(ServiceCacheListener listener)
+ {
+ listener.cacheChanged();
+
- if(listener instanceof ServiceCacheEventListener) {
- ((ServiceCacheEventListener) listener).cacheUpdated(tuple.oldInstance, tuple.newInstance);
++ if( listener instanceof ServiceCacheEventListener )
++ {
++ //noinspection unchecked
++ ((ServiceCacheEventListener) listener).cacheUpdated(tuple.oldInstance, tuple.newInstance);
+ }
+
+ return null;
+ }
+ }
+ );
break;
}
case CHILD_REMOVED:
{
- instances.remove(instanceIdFromData(event.getData()));
- notifyListeners = true;
+ final ServiceInstance<T> serviceInstance = instances.remove(instanceIdFromData(event.getData()));
+ listenerContainer.forEach(
+ new Function<ServiceCacheListener, Void>()
+ {
+ @Override
+ public Void apply(ServiceCacheListener listener)
+ {
+ listener.cacheChanged();
+
- if(listener instanceof ServiceCacheEventListener) {
- ((ServiceCacheEventListener) listener).cacheDeleted(serviceInstance);
++ if ( listener instanceof ServiceCacheEventListener )
++ {
++ //noinspection unchecked
++ ((ServiceCacheEventListener) listener).cacheDeleted(serviceInstance);
+ }
+
+ return null;
+ }
+ }
+ );
break;
}
}
[4/4] curator git commit: CURATOR-470
Posted by ra...@apache.org.
CURATOR-470
Introduces an alternate cache listener, ServiceCacheEventListener, that gives more detail about changes to the cache as opposed to the original version which merely denotes a change.
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/c0c0ecad
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/c0c0ecad
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/c0c0ecad
Branch: refs/heads/CURATOR-470
Commit: c0c0ecad59af4ca20c277790bb0e877cfb3fb5bd
Parents: 2ebd456
Author: randgalt <ra...@apache.org>
Authored: Sun Jun 24 10:48:36 2018 -0500
Committer: randgalt <ra...@apache.org>
Committed: Sun Jun 24 10:50:31 2018 -0500
----------------------------------------------------------------------
.../details/ServiceCacheEventListener.java | 9 +-
.../x/discovery/details/ServiceCacheImpl.java | 156 +++++++++----------
.../details/ServiceCacheListenerWrapper.java | 53 -------
.../curator/x/discovery/TestServiceCache.java | 43 +++--
4 files changed, 95 insertions(+), 166 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/c0c0ecad/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java
index e6345c4..7a3b570 100644
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java
@@ -18,30 +18,29 @@
*/
package org.apache.curator.x.discovery.details;
-import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.x.discovery.ServiceInstance;
/**
* Listener for events (addition/update/deletion) that happen to a service cache
*/
-public interface ServiceCacheEventListener<T> extends ConnectionStateListener
+public interface ServiceCacheEventListener<T>
{
/**
- * Called when a new cache is added.
+ * Called when a new instance is added.
*
* @param added instance added
*/
void cacheAdded(ServiceInstance<T> added);
/**
- * Called when a cache is deleted.
+ * Called when an instance is deleted.
*
* @param deleted instance deleted
*/
void cacheDeleted(ServiceInstance<T> deleted);
/**
- * Called when a cache is updated.
+ * Called when an instance is updated.
*
* @param old old instance
* @param updated updated instance
http://git-wip-us.apache.org/repos/asf/curator/blob/c0c0ecad/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
index 47449e8..715182b 100644
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
@@ -37,7 +37,6 @@ import org.apache.curator.x.discovery.ServiceCache;
import org.apache.curator.x.discovery.ServiceInstance;
import java.io.IOException;
import java.util.List;
-import java.util.Objects;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
@@ -47,7 +46,8 @@ import java.util.concurrent.atomic.AtomicReference;
public class ServiceCacheImpl<T> implements ServiceCache<T>, PathChildrenCacheListener
{
- private final ListenerContainer<ServiceCacheEventListener<T>> listenerContainer = new ListenerContainer<>();
+ private final ListenerContainer<ServiceCacheEventListener<T>> eventListenerContainer = new ListenerContainer<>();
+ private final ListenerContainer<ServiceCacheListener> listenerContainer = new ListenerContainer<>();
private final ServiceDiscoveryImpl<T> discovery;
private final AtomicReference<State> state = new AtomicReference<>(State.LATENT);
private final PathChildrenCache cache;
@@ -126,18 +126,19 @@ public class ServiceCacheImpl<T> implements ServiceCache<T>, PathChildrenCacheLi
Preconditions.checkState(state.compareAndSet(State.STARTED, State.STOPPED), "Already closed or has not been started");
listenerContainer.forEach
- (
- new Function<ServiceCacheEventListener<T>, Void>()
+ (
+ new Function<ServiceCacheListener, Void>()
+ {
+ @Override
+ public Void apply(ServiceCacheListener listener)
{
- @Override
- public Void apply(ServiceCacheEventListener<T> listener)
- {
- discovery.getClient().getConnectionStateListenable().removeListener(listener);
- return null;
- }
+ discovery.getClient().getConnectionStateListenable().removeListener(listener);
+ return null;
}
- );
+ }
+ );
listenerContainer.clear();
+ eventListenerContainer.clear();
CloseableUtils.closeQuietly(cache);
@@ -147,115 +148,98 @@ public class ServiceCacheImpl<T> implements ServiceCache<T>, PathChildrenCacheLi
@Override
public Listenable<ServiceCacheEventListener<T>> getCacheEventListenable()
{
- return listenerContainer;
+ return eventListenerContainer;
}
@Override
public void addListener(ServiceCacheListener listener)
{
- ServiceCacheListenerWrapper<T> wrapped = ServiceCacheListenerWrapper.<T>wrap(listener);
- listenerContainer.addListener(wrapped);
- discovery.getClient().getConnectionStateListenable().addListener(wrapped);
+ listenerContainer.addListener(listener);
+ discovery.getClient().getConnectionStateListenable().addListener(listener);
}
@Override
public void addListener(ServiceCacheListener listener, Executor executor)
{
- ServiceCacheListenerWrapper<T> wrapped = ServiceCacheListenerWrapper.<T>wrap(listener);
- listenerContainer.addListener(wrapped, executor);
- discovery.getClient().getConnectionStateListenable().addListener(wrapped, executor);
+ listenerContainer.addListener(listener, executor);
+ discovery.getClient().getConnectionStateListenable().addListener(listener, executor);
}
@Override
- public void removeListener(final ServiceCacheListener listener)
+ public void removeListener(ServiceCacheListener listener)
{
- listenerContainer.forEach
- (
- new Function<ServiceCacheEventListener<T>, Void>()
- {
- @Override
- public Void apply(ServiceCacheEventListener<T> eventListener)
- {
- if ( Objects.equals(ServiceCacheListenerWrapper.unwrap(eventListener), listener) )
- {
- listenerContainer.removeListener(eventListener);
- discovery.getClient().getConnectionStateListenable().removeListener(eventListener);
- }
- return null;
- }
- }
- );
+ listenerContainer.removeListener(listener);
+ discovery.getClient().getConnectionStateListenable().removeListener(listener);
}
@Override
public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception
{
+ boolean notifyListeners = false;
switch ( event.getType() )
{
case CHILD_ADDED:
+ case CHILD_UPDATED:
{
- final Tuple<T> tuple = addOrUpdateInstance(event.getData());
- listenerContainer.forEach
- (
- new Function<ServiceCacheEventListener<T>, Void>()
- {
- @Override
- public Void apply(ServiceCacheEventListener<T> listener)
- {
- listener.cacheAdded(tuple.newInstance);
- return null;
- }
- }
- );
+ notifyListeners = true;
+ applyTuple(addOrUpdateInstance(event.getData()));
break;
}
- case CHILD_UPDATED:
+ case CHILD_REMOVED:
{
- final Tuple<T> tuple = addOrUpdateInstance(event.getData());
- listenerContainer.forEach
- (
- new Function<ServiceCacheEventListener<T>, Void>()
- {
- @Override
- public Void apply(ServiceCacheEventListener<T> listener)
- {
- if ( tuple.oldInstance != null )
- {
- listener.cacheUpdated(tuple.oldInstance, tuple.newInstance);
- }
- else
- {
- listener.cacheAdded(tuple.newInstance);
- }
- return null;
- }
- }
- );
+ notifyListeners = true;
+ final ServiceInstance<T> serviceInstance = instances.remove(instanceIdFromData(event.getData()));
+ applyTuple(new Tuple<T>(serviceInstance, null));
break;
}
+ }
- case CHILD_REMOVED:
+ if ( notifyListeners )
+ {
+ listenerContainer.forEach
+ (
+ new Function<ServiceCacheListener, Void>()
+ {
+ @Override
+ public Void apply(ServiceCacheListener listener)
+ {
+ listener.cacheChanged();
+ return null;
+ }
+ }
+ );
+ }
+ }
+
+ private void applyTuple(final Tuple<T> tuple)
+ {
+ eventListenerContainer.forEach
+ (
+ new Function<ServiceCacheEventListener<T>, Void>()
{
- final ServiceInstance<T> serviceInstance = instances.remove(instanceIdFromData(event.getData()));
- if ( serviceInstance != null )
+ @Override
+ public Void apply(ServiceCacheEventListener<T> listener)
{
- listenerContainer.forEach
- (
- new Function<ServiceCacheEventListener<T>, Void>()
+ if ( tuple.oldInstance != null )
+ {
+ if ( tuple.newInstance != null )
+ {
+ listener.cacheUpdated(tuple.oldInstance, tuple.newInstance);
+ }
+ else
{
- @Override
- public Void apply(ServiceCacheEventListener<T> listener)
- {
- listener.cacheDeleted(serviceInstance);
- return null;
- }
+ listener.cacheDeleted(tuple.oldInstance);
}
- );
+ }
+ else if ( tuple.newInstance != null )
+ {
+ listener.cacheAdded(tuple.newInstance);
+ }
+ return null;
}
- break;
}
- }
+ );
}
private String instanceIdFromData(ChildData childData)
@@ -282,8 +266,8 @@ public class ServiceCacheImpl<T> implements ServiceCache<T>, PathChildrenCacheLi
}
private static class Tuple<T> {
- public final ServiceInstance<T> oldInstance;
- public final ServiceInstance<T> newInstance;
+ private final ServiceInstance<T> oldInstance;
+ private final ServiceInstance<T> newInstance;
private Tuple(final ServiceInstance<T> oldInstance, final ServiceInstance<T> newInstance) {
this.oldInstance = oldInstance;
http://git-wip-us.apache.org/repos/asf/curator/blob/c0c0ecad/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheListenerWrapper.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheListenerWrapper.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheListenerWrapper.java
deleted file mode 100644
index 6f14178..0000000
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheListenerWrapper.java
+++ /dev/null
@@ -1,53 +0,0 @@
-package org.apache.curator.x.discovery.details;
-
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.state.ConnectionState;
-import org.apache.curator.x.discovery.ServiceInstance;
-
-class ServiceCacheListenerWrapper<T> implements ServiceCacheEventListener<T>
-{
- private final ServiceCacheListener listener;
-
- static <T> ServiceCacheListenerWrapper<T> wrap(ServiceCacheListener listener)
- {
- return new ServiceCacheListenerWrapper<>(listener);
- }
-
- static ServiceCacheListener unwrap(ServiceCacheEventListener<?> eventListener)
- {
- if ( eventListener instanceof ServiceCacheListenerWrapper )
- {
- return ((ServiceCacheListenerWrapper)eventListener).listener;
- }
- return null;
- }
-
- private ServiceCacheListenerWrapper(ServiceCacheListener listener)
- {
- this.listener = listener;
- }
-
- @Override
- public void stateChanged(CuratorFramework client, ConnectionState newState)
- {
- listener.stateChanged(client, newState);
- }
-
- @Override
- public void cacheAdded(ServiceInstance<T> added)
- {
- listener.cacheChanged();
- }
-
- @Override
- public void cacheDeleted(ServiceInstance<T> deleted)
- {
- listener.cacheChanged();
- }
-
- @Override
- public void cacheUpdated(ServiceInstance<T> old, ServiceInstance<T> updated)
- {
- listener.cacheChanged();
- }
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/c0c0ecad/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java b/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java
index 70b41e3..509df7b 100644
--- a/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java
+++ b/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java
@@ -44,6 +44,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
public class TestServiceCache extends BaseClassForTests
{
+ private final Timing timing = new Timing();
+
@Test
public void testInitialLoad() throws Exception
{
@@ -106,8 +108,6 @@ public class TestServiceCache extends BaseClassForTests
@Test
public void testViaProvider() throws Exception
{
- Timing timing = new Timing();
-
List<Closeable> closeables = Lists.newArrayList();
try
{
@@ -330,36 +330,33 @@ public class TestServiceCache extends BaseClassForTests
ServiceCache<String> cache = discovery.serviceCacheBuilder().name("test").build();
closeables.add(cache);
- final CountDownLatch latch = new CountDownLatch(3);
-
- final AtomicBoolean notifyError = new AtomicBoolean(false);
+ final Semaphore latch = new Semaphore(0);
+ final AtomicBoolean validation = new AtomicBoolean(true);
ServiceCacheEventListener<String> listener = new ServiceCacheEventListener<String>()
{
@Override
- public void cacheAdded(final ServiceInstance<String> added) {
- latch.countDown();
-
- notifyError.compareAndSet(false,added == null);
- }
-
- @Override
- public void cacheDeleted(final ServiceInstance<String> deleted) {
- latch.countDown();
+ public void cacheAdded(final ServiceInstance<String> added)
+ {
+ latch.release();
- notifyError.compareAndSet(false,deleted == null);
+ validation.compareAndSet(true,added != null);
}
@Override
- public void cacheUpdated(final ServiceInstance<String> before, final ServiceInstance<String> after) {
- latch.countDown();
+ public void cacheDeleted(final ServiceInstance<String> deleted)
+ {
+ latch.release();
- notifyError.compareAndSet(false, !"before".equals(before.getPayload()));
- notifyError.compareAndSet(false, !"after".equals(after.getPayload()));
+ validation.compareAndSet(true,deleted != null);
}
@Override
- public void stateChanged(CuratorFramework client, ConnectionState newState)
+ public void cacheUpdated(final ServiceInstance<String> before, final ServiceInstance<String> after)
{
+ latch.release();
+
+ validation.compareAndSet(true, "before".equals(before.getPayload()));
+ validation.compareAndSet(true, "after".equals(after.getPayload()));
}
};
cache.getCacheEventListenable().addListener(listener);
@@ -367,12 +364,14 @@ public class TestServiceCache extends BaseClassForTests
ServiceInstance<String> instance = ServiceInstance.<String>builder().payload("before").name("test").port(10064).build();
discovery.registerService(instance);
+ Assert.assertTrue(timing.acquireSemaphore(latch));
instance = ServiceInstance.<String>builder().id(instance.getId()).payload("after").name("test").port(10064).build();
discovery.updateService(instance);
+ Assert.assertTrue(timing.acquireSemaphore(latch));
discovery.unregisterService(instance);
- Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
+ Assert.assertTrue(timing.acquireSemaphore(latch));
- Assert.assertFalse(notifyError.get());
+ Assert.assertTrue(validation.get());
}
finally
{
[3/4] curator git commit: wip
Posted by ra...@apache.org.
wip
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/2ebd4569
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/2ebd4569
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/2ebd4569
Branch: refs/heads/CURATOR-470
Commit: 2ebd456930d5970380d68af9c6b7150a2cf24abf
Parents: 6b1522c
Author: randgalt <ra...@apache.org>
Authored: Sun Jun 24 10:21:04 2018 -0500
Committer: randgalt <ra...@apache.org>
Committed: Sun Jun 24 10:21:04 2018 -0500
----------------------------------------------------------------------
.../curator/x/discovery/ServiceCache.java | 12 +-
.../details/ServiceCacheEventListener.java | 10 +-
.../x/discovery/details/ServiceCacheImpl.java | 140 +++++++++++--------
.../details/ServiceCacheListenerWrapper.java | 53 +++++++
.../curator/x/discovery/TestServiceCache.java | 13 +-
5 files changed, 150 insertions(+), 78 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/2ebd4569/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceCache.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceCache.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceCache.java
index a122d69..0214bf8 100644
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceCache.java
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceCache.java
@@ -20,6 +20,7 @@ package org.apache.curator.x.discovery;
import org.apache.curator.framework.listen.Listenable;
import org.apache.curator.x.discovery.details.InstanceProvider;
+import org.apache.curator.x.discovery.details.ServiceCacheEventListener;
import org.apache.curator.x.discovery.details.ServiceCacheListener;
import java.io.Closeable;
import java.util.List;
@@ -33,12 +34,19 @@ public interface ServiceCache<T> extends Closeable, Listenable<ServiceCacheListe
*
* @return the list
*/
- public List<ServiceInstance<T>> getInstances();
+ List<ServiceInstance<T>> getInstances();
/**
* The cache must be started before use
*
* @throws Exception errors
*/
- public void start() throws Exception;
+ void start() throws Exception;
+
+ /**
+ * Returns the listenable container over the newer {@link org.apache.curator.x.discovery.details.ServiceCacheEventListener}
+ *
+ * @return listenable
+ */
+ Listenable<ServiceCacheEventListener<T>> getCacheEventListenable();
}
http://git-wip-us.apache.org/repos/asf/curator/blob/2ebd4569/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java
index 1f783f3..e6345c4 100644
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheEventListener.java
@@ -18,27 +18,27 @@
*/
package org.apache.curator.x.discovery.details;
+import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.x.discovery.ServiceInstance;
/**
* Listener for events (addition/update/deletion) that happen to a service cache
*/
-public interface ServiceCacheEventListener<T> extends ServiceCacheListener
+public interface ServiceCacheEventListener<T> extends ConnectionStateListener
{
-
/**
* Called when a new cache is added.
*
* @param added instance added
*/
- public void cacheAdded(ServiceInstance<T> added);
+ void cacheAdded(ServiceInstance<T> added);
/**
* Called when a cache is deleted.
*
* @param deleted instance deleted
*/
- public void cacheDeleted(ServiceInstance<T> deleted);
+ void cacheDeleted(ServiceInstance<T> deleted);
/**
* Called when a cache is updated.
@@ -46,5 +46,5 @@ public interface ServiceCacheEventListener<T> extends ServiceCacheListener
* @param old old instance
* @param updated updated instance
*/
- public void cacheUpdated(ServiceInstance<T> old, ServiceInstance<T> updated);
+ void cacheUpdated(ServiceInstance<T> old, ServiceInstance<T> updated);
}
http://git-wip-us.apache.org/repos/asf/curator/blob/2ebd4569/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
index ffaf1a4..47449e8 100644
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheImpl.java
@@ -23,6 +23,7 @@ import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import org.apache.curator.framework.listen.Listenable;
import org.apache.curator.utils.CloseableExecutorService;
import org.apache.curator.utils.CloseableUtils;
import org.apache.curator.framework.CuratorFramework;
@@ -36,6 +37,7 @@ import org.apache.curator.x.discovery.ServiceCache;
import org.apache.curator.x.discovery.ServiceInstance;
import java.io.IOException;
import java.util.List;
+import java.util.Objects;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
@@ -45,9 +47,9 @@ import java.util.concurrent.atomic.AtomicReference;
public class ServiceCacheImpl<T> implements ServiceCache<T>, PathChildrenCacheListener
{
- private final ListenerContainer<ServiceCacheListener> listenerContainer = new ListenerContainer<ServiceCacheListener>();
+ private final ListenerContainer<ServiceCacheEventListener<T>> listenerContainer = new ListenerContainer<>();
private final ServiceDiscoveryImpl<T> discovery;
- private final AtomicReference<State> state = new AtomicReference<State>(State.LATENT);
+ private final AtomicReference<State> state = new AtomicReference<>(State.LATENT);
private final PathChildrenCache cache;
private final ConcurrentMap<String, ServiceInstance<T>> instances = Maps.newConcurrentMap();
@@ -125,10 +127,10 @@ public class ServiceCacheImpl<T> implements ServiceCache<T>, PathChildrenCacheLi
listenerContainer.forEach
(
- new Function<ServiceCacheListener, Void>()
+ new Function<ServiceCacheEventListener<T>, Void>()
{
@Override
- public Void apply(ServiceCacheListener listener)
+ public Void apply(ServiceCacheEventListener<T> listener)
{
discovery.getClient().getConnectionStateListenable().removeListener(listener);
return null;
@@ -143,98 +145,114 @@ public class ServiceCacheImpl<T> implements ServiceCache<T>, PathChildrenCacheLi
}
@Override
+ public Listenable<ServiceCacheEventListener<T>> getCacheEventListenable()
+ {
+ return listenerContainer;
+ }
+
+ @Override
public void addListener(ServiceCacheListener listener)
{
- listenerContainer.addListener(listener);
- discovery.getClient().getConnectionStateListenable().addListener(listener);
+ ServiceCacheListenerWrapper<T> wrapped = ServiceCacheListenerWrapper.<T>wrap(listener);
+ listenerContainer.addListener(wrapped);
+ discovery.getClient().getConnectionStateListenable().addListener(wrapped);
}
@Override
public void addListener(ServiceCacheListener listener, Executor executor)
{
- listenerContainer.addListener(listener, executor);
- discovery.getClient().getConnectionStateListenable().addListener(listener, executor);
+ ServiceCacheListenerWrapper<T> wrapped = ServiceCacheListenerWrapper.<T>wrap(listener);
+ listenerContainer.addListener(wrapped, executor);
+ discovery.getClient().getConnectionStateListenable().addListener(wrapped, executor);
}
@Override
- public void removeListener(ServiceCacheListener listener)
+ public void removeListener(final ServiceCacheListener listener)
{
- listenerContainer.removeListener(listener);
- discovery.getClient().getConnectionStateListenable().removeListener(listener);
+ listenerContainer.forEach
+ (
+ new Function<ServiceCacheEventListener<T>, Void>()
+ {
+ @Override
+ public Void apply(ServiceCacheEventListener<T> eventListener)
+ {
+ if ( Objects.equals(ServiceCacheListenerWrapper.unwrap(eventListener), listener) )
+ {
+ listenerContainer.removeListener(eventListener);
+ discovery.getClient().getConnectionStateListenable().removeListener(eventListener);
+ }
+ return null;
+ }
+ }
+ );
}
@Override
public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception
{
- final Tuple<T> tuple;
switch ( event.getType() )
{
case CHILD_ADDED:
- tuple = addOrUpdateInstance(event.getData());
- listenerContainer.forEach(
- new Function<ServiceCacheListener, Void>()
+ {
+ final Tuple<T> tuple = addOrUpdateInstance(event.getData());
+ listenerContainer.forEach
+ (
+ new Function<ServiceCacheEventListener<T>, Void>()
{
@Override
- public Void apply(ServiceCacheListener listener)
+ public Void apply(ServiceCacheEventListener<T> listener)
{
- listener.cacheChanged();
+ listener.cacheAdded(tuple.newInstance);
+ return null;
+ }
+ }
+ );
+ break;
+ }
- if ( listener instanceof ServiceCacheEventListener )
+ case CHILD_UPDATED:
+ {
+ final Tuple<T> tuple = addOrUpdateInstance(event.getData());
+ listenerContainer.forEach
+ (
+ new Function<ServiceCacheEventListener<T>, Void>()
+ {
+ @Override
+ public Void apply(ServiceCacheEventListener<T> listener)
+ {
+ if ( tuple.oldInstance != null )
{
- //noinspection unchecked
- ((ServiceCacheEventListener) listener).cacheAdded(tuple.newInstance);
+ listener.cacheUpdated(tuple.oldInstance, tuple.newInstance);
+ }
+ else
+ {
+ listener.cacheAdded(tuple.newInstance);
}
-
return null;
}
}
);
break;
- case CHILD_UPDATED:
- {
- tuple = addOrUpdateInstance(event.getData());
- listenerContainer.forEach(
- new Function<ServiceCacheListener, Void>()
- {
- @Override
- public Void apply(ServiceCacheListener listener)
- {
- listener.cacheChanged();
-
- if( listener instanceof ServiceCacheEventListener )
- {
- //noinspection unchecked
- ((ServiceCacheEventListener) listener).cacheUpdated(tuple.oldInstance, tuple.newInstance);
- }
-
- return null;
- }
- }
- );
- break;
}
case CHILD_REMOVED:
{
final ServiceInstance<T> serviceInstance = instances.remove(instanceIdFromData(event.getData()));
- listenerContainer.forEach(
- new Function<ServiceCacheListener, Void>()
- {
- @Override
- public Void apply(ServiceCacheListener listener)
- {
- listener.cacheChanged();
-
- if ( listener instanceof ServiceCacheEventListener )
- {
- //noinspection unchecked
- ((ServiceCacheEventListener) listener).cacheDeleted(serviceInstance);
- }
-
- return null;
- }
- }
- );
+ if ( serviceInstance != null )
+ {
+ listenerContainer.forEach
+ (
+ new Function<ServiceCacheEventListener<T>, Void>()
+ {
+ @Override
+ public Void apply(ServiceCacheEventListener<T> listener)
+ {
+ listener.cacheDeleted(serviceInstance);
+ return null;
+ }
+ }
+ );
+ }
break;
}
}
http://git-wip-us.apache.org/repos/asf/curator/blob/2ebd4569/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheListenerWrapper.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheListenerWrapper.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheListenerWrapper.java
new file mode 100644
index 0000000..6f14178
--- /dev/null
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceCacheListenerWrapper.java
@@ -0,0 +1,53 @@
+package org.apache.curator.x.discovery.details;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.curator.x.discovery.ServiceInstance;
+
+class ServiceCacheListenerWrapper<T> implements ServiceCacheEventListener<T>
+{
+ private final ServiceCacheListener listener;
+
+ static <T> ServiceCacheListenerWrapper<T> wrap(ServiceCacheListener listener)
+ {
+ return new ServiceCacheListenerWrapper<>(listener);
+ }
+
+ static ServiceCacheListener unwrap(ServiceCacheEventListener<?> eventListener)
+ {
+ if ( eventListener instanceof ServiceCacheListenerWrapper )
+ {
+ return ((ServiceCacheListenerWrapper)eventListener).listener;
+ }
+ return null;
+ }
+
+ private ServiceCacheListenerWrapper(ServiceCacheListener listener)
+ {
+ this.listener = listener;
+ }
+
+ @Override
+ public void stateChanged(CuratorFramework client, ConnectionState newState)
+ {
+ listener.stateChanged(client, newState);
+ }
+
+ @Override
+ public void cacheAdded(ServiceInstance<T> added)
+ {
+ listener.cacheChanged();
+ }
+
+ @Override
+ public void cacheDeleted(ServiceInstance<T> deleted)
+ {
+ listener.cacheChanged();
+ }
+
+ @Override
+ public void cacheUpdated(ServiceInstance<T> old, ServiceInstance<T> updated)
+ {
+ listener.cacheChanged();
+ }
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/2ebd4569/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java b/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java
index 65b7cb9..70b41e3 100644
--- a/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java
+++ b/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/TestServiceCache.java
@@ -330,10 +330,10 @@ public class TestServiceCache extends BaseClassForTests
ServiceCache<String> cache = discovery.serviceCacheBuilder().name("test").build();
closeables.add(cache);
- final CountDownLatch latch = new CountDownLatch(6);
+ final CountDownLatch latch = new CountDownLatch(3);
final AtomicBoolean notifyError = new AtomicBoolean(false);
- ServiceCacheListener listener = new ServiceCacheEventListener<String>()
+ ServiceCacheEventListener<String> listener = new ServiceCacheEventListener<String>()
{
@Override
public void cacheAdded(final ServiceInstance<String> added) {
@@ -358,18 +358,11 @@ public class TestServiceCache extends BaseClassForTests
}
@Override
- public void cacheChanged()
- {
- latch.countDown();
- }
-
- @Override
public void stateChanged(CuratorFramework client, ConnectionState newState)
{
-
}
};
- cache.addListener(listener);
+ cache.getCacheEventListenable().addListener(listener);
cache.start();
ServiceInstance<String> instance = ServiceInstance.<String>builder().payload("before").name("test").port(10064).build();