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 2017/05/02 03:57:04 UTC
[20/20] curator git commit: Major rework of caching. Having the
wrapped caches adds little value. Focus on the integrated caching in the
modeled client instance
Major rework of caching. Having the wrapped caches adds little value. Focus on the integrated caching in the modeled client instance
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/2cbbf999
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/2cbbf999
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/2cbbf999
Branch: refs/heads/CURATOR-397
Commit: 2cbbf999294181dbefeccebabd9c6de867142e2c
Parents: e40ed18
Author: randgalt <ra...@apache.org>
Authored: Mon May 1 22:56:36 2017 -0500
Committer: randgalt <ra...@apache.org>
Committed: Mon May 1 22:56:36 2017 -0500
----------------------------------------------------------------------
.../main/java/modeled/ModeledCacheExamples.java | 57 ----
.../modeled/CachedModeledCuratorFramework.java | 43 ---
.../async/modeled/ModeledCuratorFramework.java | 14 +-
.../apache/curator/x/async/modeled/ZPath.java | 9 +
.../cached/CachedModeledCuratorFramework.java | 48 ++++
.../x/async/modeled/cached/ModeledCache.java | 53 ++++
.../modeled/cached/ModeledCacheEventType.java | 42 +++
.../modeled/cached/ModeledCacheListener.java | 58 ++++
.../async/modeled/cached/ModeledCachedNode.java | 49 ++++
.../CachedModeledCuratorFrameworkImpl.java | 140 +++-------
.../x/async/modeled/details/ModelStage.java | 6 +
.../async/modeled/details/ModeledCacheImpl.java | 173 ++++++++++++
.../details/ModeledCuratorFrameworkImpl.java | 34 +--
.../x/async/modeled/details/ZPathImpl.java | 13 +-
.../details/recipes/ModeledCachedNodeImpl.java | 107 --------
.../details/recipes/ModeledNodeCacheImpl.java | 232 ----------------
.../recipes/ModeledPathChildrenCacheImpl.java | 266 -------------------
.../details/recipes/ModeledTreeCacheImpl.java | 179 -------------
.../x/async/modeled/recipes/ModeledCache.java | 51 ----
.../modeled/recipes/ModeledCacheEvent.java | 39 ---
.../modeled/recipes/ModeledCacheEventType.java | 57 ----
.../modeled/recipes/ModeledCacheListener.java | 84 ------
.../modeled/recipes/ModeledCachedNode.java | 49 ----
.../async/modeled/recipes/ModeledNodeCache.java | 91 -------
.../recipes/ModeledPathChildrenCache.java | 114 --------
.../async/modeled/recipes/ModeledTreeCache.java | 70 -----
.../TestCachedModeledCuratorFramework.java | 4 +-
.../details/recipes/TestEventTypeMappings.java | 44 ---
.../modeled/recipes/TestModeledCaches.java | 193 --------------
29 files changed, 497 insertions(+), 1822 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-examples/src/main/java/modeled/ModeledCacheExamples.java
----------------------------------------------------------------------
diff --git a/curator-examples/src/main/java/modeled/ModeledCacheExamples.java b/curator-examples/src/main/java/modeled/ModeledCacheExamples.java
deleted file mode 100644
index 3157f7e..0000000
--- a/curator-examples/src/main/java/modeled/ModeledCacheExamples.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/**
- * 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 modeled;
-
-import org.apache.curator.framework.recipes.cache.TreeCache;
-import org.apache.curator.x.async.modeled.JacksonModelSerializer;
-import org.apache.curator.x.async.modeled.recipes.ModeledCacheEventType;
-import org.apache.curator.x.async.modeled.recipes.ModeledCacheListener;
-import org.apache.curator.x.async.modeled.recipes.ModeledTreeCache;
-import java.util.function.Consumer;
-
-public class ModeledCacheExamples
-{
- public static ModeledTreeCache<PersonModel> wrap(TreeCache cache)
- {
- JacksonModelSerializer<PersonModel> serializer = JacksonModelSerializer.build(PersonModel.class);
-
- // wrap a TreeCache instance so that it can be used "modeled".
- return ModeledTreeCache.wrap(cache, serializer);
- }
-
- public static void watchForChanges(TreeCache cache, Consumer<PersonModel> deletePersonReceiver, Consumer<PersonModel> updatedPersonReceiver)
- {
- ModeledTreeCache<PersonModel> modeledCache = wrap(cache);
- ModeledCacheListener<PersonModel> listener = event -> {
- PersonModel person = event.getNode().getModel();
- if ( event.getType() == ModeledCacheEventType.NODE_REMOVED )
- {
- deletePersonReceiver.accept(person);
- }
- else
- {
- updatedPersonReceiver.accept(person);
- }
- };
-
- // take a standard listener and filter so that only events that have a valid model instance are sent to the listener
- ModeledCacheListener<PersonModel> filteredListener = ModeledCacheListener.filtered(listener, ModeledCacheListener.hasModelFilter());
- modeledCache.getListenable().addListener(filteredListener);
- }
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/CachedModeledCuratorFramework.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/CachedModeledCuratorFramework.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/CachedModeledCuratorFramework.java
deleted file mode 100644
index b4c6fef..0000000
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/CachedModeledCuratorFramework.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/**
- * 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.async.modeled;
-
-import org.apache.curator.x.async.modeled.recipes.ModeledCache;
-import java.io.Closeable;
-
-public interface CachedModeledCuratorFramework<T> extends ModeledCuratorFramework<T>, Closeable
-{
- /**
- * Return the cache instance
- *
- * @return cache
- */
- ModeledCache<T> getCache();
-
- /**
- * Start the internally created via {@link #cached()}
- */
- void start();
-
- /**
- * Close the internally created via {@link #cached()}
- */
- @Override
- void close();
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/ModeledCuratorFramework.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/ModeledCuratorFramework.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/ModeledCuratorFramework.java
index ac4fd8d..31b630d 100644
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/ModeledCuratorFramework.java
+++ b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/ModeledCuratorFramework.java
@@ -22,7 +22,7 @@ import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.transaction.CuratorOp;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import org.apache.curator.x.async.AsyncStage;
-import org.apache.curator.x.async.modeled.recipes.ModeledCache;
+import org.apache.curator.x.async.modeled.cached.CachedModeledCuratorFramework;
import org.apache.zookeeper.data.Stat;
import java.util.List;
import java.util.Map;
@@ -54,19 +54,9 @@ public interface ModeledCuratorFramework<T>
}
/**
- * Use the given cache as a front for this modeled instance. All read APIs check the cache
- * first and, if available, return the values from the cache.
- * the cache
- *
- * @param cache cache to use
- * @return wrapped instance
- */
- CachedModeledCuratorFramework<T> cached(ModeledCache<T> cache);
-
- /**
* Use an internally created cache as a front for this modeled instance. All read APIs check the cache
* first and, if available, return the values from the cache. Note: you must call
- * {@link org.apache.curator.x.async.modeled.CachedModeledCuratorFramework#start()} and
+ * {@link org.apache.curator.x.async.modeled.cached.CachedModeledCuratorFramework#start()} and
* {@link CachedModeledCuratorFramework#close()} to start/stop
*
* @return wrapped instance
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/ZPath.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/ZPath.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/ZPath.java
index 78cd40f..1b10a40 100644
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/ZPath.java
+++ b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/ZPath.java
@@ -176,6 +176,15 @@ public interface ZPath
boolean isRoot();
/**
+ * Return true if this path starts with the given path. i.e.
+ * <code>ZPath.from("/one/two/three").startsWith(ZPath.from("/one/two"))</code> returns true
+ *
+ * @param path base path
+ * @return true/false
+ */
+ boolean startsWith(ZPath path);
+
+ /**
* The string full path that this ZPath represents
*
* @return full path
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/CachedModeledCuratorFramework.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/CachedModeledCuratorFramework.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/CachedModeledCuratorFramework.java
new file mode 100644
index 0000000..8718e60
--- /dev/null
+++ b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/CachedModeledCuratorFramework.java
@@ -0,0 +1,48 @@
+/**
+ * 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.async.modeled.cached;
+
+import org.apache.curator.x.async.modeled.ModeledCuratorFramework;
+import java.io.Closeable;
+
+public interface CachedModeledCuratorFramework<T> extends ModeledCuratorFramework<T>, Closeable
+{
+ /**
+ * Return the cache instance
+ *
+ * @return cache
+ */
+ ModeledCache<T> getCache();
+
+ /**
+ * Start the internally created via {@link #cached()}
+ */
+ void start();
+
+ /**
+ * Close the internally created via {@link #cached()}
+ */
+ @Override
+ void close();
+
+ /**
+ * {@inheritDoc}
+ */
+ CachedModeledCuratorFramework<T> at(String child);
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCache.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCache.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCache.java
new file mode 100644
index 0000000..3536a65
--- /dev/null
+++ b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCache.java
@@ -0,0 +1,53 @@
+/**
+ * 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.async.modeled.cached;
+
+import org.apache.curator.framework.listen.Listenable;
+import org.apache.curator.x.async.modeled.ZPath;
+import java.util.Map;
+import java.util.Optional;
+
+public interface ModeledCache<T>
+{
+ /**
+ * Return the modeled current data for the given path. There are no guarantees of accuracy. This is
+ * merely the most recent view of the data. If there is no node at the given path,
+ * {@link java.util.Optional#empty()} is returned.
+ *
+ * @param path path to the node to check
+ * @return data if the node is alive, or null
+ */
+ Optional<ModeledCachedNode<T>> getCurrentData(ZPath path);
+
+ /**
+ * Return the modeled current set of children at the given path, mapped by child name. There are no
+ * guarantees of accuracy; this is merely the most recent view of the data.
+ *
+ * @param path path to the node to check
+ * @return a possibly-empty list of children if the node is alive, or null
+ */
+ Map<ZPath, ModeledCachedNode<T>> getCurrentChildren(ZPath path);
+
+ /**
+ * Return the listener container so that you can add/remove listeners
+ *
+ * @return listener container
+ */
+ Listenable<ModeledCacheListener<T>> getListenable();
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCacheEventType.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCacheEventType.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCacheEventType.java
new file mode 100644
index 0000000..e7754ea
--- /dev/null
+++ b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCacheEventType.java
@@ -0,0 +1,42 @@
+/**
+ * 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.async.modeled.cached;
+
+public enum ModeledCacheEventType
+{
+ /**
+ * A child was added to the path
+ */
+ NODE_ADDED,
+
+ /**
+ * A child's data was changed
+ */
+ NODE_UPDATED,
+
+ /**
+ * A child was removed from the path
+ */
+ NODE_REMOVED,
+
+ /**
+ * Signals that the initial cache has been populated.
+ */
+ INITIALIZED
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCacheListener.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCacheListener.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCacheListener.java
new file mode 100644
index 0000000..9ddef87
--- /dev/null
+++ b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCacheListener.java
@@ -0,0 +1,58 @@
+package org.apache.curator.x.async.modeled.cached;
+
+import org.apache.curator.x.async.modeled.ZPath;
+import org.apache.zookeeper.data.Stat;
+
+@FunctionalInterface
+public interface ModeledCacheListener<T>
+{
+ /**
+ * The given path was added, updated or removed
+ *
+ * @param type action type
+ * @param path the path
+ * @param stat the node's stat (previous stat for removal)
+ * @param model the node's model (previous model for removal)
+ */
+ void accept(ModeledCacheEventType type, ZPath path, Stat stat, T model);
+
+ /**
+ * The cache has finished initializing
+ */
+ default void initialized()
+ {
+ // NOP
+ }
+
+ /**
+ * Returns a version of this listener that only begins calling
+ * {@link #accept(ModeledCacheEventType, org.apache.curator.x.async.modeled.ZPath, org.apache.zookeeper.data.Stat, Object)}
+ * once {@link #initialized()} has been called. i.e. changes that occur as the cache is initializing are not sent
+ * to the listener
+ *
+ * @return wrapped listener
+ */
+ default ModeledCacheListener<T> postInitializedOnly()
+ {
+ return new ModeledCacheListener<T>()
+ {
+ private volatile boolean isInitialized = false;
+
+ @Override
+ public void accept(ModeledCacheEventType type, ZPath path, Stat stat, T model)
+ {
+ if ( isInitialized )
+ {
+ ModeledCacheListener.this.accept(type, path, stat, model);
+ }
+ }
+
+ @Override
+ public void initialized()
+ {
+ isInitialized = true;
+ ModeledCacheListener.this.initialized();
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCachedNode.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCachedNode.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCachedNode.java
new file mode 100644
index 0000000..3a8e742
--- /dev/null
+++ b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/cached/ModeledCachedNode.java
@@ -0,0 +1,49 @@
+/**
+ * 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.async.modeled.cached;
+
+import org.apache.curator.x.async.modeled.ZPath;
+import org.apache.zookeeper.data.Stat;
+
+/**
+ * Abstracts a cached node
+ */
+public interface ModeledCachedNode<T>
+{
+ /**
+ * The path of the node
+ *
+ * @return path
+ */
+ ZPath getPath();
+
+ /**
+ * The node's last known stat if available
+ *
+ * @return stat
+ */
+ Stat getStat();
+
+ /**
+ * The node's current model
+ *
+ * @return model
+ */
+ T getModel();
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/CachedModeledCuratorFrameworkImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/CachedModeledCuratorFrameworkImpl.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/CachedModeledCuratorFrameworkImpl.java
index 6192f3a..3318403 100644
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/CachedModeledCuratorFrameworkImpl.java
+++ b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/CachedModeledCuratorFrameworkImpl.java
@@ -18,44 +18,43 @@
*/
package org.apache.curator.x.async.modeled.details;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.transaction.CuratorOp;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import org.apache.curator.x.async.AsyncStage;
-import org.apache.curator.x.async.modeled.CachedModeledCuratorFramework;
+import org.apache.curator.x.async.api.CreateOption;
import org.apache.curator.x.async.modeled.CuratorModelSpec;
import org.apache.curator.x.async.modeled.ModeledCuratorFramework;
import org.apache.curator.x.async.modeled.ZPath;
-import org.apache.curator.x.async.modeled.recipes.ModeledCache;
-import org.apache.curator.x.async.modeled.recipes.ModeledCachedNode;
+import org.apache.curator.x.async.modeled.cached.CachedModeledCuratorFramework;
+import org.apache.curator.x.async.modeled.cached.ModeledCache;
+import org.apache.curator.x.async.modeled.cached.ModeledCachedNode;
+import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.server.DataTree;
import java.util.AbstractMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Optional;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Function;
+import java.util.Set;
import java.util.stream.Collectors;
class CachedModeledCuratorFrameworkImpl<T> implements CachedModeledCuratorFramework<T>
{
private final ModeledCuratorFramework<T> client;
- private final ModeledCache<T> cache;
- private final ZPath path;
+ private final ModeledCacheImpl<T> cache;
- CachedModeledCuratorFrameworkImpl(ModeledCuratorFramework<T> client, ModeledCache<T> cache, ZPath path)
+ CachedModeledCuratorFrameworkImpl(ModeledCuratorFramework<T> client)
{
- this.client = Objects.requireNonNull(client, "client cannot be null");
- this.cache = Objects.requireNonNull(cache, "cache cannot be null");
- this.path = Objects.requireNonNull(path, "path cannot be null");
+ this(client, new ModeledCacheImpl<>(client.unwrap(), client.modelSpec().path(), client.modelSpec().serializer(), client.modelSpec().createOptions().contains(CreateOption.compress)));
}
- @VisibleForTesting
- volatile AtomicInteger debugCachedReadCount = null;
+ private CachedModeledCuratorFrameworkImpl(ModeledCuratorFramework<T> client, ModeledCacheImpl<T> cache)
+ {
+ this.client = client;
+ this.cache = cache;
+ }
@Override
public ModeledCache<T> getCache()
@@ -66,55 +65,49 @@ class CachedModeledCuratorFrameworkImpl<T> implements CachedModeledCuratorFramew
@Override
public void start()
{
- throw new UnsupportedOperationException();
+ cache.start();
}
@Override
public void close()
{
- throw new UnsupportedOperationException();
- }
-
- @Override
- public CuratorModelSpec<T> modelSpec()
- {
- return client.modelSpec();
+ cache.close();
}
@Override
- public CachedModeledCuratorFramework<T> cached(ModeledCache<T> cache)
+ public CachedModeledCuratorFramework<T> cached()
{
- throw new UnsupportedOperationException();
+ return this;
}
@Override
- public CachedModeledCuratorFramework<T> cached()
+ public CuratorFramework unwrap()
{
- throw new UnsupportedOperationException();
+ return client.unwrap();
}
@Override
- public CuratorFramework unwrap()
+ public CuratorModelSpec<T> modelSpec()
{
- return client.unwrap();
+ return client.modelSpec();
}
@Override
- public ModeledCuratorFramework<T> at(String child)
+ public CachedModeledCuratorFramework<T> at(String child)
{
- return new CachedModeledCuratorFrameworkImpl<>(client.at(child), cache, path.at(child));
+ return new CachedModeledCuratorFrameworkImpl<>(client.at(child), cache);
}
@Override
public AsyncStage<String> set(T model)
{
- return client.set(model);
+ return client.set(model); // TODO - update cache?
}
@Override
public AsyncStage<String> set(T model, Stat storingStatIn)
{
- return client.set(model, storingStatIn);
+ return client.set(model, storingStatIn); // TODO - update cache?
}
@Override
@@ -126,25 +119,15 @@ class CachedModeledCuratorFrameworkImpl<T> implements CachedModeledCuratorFramew
@Override
public AsyncStage<T> read(Stat storingStatIn)
{
+ ZPath path = client.modelSpec().path();
Optional<ModeledCachedNode<T>> data = cache.getCurrentData(path);
- if ( data.isPresent() )
- {
- ModeledCachedNode<T> localData = data.get();
- T model = localData.getModel();
- if ( model != null )
+ return data.map(node -> {
+ if ( storingStatIn != null )
{
- if ( (storingStatIn != null) && (localData.getStat() != null) )
- {
- DataTree.copyStat(localData.getStat(), storingStatIn);
- }
- if ( debugCachedReadCount != null )
- {
- debugCachedReadCount.incrementAndGet();
- }
- return new ModelStage<>(model);
+ DataTree.copyStat(node.getStat(), storingStatIn);
}
- }
- return (storingStatIn != null) ? client.read(storingStatIn) : client.read();
+ return new ModelStage<>(node.getModel());
+ }).orElseGet(() -> new ModelStage<>(new KeeperException.NoNodeException(path.fullPath())));
}
@Override
@@ -174,67 +157,26 @@ class CachedModeledCuratorFrameworkImpl<T> implements CachedModeledCuratorFramew
@Override
public AsyncStage<Stat> checkExists()
{
+ ZPath path = client.modelSpec().path();
Optional<ModeledCachedNode<T>> data = cache.getCurrentData(path);
- return data.map(node -> {
- AsyncStage<Stat> stage = new ModelStage<>(node.getStat());
- if ( debugCachedReadCount != null )
- {
- debugCachedReadCount.incrementAndGet();
- }
- return stage;
- }).orElseGet(client::checkExists);
+ return data.map(node -> new ModelStage<>(node.getStat())).orElseGet(() -> new ModelStage<>((Stat)null));
}
@Override
public AsyncStage<List<ZPath>> getChildren()
{
- Map<ZPath, ModeledCachedNode<T>> currentChildren = cache.getCurrentChildren(path);
- if ( currentChildren != cache.noChildrenValue() )
- {
- if ( debugCachedReadCount != null )
- {
- debugCachedReadCount.incrementAndGet();
- }
- return new ModelStage<>(Lists.newArrayList(currentChildren.keySet()));
- }
- return client.getChildren();
+ Set<ZPath> paths = cache.getCurrentChildren(client.modelSpec().path()).keySet();
+ return new ModelStage<>(Lists.newArrayList(paths));
}
@Override
public AsyncStage<Map<ZPath, AsyncStage<T>>> readChildren()
{
- Map<ZPath, ModeledCachedNode<T>> currentChildren = cache.getCurrentChildren(path);
- if ( currentChildren != cache.noChildrenValue() )
- {
- if ( debugCachedReadCount != null )
- {
- debugCachedReadCount.incrementAndGet();
- }
- Map<ZPath, AsyncStage<T>> children = currentChildren.entrySet()
- .stream()
- .map(e -> new AbstractMap.SimpleEntry<>(e.getKey(), e.getValue().getModel()))
- .filter(e -> e.getValue() != null)
- .collect(Collectors.toMap(Map.Entry::getKey, e -> new ModelStage<>(e.getValue())));
- return new ModelStage<>(children);
- }
-
- ModelStage<Map<ZPath, AsyncStage<T>>> modelStage = new ModelStage<>();
- client.getChildren().whenComplete((children, e) -> {
- if ( e != null )
- {
- modelStage.completeExceptionally(e);
- }
- else
- {
- if ( debugCachedReadCount != null )
- {
- debugCachedReadCount.incrementAndGet();
- }
- Map<ZPath, AsyncStage<T>> map = children.stream().collect(Collectors.toMap(Function.identity(), path1 -> at(path1.nodeName()).read()));
- modelStage.complete(map);
- }
- });
- return modelStage;
+ Map<ZPath, AsyncStage<T>> map = cache.getCurrentChildren(client.modelSpec().path()).entrySet()
+ .stream()
+ .map(entry -> new AbstractMap.SimpleEntry<>(entry.getKey(), new ModelStage<>(entry.getValue().getModel())))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ return new ModelStage<>(map);
}
@Override
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModelStage.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModelStage.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModelStage.java
index c28b133..77caed1 100644
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModelStage.java
+++ b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModelStage.java
@@ -43,6 +43,12 @@ class ModelStage<T> extends CompletableFuture<T> implements AsyncStage<T>
complete(value);
}
+ ModelStage(Exception e)
+ {
+ event = null;
+ completeExceptionally(e);
+ }
+
@Override
public CompletionStage<WatchedEvent> event()
{
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModeledCacheImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModeledCacheImpl.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModeledCacheImpl.java
new file mode 100644
index 0000000..c1ab8cd
--- /dev/null
+++ b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModeledCacheImpl.java
@@ -0,0 +1,173 @@
+package org.apache.curator.x.async.modeled.details;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.listen.Listenable;
+import org.apache.curator.framework.listen.ListenerContainer;
+import org.apache.curator.framework.recipes.cache.TreeCache;
+import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
+import org.apache.curator.framework.recipes.cache.TreeCacheListener;
+import org.apache.curator.x.async.modeled.ModelSerializer;
+import org.apache.curator.x.async.modeled.ZPath;
+import org.apache.curator.x.async.modeled.cached.ModeledCache;
+import org.apache.curator.x.async.modeled.cached.ModeledCacheEventType;
+import org.apache.curator.x.async.modeled.cached.ModeledCacheListener;
+import org.apache.curator.x.async.modeled.cached.ModeledCachedNode;
+import org.apache.zookeeper.data.Stat;
+import java.util.AbstractMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+class ModeledCacheImpl<T> implements TreeCacheListener, ModeledCache<T>
+{
+ private final TreeCache cache;
+ private final Map<ZPath, Entry<T>> entries = new ConcurrentHashMap<>();
+ private final ModelSerializer<T> serializer;
+ private final ListenerContainer<ModeledCacheListener<T>> listenerContainer = new ListenerContainer<>();
+
+ private static final class Entry<T>
+ {
+ final Stat stat;
+ final T model;
+
+ Entry(Stat stat, T model)
+ {
+ this.stat = stat;
+ this.model = model;
+ }
+ }
+
+ ModeledCacheImpl(CuratorFramework client, ZPath path, ModelSerializer<T> serializer, boolean compressed)
+ {
+ this.serializer = serializer;
+ cache = TreeCache.newBuilder(client, path.fullPath())
+ .setCacheData(false)
+ .setDataIsCompressed(compressed)
+ .build();
+ }
+
+ public void start()
+ {
+ try
+ {
+ cache.start();
+ }
+ catch ( Exception e )
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void close()
+ {
+ cache.close();
+ entries.clear();
+ }
+
+ @Override
+ public Optional<ModeledCachedNode<T>> getCurrentData(ZPath path)
+ {
+ Entry<T> entry = entries.remove(path);
+ if ( entry != null )
+ {
+ return Optional.of(new InternalCachedNode<>(path, entry));
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public Map<ZPath, ModeledCachedNode<T>> getCurrentChildren(ZPath path)
+ {
+ return entries.entrySet()
+ .stream()
+ .filter(entry -> entry.getKey().startsWith(path))
+ .map(entry -> new AbstractMap.SimpleEntry<>(entry.getKey(), new InternalCachedNode<>(entry.getKey(), entry.getValue())))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ }
+
+ @Override
+ public Listenable<ModeledCacheListener<T>> getListenable()
+ {
+ return listenerContainer;
+ }
+
+ @Override
+ public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception
+ {
+ switch ( event.getType() )
+ {
+ case NODE_ADDED:
+ case NODE_UPDATED:
+ {
+ ZPath path = ZPath.from(event.toString());
+ T model = serializer.deserialize(event.getData().getData());
+ entries.put(path, new Entry<>(event.getData().getStat(), model));
+ ModeledCacheEventType type = (event.getType() == TreeCacheEvent.Type.NODE_ADDED) ? ModeledCacheEventType.NODE_ADDED : ModeledCacheEventType.NODE_UPDATED;
+ accept(type, path, event.getData().getStat(), model);
+ break;
+ }
+
+ case NODE_REMOVED:
+ {
+ ZPath path = ZPath.from(event.toString());
+ Entry<T> entry = entries.remove(path);
+ T model = (entry != null) ? entry.model : serializer.deserialize(event.getData().getData());
+ Stat stat = (entry != null) ? entry.stat : event.getData().getStat();
+ accept(ModeledCacheEventType.NODE_REMOVED, path, stat, model);
+ break;
+ }
+
+ case INITIALIZED:
+ {
+ listenerContainer.forEach(l -> {
+ l.initialized();
+ return null;
+ });
+ break;
+ }
+
+ default:
+ // ignore
+ break;
+ }
+ }
+
+ private void accept(ModeledCacheEventType type, ZPath path, Stat stat, T model)
+ {
+ listenerContainer.forEach(l -> {
+ l.accept(type, path, stat, model);
+ return null;
+ });
+ }
+
+ private static class InternalCachedNode<U> implements ModeledCachedNode<U>
+ {
+ private final ZPath path;
+ private final Entry<U> entry;
+
+ private InternalCachedNode(ZPath path, Entry<U> entry)
+ {
+ this.path = path;
+ this.entry = entry;
+ }
+
+ @Override
+ public ZPath getPath()
+ {
+ return path;
+ }
+
+ @Override
+ public Stat getStat()
+ {
+ return entry.stat;
+ }
+
+ @Override
+ public U getModel()
+ {
+ return entry.model;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModeledCuratorFrameworkImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModeledCuratorFrameworkImpl.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModeledCuratorFrameworkImpl.java
index 6ed0649..aba87f3 100644
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModeledCuratorFrameworkImpl.java
+++ b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModeledCuratorFrameworkImpl.java
@@ -23,7 +23,6 @@ import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.UnhandledErrorListener;
import org.apache.curator.framework.api.transaction.CuratorOp;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
-import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.x.async.AsyncCuratorFramework;
import org.apache.curator.x.async.AsyncStage;
import org.apache.curator.x.async.WatchMode;
@@ -33,12 +32,10 @@ import org.apache.curator.x.async.api.AsyncPathable;
import org.apache.curator.x.async.api.AsyncTransactionSetDataBuilder;
import org.apache.curator.x.async.api.CreateOption;
import org.apache.curator.x.async.api.WatchableAsyncCuratorFramework;
-import org.apache.curator.x.async.modeled.CachedModeledCuratorFramework;
import org.apache.curator.x.async.modeled.CuratorModelSpec;
import org.apache.curator.x.async.modeled.ModeledCuratorFramework;
import org.apache.curator.x.async.modeled.ZPath;
-import org.apache.curator.x.async.modeled.recipes.ModeledCache;
-import org.apache.curator.x.async.modeled.recipes.ModeledTreeCache;
+import org.apache.curator.x.async.modeled.cached.CachedModeledCuratorFramework;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
@@ -98,36 +95,9 @@ public class ModeledCuratorFrameworkImpl<T> implements ModeledCuratorFramework<T
}
@Override
- public CachedModeledCuratorFramework<T> cached(ModeledCache<T> cache)
- {
- return new CachedModeledCuratorFrameworkImpl<>(this, cache, modelSpec.path());
- }
-
- @Override
public CachedModeledCuratorFramework<T> cached()
{
- TreeCache.Builder builder = TreeCache.newBuilder(client.unwrap(), modelSpec.path().fullPath());
- builder = builder.setCacheData(true);
- if ( modelSpec.createOptions().contains(CreateOption.compress) )
- {
- builder = builder.setDataIsCompressed(true);
- }
- TreeCache cache = builder.build();
- ModeledTreeCache<T> wrapped = ModeledTreeCache.wrap(cache, modelSpec.serializer());
- return new CachedModeledCuratorFrameworkImpl<T>(this, wrapped, modelSpec.path())
- {
- @Override
- public void start()
- {
- wrapped.start();
- }
-
- @Override
- public void close()
- {
- wrapped.close();
- }
- };
+ return new CachedModeledCuratorFrameworkImpl<>(this);
}
@Override
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ZPathImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ZPathImpl.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ZPathImpl.java
index 8072785..36c9d1a 100644
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ZPathImpl.java
+++ b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ZPathImpl.java
@@ -73,7 +73,7 @@ public class ZPathImpl implements ZPath
{
names = Objects.requireNonNull(names, "names cannot be null");
names.forEach(ZPathImpl::validate);
- ImmutableList.Builder<String> builder = ImmutableList.<String>builder();
+ ImmutableList.Builder<String> builder = ImmutableList.builder();
if ( base != null )
{
if ( base instanceof ZPathImpl )
@@ -113,6 +113,17 @@ public class ZPathImpl implements ZPath
}
@Override
+ public boolean startsWith(ZPath path)
+ {
+ if ( path instanceof ZPathImpl )
+ {
+ ZPathImpl rhs = (ZPathImpl)path;
+ return (nodes.size() >= rhs.nodes.size()) && nodes.subList(0, rhs.nodes.size()).equals(rhs);
+ }
+ return false;
+ }
+
+ @Override
public Pattern toSchemaPathPattern()
{
return Pattern.compile(fullPath() + ZKPaths.PATH_SEPARATOR + ".*");
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledCachedNodeImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledCachedNodeImpl.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledCachedNodeImpl.java
deleted file mode 100644
index e66fd8a..0000000
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledCachedNodeImpl.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/**
- * 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.async.modeled.details.recipes;
-
-import org.apache.curator.x.async.modeled.ZPath;
-import org.apache.curator.x.async.modeled.recipes.ModeledCachedNode;
-import org.apache.zookeeper.data.Stat;
-import java.util.Objects;
-
-public class ModeledCachedNodeImpl<T> implements ModeledCachedNode<T>
-{
- private final ZPath path;
- private final Stat stat;
- private final T data;
-
- public ModeledCachedNodeImpl(ZPath path)
- {
- this(path, null, new Stat());
- }
-
- public ModeledCachedNodeImpl(ZPath path, T data)
- {
- this(path, data, new Stat());
- }
-
- public ModeledCachedNodeImpl(ZPath path, T data, Stat stat)
- {
- this.path = Objects.requireNonNull(path, "path cannot be null");
- this.stat = Objects.requireNonNull(stat, "stat cannot be null");
- this.data = data;
- }
-
- @Override
- public ZPath getPath()
- {
- return path;
- }
-
- @Override
- public Stat getStat()
- {
- return stat;
- }
-
- @Override
- public T getModel()
- {
- return data;
- }
-
- @Override
- public boolean equals(Object o)
- {
- if ( this == o )
- {
- return true;
- }
- if ( o == null || getClass() != o.getClass() )
- {
- return false;
- }
-
- ModeledCachedNodeImpl<?> that = (ModeledCachedNodeImpl<?>)o;
-
- if ( !path.equals(that.path) )
- {
- return false;
- }
- //noinspection SimplifiableIfStatement
- if ( !stat.equals(that.stat) )
- {
- return false;
- }
- return data != null ? data.equals(that.data) : that.data == null;
- }
-
- @Override
- public int hashCode()
- {
- int result = path.hashCode();
- result = 31 * result + stat.hashCode();
- result = 31 * result + (data != null ? data.hashCode() : 0);
- return result;
- }
-
- @Override
- public String toString()
- {
- return "ModeledCachedNode{" + "stat=" + stat + ", data=" + data + '}';
- }
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledNodeCacheImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledNodeCacheImpl.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledNodeCacheImpl.java
deleted file mode 100644
index 5b89faf..0000000
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledNodeCacheImpl.java
+++ /dev/null
@@ -1,232 +0,0 @@
-/**
- * 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.async.modeled.details.recipes;
-
-import com.google.common.util.concurrent.MoreExecutors;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.listen.Listenable;
-import org.apache.curator.framework.recipes.cache.ChildData;
-import org.apache.curator.framework.recipes.cache.NodeCache;
-import org.apache.curator.framework.recipes.cache.NodeCacheListener;
-import org.apache.curator.framework.state.ConnectionState;
-import org.apache.curator.framework.state.ConnectionStateListener;
-import org.apache.curator.utils.CloseableUtils;
-import org.apache.curator.x.async.modeled.ModelSerializer;
-import org.apache.curator.x.async.modeled.ZPath;
-import org.apache.curator.x.async.modeled.recipes.ModeledCacheEvent;
-import org.apache.curator.x.async.modeled.recipes.ModeledCacheEventType;
-import org.apache.curator.x.async.modeled.recipes.ModeledCacheListener;
-import org.apache.curator.x.async.modeled.recipes.ModeledCachedNode;
-import org.apache.curator.x.async.modeled.recipes.ModeledNodeCache;
-import org.apache.zookeeper.data.Stat;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
-
-public class ModeledNodeCacheImpl<T> implements ModeledNodeCache<T>, ConnectionStateListener
-{
- private final NodeCache cache;
- private final ModelSerializer<T> serializer;
- private final ZPath path;
- private final Map<ModeledCacheListener<T>, NodeCacheListener> listenerMap = new ConcurrentHashMap<>();
-
- public ModeledNodeCacheImpl(NodeCache cache, ModelSerializer<T> serializer)
- {
- this.cache = Objects.requireNonNull(cache, "cache cannot be null");
- this.serializer = Objects.requireNonNull(serializer, "serializer cannot be null");
- path = ZPath.parse(cache.getPath());
- }
-
- @Override
- public void stateChanged(CuratorFramework client, ConnectionState newState)
- {
- ModeledCacheEventType mappedType;
- switch ( newState )
- {
- default:
- {
- mappedType = null;
- break;
- }
-
- case RECONNECTED:
- case CONNECTED:
- {
- mappedType = ModeledCacheEventType.CONNECTION_RECONNECTED;
- break;
- }
-
- case SUSPENDED:
- {
- mappedType = ModeledCacheEventType.CONNECTION_SUSPENDED;
- break;
- }
-
- case LOST:
- {
- mappedType = ModeledCacheEventType.CONNECTION_LOST;
- break;
- }
- }
-
- if ( mappedType != null )
- {
- ModeledCacheEvent<T> event = new ModeledCacheEvent<T>()
- {
- @Override
- public ModeledCacheEventType getType()
- {
- return mappedType;
- }
-
- @Override
- public ModeledCachedNode<T> getNode()
- {
- return null;
- }
- };
- listenerMap.keySet().forEach(l -> l.event(null));
- }
- }
-
- @Override
- public NodeCache unwrap()
- {
- return cache;
- }
-
- @Override
- public void start()
- {
- try
- {
- cache.start();
- }
- catch ( Exception e )
- {
- throw new RuntimeException("Could not start", e);
- }
- cache.getClient().getConnectionStateListenable().addListener(this);
- }
-
- @Override
- public void start(boolean buildInitial)
- {
- cache.getClient().getConnectionStateListenable().removeListener(this);
- try
- {
- cache.start(buildInitial);
- }
- catch ( Exception e )
- {
- throw new RuntimeException("Could not start", e);
- }
- }
-
- @Override
- public void rebuild()
- {
- try
- {
- cache.rebuild();
- }
- catch ( Exception e )
- {
- throw new RuntimeException("Could not rebuild", e);
- }
- }
-
- @Override
- public Listenable<ModeledCacheListener<T>> getListenable()
- {
- return new Listenable<ModeledCacheListener<T>>()
- {
- @Override
- public void addListener(ModeledCacheListener<T> listener)
- {
- addListener(listener, MoreExecutors.sameThreadExecutor());
- }
-
- @Override
- public void addListener(ModeledCacheListener<T> listener, Executor executor)
- {
- NodeCacheListener nodeCacheListener = () ->
- {
- Optional<ModeledCachedNode<T>> currentData = getCurrentData();
- ModeledCacheEvent<T> event = new ModeledCacheEvent<T>()
- {
- @Override
- public ModeledCacheEventType getType()
- {
- return currentData.isPresent() ? ModeledCacheEventType.NODE_UPDATED : ModeledCacheEventType.NODE_REMOVED;
- }
-
- @Override
- public ModeledCachedNode<T> getNode()
- {
- return currentData.orElse(null);
- }
- };
- listener.event(event);
- };
- listenerMap.put(listener, nodeCacheListener);
- cache.getListenable().addListener(nodeCacheListener, executor);
- }
-
- @Override
- public void removeListener(ModeledCacheListener<T> listener)
- {
- NodeCacheListener nodeCacheListener = listenerMap.remove(listener);
- if ( nodeCacheListener != null )
- {
- cache.getListenable().removeListener(nodeCacheListener);
- }
- }
- };
- }
-
- @Override
- public Optional<ModeledCachedNode<T>> getCurrentData()
- {
- ChildData currentData = cache.getCurrentData();
- if ( currentData == null )
- {
- return Optional.empty();
- }
- byte[] data = currentData.getData();
- Stat stat = currentData.getStat();
- if ( stat == null )
- {
- stat = new Stat();
- }
- if ( (data == null) || (data.length == 0) )
- {
- return Optional.of(new ModeledCachedNodeImpl<T>(path, null, stat));
- }
- return Optional.of(new ModeledCachedNodeImpl<>(path, serializer.deserialize(data), stat));
- }
-
- @Override
- public void close()
- {
- CloseableUtils.closeQuietly(cache);
- }
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledPathChildrenCacheImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledPathChildrenCacheImpl.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledPathChildrenCacheImpl.java
deleted file mode 100644
index ed86404..0000000
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledPathChildrenCacheImpl.java
+++ /dev/null
@@ -1,266 +0,0 @@
-/**
- * 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.async.modeled.details.recipes;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Maps;
-import com.google.common.util.concurrent.MoreExecutors;
-import org.apache.curator.framework.listen.Listenable;
-import org.apache.curator.framework.recipes.cache.ChildData;
-import org.apache.curator.framework.recipes.cache.PathChildrenCache;
-import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
-import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
-import org.apache.curator.utils.CloseableUtils;
-import org.apache.curator.x.async.modeled.ModelSerializer;
-import org.apache.curator.x.async.modeled.ZPath;
-import org.apache.curator.x.async.modeled.recipes.ModeledCacheEvent;
-import org.apache.curator.x.async.modeled.recipes.ModeledCacheEventType;
-import org.apache.curator.x.async.modeled.recipes.ModeledCacheListener;
-import org.apache.curator.x.async.modeled.recipes.ModeledCachedNode;
-import org.apache.curator.x.async.modeled.recipes.ModeledPathChildrenCache;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
-import java.util.stream.Collectors;
-
-public class ModeledPathChildrenCacheImpl<T> implements ModeledPathChildrenCache<T>
-{
- private final PathChildrenCache cache;
- private final Map<ModeledCacheListener<T>, PathChildrenCacheListener> listenerMap = new ConcurrentHashMap<>();
- private final ModelSerializer<T> serializer;
-
- public ModeledPathChildrenCacheImpl(PathChildrenCache cache, ModelSerializer<T> serializer)
- {
- this.cache = Objects.requireNonNull(cache, "cache cannot be null");
- this.serializer = Objects.requireNonNull(serializer, "serializer cannot be null");
- }
-
- @Override
- public PathChildrenCache unwrap()
- {
- return cache;
- }
-
- @Override
- public void start()
- {
- try
- {
- cache.start();
- }
- catch ( Exception e )
- {
- throw new RuntimeException("can't start cache", e);
- }
- }
-
- @Override
- public void start(PathChildrenCache.StartMode mode)
- {
- try
- {
- cache.start(mode);
- }
- catch ( Exception e )
- {
- throw new RuntimeException("can't start cache", e);
- }
- }
-
- @Override
- public void rebuild()
- {
- try
- {
- cache.rebuild();
- }
- catch ( Exception e )
- {
- throw new RuntimeException("can't rebuild cache", e);
- }
- }
-
- @Override
- public void rebuildNode(ZPath fullPath)
- {
- try
- {
- cache.rebuildNode(fullPath.fullPath());
- }
- catch ( Exception e )
- {
- throw new RuntimeException("can't rebuild cache at " + fullPath, e);
- }
- }
-
- @Override
- public Listenable<ModeledCacheListener<T>> getListenable()
- {
- return new Listenable<ModeledCacheListener<T>>()
- {
- @Override
- public void addListener(ModeledCacheListener<T> listener)
- {
- addListener(listener, MoreExecutors.sameThreadExecutor());
- }
-
- @Override
- public void addListener(ModeledCacheListener<T> listener, Executor executor)
- {
- PathChildrenCacheListener pathChildrenCacheListener = (client, event) -> {
- ModeledCacheEventType eventType = toType(event.getType());
- ModeledCachedNode<T> node = from(serializer, event.getData());
- ModeledCacheEvent<T> modeledEvent = new ModeledCacheEvent<T>()
- {
- @Override
- public ModeledCacheEventType getType()
- {
- return eventType;
- }
-
- @Override
- public ModeledCachedNode<T> getNode()
- {
- return node;
- }
- };
- listener.event(modeledEvent);
- };
- listenerMap.put(listener, pathChildrenCacheListener);
- cache.getListenable().addListener(pathChildrenCacheListener);
- }
-
- @Override
- public void removeListener(ModeledCacheListener listener)
- {
- PathChildrenCacheListener pathChildrenCacheListener = listenerMap.remove(listener);
- if ( pathChildrenCacheListener != null )
- {
- cache.getListenable().removeListener(pathChildrenCacheListener);
- }
- }
- };
- }
-
- @Override
- public List<ModeledCachedNode<T>> getCurrentData()
- {
- return cache.getCurrentData().stream()
- .map(data -> from(serializer, data))
- .collect(Collectors.toList());
- }
-
- @Override
- public Map<ZPath, ModeledCachedNode<T>> getCurrentChildren(ZPath fullPath)
- {
- ChildData currentData = cache.getCurrentData(fullPath.fullPath());
- if ( currentData == null )
- {
- return noChildrenValue();
- }
- Map<ZPath, ModeledCachedNode<T>> map = Maps.newHashMap();
- map.put(fullPath, from(serializer, currentData));
- return map;
- }
-
- @Override
- public Optional<ModeledCachedNode<T>> getCurrentData(ZPath fullPath)
- {
- return Optional.ofNullable(from(serializer, cache.getCurrentData(fullPath.fullPath())));
- }
-
- @Override
- public void clearDataBytes(ZPath fullPath)
- {
- cache.clearDataBytes(fullPath.fullPath());
- }
-
- @Override
- public boolean clearDataBytes(ZPath fullPath, int ifVersion)
- {
- return cache.clearDataBytes(fullPath.fullPath(), ifVersion);
- }
-
- @Override
- public void clearAndRefresh()
- {
- try
- {
- cache.clearAndRefresh();
- }
- catch ( Exception e )
- {
- throw new RuntimeException("could not clear and refresh", e);
- }
- }
-
- @Override
- public void clear()
- {
- cache.clear();
- }
-
- @Override
- public void close()
- {
- CloseableUtils.closeQuietly(cache);
- }
-
- static <T> ModeledCachedNode<T> from(ModelSerializer<T> serializer, ChildData data)
- {
- if ( data == null )
- {
- return null;
- }
- T model = ((data.getData() != null) && (data.getData().length > 0)) ? serializer.deserialize(data.getData()) : null;
- return new ModeledCachedNodeImpl<>(ZPath.parse(data.getPath()), model, data.getStat());
- }
-
- @VisibleForTesting
- static ModeledCacheEventType toType(PathChildrenCacheEvent.Type type)
- {
- switch ( type )
- {
- case CHILD_ADDED:
- return ModeledCacheEventType.NODE_ADDED;
-
- case CHILD_UPDATED:
- return ModeledCacheEventType.NODE_UPDATED;
-
- case CHILD_REMOVED:
- return ModeledCacheEventType.NODE_REMOVED;
-
- case CONNECTION_SUSPENDED:
- return ModeledCacheEventType.CONNECTION_SUSPENDED;
-
- case CONNECTION_RECONNECTED:
- return ModeledCacheEventType.CONNECTION_RECONNECTED;
-
- case CONNECTION_LOST:
- return ModeledCacheEventType.CONNECTION_LOST;
-
- case INITIALIZED:
- return ModeledCacheEventType.INITIALIZED;
- }
- throw new UnsupportedOperationException("Unknown type: " + type);
- }
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledTreeCacheImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledTreeCacheImpl.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledTreeCacheImpl.java
deleted file mode 100644
index 7f0aecc..0000000
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/recipes/ModeledTreeCacheImpl.java
+++ /dev/null
@@ -1,179 +0,0 @@
-/**
- * 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.async.modeled.details.recipes;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.util.concurrent.MoreExecutors;
-import org.apache.curator.framework.listen.Listenable;
-import org.apache.curator.framework.recipes.cache.ChildData;
-import org.apache.curator.framework.recipes.cache.TreeCache;
-import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
-import org.apache.curator.framework.recipes.cache.TreeCacheListener;
-import org.apache.curator.utils.CloseableUtils;
-import org.apache.curator.x.async.modeled.ModelSerializer;
-import org.apache.curator.x.async.modeled.ZPath;
-import org.apache.curator.x.async.modeled.recipes.ModeledCacheEvent;
-import org.apache.curator.x.async.modeled.recipes.ModeledCacheEventType;
-import org.apache.curator.x.async.modeled.recipes.ModeledCacheListener;
-import org.apache.curator.x.async.modeled.recipes.ModeledCachedNode;
-import org.apache.curator.x.async.modeled.recipes.ModeledTreeCache;
-import java.util.AbstractMap;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executor;
-import java.util.stream.Collectors;
-
-import static org.apache.curator.x.async.modeled.details.recipes.ModeledPathChildrenCacheImpl.from;
-
-public class ModeledTreeCacheImpl<T> implements ModeledTreeCache<T>
-{
- private final TreeCache cache;
- private final Map<ModeledCacheListener<T>, TreeCacheListener> listenerMap = new ConcurrentHashMap<>();
- private final ModelSerializer<T> serializer;
-
- public ModeledTreeCacheImpl(TreeCache cache, ModelSerializer<T> serializer)
- {
- this.cache = Objects.requireNonNull(cache, "cache cannot be null");
- this.serializer = Objects.requireNonNull(serializer, "serializer cannot be null");
- }
-
- @Override
- public TreeCache unwrap()
- {
- return cache;
- }
-
- @Override
- public void start()
- {
- try
- {
- cache.start();
- }
- catch ( Exception e )
- {
- throw new RuntimeException("Could not start", e);
- }
- }
-
- @Override
- public void close()
- {
- CloseableUtils.closeQuietly(cache);
- }
-
- @Override
- public Listenable<ModeledCacheListener<T>> getListenable()
- {
- return new Listenable<ModeledCacheListener<T>>()
- {
- @Override
- public void addListener(ModeledCacheListener<T> listener)
- {
- addListener(listener, MoreExecutors.sameThreadExecutor());
- }
-
- @Override
- public void addListener(ModeledCacheListener<T> listener, Executor executor)
- {
- TreeCacheListener treeCacheListener = (client, event) -> {
- ModeledCacheEventType eventType = toType(event.getType());
- ModeledCachedNode<T> node = from(serializer, event.getData());
- ModeledCacheEvent<T> wrappedEvent = new ModeledCacheEvent<T>()
- {
- @Override
- public ModeledCacheEventType getType()
- {
- return eventType;
- }
-
- @Override
- public ModeledCachedNode<T> getNode()
- {
- return node;
- }
- };
- listener.event(wrappedEvent);
- };
- listenerMap.put(listener, treeCacheListener);
- cache.getListenable().addListener(treeCacheListener, executor);
- }
-
- @Override
- public void removeListener(ModeledCacheListener<T> listener)
- {
- TreeCacheListener treeCacheListener = listenerMap.remove(listener);
- if ( treeCacheListener != null )
- {
- cache.getListenable().removeListener(treeCacheListener);
- }
- }
- };
- }
-
- @Override
- public Map<ZPath, ModeledCachedNode<T>> getCurrentChildren(ZPath fullPath)
- {
- Map<String, ChildData> currentChildren = cache.getCurrentChildren(fullPath.fullPath());
- if ( currentChildren == null )
- {
- return noChildrenValue();
- }
- return currentChildren.entrySet().stream()
- .map(entry -> new AbstractMap.SimpleEntry<>(ZPath.parse(entry.getKey()), from(serializer, entry.getValue())))
- .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
- }
-
- @Override
- public Optional<ModeledCachedNode<T>> getCurrentData(ZPath fullPath)
- {
- return Optional.ofNullable(from(serializer, cache.getCurrentData(fullPath.fullPath())));
- }
-
- @VisibleForTesting
- static ModeledCacheEventType toType(TreeCacheEvent.Type type)
- {
- switch ( type )
- {
- case NODE_ADDED:
- return ModeledCacheEventType.NODE_ADDED;
-
- case NODE_UPDATED:
- return ModeledCacheEventType.NODE_UPDATED;
-
- case NODE_REMOVED:
- return ModeledCacheEventType.NODE_REMOVED;
-
- case CONNECTION_SUSPENDED:
- return ModeledCacheEventType.CONNECTION_SUSPENDED;
-
- case CONNECTION_RECONNECTED:
- return ModeledCacheEventType.CONNECTION_RECONNECTED;
-
- case CONNECTION_LOST:
- return ModeledCacheEventType.CONNECTION_LOST;
-
- case INITIALIZED:
- return ModeledCacheEventType.INITIALIZED;
- }
- throw new UnsupportedOperationException("Unknown type: " + type);
- }
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCache.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCache.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCache.java
deleted file mode 100644
index b810512..0000000
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCache.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/**
- * 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.async.modeled.recipes;
-
-import org.apache.curator.x.async.modeled.ZPath;
-import java.util.Collections;
-import java.util.Map;
-import java.util.Optional;
-
-public interface ModeledCache<T>
-{
- default Map<ZPath, ModeledCachedNode<T>> noChildrenValue()
- {
- return Collections.emptyMap();
- }
-
- /**
- * Return the modeled current data for the given path. There are no guarantees of accuracy. This is
- * merely the most recent view of the data. If there is no node at the given path,
- * {@link java.util.Optional#empty()} is returned.
- *
- * @param fullPath full path to the node to check
- * @return data if the node is alive, or null
- */
- Optional<ModeledCachedNode<T>> getCurrentData(ZPath fullPath);
-
- /**
- * Return the modeled current set of children at the given path, mapped by child name. There are no
- * guarantees of accuracy; this is merely the most recent view of the data.
- *
- * @param fullPath full path to the node to check
- * @return a possibly-empty list of children if the node is alive, or null
- */
- Map<ZPath, ModeledCachedNode<T>> getCurrentChildren(ZPath fullPath);
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCacheEvent.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCacheEvent.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCacheEvent.java
deleted file mode 100644
index 469d8d6..0000000
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCacheEvent.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/**
- * 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.async.modeled.recipes;
-
-/**
- * Abstracts a cache event
- */
-public interface ModeledCacheEvent<T>
-{
- /**
- * The event type
- *
- * @return event type
- */
- ModeledCacheEventType getType();
-
- /**
- * Cached node if appropriate for the event (i.e. NODE_* events)
- *
- * @return node
- */
- ModeledCachedNode<T> getNode();
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCacheEventType.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCacheEventType.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCacheEventType.java
deleted file mode 100644
index bfdf57d..0000000
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCacheEventType.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/**
- * 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.async.modeled.recipes;
-
-public enum ModeledCacheEventType
-{
- /**
- * A child was added to the path
- */
- NODE_ADDED,
-
- /**
- * A child's data was changed
- */
- NODE_UPDATED,
-
- /**
- * A child was removed from the path
- */
- NODE_REMOVED,
-
- /**
- * Called when the connection has changed to {@link org.apache.curator.framework.state.ConnectionState#SUSPENDED}
- */
- CONNECTION_SUSPENDED,
-
- /**
- * Called when the connection has changed to {@link org.apache.curator.framework.state.ConnectionState#RECONNECTED}
- */
- CONNECTION_RECONNECTED,
-
- /**
- * Called when the connection has changed to {@link org.apache.curator.framework.state.ConnectionState#LOST}
- */
- CONNECTION_LOST,
-
- /**
- * Signals that the initial cache has been populated.
- */
- INITIALIZED
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCacheListener.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCacheListener.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCacheListener.java
deleted file mode 100644
index 7b82e72..0000000
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCacheListener.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/**
- * 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.async.modeled.recipes;
-
-import java.util.function.Predicate;
-
-/**
- * Event listener
- */
-@FunctionalInterface
-public interface ModeledCacheListener<T>
-{
- /**
- * Receive an event
- *
- * @param event the event
- */
- void event(ModeledCacheEvent<T> event);
-
- /**
- * Wrap this listener with a filter
- *
- * @param filter test for events. Only events that pass the filter are sent to the listener
- * @return filtered version of this listener
- */
- static <T> ModeledCacheListener<T> filtered(ModeledCacheListener<T> listener, Predicate<ModeledCacheEvent<T>> filter)
- {
- return event -> {
- if ( filter.test(event) )
- {
- listener.event(event);
- }
- };
- }
-
- /**
- * Filters out all but CRUD events
- *
- * @return predicate
- */
- static <T> Predicate<ModeledCacheEvent<T>> nodeEventFilter()
- {
- return event -> (event.getType() == ModeledCacheEventType.NODE_ADDED)
- || (event.getType() == ModeledCacheEventType.NODE_UPDATED)
- || (event.getType() == ModeledCacheEventType.NODE_REMOVED)
- ;
- }
-
- /**
- * Filters out all but {@link ModeledCacheEventType#NODE_REMOVED} events
- *
- * @return predicate
- */
- static <T> Predicate<ModeledCacheEvent<T>> nodeRemovedFilter()
- {
- return event -> event.getType() == ModeledCacheEventType.NODE_REMOVED;
- }
-
- /**
- * Filters out all but events that have valid model instances
- *
- * @return predicate
- */
- static <T> Predicate<ModeledCacheEvent<T>> hasModelFilter()
- {
- return event -> (event.getNode() != null) && (event.getNode().getModel() != null);
- }
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCachedNode.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCachedNode.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCachedNode.java
deleted file mode 100644
index a54f7b7..0000000
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledCachedNode.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/**
- * 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.async.modeled.recipes;
-
-import org.apache.curator.x.async.modeled.ZPath;
-import org.apache.zookeeper.data.Stat;
-
-/**
- * Abstracts a cached node
- */
-public interface ModeledCachedNode<T>
-{
- /**
- * The path of the node
- *
- * @return path
- */
- ZPath getPath();
-
- /**
- * The node's last known stat if available
- *
- * @return stat
- */
- Stat getStat();
-
- /**
- * The node's current model
- *
- * @return model
- */
- T getModel();
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cbbf999/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledNodeCache.java
----------------------------------------------------------------------
diff --git a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledNodeCache.java b/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledNodeCache.java
deleted file mode 100644
index a5fb598..0000000
--- a/curator-x-async/src/main/java/org/apache/curator/x/async/modeled/recipes/ModeledNodeCache.java
+++ /dev/null
@@ -1,91 +0,0 @@
-/**
- * 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.async.modeled.recipes;
-
-import org.apache.curator.framework.listen.Listenable;
-import org.apache.curator.framework.recipes.cache.NodeCache;
-import org.apache.curator.x.async.modeled.ModelSerializer;
-import org.apache.curator.x.async.modeled.details.recipes.ModeledNodeCacheImpl;
-import java.io.Closeable;
-import java.util.Optional;
-
-/**
- * Wraps a {@link org.apache.curator.framework.recipes.cache.NodeCache} so that
- * node data can be viewed as strongly typed models.
- */
-public interface ModeledNodeCache<T> extends Closeable
-{
- /**
- * Return a newly wrapped cache
- *
- * @param cache the cache to wrap
- * @param serializer for the model
- * @return new wrapped cache
- */
- static <T> ModeledNodeCache<T> wrap(NodeCache cache, ModelSerializer<T> serializer)
- {
- return new ModeledNodeCacheImpl<>(cache, serializer);
- }
-
- /**
- * Return the original cache that was wrapped
- *
- * @return cache
- */
- NodeCache unwrap();
-
- /**
- * Forwards to {@link org.apache.curator.framework.recipes.cache.NodeCache#start()}
- */
- void start();
-
- /**
- * Forwards to {@link org.apache.curator.framework.recipes.cache.NodeCache#start(boolean)}
- */
- void start(boolean buildInitial);
-
- /**
- * Forwards to {@link org.apache.curator.framework.recipes.cache.NodeCache#rebuild()}
- */
- void rebuild();
-
- /**
- * Return the listener container so that you can add/remove listeners. Note:
- * {@link org.apache.curator.x.async.modeled.recipes.ModeledCacheEventType#INITIALIZED}
- * and {@link org.apache.curator.x.async.modeled.recipes.ModeledCacheEventType#NODE_ADDED} are not
- * used.
- *
- * @return listener container
- */
- Listenable<ModeledCacheListener<T>> getListenable();
-
- /**
- * Return the modeled current data. There are no guarantees of accuracy. This is
- * merely the most recent view of the data. If the node does not exist,
- * this returns {@link java.util.Optional#empty()} is returned
- *
- * @return node data
- */
- Optional<ModeledCachedNode<T>> getCurrentData();
-
- /**
- * Forwards to {@link org.apache.curator.framework.recipes.cache.NodeCache#close()}
- */
- void close();
-}