More operators, prepare release 0.3.0

README.md

@@ -9,7 +9,7 @@ Prototype Java 9 library based on the asynchronous enumerable concept (where mov
 ### Gradle
 
 ```groovy
-compile "com.github.akarnokd:async-enumerable:0.2.0"
+compile "com.github.akarnokd:async-enumerable:0.3.0"
 ```
 
 ### Getting started

gradle.properties

@@ -1,3 +1,3 @@
-version=0.2.0
+version=0.3.0
 org.gradle.jvmargs=-XX:+IgnoreUnrecognizedVMOptions --permit-illegal-access --show-version
 

src/main/java/hu/akarnokd/asyncenum/AsyncDoOnCancel.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2017 David Karnok
+ *
+ * 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 hu.akarnokd.asyncenum;
+
+import java.util.concurrent.CompletionStage;
+
+final class AsyncDoOnCancel<T> implements AsyncEnumerable<T> {
+
+    final AsyncEnumerable<T> source;
+
+    final Runnable onCancel;
+
+    AsyncDoOnCancel(AsyncEnumerable<T> source, Runnable onCancel) {
+        this.source = source;
+        this.onCancel = onCancel;
+    }
+
+    @Override
+    public AsyncEnumerator<T> enumerator() {
+        return new DoOnCancelEnumerator<>(source.enumerator(), onCancel);
+    }
+
+    static final class DoOnCancelEnumerator<T> implements AsyncEnumerator<T> {
+
+        final AsyncEnumerator<T> source;
+
+        final Runnable onCancel;
+
+        DoOnCancelEnumerator(AsyncEnumerator<T> source, Runnable onCancel) {
+            this.source = source;
+            this.onCancel = onCancel;
+        }
+
+        @Override
+        public CompletionStage<Boolean> moveNext() {
+            return source.moveNext();
+        }
+
+        @Override
+        public T current() {
+            return source.current();
+        }
+
+        @Override
+        public void cancel() {
+            onCancel.run();
+            source.cancel();
+        }
+    }
+}

src/main/java/hu/akarnokd/asyncenum/AsyncEnumerable.java

@@ -120,6 +120,18 @@ static AsyncEnumerable<Long> interval(long initialDelay, long period, TimeUnit u
         return new AsyncInterval(initialDelay, period, unit, executor);
     }
 
+    static <T> AsyncEnumerable<T> fromCallable(Callable<? extends T> callable) {
+        return new AsyncFromCallable<>(callable);
+    }
+
+    static <T> AsyncEnumerable<T> repeatItem(T item) {
+        return new AsyncRepeatItem<>(item);
+    }
+
+    static <T> AsyncEnumerable<T> repeatCallable(Callable<? extends T> callable) {
+        return new AsyncRepeatCallable<>(callable);
+    }
+
     // -------------------------------------------------------------------------------------
     // Instance transformations
 
@@ -238,6 +250,54 @@ default AsyncEnumerable<T> doFinally(Runnable onFinally) {
         return new AsyncDoFinally<>(this, onFinally);
     }
 
+    default AsyncEnumerable<T> ignoreElements() {
+        return new AsyncIgnoreElements<>(this);
+    }
+
+    default AsyncEnumerable<T> doOnCancel(Runnable onCancel) {
+        return new AsyncDoOnCancel<>(this, onCancel);
+    }
+
+    default AsyncEnumerable<T> repeat(long times) {
+        return repeat(times, () -> false);
+    }
+
+    default AsyncEnumerable<T> repeat(BooleanSupplier stop) {
+        return repeat(Long.MAX_VALUE, stop);
+    }
+
+    default AsyncEnumerable<T> repeat(long times, BooleanSupplier stop) {
+        return new AsyncRepeat<>(this, times, stop);
+    }
+
+    default AsyncEnumerable<T> retry(long times) {
+        return retry(times, e -> true);
+    }
+
+    default AsyncEnumerable<T> retry(Predicate<? super Throwable> predicate) {
+        return retry(Long.MAX_VALUE, predicate);
+    }
+
+    default AsyncEnumerable<T> retry(long times, Predicate<? super Throwable> predicate) {
+        return new AsyncRetry<>(this, times, predicate);
+    }
+
+    default AsyncEnumerable<T> repeatWhen(Supplier<? extends CompletionStage<Boolean>> completer) {
+        return repeatWhen(() -> null, s -> completer.get());
+    }
+
+    default <S> AsyncEnumerable<T> repeatWhen(Supplier<S> stateSupplier, Function<? super S, ? extends CompletionStage<Boolean>> completer) {
+        return new AsyncRepeatWhen<>(this, stateSupplier, completer);
+    }
+
+    default <S> AsyncEnumerable<T> retryWhen(Function<? super Throwable, ? extends CompletionStage<Boolean>> completer) {
+        return retryWhen(() -> null, (s, e) -> completer.apply(e));
+    }
+
+    default <S> AsyncEnumerable<T> retryWhen(Supplier<S> stateSupplier, BiFunction<? super S, ? super Throwable, ? extends CompletionStage<Boolean>> completer) {
+        return new AsyncRetryWhen<T, S>(this, stateSupplier, completer);
+    }
+
     // -------------------------------------------------------------------------------------
     // Instance consumers
 

src/main/java/hu/akarnokd/asyncenum/AsyncFromCallable.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2017 David Karnok
+ *
+ * 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 hu.akarnokd.asyncenum;
+
+import java.util.concurrent.*;
+
+final class AsyncFromCallable<T> implements AsyncEnumerable<T> {
+
+    final Callable<? extends T> callable;
+
+    AsyncFromCallable(Callable<? extends T> callable) {
+        this.callable = callable;
+    }
+
+    @Override
+    public AsyncEnumerator<T> enumerator() {
+        return new FromCallableEnumerator<>(callable);
+    }
+
+    static final class FromCallableEnumerator<T> implements AsyncEnumerator<T> {
+
+        final Callable<? extends T> callable;
+
+        T result;
+
+        boolean once;
+
+        FromCallableEnumerator(Callable<? extends T> callable) {
+            this.callable = callable;
+        }
+
+        @Override
+        public CompletionStage<Boolean> moveNext() {
+            if (once) {
+                result = null;
+                return FALSE;
+            }
+            once = true;
+            try {
+                result = callable.call();
+            } catch (Exception ex) {
+                return CompletableFuture.failedStage(ex);
+            }
+            return TRUE;
+        }
+
+        @Override
+        public T current() {
+            return result;
+        }
+    }
+}

src/main/java/hu/akarnokd/asyncenum/AsyncIgnoreElements.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2017 David Karnok
+ *
+ * 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 hu.akarnokd.asyncenum;
+
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiConsumer;
+
+final class AsyncIgnoreElements<T> implements AsyncEnumerable<T> {
+
+    final AsyncEnumerable<T> source;
+
+    AsyncIgnoreElements(AsyncEnumerable<T> source) {
+        this.source = source;
+    }
+
+    @Override
+    public AsyncEnumerator<T> enumerator() {
+        return new IgnoreElementsEnumerator<>(source.enumerator());
+    }
+
+    static final class IgnoreElementsEnumerator<T>
+            extends AtomicInteger
+            implements AsyncEnumerator<T>, BiConsumer<Boolean, Throwable> {
+
+        final AsyncEnumerator<T> source;
+
+        CompletableFuture<Boolean> completable;
+
+        IgnoreElementsEnumerator(AsyncEnumerator<T> source) {
+            this.source = source;
+        }
+
+        @Override
+        public CompletionStage<Boolean> moveNext() {
+            CompletableFuture<Boolean> cf = new CompletableFuture<>();
+            completable = cf;
+            nextSource();
+            return cf;
+        }
+
+        void nextSource() {
+            if (getAndIncrement() == 0) {
+                do {
+                    source.moveNext().whenComplete(this);
+                } while (decrementAndGet() != 0);
+            }
+        }
+
+        @Override
+        public T current() {
+            return null; // elements are ignored
+        }
+
+        @Override
+        public void cancel() {
+            source.cancel();
+        }
+
+        @Override
+        public void accept(Boolean aBoolean, Throwable throwable) {
+            if (throwable != null) {
+                completable.completeExceptionally(throwable);
+                return;
+            }
+            if (aBoolean) {
+                nextSource();
+            } else {
+                completable.complete(false);
+            }
+        }
+    }
+}