Upgrade *extras* to JDK 1.8; add std. @Continuable forEach to Continuations class for Stream/Iterable/ContinuableRunnable(coroutine)

Author: vsilaev

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

@@ -1,10 +1,6 @@
 package org.apache.commons.javaflow.examples.lambdas;
 
-import java.util.Iterator;
-import java.util.function.Consumer;
-import java.util.stream.Stream;
-
-import org.apache.commons.javaflow.api.continuable;
+import org.apache.commons.javaflow.extras.ContinuableConsumer;
 import org.apache.commons.javaflow.extras.ContinuableRunnable;
 
 public class ContinuableAdapters {
@@ -17,21 +13,7 @@ public static <T> ContinuableConsumer<T> accept(ContinuableConsumer<T> o) {
         return o;
     }
 
-    public @continuable static <T> void forEach(Stream<T> o, ContinuableConsumer<? super T> action) {
-        forEach(o.iterator(), action);
-    }
-
-    public @continuable static <T> void forEach(Iterable<T> o, ContinuableConsumer<? super T> action) {
-        forEach(o.iterator(), action);
-    }
-
-    public @continuable static <T> void forEach(Iterator<T> it, ContinuableConsumer<? super T> action) {
-        while (it.hasNext()) {
-            T v = it.next();
-            action.accept(v);
-        }
-    }
-
+    /*
     interface ContinuableConsumer<T> extends Consumer<T> {
         @continuable void accept(T t);
 
@@ -47,4 +29,5 @@ default ContinuableConsumer<T> andThen(Consumer<? super T> after) {
             return andThen((ContinuableConsumer<? super T>)after);
         }
     }
+    */
 }

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

@@ -2,7 +2,6 @@
 
 import static org.apache.commons.javaflow.examples.lambdas.ContinuableAdapters.accept;
 import static org.apache.commons.javaflow.examples.lambdas.ContinuableAdapters.exec;
-import static org.apache.commons.javaflow.examples.lambdas.ContinuableAdapters.forEach;
 
 import java.util.Arrays;
 import java.util.List;
@@ -16,10 +15,9 @@
 public class LambdasExample {
     public static void main(final String[] argv) throws Exception {
         LambdasExample example = new LambdasExample();
-
-        for (Continuation cc = Continuations.start(example::executeAll); null != cc; cc = cc.resume()) {
-            System.out.println("Interrupted " + cc.value());
-        }
+        Continuations.<String>forEach(example::executeAll, s -> {
+            System.out.println("Interrupted " + s);
+        });
 
         System.out.println("===");
 
@@ -53,7 +51,7 @@ public static void main(final String[] argv) throws Exception {
         // See org.apache.commons.javaflow.examples.lambdas.ContinuableAdapters for
         // definition of "accept" & "forEach"
         List<String> listOfStrings = Arrays.asList("A", null, "B", null, "C"); 
-        forEach( 
+        Continuations.forEach( 
                 listOfStrings.stream().filter(s -> s != null).map(s -> s + s), 
                 accept(this::yieldString1).andThen(this::yieldString2) 
         );

net.tascalate.javaflow.extras/.classpath

@@ -22,7 +22,7 @@
 			<attribute name="maven.pomderived" value="true"/>
 		</attributes>
 	</classpathentry>
-	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/J2SE-1.5">
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.8">
 		<attributes>
 			<attribute name="maven.pomderived" value="true"/>
 		</attributes>

net.tascalate.javaflow.extras/.settings/org.eclipse.jdt.core.prefs

@@ -1,5 +1,5 @@
 eclipse.preferences.version=1
-org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.5
-org.eclipse.jdt.core.compiler.compliance=1.5
+org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.8
+org.eclipse.jdt.core.compiler.compliance=1.8
 org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning
-org.eclipse.jdt.core.compiler.source=1.5
+org.eclipse.jdt.core.compiler.source=1.8

net.tascalate.javaflow.extras/pom.xml

@@ -33,6 +33,14 @@
 	</dependencies>
 	<build>
 		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-compiler-plugin</artifactId>
+				<configuration>
+					<source>1.8</source>
+					<target>1.8</target>
+				</configuration>
+			</plugin>
 			<plugin>
 				<groupId>net.tascalate.javaflow</groupId>
 				<artifactId>net.tascalate.javaflow.tools.maven</artifactId>

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

@@ -0,0 +1,49 @@
+package org.apache.commons.javaflow.extras;
+
+import java.util.Objects;
+
+import org.apache.commons.javaflow.api.continuable;
+
+/**
+ * Represents an continuable operation that accepts a single input argument and returns no
+ * result. Unlike most other functional interfaces, {@code ContinuableConsumer} is expected
+ * to operate via side-effects.
+ *
+ * <p>This is a functional interface whose functional method is {@link #accept(Object)}.
+ *
+ * @param <T> the type of the input to the operation
+ *
+ */
+@FunctionalInterface
+public interface ContinuableConsumer<T> {
+    
+    /**
+     * Performs this operation on the given argument.
+     *
+     * @param t the input argument
+     */
+    @continuable void accept(T t);
+    
+    /**
+     * Returns a composed {@code ContinuableConsumer} that performs, in sequence, this
+     * operation followed by the {@code after} operation. If performing either
+     * operation throws an exception, it is relayed to the caller of the
+     * composed operation.  If performing this operation throws an exception,
+     * the {@code after} operation will not be performed.
+     *
+     * @param after the operation to perform after this operation
+     * @return a composed {@code ContinuableConsumer} that performs in sequence this
+     * operation followed by the {@code after} operation
+     * @throws NullPointerException if {@code after} is null
+     */
+    default ContinuableConsumer<T> andThen(ContinuableConsumer<? super T> after) {
+        Objects.requireNonNull(after);
+        return new ContinuableConsumer<T>() {
+            public @continuable void accept(T t) {
+                ContinuableConsumer.this.accept(t);
+                after.accept(t);
+            }
+        };
+    }
+
+}

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

@@ -2,7 +2,13 @@
 
 import org.apache.commons.javaflow.api.continuable;
 
+/**
+ * Continuable version of Runnable 
+ */
+@FunctionalInterface
 public interface ContinuableRunnable extends Runnable {
-	//Re-declare to mark as continuable
+	/**
+	 * Run method re-declared as continuable
+	 */
 	@continuable void run();
 }

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

@@ -1,19 +1,41 @@
 package org.apache.commons.javaflow.extras;
 
+import java.util.Iterator;
+import java.util.stream.Stream;
+
 import org.apache.commons.javaflow.api.Continuation;
+import org.apache.commons.javaflow.api.continuable;
 
 final public class Continuations {
-	private Continuations() {}
-	
-	public static Continuation create(ContinuableRunnable o) {
-		return Continuation.startSuspendedWith(o);
-	}
-	
-	public static Continuation start(ContinuableRunnable o, Object ctx) {
-		return Continuation.startWith(o, ctx);
-	}
-	
-	public static Continuation start(ContinuableRunnable o) {
-		return Continuation.startWith(o);
-	}
+    private Continuations() {}
+
+    public static Continuation create(ContinuableRunnable o) {
+        return Continuation.startSuspendedWith(o);
+    }
+
+    public static Continuation start(ContinuableRunnable o, Object ctx) {
+        return Continuation.startWith(o, ctx);
+    }
+
+    public static Continuation start(ContinuableRunnable o) {
+        return Continuation.startWith(o);
+    }
+    
+    public @continuable static <T> void forEach(ContinuableRunnable coroutine, ContinuableConsumer<? super T> action) {
+        forEach(new CoIterator<>(coroutine), action);
+    }
+
+    public @continuable static <T> void forEach(Stream<T> iterable, ContinuableConsumer<? super T> action) {
+        forEach(iterable.iterator(), action);
+    }
+
+    public @continuable static <T> void forEach(Iterable<T> iterable, ContinuableConsumer<? super T> action) {
+        forEach(iterable.iterator(), action);
+    }
+
+    public @continuable static <T> void forEach(Iterator<T> iterator, ContinuableConsumer<? super T> action) {
+        while (iterator.hasNext()) {
+            action.accept( iterator.next() );
+        }
+    }
 }