added iddle for fold

Author: rssh

build.sbt

@@ -9,7 +9,7 @@ resolvers += Resolver.sonatypeRepo("snapshots")
 
 resolvers += "Typesafe Repository" at "http://repo.typesafe.com/typesafe/releases/"
 
-scalacOptions ++= Seq("-unchecked","-deprecation", "-feature" /* ,   "-Ymacro-debug-lite"  ,  "-Ydebug"  , "-Ylog:lambdalift" */ )
+scalacOptions ++= Seq("-unchecked","-deprecation", "-feature" /* , "-Ymacro-debug-lite"  ,  "-Ydebug"  , "-Ylog:lambdalift" */ )
 
 libraryDependencies <+= scalaVersion( "org.scala-lang" % "scala-reflect" % _ )
 

notes/0.99.7.markdown

@@ -5,7 +5,7 @@
 - implemented EffectedInput/EffectedOutput
  (more detailed about fold over selected and Effected pattern can
   be readed at https://github.com/rssh/notes/blob/master/2016_03_05_see-ma-no-vars.md )
-- akka 2.4.2
+- akka 2.4.2, scala-2.11.8
 - channels by defaults now unbuffered. (as in Go)
 
 

notes/papers.markdown

@@ -16,5 +16,7 @@ CSP for scala: implementation, papers, ect ..
 
 * Minimalistics implementation of clojure core-async in scala: http://blog.podsnap.com/scasync.html
 
+* implementation over hawtdipath: https://github.com/oplohmann/CoSelect.Scala
+
 ------------------
   part of [scala-gopher](https://github.com/rssh/scala-gopher)

src/main/scala/gopher/channels/FoldSelectorBuilder.scala

@@ -33,7 +33,12 @@ trait FoldSelectorBuilder[T] extends SelectorBuilder[T]
    def writingWithFlowTerminationAsync[A](ch:Output[A], x: =>A, f: (ExecutionContext, FlowTermination[T], A) => Future[T] ): this.type =
        withWriter[A](ch,   { cw => Some(x,f(ec,cw.flowTermination, x) map Function.const(cw)) } )
 
+  def idle(body:T): FoldSelectorBuilder[T] =
+         macro SelectorBuilder.idleImpl[T,FoldSelectorBuilder[T]]
 
+   @inline
+   def idleWithFlowTerminationAsync[A](f: (ExecutionContext, FlowTermination[T]) => Future[T] ): this.type =
+       withIdle{ st => Some(f(ec,st.flowTermination) map Function.const(st)) }
 
 }
 
@@ -239,6 +244,8 @@ class FoldSelectorBuilderImpl(val c:Context)
                                      c.abort(cd.pat.pos,"x:channel.read or x:channel.write form is required")
                             }
                        }
+                     case Ident(TermName("_")) => (cd.pat, cd.guard)
+                     case _ => c.abort(cd.pat.pos,"expected Bind or Default in pattern, have:"+cd.pat)
                    }
        atPos(cd.pos)(CaseDef(pat,substProj(fp,guard),preTransformCaseDefBody(fp,patSymbol,cd.body)))
      }else{

src/test/scala/gopher/channels/InputOpsSuite.scala

@@ -259,8 +259,8 @@ class InputOpsSuite extends FunSuite with AsyncAssertions {
 
   test("append for finite stream") {
       val w = new Waiter
-      val ch1 = gopherApi.makeChannel[Int]() 
-      val ch2 = gopherApi.makeChannel[Int]() 
+      val ch1 = gopherApi.makeChannel[Int](10) 
+      val ch2 = gopherApi.makeChannel[Int](10) 
       val appended = ch1 append ch2
       var sum = 0
       var prev = 0

src/test/scala/gopher/channels/MacroSelectSuite.scala

@@ -219,6 +219,46 @@ class MacroSelectSuite extends FunSuite
     }
    }
 
+   test("fold over selector with idle")  {
+     import gopherApi._
+     val ch1 = makeChannel[Int](10)
+     val ch2 = makeChannel[Int](10)
+     ch1.awrite(1)
+     val sf = select.afold((0,0,0)){ case ((n1,n2,nIdle),s) =>
+       s match {
+         case x:ch1.read =>
+                 val nn1 = n1+1
+                 if (nn1 > 100) {
+                    CurrentFlowTermination.exit((nn1,n2,nIdle))
+                 }else{
+                     ch2.write(x)
+                     (nn1,n2,nIdle)
+                 }
+         case x:ch2.read =>
+                ch1.write(x)
+                (n1,n2+1,nIdle)
+         case _ =>
+                (n1,n2,nIdle+1)
+ 
+       }
+     }
+     val (n1,n2,ni) = Await.result(sf, 10 seconds)
+     assert (n1+n2+ni > 100)
+     val sf2 = select.afold((0,0)){ case ((n1,nIdle),s) =>
+       s match {
+         case x:ch1.read =>
+                      (n1+1,nIdle)
+         case _ =>
+                   val nni = nIdle+1
+                   if (nni > 3) {
+                      CurrentFlowTermination.exit((n1,nni))
+                   } else {
+                      (n1,nni)
+                   }
+     } }
+     val (n21,n2i) = Await.result(sf2, 10 seconds)
+     assert(n2i>3)
+   }
 
    lazy val gopherApi = CommonTestObjects.gopherApi