implememted fold for Inpit

Author: rssh

src/main/scala/gopher/channels/Input.scala

@@ -8,7 +8,10 @@ import scala.reflect.macros.blackbox.Context
 import scala.reflect.api._
 import scala.util._
 import java.util.concurrent.ConcurrentLinkedQueue
+
 import gopher._
+import gopher.util._
+
 
 import java.util.concurrent.atomic._
 
@@ -316,13 +319,49 @@ trait Input[A]
   }
 
 
-/*
-  def fold[S,B](s0:S)(f:(S,A)=>(S,Option[B])) = new Input[A] {
+  def fold[S,B](s0:S)(f:(S,A)=>S): S = macro InputMacro.foldImpl[A,S]
 
-      def  cbread[C](f: ContRead[A,C] => Option[ContRead.In[A]=>Future[Continuated[C]]], ft: FlowTermination[C] ): Unit =
+  def afold[S,B](s0:S)(f:(S,A)=>S): Future[S] = macro InputMacro.afoldImpl[A,S]
+     
+  
+  def afoldSync[S,B](s0:S)(f:(S,A)=>S): Future[S] =
+  {
+    val ft = PromiseFlowTermination[S]
+    var s = s0
+    def applyF(cont:ContRead[A,S]):Option[ContRead.In[A]=>Future[Continuated[S]]] =
+    {
+          val contFold = ContRead(applyF,this,ft)
+          Some{
+            case ContRead.ChannelClosed => Future successful Done(s,ft)
+            case ContRead.Value(a) => s = f(s,a)  
+                                          Future successful contFold
+            case ContRead.Skip => Future successful contFold
+            case ContRead.Failure(ex) => Future failed ex
+          }
+    }
+    cbread(applyF,ft)
+    ft.future
+  }
 
+  def afoldAsync[S,B](s0:S)(f:(S,A)=>Future[S])(implicit ec:ExecutionContext): Future[S] =
+  {
+    val ft = PromiseFlowTermination[S]
+    var s = s0
+    def applyF(cont:ContRead[A,S]):Option[ContRead.In[A]=>Future[Continuated[S]]] =
+    {
+          Some{
+            case ContRead.ChannelClosed => Future successful Done(s,ft)
+            case ContRead.Value(a) => f(s,a) map { x => 
+                                        s = x
+                                        ContRead(applyF,this,ft)
+                                      }
+            case ContRead.Skip => Future successful ContRead(applyF,this,ft)
+            case ContRead.Failure(ex) => Future failed ex
+          }
+    }
+    cbread(applyF,ft)
+    ft.future
   }
-*/
 
 }
 
@@ -391,25 +430,10 @@ object InputMacro
   def aforeachImpl[A](c:Context)(f:c.Expr[A=>Unit]): c.Expr[Future[Unit]] =
   {
    import c.universe._
-   val findAwait = new Traverser {
-      var found = false
-      override def traverse(tree:Tree):Unit =
-      {
-       if (!found) {
-         tree match {
-            case Apply(TypeApply(Select(obj,TermName("await")),objType), args) =>
-                   if (obj.tpe =:= typeOf[scala.async.Async.type]) {
-                       found=true
-                   } else super.traverse(tree)
-            case _ => super.traverse(tree)
-         }
-       }
-      }
-   }
    f.tree match {
      case Function(valdefs,body) =>
-            findAwait.traverse(body)
-            if (findAwait.found) {
+            if (MacroUtil.hasAwait(c)(body)) {
+               // TODO: add support for flow-termination (?)
                val nbody = q"scala.async.Async.async(${body})"
                val nfunction = atPos(f.tree.pos)(Function(valdefs,nbody))
                val ntree = q"${c.prefix}.foreachAsync(${nfunction})"
@@ -421,5 +445,27 @@ object InputMacro
    }
   }
 
+  def foldImpl[A,S](c:Context)(s0:c.Expr[S])(f:c.Expr[(S,A)=>S]): c.Expr[S] =
+  {
+   import c.universe._
+   c.Expr[S](q"scala.async.Async.await(${afoldImpl(c)(s0)(f)})")
+  }
+
+  def afoldImpl[A,S](c:Context)(s0:c.Expr[S])(f:c.Expr[(S,A)=>S]): c.Expr[Future[S]] =
+  {
+   import c.universe._
+   f.tree match {
+     case Function(valdefs,body) =>
+            if (MacroUtil.hasAwait(c)(body)) {
+               val nbody = atPos(body.pos)(q"scala.async.Async.async(${body})")
+               val nfunction = atPos(f.tree.pos)(Function(valdefs,nbody))
+               val ntree = q"${c.prefix}.afoldAsync(${s0.tree})(${nfunction})"
+               c.Expr[Future[S]](c.untypecheck(ntree))
+            } else {
+               c.Expr[Future[S]](q"${c.prefix}.afoldSync(${s0.tree})(${f.tree})")
+            }
+   }
+  }
+
 
 }

src/main/scala/gopher/util/MacroUtil.scala

@@ -2,6 +2,7 @@ package gopher.util
 
 import scala.reflect.macros.blackbox.Context
 import scala.reflect.api._
+import scala.language.reflectiveCalls
 
 
 object MacroUtil
@@ -30,5 +31,28 @@ object MacroUtil
      }
   }
 
+  def hasAwait(c:Context)(x: c.Tree):Boolean =
+  {
+    import c.universe._
+    val findAwait = new Traverser {
+      var found = false
+      override def traverse(tree:Tree):Unit =
+      {
+       if (!found) {
+         tree match {
+            case Apply(TypeApply(Select(obj,TermName("await")),objType), args) =>
+                   if (obj.tpe =:= typeOf[scala.async.Async.type]) {
+                       found=true
+                   } else super.traverse(tree)
+            case _ => super.traverse(tree)
+         }
+       }
+      }
+   }
+   findAwait.traverse(x)
+   findAwait.found
+  }
+
+
   final val SHORT_LEN = 80
 }