GH-1246. Asynchronously initialize cache before reading

curator-x-async/pom.xml

@@ -64,6 +64,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-params</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-log4j12</artifactId>

curator-x-async/src/main/java/org/apache/curator/x/async/modeled/ModeledFramework.java

@@ -93,6 +93,13 @@ static <T> ModeledFrameworkBuilder<T> builder() {
      */
     CachedModeledFramework<T> cached(ExecutorService executor);
 
+    /**
+     * Return a cached framework which waits for cache to be initialized before accessing it.
+     *
+     * @return wrapped cached framework that waits for initialization.
+     */
+    CachedModeledFramework<T> initialized();
+
     /**
      * Return mutator APIs that work with {@link org.apache.curator.x.async.modeled.versioned.Versioned} containers
      *

curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/CachedModeledFrameworkImpl.java

@@ -184,6 +184,11 @@ public AsyncStage<List<T>> list() {
         return ModelStage.completed(children);
     }
 
+    @Override
+    public CachedModeledFramework<T> initialized() {
+        return new InitializedCachedModeledFramework<>(this);
+    }
+
     @Override
     public AsyncStage<Stat> update(T model) {
         return client.update(model);

curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/InitializedCachedModeledFramework.java

@@ -0,0 +1,287 @@
+/*
+ * 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;
+
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
+import org.apache.curator.framework.api.transaction.CuratorOp;
+import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
+import org.apache.curator.framework.listen.Listenable;
+import org.apache.curator.x.async.AsyncCuratorFramework;
+import org.apache.curator.x.async.AsyncStage;
+import org.apache.curator.x.async.modeled.ModelSpec;
+import org.apache.curator.x.async.modeled.ModeledFramework;
+import org.apache.curator.x.async.modeled.ZNode;
+import org.apache.curator.x.async.modeled.ZPath;
+import org.apache.curator.x.async.modeled.cached.CachedModeledFramework;
+import org.apache.curator.x.async.modeled.cached.ModeledCache;
+import org.apache.curator.x.async.modeled.cached.ModeledCacheListener;
+import org.apache.curator.x.async.modeled.versioned.VersionedModeledFramework;
+import org.apache.zookeeper.data.Stat;
+
+public class InitializedCachedModeledFramework<T> implements CachedModeledFramework<T> {
+
+    private final CachedModeledFramework<T> client;
+    private final ModelStage<Void> init;
+
+    private InitializedCachedModeledFramework(CachedModeledFramework<T> client, ModelStage<Void> init) {
+        this.client = client;
+        this.init = init;
+    }
+
+    InitializedCachedModeledFramework(CachedModeledFramework<T> client) {
+        this.client = client;
+        init = ModelStage.make();
+        listenable().addListener(new ModeledCacheListener<T>() {
+
+            @Override
+            public void accept(Type type, ZPath path, Stat stat, Object model) {
+                // NOP
+            }
+
+            @Override
+            public void initialized() {
+                init.complete(null);
+                ModeledCacheListener.super.initialized();
+            }
+
+            @Override
+            public void handleException(Exception e) {
+                init.completeExceptionally(e);
+                ModeledCacheListener.super.handleException(e);
+            }
+        });
+    }
+
+    @Override
+    public ModeledCache<T> cache() {
+        return client.cache();
+    }
+
+    @Override
+    public void start() {
+        client.start();
+    }
+
+    @Override
+    public void close() {
+        client.close();
+    }
+
+    @Override
+    public Listenable<ModeledCacheListener<T>> listenable() {
+        return client.listenable();
+    }
+
+    @Override
+    public AsyncStage<List<ZNode<T>>> childrenAsZNodes() {
+        return internalRead(client::childrenAsZNodes);
+    }
+
+    @Override
+    public CuratorOp createOp(T model) {
+        return client.createOp(model);
+    }
+
+    @Override
+    public CuratorOp updateOp(T model) {
+        return client.updateOp(model);
+    }
+
+    @Override
+    public CuratorOp updateOp(T model, int version) {
+        return client.updateOp(model, version);
+    }
+
+    @Override
+    public CuratorOp deleteOp() {
+        return client.deleteOp();
+    }
+
+    @Override
+    public CuratorOp deleteOp(int version) {
+        return client.deleteOp(version);
+    }
+
+    @Override
+    public CuratorOp checkExistsOp() {
+        return client.checkExistsOp();
+    }
+
+    @Override
+    public CuratorOp checkExistsOp(int version) {
+        return client.checkExistsOp(version);
+    }
+
+    @Override
+    public AsyncStage<List<CuratorTransactionResult>> inTransaction(List<CuratorOp> operations) {
+        return client.inTransaction(operations);
+    }
+
+    @Override
+    public CachedModeledFramework<T> cached() {
+        throw new UnsupportedOperationException("Already a cached instance");
+    }
+
+    @Override
+    public CachedModeledFramework<T> cached(ExecutorService executor) {
+        throw new UnsupportedOperationException("Already a cached instance");
+    }
+
+    @Override
+    public VersionedModeledFramework<T> versioned() {
+        return new VersionedModeledFrameworkImpl<>(this);
+    }
+
+    @Override
+    public AsyncCuratorFramework unwrap() {
+        return client.unwrap();
+    }
+
+    @Override
+    public ModelSpec<T> modelSpec() {
+        return client.modelSpec();
+    }
+
+    @Override
+    public CachedModeledFramework<T> child(Object child) {
+        return new InitializedCachedModeledFramework<>(client.child(child), init);
+    }
+
+    @Override
+    public ModeledFramework<T> parent() {
+        throw new UnsupportedOperationException(
+                "Not supported for CachedModeledFramework. Instead, call parent() on the ModeledFramework before calling cached()");
+    }
+
+    @Override
+    public CachedModeledFramework<T> withPath(ZPath path) {
+        return new InitializedCachedModeledFramework<>(client.withPath(path), init);
+    }
+
+    @Override
+    public AsyncStage<String> set(T model) {
+        return client.set(model);
+    }
+
+    @Override
+    public AsyncStage<String> set(T model, int version) {
+        return client.set(model, version);
+    }
+
+    @Override
+    public AsyncStage<String> set(T model, Stat storingStatIn) {
+        return client.set(model, storingStatIn);
+    }
+
+    @Override
+    public AsyncStage<String> set(T model, Stat storingStatIn, int version) {
+        return client.set(model, storingStatIn, version);
+    }
+
+    @Override
+    public AsyncStage<T> read() {
+        return internalRead(client::read);
+    }
+
+    @Override
+    public AsyncStage<T> read(Stat storingStatIn) {
+        return internalRead(() -> client.read(storingStatIn));
+    }
+
+    @Override
+    public AsyncStage<ZNode<T>> readAsZNode() {
+        return internalRead(client::readAsZNode);
+    }
+
+    @Override
+    public AsyncStage<Stat> update(T model) {
+        return client.update(model);
+    }
+
+    @Override
+    public AsyncStage<Stat> update(T model, int version) {
+        return client.update(model, version);
+    }
+
+    @Override
+    public AsyncStage<Void> delete() {
+        return client.delete();
+    }
+
+    @Override
+    public AsyncStage<Void> delete(int version) {
+        return client.delete(version);
+    }
+
+    @Override
+    public AsyncStage<Stat> checkExists() {
+        return client.checkExists();
+    }
+
+    @Override
+    public AsyncStage<List<ZPath>> children() {
+        return internalRead(client::children);
+    }
+
+    @Override
+    public AsyncStage<T> readThrough() {
+        return internalRead(client::readThrough);
+    }
+
+    @Override
+    public AsyncStage<T> readThrough(Stat storingStatIn) {
+        return internalRead(() -> client.readThrough(storingStatIn));
+    }
+
+    @Override
+    public AsyncStage<ZNode<T>> readThroughAsZNode() {
+        return internalRead(client::readThroughAsZNode);
+    }
+
+    @Override
+    public AsyncStage<List<T>> list() {
+        return internalRead(client::list);
+    }
+
+    @Override
+    public CachedModeledFramework<T> initialized() {
+        throw new UnsupportedOperationException("Already an initialized instance");
+    }
+
+    private <U> AsyncStage<U> internalRead(Supplier<AsyncStage<U>> innerSupplier) {
+        ModelStage<U> stage = ModelStage.make();
+        init.whenComplete((data, throwable) -> {
+            if (throwable == null) {
+                innerSupplier.get().whenComplete((data1, throwable1) -> {
+                    if (throwable1 == null) {
+                        stage.complete(data1);
+                    } else {
+                        stage.completeExceptionally(throwable1);
+                    }
+                });
+            } else {
+                stage.completeExceptionally(throwable);
+            }
+        });
+        return stage;
+    }
+}

curator-x-async/src/main/java/org/apache/curator/x/async/modeled/details/ModeledFrameworkImpl.java

@@ -131,6 +131,11 @@ public CachedModeledFramework<T> cached() {
         return cached(ThreadUtils.newSingleThreadExecutor("CachedModeledFramework"));
     }
 
+    @Override
+    public CachedModeledFramework<T> initialized() {
+        return new InitializedCachedModeledFramework<>(cached());
+    }
+
     @Override
     public CachedModeledFramework<T> cached(ExecutorService executor) {
         Preconditions.checkState(