Extensions to Continuations class

Author: vsilaev

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

@@ -29,7 +29,7 @@
 public class LambdasExample {
     public static void main(String[] argv) throws Exception {
         LambdasExample demo = new LambdasExample();
-        Continuations.execute(demo::runExamples, s -> System.out.println("Interrupted " + s));
+        Continuations.execute(demo::runExamples, (String s) -> System.out.println("Interrupted " + s));
 
         System.out.println("===");
 

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

@@ -0,0 +1,58 @@
+/**
+ * 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.examples.lambdas;
+
+import org.apache.commons.javaflow.api.Continuation;
+
+import org.apache.commons.javaflow.extras.CoIterator;
+import org.apache.commons.javaflow.extras.Continuations;
+
+public class LambdasExampleMinimalIterator {
+
+    public static void main(String[] argv) throws Exception {
+
+        // Create suspended continuation
+        Continuation cc = Continuations.create(() -> {
+            try {
+                for (int i = 1; i <= 5; i++) {
+                    System.out.println("Exe before suspend");
+                    Continuation.suspend(i);
+                    System.out.println("Exe after suspend");
+                }
+            } finally {
+                System.out.println("Continuation gracefully exited");
+            }
+        });
+        
+        // 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)) {
+            int c = 0;
+            while (i.hasNext()) {
+                System.out.println("Interrupted " + i.next());
+                c++;
+                if (c == 3) {
+                    // Emulate early exit
+                    break;
+                }
+            }
+        }
+
+        System.out.println("===");
+    }
+
+}

net.tascalate.javaflow.examples/src/main/java/org/apache/commons/javaflow/examples/lambdas/LambdasExampleMinimal.java renamed to net.tascalate.javaflow.examples/src/main/java/org/apache/commons/javaflow/examples/lambdas/LambdasExampleMinimalLoop.java

@@ -18,7 +18,7 @@
 import org.apache.commons.javaflow.api.Continuation;
 import org.apache.commons.javaflow.extras.Continuations;
 
-public class LambdasExampleMinimal {
+public class LambdasExampleMinimalLoop {
 
     public static void main(String[] argv) throws Exception {
 

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

@@ -0,0 +1,50 @@
+/**
+ * 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.examples.lambdas;
+
+import java.util.Optional;
+import java.util.stream.Stream;
+
+import org.apache.commons.javaflow.api.Continuation;
+import org.apache.commons.javaflow.extras.Continuations;
+
+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(() -> {
+            try {
+                for (int i = 1; i <= 5; i++) {
+                    System.out.println("Exe before suspend");
+                    Continuation.suspend(i);
+                    System.out.println("Exe after suspend");
+                }
+            } finally {
+                System.out.println("Continuation gracefully exited");
+            }
+        })) {
+            Optional<Integer> firstDividableByThree = 
+                stream.peek(v -> System.out.println("Interrupted " + v))
+                      .filter(v -> v% 3 == 0)
+                      .findFirst()
+                ;
+            System.out.println(firstDividableByThree);
+        }
+    }
+
+}

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

@@ -18,8 +18,10 @@
 import static org.apache.commons.javaflow.extras.ContinuationSupport.toRunnable;
 
 import java.util.Iterator;
-import java.util.function.Supplier;
+import java.util.Spliterators;
+import java.util.function.Consumer;
 import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 
 import org.apache.commons.javaflow.api.Continuation;
 import org.apache.commons.javaflow.api.continuable;
@@ -64,29 +66,56 @@ 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> CoIterator<T> iterate(ContinuableRunnable generator) {
+        return iterate(create(generator));
+    }
+    
+    public static <T> Stream<T> stream(Continuation continuation) {
+        CoIterator<T> iterator = iterate(continuation);
+        return StreamSupport
+               .stream(Spliterators.spliteratorUnknownSize(iterator, 0), false)
+               .onClose(iterator::close);
+    }
+
+    public static <T> Stream<T> stream(ContinuableRunnable generator) {
+        return stream(create(generator));
+    }
+
     /**
      * Executes the suspended continuation from the point specified till the end 
-     * of the corresponding code block and performs a potentially suspendable action 
+     * 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 valueType a type of the values yielded from code block
      * @param action a continuable action to perform on the values yielded
-     */  
-    public @continuable static void execute(Continuation continuation, ContinuableConsumer<? super Object> action) {
-        execute(continuation, Object.class, action);
+     */
+    public static <T> void execute(Continuation continuation, Consumer<? super T> action) {
+        try (CoIterator<T> iter = new CoIterator<>(continuation)) {
+            while (iter.hasNext()) {
+                action.accept(iter.next());
+            }
+        }
     }
 
     /**
-     * Fully executes the continuable code block and performs a potentially suspendable 
+     * Fully executes the continuable code block and performs a non-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 void execute(ContinuableRunnable generator, ContinuableConsumer<? super Object> action) {
-        execute(generator, Object.class, action);
+    public static <T> void execute(ContinuableRunnable generator, Consumer<? super T> action) {
+        execute(create(generator), action);
     }
-    
+
 
     /**
      * Executes the suspended continuation from the point specified till the end 
@@ -98,7 +127,7 @@ public static Continuation start(ContinuableRunnable o) {
      * @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 execute(Continuation continuation, Class<T> valueType, ContinuableConsumer<? super T> action) {
+    public @continuable static <T> void executeContinuable(Continuation continuation, ContinuableConsumer<? super T> action) {
         forEach(()-> new CoIterator<>(continuation), action);
     }
 
@@ -111,17 +140,10 @@ public static Continuation start(ContinuableRunnable o) {
      * @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 execute(ContinuableRunnable generator, Class<T> valueType, ContinuableConsumer<? super T> action) {
-        forEach(()-> new CoIterator<>(generator), action);
+    public @continuable static <T> void executeContinuable(ContinuableRunnable generator, ContinuableConsumer<? super T> action) {
+        executeContinuable(create(generator), action);
     }
 
-    private @continuable static <T> void execute(Supplier<CoIterator<T>> iteratorProvider, Class<T> valueType, ContinuableConsumer<? super T> action) {
-        try (CoIterator<T> iterator = iteratorProvider.get()) {
-            forEach(iterator, action);
-        }
-    }
-    
-
     /**
      * Performs an continuable action for each element of the {@link Stream} supplied.
      *