http://git-wip-us.apache.org/repos/asf/curator/blob/9b84ba39/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/details/CrimpledPathAndBytesableImpl.java ---------------------------------------------------------------------- diff --git a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/details/CrimpledPathAndBytesableImpl.java b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/details/CrimpledPathAndBytesableImpl.java new file mode 100644 index 0000000..22408ea --- /dev/null +++ b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/details/CrimpledPathAndBytesableImpl.java @@ -0,0 +1,92 @@ +/** + * 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.details; + +import org.apache.curator.framework.api.PathAndBytesable; +import org.apache.curator.framework.api.Pathable; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; + +class CrimpledPathAndBytesableImpl<MAIN, RESULT extends CompletionStage<MAIN>> implements CrimpledPathAndBytesable<RESULT> +{ + private final Pathable<MAIN> pathable; + private final PathAndBytesable<MAIN> pathAndBytesable; + private final RESULT result; + private final CompletableFuture second; + + CrimpledPathAndBytesableImpl(Pathable<MAIN> pathable, RESULT result, CompletableFuture second) + { + this.result = result; + this.second = second; + this.pathAndBytesable = null; + this.pathable = pathable; + } + + CrimpledPathAndBytesableImpl(PathAndBytesable<MAIN> pathAndBytesable, RESULT result, CompletableFuture second) + { + this.pathAndBytesable = pathAndBytesable; + this.result = result; + this.second = second; + this.pathable = null; + } + + @Override + public RESULT forPath(String path) + { + try + { + if ( pathable != null ) + { + pathable.forPath(path); + } + else + { + pathAndBytesable.forPath(path); + } + } + catch ( Exception e ) + { + setException(e); + } + return result; + } + + @Override + public RESULT forPath(String path, byte[] data) + { + try + { + pathAndBytesable.forPath(path, data); + } + catch ( Exception e ) + { + setException(e); + } + return result; + } + + private void setException(Exception e) + { + result.toCompletableFuture().completeExceptionally(e); + if ( second != null ) + { + second.completeExceptionally(e); + } + } +}
http://git-wip-us.apache.org/repos/asf/curator/blob/9b84ba39/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncCuratorFrameworkImpl.java ---------------------------------------------------------------------- diff --git a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncCuratorFrameworkImpl.java b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncCuratorFrameworkImpl.java new file mode 100644 index 0000000..74adaf1 --- /dev/null +++ b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncCuratorFrameworkImpl.java @@ -0,0 +1,144 @@ +package org.apache.curator.x.crimps.async.imps; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.api.*; +import org.apache.curator.framework.api.transaction.TransactionOp; +import org.apache.curator.x.crimps.async.AsyncCuratorFramework; +import org.apache.curator.x.crimps.async.AsyncCuratorFrameworkDsl; +import org.apache.curator.x.crimps.async.AsyncDeleteBuilder; +import org.apache.curator.x.crimps.async.AsyncGetChildrenBuilder; +import org.apache.curator.x.crimps.async.AsyncMultiTransaction; +import org.apache.curator.x.crimps.async.AsyncSetDataBuilder; +import org.apache.curator.x.crimps.async.WatchedAsyncCuratorFramework; +import org.apache.curator.x.crimps.async.details.AsyncCrimps; +import java.util.List; +import java.util.concurrent.CompletionStage; + +public class AsyncCuratorFrameworkImpl implements AsyncCuratorFramework +{ + private final CuratorFramework client; + private final AsyncCrimps async; + private final UnhandledErrorListener unhandledErrorListener; + + public AsyncCuratorFrameworkImpl(CuratorFramework client, AsyncCrimps async, UnhandledErrorListener unhandledErrorListener) + { + this.client = client; + this.async = async; + this.unhandledErrorListener = unhandledErrorListener; + } + + @Override + public AsyncCuratorFrameworkDsl withUnhandledErrorListener(UnhandledErrorListener listener) + { + return new AsyncCuratorFrameworkImpl(client, async, listener); + } + + @Override + public CuratorFramework getCuratorFramework() + { + return client; + } + + @Override + public WatchedAsyncCuratorFramework watched() + { + return new WatchedAsyncCuratorFrameworkImpl(client, async, unhandledErrorListener); + } + + @Override + public CreateBuilder create() + { + return null; + } + + @Override + public AsyncDeleteBuilder delete() + { + return new AsyncDeleteBuilderImpl(client, async, unhandledErrorListener); + } + + @Override + public ExistsBuilder checkExists() + { + return null; + } + + @Override + public GetDataBuilder getData() + { + return null; + } + + @Override + public AsyncSetDataBuilder setData() + { + return new AsyncSetDataBuilderImpl(client, async, unhandledErrorListener); + } + + @Override + public AsyncGetChildrenBuilder<CompletionStage<List<String>>> getChildren() + { + return new AsyncGetChildrenBuilderImpl<CompletionStage<List<String>>>(client, unhandledErrorListener) + { + @Override + protected CompletionStage<List<String>> finish(BackgroundPathable<List<String>> builder, String path) + { + return async.children(builder).forPath(path); + } + }; + } + + @Override + public GetACLBuilder getACL() + { + return null; + } + + @Override + public SetACLBuilder setACL() + { + return null; + } + + @Override + public ReconfigBuilder reconfig() + { + return null; + } + + @Override + public GetConfigBuilder getConfig() + { + return null; + } + + @Override + public AsyncMultiTransaction transaction() + { + return operations -> async.opResults(client.transaction()).forOperations(operations); + } + + @Override + public TransactionOp transactionOp() + { + return null; + } + + @Override + public SyncBuilder sync() + { + return null; + } + + @Override + public RemoveWatchesBuilder watches() + { + return null; + } + + public UnhandledErrorListener getUnhandledErrorListener() + { + return unhandledErrorListener; + } + +} http://git-wip-us.apache.org/repos/asf/curator/blob/9b84ba39/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncDeleteBuilderImpl.java ---------------------------------------------------------------------- diff --git a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncDeleteBuilderImpl.java b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncDeleteBuilderImpl.java new file mode 100644 index 0000000..783c08f --- /dev/null +++ b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncDeleteBuilderImpl.java @@ -0,0 +1,66 @@ +package org.apache.curator.x.crimps.async.imps; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.api.BackgroundPathable; +import org.apache.curator.framework.api.BackgroundVersionable; +import org.apache.curator.framework.api.ChildrenDeletable; +import org.apache.curator.framework.api.DeleteBuilder; +import org.apache.curator.framework.api.DeleteBuilderMain; +import org.apache.curator.framework.api.UnhandledErrorListener; +import org.apache.curator.x.crimps.async.AsyncDeleteBuilder; +import org.apache.curator.x.crimps.async.AsyncPathable; +import org.apache.curator.x.crimps.async.DeleteOption; +import org.apache.curator.x.crimps.async.details.AsyncCrimps; +import java.util.Collections; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.CompletionStage; + +class AsyncDeleteBuilderImpl implements AsyncDeleteBuilder +{ + private final CuratorFramework client; + private final AsyncCrimps async; + private final UnhandledErrorListener unhandledErrorListener; + private Set<DeleteOption> options = Collections.emptySet(); + private int version = -1; + + public AsyncDeleteBuilderImpl(CuratorFramework client, AsyncCrimps async, UnhandledErrorListener unhandledErrorListener) + { + this.client = client; + this.async = async; + this.unhandledErrorListener = unhandledErrorListener; + } + + @Override + public AsyncPathable<CompletionStage<Void>> withOptions(Set<DeleteOption> options) + { + this.options = Objects.requireNonNull(options, "options cannot be null"); + return this; + } + + @Override + public AsyncPathable<CompletionStage<Void>> withOptionsAndVersion(Set<DeleteOption> options, int version) + { + this.options = Objects.requireNonNull(options, "options cannot be null"); + this.version = version; + return this; + } + + @Override + public AsyncPathable<CompletionStage<Void>> withVersion(int version) + { + this.version = version; + return this; + } + + @Override + public CompletionStage<Void> forPath(String path) + { + DeleteBuilder builder1 = client.delete(); + DeleteBuilderMain builder2 = options.contains(DeleteOption.quietly) ? builder1.quietly() : builder1; + ChildrenDeletable builder3 = options.contains(DeleteOption.guaranteed) ? builder2.guaranteed() : builder2; + BackgroundVersionable builder4 = options.contains(DeleteOption.deletingChildrenIfNeeded) ? builder3.deletingChildrenIfNeeded() : builder3; + BackgroundPathable<Void> builder5 = (version >= 0) ? builder4.withVersion(version) : builder4; + return async.ignored(builder5).forPath(path); + } +} http://git-wip-us.apache.org/repos/asf/curator/blob/9b84ba39/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncGetChildrenBuilderImpl.java ---------------------------------------------------------------------- diff --git a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncGetChildrenBuilderImpl.java b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncGetChildrenBuilderImpl.java new file mode 100644 index 0000000..67f9e23 --- /dev/null +++ b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncGetChildrenBuilderImpl.java @@ -0,0 +1,42 @@ +package org.apache.curator.x.crimps.async.imps; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.api.BackgroundPathable; +import org.apache.curator.framework.api.GetChildrenBuilder; +import org.apache.curator.framework.api.Pathable; +import org.apache.curator.framework.api.UnhandledErrorListener; +import org.apache.curator.framework.api.Watchable; +import org.apache.curator.x.crimps.async.AsyncGetChildrenBuilder; +import org.apache.curator.x.crimps.async.AsyncPathable; +import org.apache.zookeeper.data.Stat; +import java.util.List; + +abstract class AsyncGetChildrenBuilderImpl<T> implements AsyncGetChildrenBuilder<T> +{ + private final CuratorFramework client; + private final UnhandledErrorListener unhandledErrorListener; + private Stat stat; + + public AsyncGetChildrenBuilderImpl(CuratorFramework client, UnhandledErrorListener unhandledErrorListener) + { + this.client = client; + this.unhandledErrorListener = unhandledErrorListener; + } + + @Override + public AsyncPathable<T> storingStatIn(Stat stat) + { + this.stat = stat; + return this; + } + + @Override + public T forPath(String path) + { + GetChildrenBuilder builder1 = client.getChildren(); + Watchable<? extends Pathable<List<String>>> builder2 = (stat != null) ? builder1.storingStatIn(stat) : builder1; // TODO + return finish(builder1, path); + } + + protected abstract T finish(BackgroundPathable<List<String>> builder, String path); +} http://git-wip-us.apache.org/repos/asf/curator/blob/9b84ba39/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncSetDataBuilderImpl.java ---------------------------------------------------------------------- diff --git a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncSetDataBuilderImpl.java b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncSetDataBuilderImpl.java new file mode 100644 index 0000000..906c955 --- /dev/null +++ b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/AsyncSetDataBuilderImpl.java @@ -0,0 +1,68 @@ +package org.apache.curator.x.crimps.async.imps; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.api.BackgroundPathAndBytesable; +import org.apache.curator.framework.api.SetDataBuilder; +import org.apache.curator.framework.api.UnhandledErrorListener; +import org.apache.curator.x.crimps.async.AsyncPathAndBytesable; +import org.apache.curator.x.crimps.async.AsyncSetDataBuilder; +import org.apache.curator.x.crimps.async.details.AsyncCrimps; +import org.apache.zookeeper.data.Stat; +import java.util.concurrent.CompletionStage; + +class AsyncSetDataBuilderImpl implements AsyncSetDataBuilder +{ + private final CuratorFramework client; + private final AsyncCrimps async; + private final UnhandledErrorListener unhandledErrorListener; + private boolean compressed = false; + private int version = -1; + + AsyncSetDataBuilderImpl(CuratorFramework client, AsyncCrimps async, UnhandledErrorListener unhandledErrorListener) + { + this.client = client; + this.async = async; + this.unhandledErrorListener = unhandledErrorListener; + } + + @Override + public CompletionStage<Stat> forPath(String path) + { + return forPath(path, null); + } + + @Override + public CompletionStage<Stat> forPath(String path, byte[] data) + { + SetDataBuilder builder1 = client.setData(); + BackgroundPathAndBytesable<Stat> builder2 = (version >= 0) ? builder1.withVersion(version) : builder1; + // compressed ? builder2.compressed() : builder2; TODO compressed with version + if ( data != null ) + { + return async.stat(builder2).forPath(path, data); + } + return async.stat(builder2).forPath(path); + } + + @Override + public AsyncPathAndBytesable<CompletionStage<Stat>> compressed() + { + compressed = true; + return this; + } + + @Override + public AsyncPathAndBytesable<CompletionStage<Stat>> compressedWithVersion(int version) + { + compressed = true; + this.version = version; + return this; + } + + @Override + public AsyncPathAndBytesable<CompletionStage<Stat>> withVersion(int version) + { + this.version = version; + return this; + } +} http://git-wip-us.apache.org/repos/asf/curator/blob/9b84ba39/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/WatchedAsyncCuratorFrameworkImpl.java ---------------------------------------------------------------------- diff --git a/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/WatchedAsyncCuratorFrameworkImpl.java b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/WatchedAsyncCuratorFrameworkImpl.java new file mode 100644 index 0000000..9fc5e3c --- /dev/null +++ b/curator-x-crimps/src/main/java/org/apache/curator/x/crimps/async/imps/WatchedAsyncCuratorFrameworkImpl.java @@ -0,0 +1,58 @@ +package org.apache.curator.x.crimps.async.imps; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.api.BackgroundPathable; +import org.apache.curator.framework.api.ExistsBuilder; +import org.apache.curator.framework.api.GetConfigBuilder; +import org.apache.curator.framework.api.GetDataBuilder; +import org.apache.curator.framework.api.UnhandledErrorListener; +import org.apache.curator.x.crimps.async.AsyncGetChildrenBuilder; +import org.apache.curator.x.crimps.async.WatchedAsyncCuratorFramework; +import org.apache.curator.x.crimps.async.details.AsyncCrimps; +import org.apache.curator.x.crimps.async.details.Crimped; +import java.util.List; + +public class WatchedAsyncCuratorFrameworkImpl implements WatchedAsyncCuratorFramework +{ + private final CuratorFramework client; + private final AsyncCrimps async; + private final UnhandledErrorListener unhandledErrorListener; + + public WatchedAsyncCuratorFrameworkImpl(CuratorFramework client, AsyncCrimps async, UnhandledErrorListener unhandledErrorListener) + { + this.client = client; + this.async = async; + this.unhandledErrorListener = unhandledErrorListener; + } + + @Override + public ExistsBuilder checkExists() + { + return null; + } + + @Override + public GetDataBuilder getData() + { + return null; + } + + @Override + public AsyncGetChildrenBuilder<Crimped<List<String>>> getChildren() + { + return new AsyncGetChildrenBuilderImpl<Crimped<List<String>>>(client, unhandledErrorListener) + { + @Override + protected Crimped<List<String>> finish(BackgroundPathable<List<String>> builder, String path) + { + return null; // TODO + } + }; + } + + @Override + public GetConfigBuilder getConfig() + { + return null; + } +} http://git-wip-us.apache.org/repos/asf/curator/blob/9b84ba39/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 62065c3..f7fc930 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 @@ -20,6 +20,13 @@ package org.apache.curator.x.crimps.async; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.api.BackgroundPathAndBytesable; +import org.apache.curator.framework.api.BackgroundPathable; +import org.apache.curator.framework.api.BackgroundVersionable; +import org.apache.curator.framework.api.ChildrenDeletable; +import org.apache.curator.framework.api.DeleteBuilderMain; +import org.apache.curator.framework.api.SetDataBackgroundVersionable; +import org.apache.curator.framework.api.SetDataBuilder; import org.apache.curator.framework.api.transaction.CuratorOp; import org.apache.curator.framework.api.transaction.CuratorTransactionResult; import org.apache.curator.framework.api.transaction.OperationType; @@ -27,6 +34,8 @@ import org.apache.curator.retry.RetryOneTime; import org.apache.curator.test.BaseClassForTests; import org.apache.curator.test.Timing; import org.apache.curator.x.crimps.Crimps; +import org.apache.curator.x.crimps.async.details.AsyncCrimps; +import org.apache.curator.x.crimps.async.details.Crimped; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.data.Stat; @@ -48,6 +57,11 @@ public class TestCrimps extends BaseClassForTests try ( CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1)) ) { client.start(); + + SetDataBuilder setDataBuilder = client.setData(); + BackgroundPathAndBytesable<Stat> withVersion = client.setData().withVersion(0); + SetDataBackgroundVersionable compressed = client.setData().compressed(); + CompletionStage<String> f = async.name(client.create().creatingParentContainersIfNeeded().withMode(CreateMode.PERSISTENT_SEQUENTIAL)).forPath("/a/b/c"); complete(f.handle((path, e) -> { Assert.assertEquals(path, "/a/b/c0000000000"); http://git-wip-us.apache.org/repos/asf/curator/blob/9b84ba39/pom.xml ---------------------------------------------------------------------- diff --git a/pom.xml b/pom.xml index d56386f..f96c7c0 100644 --- a/pom.xml +++ b/pom.xml @@ -287,6 +287,7 @@ <module>curator-x-discovery-server</module> <module>curator-x-rpc</module> <module>curator-x-crimps</module> + <module>curator-x-async</module> </modules> <dependencyManagement>