watched version of getConfig

Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/3aa51d50
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/3aa51d50
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/3aa51d50

Branch: refs/heads/CURATOR-3.0
Commit: 3aa51d503afe2f5952686976681194418afe171f
Parents: 81f0ab2
Author: randgalt <randg...@apache.org>
Authored: Thu Jan 5 23:17:03 2017 -0500
Committer: randgalt <randg...@apache.org>
Committed: Thu Jan 5 23:17:03 2017 -0500

----------------------------------------------------------------------
 .../curator/x/crimps/async/AsyncCrimps.java     | 69 ++++++++--------
 .../crimps/async/CrimpedBackgroundCallback.java |  6 +-
 .../crimps/async/CrimpedConfigEnsembleable.java |  9 +--
 .../x/crimps/async/CrimpedEnsembleable.java     |  4 +-
 .../x/crimps/async/CrimpedEnsembleableImpl.java |  2 +-
 .../async/CrimpedWatchedEnsembleable.java       | 29 +++++++
 .../async/CrimpedWatchedEnsembleableImpl.java   | 83 ++++++++++++++++++++
 .../curator/x/crimps/async/TestCrimps.java      |  9 +--
 8 files changed, 155 insertions(+), 56 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/curator/blob/3aa51d50/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/AsyncCrimps.java
----------------------------------------------------------------------
diff --git 
a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/AsyncCrimps.java
 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/AsyncCrimps.java
index 639deca..bc3a2c3 100644
--- 
a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/AsyncCrimps.java
+++ 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/AsyncCrimps.java
@@ -22,7 +22,6 @@ import org.apache.curator.framework.api.*;
 import 
org.apache.curator.framework.api.transaction.CuratorMultiTransactionMain;
 import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
 import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Stat;
 import java.util.List;
@@ -98,28 +97,28 @@ public class AsyncCrimps
         return build(builder, dataProc);
     }
 
+    public CrimpedPathable<Crimped<byte[]>> 
dataWatched(Watchable<BackgroundPathable<byte[]>> builder)
+    {
+        return build(builder, dataProc);
+    }
+
     public CrimpedPathable<CompletionStage<List<String>>> 
children(BackgroundPathable<List<String>> builder)
     {
         return build(builder, childrenProc);
     }
 
-    public CrimpedPathable<CompletionStage<Stat>> 
stat(BackgroundPathable<Stat> builder)
+    public CrimpedPathable<Crimped<List<String>>> 
childrenWatched(Watchable<BackgroundPathable<List<String>>> builder)
     {
-        return build(builder, statProc);
+        return build(builder, childrenProc);
     }
 
-    public CrimpedPathable<CompletionStage<Stat>> 
safeStat(BackgroundPathable<Stat> builder)
+    public CrimpedPathable<CompletionStage<Stat>> 
stat(BackgroundPathable<Stat> builder)
     {
         return build(builder, safeStatProc);
     }
 
     public CrimpedPathable<Crimped<Stat>> 
statWatched(Watchable<BackgroundPathable<Stat>> builder)
     {
-        return build(builder, statProc);
-    }
-
-    public CrimpedPathable<Crimped<Stat>> 
safeStatWatched(Watchable<BackgroundPathable<Stat>> builder)
-    {
         return build(builder, safeStatProc);
     }
 
@@ -133,26 +132,23 @@ public class AsyncCrimps
         return build(builder, statProc);
     }
 
-    public CrimpedEnsembleable 
ensemble(Backgroundable<ErrorListenerEnsembleable<byte[]>> builder)
+    public CrimpedWatchedEnsembleable 
ensembleWatched(Watchable<BackgroundEnsembleable<byte[]>> builder)
     {
-        CrimpedBackgroundCallback<byte[]> callback = new 
CrimpedBackgroundCallback<>(dataProc);
-
-        Ensembleable<byte[]> main;
-        if ( unhandledErrorListener != null )
-        {
-            main = 
builder.inBackground(callback).withUnhandledErrorListener(unhandledErrorListener);
-        }
-        else
-        {
-            main = builder.inBackground(callback);
-        }
+        CrimpedWatcher crimpedWatcher = new CrimpedWatcher();
+        CrimpedBackgroundCallback<byte[]> callback = new 
CrimpedBackgroundCallback<>(dataProc, crimpedWatcher);
+        BackgroundEnsembleable<byte[]> localBuilder = 
builder.usingWatcher(crimpedWatcher);
+        return new CrimpedWatchedEnsembleableImpl(toFinalBuilder(callback, 
localBuilder), callback);
+    }
 
-        return new CrimpedEnsembleableImpl(main, callback);
+    public CrimpedEnsembleable 
ensemble(Backgroundable<ErrorListenerEnsembleable<byte[]>> builder)
+    {
+        CrimpedBackgroundCallback<byte[]> callback = new 
CrimpedBackgroundCallback<>(dataProc, null);
+        return new CrimpedEnsembleableImpl(toFinalBuilder(callback, builder), 
callback);
     }
 
     public CrimpedEnsembleable 
ensemble(Backgroundable<ErrorListenerReconfigBuilderMain> builder, List<String> 
newMembers)
     {
-        CrimpedBackgroundCallback<byte[]> callback = new 
CrimpedBackgroundCallback<>(dataProc);
+        CrimpedBackgroundCallback<byte[]> callback = new 
CrimpedBackgroundCallback<>(dataProc, null);
 
         ReconfigBuilderMain main;
         if ( unhandledErrorListener != null )
@@ -169,7 +165,7 @@ public class AsyncCrimps
 
     public CrimpedEnsembleable 
ensemble(Backgroundable<ErrorListenerReconfigBuilderMain> builder, List<String> 
joining, List<String> leaving)
     {
-        CrimpedBackgroundCallback<byte[]> callback = new 
CrimpedBackgroundCallback<>(dataProc);
+        CrimpedBackgroundCallback<byte[]> callback = new 
CrimpedBackgroundCallback<>(dataProc, null);
 
         ReconfigBuilderMain main;
         if ( unhandledErrorListener != null )
@@ -204,7 +200,7 @@ public class AsyncCrimps
 
     public CrimpedMultiTransaction 
opResults(Backgroundable<ErrorListenerMultiTransactionMain> builder)
     {
-        CrimpedBackgroundCallback<List<CuratorTransactionResult>> callback = 
new CrimpedBackgroundCallback<>(opResultsProc);
+        CrimpedBackgroundCallback<List<CuratorTransactionResult>> callback = 
new CrimpedBackgroundCallback<>(opResultsProc, null);
         ErrorListenerMultiTransactionMain main = 
builder.inBackground(callback);
         CuratorMultiTransactionMain finalBuilder = (unhandledErrorListener != 
null) ? main.withUnhandledErrorListener(unhandledErrorListener) : main;
         return ops -> {
@@ -222,7 +218,7 @@ public class AsyncCrimps
 
     public <T> CrimpledPathAndBytesable<CompletionStage<T>> 
build(BackgroundPathAndBytesable<T> builder, BackgroundProc<T> backgroundProc)
     {
-        CrimpedBackgroundCallback<T> callback = new 
CrimpedBackgroundCallback<T>(backgroundProc);
+        CrimpedBackgroundCallback<T> callback = new 
CrimpedBackgroundCallback<T>(backgroundProc, null);
         ErrorListenerPathAndBytesable<T> localBuilder = 
builder.inBackground(callback);
         PathAndBytesable<T> finalLocalBuilder = (unhandledErrorListener != 
null) ? localBuilder.withUnhandledErrorListener(unhandledErrorListener) : 
localBuilder;
         return new CrimpledPathAndBytesableImpl<>(finalLocalBuilder, callback, 
null);
@@ -231,28 +227,27 @@ public class AsyncCrimps
     public <T> CrimpedPathable<Crimped<T>> 
build(Watchable<BackgroundPathable<T>> builder, BackgroundProc<T> 
backgroundProc)
     {
         CrimpedWatcher crimpedWatcher = new CrimpedWatcher();
-        CrimpedBackgroundCallback<T> callback = new 
CrimpedBackgroundCallback<T>(backgroundProc)
-        {
-            @Override
-            public CompletionStage<WatchedEvent> event()
-            {
-                return crimpedWatcher;
-            }
-        };
+        CrimpedBackgroundCallback<T> callback = new 
CrimpedBackgroundCallback<>(backgroundProc, crimpedWatcher);
         Pathable<T> finalLocalBuilder = toFinalBuilder(callback, 
builder.usingWatcher(crimpedWatcher));
         return new CrimpledPathAndBytesableImpl<T, 
Crimped<T>>(finalLocalBuilder, callback, crimpedWatcher);
     }
 
     public <T> CrimpedPathable<CompletionStage<T>> build(BackgroundPathable<T> 
builder, BackgroundProc<T> backgroundProc)
     {
-        CrimpedBackgroundCallback<T> callback = new 
CrimpedBackgroundCallback<T>(backgroundProc);
+        CrimpedBackgroundCallback<T> callback = new 
CrimpedBackgroundCallback<T>(backgroundProc, null);
         Pathable<T> finalLocalBuilder = toFinalBuilder(callback, builder);
         return new CrimpledPathAndBytesableImpl<>(finalLocalBuilder, callback, 
null);
     }
 
-    private <T> Pathable<T> toFinalBuilder(CrimpedBackgroundCallback<T> 
callback, BackgroundPathable<T> backgroundPathable)
+    private Ensembleable<byte[]> 
toFinalBuilder(CrimpedBackgroundCallback<byte[]> callback, 
Backgroundable<ErrorListenerEnsembleable<byte[]>> builder)
+    {
+        ErrorListenerEnsembleable<byte[]> localBuilder = 
builder.inBackground(callback);
+        return (unhandledErrorListener != null) ? 
localBuilder.withUnhandledErrorListener(unhandledErrorListener) : localBuilder;
+    }
+
+    private <T> Pathable<T> toFinalBuilder(CrimpedBackgroundCallback<T> 
callback, BackgroundPathable<T> builder)
     {
-        ErrorListenerPathable<T> localBuilder = 
backgroundPathable.inBackground(callback);
+        ErrorListenerPathable<T> localBuilder = builder.inBackground(callback);
         return (unhandledErrorListener != null) ? 
localBuilder.withUnhandledErrorListener(unhandledErrorListener) : localBuilder;
     }
 

http://git-wip-us.apache.org/repos/asf/curator/blob/3aa51d50/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedBackgroundCallback.java
----------------------------------------------------------------------
diff --git 
a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedBackgroundCallback.java
 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedBackgroundCallback.java
index b3c20d2..9d9a6fe 100644
--- 
a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedBackgroundCallback.java
+++ 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedBackgroundCallback.java
@@ -28,16 +28,18 @@ import java.util.concurrent.CompletionStage;
 class CrimpedBackgroundCallback<T> extends CompletableFuture<T> implements 
BackgroundCallback, Crimped<T>
 {
     private final BackgroundProc<T> resultFunction;
+    private final CrimpedWatcher watcher;
 
-    CrimpedBackgroundCallback(BackgroundProc<T> resultFunction)
+    CrimpedBackgroundCallback(BackgroundProc<T> resultFunction, CrimpedWatcher 
watcher)
     {
         this.resultFunction = resultFunction;
+        this.watcher = watcher;
     }
 
     @Override
     public CompletionStage<WatchedEvent> event()
     {
-        return null;
+        return watcher;
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/curator/blob/3aa51d50/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedConfigEnsembleable.java
----------------------------------------------------------------------
diff --git 
a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedConfigEnsembleable.java
 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedConfigEnsembleable.java
index 10c0c35..6d8811f 100644
--- 
a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedConfigEnsembleable.java
+++ 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedConfigEnsembleable.java
@@ -18,16 +18,13 @@
  */
 package org.apache.curator.x.crimps.async;
 
-import org.apache.curator.framework.api.Ensembleable;
-import java.util.concurrent.CompletionStage;
-
-public interface CrimpedConfigEnsembleable extends
-    CrimpledEnsembleable<CompletionStage<byte[]>>
+public interface CrimpedConfigEnsembleable<T> extends
+    CrimpledEnsembleable<T>
 {
     /**
      * Sets the configuration version to use.
      * @param config The version of the configuration.
      * @return this
      */
-    CrimpledEnsembleable<CompletionStage<byte[]>> fromConfig(long config);
+    CrimpledEnsembleable<T> fromConfig(long config);
 }

http://git-wip-us.apache.org/repos/asf/curator/blob/3aa51d50/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedEnsembleable.java
----------------------------------------------------------------------
diff --git 
a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedEnsembleable.java
 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedEnsembleable.java
index c8166e7..7824fb6 100644
--- 
a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedEnsembleable.java
+++ 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedEnsembleable.java
@@ -22,8 +22,8 @@ import org.apache.curator.framework.api.Statable;
 import java.util.concurrent.CompletionStage;
 
 public interface CrimpedEnsembleable extends
-    CrimpedConfigEnsembleable,
-    Statable<CrimpedConfigEnsembleable>,
+    CrimpedConfigEnsembleable<CompletionStage<byte[]>>,
+    Statable<CrimpedConfigEnsembleable<CompletionStage<byte[]>>>,
     CrimpledEnsembleable<CompletionStage<byte[]>>
 {
 }

http://git-wip-us.apache.org/repos/asf/curator/blob/3aa51d50/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedEnsembleableImpl.java
----------------------------------------------------------------------
diff --git 
a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedEnsembleableImpl.java
 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedEnsembleableImpl.java
index d94d242..b5b7a43 100644
--- 
a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedEnsembleableImpl.java
+++ 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedEnsembleableImpl.java
@@ -62,7 +62,7 @@ class CrimpedEnsembleableImpl implements CrimpedEnsembleable
     }
 
     @Override
-    public CrimpedConfigEnsembleable storingStatIn(Stat stat)
+    public CrimpedConfigEnsembleable<CompletionStage<byte[]>> 
storingStatIn(Stat stat)
     {
         ensembleable = configureEnsembleable = 
configBuilder.storingStatIn(stat);
         return this;

http://git-wip-us.apache.org/repos/asf/curator/blob/3aa51d50/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedWatchedEnsembleable.java
----------------------------------------------------------------------
diff --git 
a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedWatchedEnsembleable.java
 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedWatchedEnsembleable.java
new file mode 100644
index 0000000..be0cc58
--- /dev/null
+++ 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedWatchedEnsembleable.java
@@ -0,0 +1,29 @@
+/**
+ * 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.crimps.async;
+
+import org.apache.curator.framework.api.Statable;
+import java.util.concurrent.CompletionStage;
+
+public interface CrimpedWatchedEnsembleable extends
+    CrimpedConfigEnsembleable<Crimped<byte[]>>,
+    Statable<CrimpedConfigEnsembleable<Crimped<byte[]>>>,
+    CrimpledEnsembleable<Crimped<byte[]>>
+{
+}

http://git-wip-us.apache.org/repos/asf/curator/blob/3aa51d50/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedWatchedEnsembleableImpl.java
----------------------------------------------------------------------
diff --git 
a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedWatchedEnsembleableImpl.java
 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedWatchedEnsembleableImpl.java
new file mode 100644
index 0000000..78c7304
--- /dev/null
+++ 
b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/CrimpedWatchedEnsembleableImpl.java
@@ -0,0 +1,83 @@
+/**
+ * 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.crimps.async;
+
+import org.apache.curator.framework.api.ConfigureEnsembleable;
+import org.apache.curator.framework.api.Ensembleable;
+import org.apache.curator.framework.api.Statable;
+import org.apache.zookeeper.data.Stat;
+
+class CrimpedWatchedEnsembleableImpl implements CrimpedWatchedEnsembleable
+{
+    private final CrimpedBackgroundCallback<byte[]> callback;
+    private final Statable<ConfigureEnsembleable> configBuilder;
+    private Ensembleable<byte[]> ensembleable;
+    private ConfigureEnsembleable configureEnsembleable;
+
+    CrimpedWatchedEnsembleableImpl(Statable<ConfigureEnsembleable> 
configBuilder, CrimpedBackgroundCallback<byte[]> callback)
+    {
+        this.configBuilder = configBuilder;
+        this.callback = callback;
+        configureEnsembleable = configBuilder.storingStatIn(new Stat());
+        ensembleable = configureEnsembleable;
+    }
+
+    CrimpedWatchedEnsembleableImpl(Ensembleable<byte[]> ensembleable, 
CrimpedBackgroundCallback<byte[]> callback)
+    {
+        this.ensembleable = ensembleable;
+        this.configBuilder = null;
+        this.callback = callback;
+        configureEnsembleable = null;
+    }
+
+    @Override
+    public Crimped<byte[]> forEnsemble()
+    {
+        try
+        {
+            ensembleable.forEnsemble();
+        }
+        catch ( Exception e )
+        {
+            callback.completeExceptionally(e);
+        }
+        return callback;
+    }
+
+    @Override
+    public CrimpedConfigEnsembleable<Crimped<byte[]>> storingStatIn(Stat stat)
+    {
+        ensembleable = configureEnsembleable = 
configBuilder.storingStatIn(stat);
+        return this;
+    }
+
+    @Override
+    public CrimpledEnsembleable<Crimped<byte[]>> fromConfig(long config)
+    {
+        try
+        {
+            ensembleable = configureEnsembleable.fromConfig(config);
+        }
+        catch ( Exception e )
+        {
+            callback.completeExceptionally(e);
+        }
+        return this;
+    }
+}

http://git-wip-us.apache.org/repos/asf/curator/blob/3aa51d50/curator-x-crimps/src/test/java/org/apache/curator/x/crimps/async/TestCrimps.java
----------------------------------------------------------------------
diff --git 
a/curator-x-crimps/src/test/java/org/apache/curator/x/crimps/async/TestCrimps.java
 
b/curator-x-crimps/src/test/java/org/apache/curator/x/crimps/async/TestCrimps.java
index d5b7a6d..62065c3 100644
--- 
a/curator-x-crimps/src/test/java/org/apache/curator/x/crimps/async/TestCrimps.java
+++ 
b/curator-x-crimps/src/test/java/org/apache/curator/x/crimps/async/TestCrimps.java
@@ -108,20 +108,13 @@ public class TestCrimps extends BaseClassForTests
         {
             client.start();
 
-            CompletionStage<Stat> f = 
async.safeStat(client.checkExists()).forPath("/test");
+            CompletionStage<Stat> f = 
async.stat(client.checkExists()).forPath("/test");
             complete(f.handle((stat, e) -> {
                 Assert.assertNull(e);
                 Assert.assertNull(stat);
                 return null;
             }));
 
-            f = async.stat(client.checkExists()).forPath("/test");
-            complete(f.handle((stat, e) -> {
-                Assert.assertNotNull(e);
-                Assert.assertNull(stat);
-                return null;
-            }));
-
             
async.path(client.create()).forPath("/test").toCompletableFuture().get();
             f = async.stat(client.checkExists()).forPath("/test");
             complete(f.handle((stat, e) -> {

Reply via email to