Async version of Query and Traversal. Filter, Map, FlatMap, Fork, Of, SKip and TakeWhile.

Author: fmcarvalho

src/main/java/org/jayield/AsyncQuery.java

@@ -0,0 +1,125 @@
+/*
+ * Copyright (c) 2020, Fernando Miguel Carvalho, mcarvalho@cc.isel.ipl.pt
+ *
+ * Licensed 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.jayield;
+
+import org.jayield.async.AsyncQueryFilter;
+import org.jayield.async.AsyncQueryFlatMapConcat;
+import org.jayield.async.AsyncQueryFlatMapMerge;
+import org.jayield.async.AsyncQueryFork;
+import org.jayield.async.AsyncQueryMap;
+import org.jayield.async.AsyncQueryOf;
+import org.jayield.async.AsyncQueryOfIterator;
+import org.jayield.async.AsyncQueryOnNext;
+import org.jayield.async.AsyncQuerySkip;
+import org.jayield.async.AsyncQueryTakeWhile;
+
+import java.util.Iterator;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+import java.util.function.Predicate;
+
+/**
+ * An asynchronous sequence of elements supporting sequential operations.
+ * Query operations are composed into a pipeline to perform computation.
+ *
+ * @author Miguel Gamboa
+ *         created on 07-07-2020
+ */
+public abstract class AsyncQuery<T> implements AsyncTraverser<T>{
+
+    /**
+     * Returns an asynchronous sequential ordered query whose elements
+     * are the specified values in data parameter.
+     */
+    public static <U> AsyncQuery<U> of(U...data) {
+        return new AsyncQueryOf<>(data);
+    }
+    /**
+     * Returns an asynchronous sequential ordered query whose elements
+     * are the specified values in the Iterator parameter.
+     */
+    public static <U> AsyncQuery<U> of(Iterator<U> iter) {
+        return new AsyncQueryOfIterator<>(iter);
+    }
+    /**
+     * Returns an asynchronous sequential ordered query whose elements
+     * are the specified values in data parameter running on thread pool.
+     */
+    public static <U> AsyncQuery<U> fork(U...data) {
+        return new AsyncQueryFork<>(data);
+    }
+
+    /**
+     * Returns a new asynchronous query emitting the same items of this query,
+     * additionally performing the provided action on each element as elements are consumed
+     * from the resulting query.
+     */
+    public final AsyncQuery<T> onNext(BiConsumer<? super T, ? super Throwable> action) {
+        return new AsyncQueryOnNext<>(this, action);
+    }
+
+    /**
+     * Returns a new asynchronous query consisting of the remaining elements of
+     * this query after discarding the first {@code n} elements of the query.
+     */
+    public final AsyncQuery<T> skip(int n) {
+        return new AsyncQuerySkip<>(this, n);
+    }
+
+    /**
+     * Returns an asynchronous query consisting of the elements of this query that match
+     * the given predicate.
+     */
+    public final AsyncQuery<T> filter(Predicate<? super T> p) {
+        return new AsyncQueryFilter<>(this, p);
+    }
+
+    /**
+     * Returns an asynchronous query consisting of the results of applying the given
+     * function to the elements of this query.
+     */
+    public final <R> AsyncQuery<R> map(Function<? super T,? extends R> mapper) {
+        return new AsyncQueryMap<>(this, mapper);
+    }
+
+    /**
+     * Returns a query consisting of the longest prefix of elements taken from
+     * this query that match the given predicate.
+     */
+    public final AsyncQuery<T> takeWhile(Predicate<? super T> predicate){
+        return new AsyncQueryTakeWhile<>(this, predicate);
+    }
+    /**
+     * Returns an asynchronous query consisting of the results of replacing each element of
+     * this query with the contents of a mapped query produced by applying
+     * the provided mapping function to each element.
+     * It waits for the inner flow to complete before starting to collect the next one.
+     */
+    public final <R> AsyncQuery<R> flatMapConcat(Function<? super T,? extends AsyncQuery<? extends R>> mapper) {
+        return new AsyncQueryFlatMapConcat<>(this, mapper);
+    }
+
+    public final <R> AsyncQuery<R> flatMapMerge(Function<? super T,? extends AsyncQuery<? extends R>> mapper) {
+        return new AsyncQueryFlatMapMerge<>(this, mapper);
+    }
+
+    public final void blockingSubscribe() {
+        this
+            .subscribe((item, err) -> { })
+            .join(); // In both previous cases cf will raise an exception.
+    }
+}

src/main/java/org/jayield/AsyncTraverser.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) 2018, Fernando Miguel Carvalho, mcarvalho@cc.isel.ipl.pt
+ *
+ * Licensed 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.jayield;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
+
+import static java.util.concurrent.CompletableFuture.supplyAsync;
+
+/**
+ * Asynchronous traversal.
+ * Jayield uses traverse method as its first choice to
+ * implement AsyncQuery operations.
+ * This is a special kind of traversal that disallows individually access.
+ */
+public interface AsyncTraverser<T> {
+    /**
+     * Yields elements sequentially until all elements have been
+     * processed or an exception is thrown.
+     * The given consumer is invoked with the result (or null if none)
+     * and the exception (or null if none).
+     *
+     * @return A CompletableFuture to signal finish to enable cancellation
+     * through its cancel() method.
+     */
+    CompletableFuture<Void> subscribe(BiConsumer<? super T,? super Throwable> cons);
+}

src/main/java/org/jayield/async/AsyncQueryFilter.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) 2020, Fernando Miguel Carvalho, mcarvalho@cc.isel.ipl.pt
+ *
+ * Licensed 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.jayield.async;
+
+import org.jayield.AsyncQuery;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
+import java.util.function.Predicate;
+
+public class AsyncQueryFilter<T> extends AsyncQuery<T> {
+    private final AsyncQuery<T> upstream;
+    private final Predicate<? super T> p;
+
+    public AsyncQueryFilter(AsyncQuery<T> upstream, Predicate<? super T> p) {
+        this.upstream = upstream;
+        this.p = p;
+    }
+
+    @Override
+    public CompletableFuture<Void> subscribe(BiConsumer<? super T, ? super Throwable> cons) {
+        return upstream.subscribe((item, err) -> {
+            if(err != null) {
+                cons.accept(null, err);
+                return;
+            }
+            if(p.test(item)) cons.accept(item, null);
+        });
+    }
+}

src/main/java/org/jayield/async/AsyncQueryFlatMapConcat.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2020, Fernando Miguel Carvalho, mcarvalho@cc.isel.ipl.pt
+ *
+ * Licensed 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.jayield.async;
+
+import org.jayield.AsyncQuery;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+public class AsyncQueryFlatMapConcat<T, R> extends AsyncQuery<R> {
+    private final AsyncQuery<T> upstream;
+    private final Function<? super T, ? extends AsyncQuery<? extends R>> mapper;
+
+
+    public AsyncQueryFlatMapConcat(AsyncQuery<T> upstream, Function<? super T, ? extends AsyncQuery<? extends R>> mapper) {
+        this.upstream = upstream;
+        this.mapper = mapper;
+    }
+
+    @Override
+    public CompletableFuture<Void> subscribe(BiConsumer<? super R, ? super Throwable> cons) {
+        return upstream.subscribe((item, err) -> {
+            if(err != null) {
+                cons.accept(null, err);
+                return;
+            }
+            mapper
+                .apply(item)
+                .subscribe(cons::accept)
+                .join(); // !!!! Replace this by Continuation !!!
+        });
+    }
+}

src/main/java/org/jayield/async/AsyncQueryFlatMapMerge.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright (c) 2020, Fernando Miguel Carvalho, mcarvalho@cc.isel.ipl.pt
+ *
+ * Licensed 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.jayield.async;
+
+import org.jayield.AsyncQuery;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+public class AsyncQueryFlatMapMerge<T, R> extends AsyncQuery<R> {
+    private final AsyncQuery<T> upstream;
+    private final Function<? super T, ? extends AsyncQuery<? extends R>> mapper;
+
+
+    public AsyncQueryFlatMapMerge(AsyncQuery<T> upstream, Function<? super T, ? extends AsyncQuery<? extends R>> mapper) {
+        this.upstream = upstream;
+        this.mapper = mapper;
+    }
+
+    @Override
+    public CompletableFuture<Void> subscribe(BiConsumer<? super R, ? super Throwable> cons) {
+        List<CompletableFuture<Void>> cfs = new ArrayList<>();
+        return upstream
+            .subscribe((item, err) -> {
+                if (err != null) {
+                    cons.accept(null, err);
+                    return;
+                }
+                cfs.add(mapper
+                    .apply(item)
+                    .subscribe(cons::accept));
+            })
+            .thenCompose(ignore -> CompletableFuture.allOf(cfs.toArray(new CompletableFuture[cfs.size()])));
+    }
+}

src/main/java/org/jayield/async/AsyncQueryFork.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2020, Fernando Miguel Carvalho, mcarvalho@cc.isel.ipl.pt
+ *
+ * Licensed 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.jayield.async;
+
+import org.jayield.AsyncQuery;
+import org.jayield.Query;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
+
+import static java.util.concurrent.CompletableFuture.runAsync;
+
+public class AsyncQueryFork<U> extends AsyncQuery<U> {
+    private final U[] data;
+
+    public AsyncQueryFork(U[] data) {
+        this.data = data;
+    }
+
+    @Override
+    public CompletableFuture<Void> subscribe(BiConsumer<? super U, ? super Throwable> cons) {
+        return runAsync(() -> Query
+            .of(data)
+            .traverse(item -> cons.accept(item, null)));
+    }
+}