first pass at implementation and a simple test
Project: http://git-wip-us.apache.org/repos/asf/curator/repo Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/58fcc548 Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/58fcc548 Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/58fcc548 Branch: refs/heads/CURATOR-3.0 Commit: 58fcc548f87e3c787f82d48931e2e9992f6c3366 Parents: 4c2ba37 Author: randgalt <randg...@apache.org> Authored: Mon May 11 15:29:09 2015 -0500 Committer: randgalt <randg...@apache.org> Committed: Mon May 11 15:29:09 2015 -0500 ---------------------------------------------------------------------- .../curator/framework/CuratorFramework.java | 2 + .../WatcherRemoveCuratorFramework.java | 6 + .../framework/imps/CuratorFrameworkImpl.java | 26 ++- .../framework/imps/ExistsBuilderImpl.java | 4 +- .../framework/imps/GetChildrenBuilderImpl.java | 4 +- .../framework/imps/GetDataBuilderImpl.java | 4 +- .../curator/framework/imps/NamespaceFacade.java | 7 + .../imps/RemoveWatchesBuilderImpl.java | 10 +- .../framework/imps/WatcherRemovalFacade.java | 170 +++++++++++++++++++ .../framework/imps/WatcherRemovalManager.java | 67 ++++++++ .../apache/curator/framework/imps/Watching.java | 6 +- .../imps/TestWatcherRemovalManager.java | 54 ++++++ .../org/apache/curator/test/WatchersDebug.java | 74 ++++++++ 13 files changed, 419 insertions(+), 15 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-framework/src/main/java/org/apache/curator/framework/CuratorFramework.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/CuratorFramework.java b/curator-framework/src/main/java/org/apache/curator/framework/CuratorFramework.java index 4b30fd4..75b434a 100644 --- a/curator-framework/src/main/java/org/apache/curator/framework/CuratorFramework.java +++ b/curator-framework/src/main/java/org/apache/curator/framework/CuratorFramework.java @@ -235,4 +235,6 @@ public interface CuratorFramework extends Closeable * @throws InterruptedException If interrupted while waiting */ public void blockUntilConnected() throws InterruptedException; + + WatcherRemoveCuratorFramework newWatcherRemoveCuratorFramework(); } http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-framework/src/main/java/org/apache/curator/framework/WatcherRemoveCuratorFramework.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/WatcherRemoveCuratorFramework.java b/curator-framework/src/main/java/org/apache/curator/framework/WatcherRemoveCuratorFramework.java new file mode 100644 index 0000000..58b4bf8 --- /dev/null +++ b/curator-framework/src/main/java/org/apache/curator/framework/WatcherRemoveCuratorFramework.java @@ -0,0 +1,6 @@ +package org.apache.curator.framework; + +public interface WatcherRemoveCuratorFramework extends CuratorFramework +{ + void removeWatchers(); +} http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java index 5caff7d..35816a4 100644 --- a/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java +++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java @@ -29,6 +29,7 @@ import org.apache.curator.RetryLoop; import org.apache.curator.TimeTrace; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.WatcherRemoveCuratorFramework; import org.apache.curator.framework.api.*; import org.apache.curator.framework.api.transaction.CuratorTransaction; import org.apache.curator.framework.listen.Listenable; @@ -150,6 +151,12 @@ public class CuratorFrameworkImpl implements CuratorFramework namespaceFacadeCache = new NamespaceFacadeCache(this); } + @Override + public WatcherRemoveCuratorFramework newWatcherRemoveCuratorFramework() + { + return new WatcherRemovalFacade(this); + } + private ZookeeperFactory makeZookeeperFactory(final ZookeeperFactory actualZookeeperFactory) { return new ZookeeperFactory() @@ -570,14 +577,14 @@ public class CuratorFrameworkImpl implements CuratorFramework final String localReason = reason; unhandledErrorListeners.forEach(new Function<UnhandledErrorListener, Void>() + { + @Override + public Void apply(UnhandledErrorListener listener) { - @Override - public Void apply(UnhandledErrorListener listener) - { - listener.unhandledError(localReason, e); - return null; - } - }); + listener.unhandledError(localReason, e); + return null; + } + }); if ( debugUnhandledErrorListener != null ) { @@ -660,6 +667,11 @@ public class CuratorFrameworkImpl implements CuratorFramework return Watcher.Event.KeeperState.fromInt(-1); } + WatcherRemovalManager getWatcherRemovalManager() + { + return null; + } + private void suspendConnection() { if ( !connectionStateManager.setToSuspended() ) http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-framework/src/main/java/org/apache/curator/framework/imps/ExistsBuilderImpl.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/ExistsBuilderImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/ExistsBuilderImpl.java index a1e2ee5..d5f7b5b 100644 --- a/curator-framework/src/main/java/org/apache/curator/framework/imps/ExistsBuilderImpl.java +++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/ExistsBuilderImpl.java @@ -129,7 +129,7 @@ class ExistsBuilderImpl implements ExistsBuilder, BackgroundOperation<String> } else { - client.getZooKeeper().exists(operationAndData.getData(), watching.getWatcher(), callback, backgrounding.getContext()); + client.getZooKeeper().exists(operationAndData.getData(), watching.getWatcher(client, operationAndData.getData()), callback, backgrounding.getContext()); } } @@ -169,7 +169,7 @@ class ExistsBuilderImpl implements ExistsBuilder, BackgroundOperation<String> } else { - returnStat = client.getZooKeeper().exists(path, watching.getWatcher()); + returnStat = client.getZooKeeper().exists(path, watching.getWatcher(client, path)); } return returnStat; } http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-framework/src/main/java/org/apache/curator/framework/imps/GetChildrenBuilderImpl.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/GetChildrenBuilderImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/GetChildrenBuilderImpl.java index 16f6d4b..f9f784f 100644 --- a/curator-framework/src/main/java/org/apache/curator/framework/imps/GetChildrenBuilderImpl.java +++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/GetChildrenBuilderImpl.java @@ -172,7 +172,7 @@ class GetChildrenBuilderImpl implements GetChildrenBuilder, BackgroundOperation< } else { - client.getZooKeeper().getChildren(operationAndData.getData(), watching.getWatcher(), callback, backgrounding.getContext()); + client.getZooKeeper().getChildren(operationAndData.getData(), watching.getWatcher(client, operationAndData.getData()), callback, backgrounding.getContext()); } } @@ -211,7 +211,7 @@ class GetChildrenBuilderImpl implements GetChildrenBuilder, BackgroundOperation< } else { - children = client.getZooKeeper().getChildren(path, watching.getWatcher(), responseStat); + children = client.getZooKeeper().getChildren(path, watching.getWatcher(client, path), responseStat); } return children; } http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-framework/src/main/java/org/apache/curator/framework/imps/GetDataBuilderImpl.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/GetDataBuilderImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/GetDataBuilderImpl.java index e994b03..7077839 100644 --- a/curator-framework/src/main/java/org/apache/curator/framework/imps/GetDataBuilderImpl.java +++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/GetDataBuilderImpl.java @@ -260,7 +260,7 @@ class GetDataBuilderImpl implements GetDataBuilder, BackgroundOperation<String> } else { - client.getZooKeeper().getData(operationAndData.getData(), watching.getWatcher(), callback, backgrounding.getContext()); + client.getZooKeeper().getData(operationAndData.getData(), watching.getWatcher(client, operationAndData.getData()), callback, backgrounding.getContext()); } } @@ -299,7 +299,7 @@ class GetDataBuilderImpl implements GetDataBuilder, BackgroundOperation<String> } else { - responseData = client.getZooKeeper().getData(path, watching.getWatcher(), responseStat); + responseData = client.getZooKeeper().getData(path, watching.getWatcher(client, path), responseStat); } return responseData; } http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-framework/src/main/java/org/apache/curator/framework/imps/NamespaceFacade.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/NamespaceFacade.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/NamespaceFacade.java index 818fe5f..8dc7ae1 100644 --- a/curator-framework/src/main/java/org/apache/curator/framework/imps/NamespaceFacade.java +++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/NamespaceFacade.java @@ -21,6 +21,7 @@ package org.apache.curator.framework.imps; import org.apache.curator.CuratorZookeeperClient; import org.apache.curator.RetryLoop; import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.WatcherRemoveCuratorFramework; import org.apache.curator.framework.api.*; import org.apache.curator.framework.listen.Listenable; import org.apache.curator.framework.state.ConnectionStateListener; @@ -41,6 +42,12 @@ class NamespaceFacade extends CuratorFrameworkImpl } @Override + public WatcherRemoveCuratorFramework newWatcherRemoveCuratorFramework() + { + throw new UnsupportedOperationException(); + } + + @Override public CuratorFramework nonNamespaceView() { return usingNamespace(null); http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-framework/src/main/java/org/apache/curator/framework/imps/RemoveWatchesBuilderImpl.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/RemoveWatchesBuilderImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/RemoveWatchesBuilderImpl.java index 1e4fb88..ab977a9 100644 --- a/curator-framework/src/main/java/org/apache/curator/framework/imps/RemoveWatchesBuilderImpl.java +++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/RemoveWatchesBuilderImpl.java @@ -40,7 +40,15 @@ public class RemoveWatchesBuilderImpl implements RemoveWatchesBuilder, RemoveWat this.quietly = false; this.backgrounding = new Backgrounding(); } - + + void prepInternalRemoval(Watcher watcher) + { + this.watcher = watcher; + watcherType = WatcherType.Any; + quietly = true; + this.backgrounding = new Backgrounding(); + } + @Override public RemoveWatchesType remove(Watcher watcher) { http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalFacade.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalFacade.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalFacade.java new file mode 100644 index 0000000..a11ca5d --- /dev/null +++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalFacade.java @@ -0,0 +1,170 @@ +/** + * 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.framework.imps; + +import org.apache.curator.CuratorZookeeperClient; +import org.apache.curator.RetryLoop; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.WatcherRemoveCuratorFramework; +import org.apache.curator.framework.api.CuratorEvent; +import org.apache.curator.framework.api.CuratorListener; +import org.apache.curator.framework.api.UnhandledErrorListener; +import org.apache.curator.framework.listen.Listenable; +import org.apache.curator.framework.state.ConnectionStateListener; +import org.apache.curator.utils.EnsurePath; +import org.apache.zookeeper.ZooKeeper; + +class WatcherRemovalFacade extends CuratorFrameworkImpl implements WatcherRemoveCuratorFramework +{ + private final CuratorFrameworkImpl client; + private final WatcherRemovalManager removalManager; + + WatcherRemovalFacade(CuratorFrameworkImpl client) + { + super(client); + this.client = client; + removalManager = new WatcherRemovalManager(client); + } + + @Override + public WatcherRemoveCuratorFramework newWatcherRemoveCuratorFramework() + { + throw new UnsupportedOperationException(); + } + + @Override + public void removeWatchers() + { + removalManager.removeWatchers(); + } + + @Override + WatcherRemovalManager getWatcherRemovalManager() + { + return removalManager; + } + + @Override + public CuratorFramework nonNamespaceView() + { + return client.usingNamespace(null); + } + + @Override + public CuratorFramework usingNamespace(String newNamespace) + { + return client.getNamespaceFacadeCache().get(newNamespace); + } + + @Override + public String getNamespace() + { + return client.getNamespace(); + } + + @Override + public void start() + { + throw new UnsupportedOperationException(); + } + + @Override + public void close() + { + throw new UnsupportedOperationException(); + } + + @Override + public Listenable<ConnectionStateListener> getConnectionStateListenable() + { + return client.getConnectionStateListenable(); + } + + @Override + public Listenable<CuratorListener> getCuratorListenable() + { + return client.getCuratorListenable(); + } + + @Override + public Listenable<UnhandledErrorListener> getUnhandledErrorListenable() + { + return client.getUnhandledErrorListenable(); + } + + @Override + public void sync(String path, Object context) + { + client.sync(path, context); + } + + @Override + public CuratorZookeeperClient getZookeeperClient() + { + return client.getZookeeperClient(); + } + + @Override + RetryLoop newRetryLoop() + { + return client.newRetryLoop(); + } + + @Override + ZooKeeper getZooKeeper() throws Exception + { + return client.getZooKeeper(); + } + + @Override + <DATA_TYPE> void processBackgroundOperation(OperationAndData<DATA_TYPE> operationAndData, CuratorEvent event) + { + client.processBackgroundOperation(operationAndData, event); + } + + @Override + void logError(String reason, Throwable e) + { + client.logError(reason, e); + } + + @Override + String unfixForNamespace(String path) + { + return client.unfixForNamespace(path); + } + + @Override + String fixForNamespace(String path) + { + return client.fixForNamespace(path); + } + + @Override + public EnsurePath newNamespaceAwareEnsurePath(String path) + { + return client.newNamespaceAwareEnsurePath(path); + } + + @Override + FailedDeleteManager getFailedDeleteManager() + { + return client.getFailedDeleteManager(); + } +} http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalManager.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalManager.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalManager.java new file mode 100644 index 0000000..9461de5 --- /dev/null +++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/WatcherRemovalManager.java @@ -0,0 +1,67 @@ +package org.apache.curator.framework.imps; + +import com.google.common.collect.Maps; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.util.Map; + +class WatcherRemovalManager +{ + private final Logger log = LoggerFactory.getLogger(getClass()); + private final CuratorFrameworkImpl client; + private final Map<Watcher, String> entries = Maps.newConcurrentMap(); + + WatcherRemovalManager(CuratorFrameworkImpl client) + { + this.client = client; + } + + Watcher add(String path, Watcher watcher) + { + Watcher wrappedWatcher = new WrappedWatcher(entries, watcher); + entries.put(wrappedWatcher, path); + return wrappedWatcher; + } + + void removeWatchers() + { + for ( Map.Entry<Watcher, String> entry : entries.entrySet() ) + { + Watcher watcher = entry.getKey(); + String path = entry.getValue(); + try + { + log.debug("Removing watcher for path: " + path); + RemoveWatchesBuilderImpl builder = new RemoveWatchesBuilderImpl(client); + builder.prepInternalRemoval(watcher); + builder.forPath(path); + } + catch ( Exception e ) + { + String message = "Could not remove watcher for path: " + path; + log.error(message); + } + } + } + + private static class WrappedWatcher implements Watcher + { + private final Map<Watcher, String> entries; + private final Watcher watcher; + + WrappedWatcher(Map<Watcher, String> entries, Watcher watcher) + { + this.entries = entries; + this.watcher = watcher; + } + + @Override + public void process(WatchedEvent event) + { + entries.remove(this); + watcher.process(event); + } + } +} http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-framework/src/main/java/org/apache/curator/framework/imps/Watching.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/Watching.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/Watching.java index a9d0ab1..ae16dfc 100644 --- a/curator-framework/src/main/java/org/apache/curator/framework/imps/Watching.java +++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/Watching.java @@ -50,8 +50,12 @@ class Watching watched = false; } - Watcher getWatcher() + Watcher getWatcher(CuratorFrameworkImpl client, String unfixedPath) { + if ( client.getWatcherRemovalManager() != null ) + { + return client.getWatcherRemovalManager().add(unfixedPath, watcher); + } return watcher; } http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-framework/src/test/java/org/apache/curator/framework/imps/TestWatcherRemovalManager.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/test/java/org/apache/curator/framework/imps/TestWatcherRemovalManager.java b/curator-framework/src/test/java/org/apache/curator/framework/imps/TestWatcherRemovalManager.java new file mode 100644 index 0000000..9a5dbdc --- /dev/null +++ b/curator-framework/src/test/java/org/apache/curator/framework/imps/TestWatcherRemovalManager.java @@ -0,0 +1,54 @@ +package org.apache.curator.framework.imps; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.WatcherRemoveCuratorFramework; +import org.apache.curator.retry.RetryOneTime; +import org.apache.curator.test.BaseClassForTests; +import org.apache.curator.test.Timing; +import org.apache.curator.test.WatchersDebug; +import org.apache.curator.utils.CloseableUtils; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.testng.Assert; +import org.testng.annotations.Test; +import java.util.List; + +public class TestWatcherRemovalManager extends BaseClassForTests +{ + @Test + public void testBasic() throws Exception + { + CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1)); + try + { + client.start(); + + WatcherRemoveCuratorFramework removerClient = client.newWatcherRemoveCuratorFramework(); + + Watcher watcher = new Watcher() + { + @Override + public void process(WatchedEvent event) + { + // NOP + } + }; + removerClient.checkExists().usingWatcher(watcher).forPath("/hey"); + + List<String> existWatches = WatchersDebug.getExistWatches(client.getZookeeperClient().getZooKeeper()); + Assert.assertEquals(existWatches.size(), 1); + + removerClient.removeWatchers(); + + new Timing().sleepABit(); + + existWatches = WatchersDebug.getExistWatches(client.getZookeeperClient().getZooKeeper()); + Assert.assertEquals(existWatches.size(), 0); + } + finally + { + CloseableUtils.closeQuietly(client); + } + } +} http://git-wip-us.apache.org/repos/asf/curator/blob/58fcc548/curator-test/src/main/java/org/apache/curator/test/WatchersDebug.java ---------------------------------------------------------------------- diff --git a/curator-test/src/main/java/org/apache/curator/test/WatchersDebug.java b/curator-test/src/main/java/org/apache/curator/test/WatchersDebug.java new file mode 100644 index 0000000..a97e5c1 --- /dev/null +++ b/curator-test/src/main/java/org/apache/curator/test/WatchersDebug.java @@ -0,0 +1,74 @@ +package org.apache.curator.test; + +import org.apache.zookeeper.ZooKeeper; +import java.lang.reflect.Method; +import java.util.List; + +public class WatchersDebug +{ + private static final Method getDataWatches; + private static final Method getExistWatches; + private static final Method getChildWatches; + static + { + Method localGetDataWatches = null; + Method localGetExistWatches = null; + Method localGetChildWatches = null; + try + { + localGetDataWatches = getMethod("getDataWatches"); + localGetExistWatches = getMethod("getExistWatches"); + localGetChildWatches = getMethod("getChildWatches"); + } + catch ( NoSuchMethodException e ) + { + e.printStackTrace(); + } + getDataWatches = localGetDataWatches; + getExistWatches = localGetExistWatches; + getChildWatches = localGetChildWatches; + } + + public static List<String> getDataWatches(ZooKeeper zooKeeper) + { + return callMethod(zooKeeper, WatchersDebug.getDataWatches); + } + + public static List<String> getExistWatches(ZooKeeper zooKeeper) + { + return callMethod(zooKeeper, getExistWatches); + } + + public static List<String> getChildWatches(ZooKeeper zooKeeper) + { + return callMethod(zooKeeper, getChildWatches); + } + + private WatchersDebug() + { + } + + private static Method getMethod(String name) throws NoSuchMethodException + { + Method m = ZooKeeper.class.getDeclaredMethod(name); + m.setAccessible(true); + return m; + } + + private static List<String> callMethod(ZooKeeper zooKeeper, Method method) + { + if ( zooKeeper == null ) + { + return null; + } + try + { + //noinspection unchecked + return (List<String>)method.invoke(zooKeeper); + } + catch ( Exception e ) + { + throw new RuntimeException(e); + } + } +}
