Refactoring Continuations class, adding parameter to specify whether to return current continuation values from streams/iterators

Author: vsilaev

net.tascalate.javaflow.examples/src/main/java/org/apache/commons/javaflow/examples/lambdas/LambdasExampleMinimalIterator.java

@@ -16,8 +16,7 @@
 package org.apache.commons.javaflow.examples.lambdas;
 
 import org.apache.commons.javaflow.api.Continuation;
-
-import org.apache.commons.javaflow.extras.CoIterator;
+import org.apache.commons.javaflow.extras.ClosableIterator;
 import org.apache.commons.javaflow.extras.Continuations;
 
 public class LambdasExampleMinimalIterator {
@@ -40,7 +39,7 @@ public static void main(String[] argv) throws Exception {
         // use try-with-resources to close the coIterator 
         // (and hence terminate underlying continuation) 
         // in case of early exit
-        try (CoIterator<Integer> i = Continuations.iterate(cc)) {
+        try (ClosableIterator<Integer> i = Continuations.iterate(cc)) {
             int c = 0;
             while (i.hasNext()) {
                 System.out.println("Interrupted " + i.next());

net.tascalate.javaflow.examples/src/main/java/org/apache/commons/javaflow/examples/lambdas/LambdasExampleMinimalStream.java

@@ -24,10 +24,8 @@
 public class LambdasExampleMinimalStream {
 
     public static void main(String[] argv) throws Exception {
-        // use try-with-resources to close the stream 
-        // (and hence terminate underlying continuation) 
-        // in case of early exit
-        try (Stream<Integer> stream = Continuations.stream(() -> {
+        // Create suspended continuation
+        Continuation cc = Continuations.create(() -> {
             try {
                 for (int i = 1; i <= 5; i++) {
                     System.out.println("Exe before suspend");
@@ -37,7 +35,12 @@ public static void main(String[] argv) throws Exception {
             } finally {
                 System.out.println("Continuation gracefully exited");
             }
-        })) {
+        });
+        
+        // use try-with-resources to close the stream 
+        // (and hence terminate underlying continuation) 
+        // in case of early exit
+        try (Stream<Integer> stream = Continuations.stream(cc)) {
             Optional<Integer> firstDividableByThree = 
                 stream.peek(v -> System.out.println("Interrupted " + v))
                       .filter(v -> v% 3 == 0)

net.tascalate.javaflow.extras/src/main/java/org/apache/commons/javaflow/extras/ClosableIterator.java

@@ -0,0 +1,32 @@
+/**
+ * Copyright 2013-2017 Valery Silaev (http://vsilaev.com)
+ *
+ * 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.apache.commons.javaflow.extras;
+
+import java.util.Iterator;
+
+/**
+ * Iterator that additionally implements {@link Closable} interface 
+ * to execute clean-up when not fully iterated
+ * 
+ * @author vsilaev
+ *
+ * @param <E>
+ * Type of objects returned by the iterator 
+ */
+public interface ClosableIterator<E> extends Iterator<E>, AutoCloseable {
+    @Override
+    void close();
+}

net.tascalate.javaflow.extras/src/main/java/org/apache/commons/javaflow/extras/CoIterator.java

@@ -16,7 +16,6 @@
 package org.apache.commons.javaflow.extras;
 
 import java.io.Serializable;
-import java.util.Iterator;
 import java.util.NoSuchElementException;
 
 import org.apache.commons.javaflow.api.Continuation;
@@ -41,10 +40,10 @@
  * it's impossible to pass value back to continuation from client code; the result of 
  * {@link Continuation#suspend(Object)} in continuation will be always <i>null</i> 
  * 
- * @param <T>
+ * @param <E>
  * Type of objects returned by the iterator
  */
-public class CoIterator<T> implements Iterator<T>, Serializable, AutoCloseable {
+public class CoIterator<E> implements ClosableIterator<E>, Serializable {
     private static final long serialVersionUID = 1L;
 
     private boolean advance;
@@ -59,7 +58,7 @@ public class CoIterator<T> implements Iterator<T>, Serializable, AutoCloseable {
      */
     public CoIterator(Runnable code) {
         cc = Continuation.startSuspendedWith(code);
-        advance = true;
+        advance = null != cc;
     }
 
     /**
@@ -74,31 +73,47 @@ public CoIterator(ContinuableRunnable code) {
     }
 
     /**
-     * Iterator constructor
+     * <p>Iterator constructor
+     * <p>Valued returned by this iterator will be results these are yielded via call to 
+     * {@link Continuation#suspend(Object)}, i.e. cc.value() is not included
      * 
      * @param cc
      * Current {@link Continuation} to start iteration from. 
-     * Valued returned by this iterator will be results these are yielded via call to 
-     * {@link Continuation#suspend(Object)}, i.e. cc.value() is not included
      */
     public CoIterator(Continuation cc) {
+        this(cc, false);
+    }
+    
+    /**
+     * <p>Iterator constructor
+     * <p>Valued returned by this iterator will be results these are yielded via call to 
+     * {@link Continuation#suspend(Object)}, i.e. cc.value() is not included unless 
+     * <code>useCurrentValue</code> is true
+     *  
+     * @param cc
+     * Current {@link Continuation} to start iteration from.
+     * 
+     * @param useCurrentValue
+     * Should the value of the supplied continuation be used as a first returned value 
+     */
+    public CoIterator(Continuation cc, boolean useCurrentValue) {
         this.cc = cc;
-        advance = true;
+        advance = !useCurrentValue && null != cc;
     }
 
     public boolean hasNext() {
         advanceIfNecessary();
         return cc != null;
     }
 
-    public T next() {
+    public E next() {
         advanceIfNecessary();
 
         if (cc == null)
             throw new NoSuchElementException();
 
         @SuppressWarnings("unchecked")
-        T result = (T) cc.value();
+        E result = (E)cc.value();
         advance = true;
 
         return result;
@@ -109,21 +124,18 @@ public void remove() {
     }
 
     public void close() {
-        if (null != cc) {
-            try {
-                cc.terminate();
-            } finally {
-                cc = null;
-                advance = false;
-            }
-        } else {
+        try {
+            cc.terminate();
+        } finally {
+            cc = null;
             advance = false;
         }
     }
 
     protected void advanceIfNecessary() {
-        if (advance)
+        if (advance) {
             cc = cc.resume();
+        }
         advance = false;
     }
 }

net.tascalate.javaflow.extras/src/main/java/org/apache/commons/javaflow/extras/Continuations.java

@@ -66,23 +66,32 @@ public static Continuation start(ContinuableRunnable o) {
         return Continuation.startWith(toRunnable(o));
     }
 
-    public static <T> CoIterator<T> iterate(Continuation continuation) {
-        return new CoIterator<>(continuation);
+    public static <T> ClosableIterator<T> iterate(Continuation continuation) {
+        return iterate(continuation, false);
     }
 
-    public static <T> CoIterator<T> iterate(ContinuableRunnable generator) {
-        return iterate(create(generator));
+    public static <T> ClosableIterator<T> iterate(Continuation continuation, boolean useCurrentValue) {
+        return new CoIterator<>(continuation, useCurrentValue);
+    }
+    
+    
+    public static <T> ClosableIterator<T> iterate(ContinuableRunnable generator) {
+        return iterate(create(generator), false);
     }
 
     public static <T> Stream<T> stream(Continuation continuation) {
-        CoIterator<T> iterator = iterate(continuation);
+        return stream(continuation, false);
+    }
+    
+    public static <T> Stream<T> stream(Continuation continuation, boolean useCurrentValue) {
+        ClosableIterator<T> iterator = iterate(continuation, useCurrentValue);
         return StreamSupport
                .stream(Spliterators.spliteratorUnknownSize(iterator, 0), false)
                .onClose(iterator::close);
     }
 
     public static <T> Stream<T> stream(ContinuableRunnable generator) {
-        return stream(create(generator));
+        return stream(create(generator), false);
     }
 
     /**
@@ -92,11 +101,24 @@ public static <T> Stream<T> stream(ContinuableRunnable generator) {
      * 
      * @param <T> a type of values  
      * @param continuation a continuation to resume a code block that yields multiple results
-     * @param valueType a type of the values yielded from code block
-     * @param action a continuable action to perform on the values yielded
+     * @param action a non-continuable action to perform on the values yielded
      */
     public static <T> void execute(Continuation continuation, Consumer<? super T> action) {
-        try (CoIterator<T> iter = new CoIterator<>(continuation)) {
+        execute(continuation, false, action);
+    }
+    
+    /**
+     * Executes the suspended continuation from the point specified till the end 
+     * of the corresponding code block and performs a non-suspendable action 
+     * on each value yielded.
+     * 
+     * @param <T> a type of values  
+     * @param continuation a continuation to resume a code block that yields multiple results
+     * @param useCurrentValue should the value of the supplied continuation be used as a first value to process
+     * @param action a non-continuable action to perform on the values yielded
+     */
+    public static <T> void execute(Continuation continuation, boolean useCurrentValue, Consumer<? super T> action) {
+        try (ClosableIterator<T> iter = iterate(continuation, useCurrentValue)) {
             while (iter.hasNext()) {
                 action.accept(iter.next());
             }
@@ -109,8 +131,7 @@ public static <T> void execute(Continuation continuation, Consumer<? super T> ac
      * 
      * @param <T> a type of values 
      * @param generator a continuable code block that yields multiple results
-     * @param valueType a type of the values yielded from code block
-     * @param action a continuable action to perform on the values yielded
+     * @param action a non-continuable action to perform on the values yielded
      */
     public static <T> void execute(ContinuableRunnable generator, Consumer<? super T> action) {
         execute(create(generator), action);
@@ -124,24 +145,38 @@ public static <T> void execute(ContinuableRunnable generator, Consumer<? super T
      * 
      * @param <T> a type of values  
      * @param continuation a continuation to resume a code block that yields multiple results
-     * @param valueType a type of the values yielded from code block
      * @param action a continuable action to perform on the values yielded
      */
     public @continuable static <T> void executeContinuable(Continuation continuation, ContinuableConsumer<? super T> action) {
-        forEach(()-> new CoIterator<>(continuation), action);
+        executeContinuable(continuation, false, action);
     }
 
+    /**
+     * Executes the suspended continuation from the point specified till the end 
+     * of the corresponding code block and performs a potentially suspendable action 
+     * on each value yielded.
+     * 
+     * @param <T> a type of values  
+     * @param useCurrentValue should the value of the supplied continuation be used as a first value to process
+     * @param action a continuable action to perform on the values yielded
+     */
+    public @continuable static <T> void executeContinuable(Continuation continuation, boolean useCurrentValue, ContinuableConsumer<? super T> action) {
+        try (ClosableIterator<T> iter = iterate(continuation, useCurrentValue)) {
+            forEach(iter, action);
+        }
+    }
+
+    
     /**
      * Fully executes the continuable code block and performs a potentially suspendable 
      * action on each value yielded.
      * 
      * @param <T> a type of values 
      * @param generator a continuable code block that yields multiple results
-     * @param valueType a type of the values yielded from code block
      * @param action a continuable action to perform on the values yielded
      */
     public @continuable static <T> void executeContinuable(ContinuableRunnable generator, ContinuableConsumer<? super T> action) {
-        executeContinuable(create(generator), action);
+        executeContinuable(create(generator), false, action);
     }
 
     /**
@@ -169,7 +204,10 @@ public static <T> void execute(ContinuableRunnable generator, Consumer<? super T
      * @param action a continuable action to perform on the elements
      */   
     public @continuable static <T> void forEach(Iterable<T> iterable, ContinuableConsumer<? super T> action) {
-        forEach(iterable.iterator(), action);
+        Iterator<T> iter = iterable.iterator();
+        try (ClosableIterator<T> closable = asClosable(iter)) {
+            forEach(iter, action);
+        }
     }
 
     /**
@@ -188,5 +226,9 @@ public static <T> void execute(ContinuableRunnable generator, Consumer<? super T
         }
     }
 
+    @SuppressWarnings("unchecked")
+    private static <E> ClosableIterator<E> asClosable(Object o) {
+        return o instanceof ClosableIterator ? (ClosableIterator<E>)o : null;
+    }
 
 }