implemented select timeout in select.once

Author: rssh

build.sbt

@@ -13,7 +13,7 @@ scalacOptions ++= Seq("-unchecked","-deprecation", "-feature" /* , "-Ymacro-debu
 
 libraryDependencies <+= scalaVersion( "org.scala-lang" % "scala-reflect" % _ )
 
-libraryDependencies += "org.scala-lang.modules" %% "scala-async" % "0.9.5"
+libraryDependencies += "org.scala-lang.modules" %% "scala-async" % "0.9.6-RC2"
 
 libraryDependencies += "org.scalatest" %% "scalatest" % "2.2.6" % "test"
 

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

@@ -6,6 +6,7 @@ import scala.reflect.api._
 import gopher._
 import gopher.util._
 import scala.concurrent._
+import scala.concurrent.duration._
 import scala.annotation.unchecked._
 
 
@@ -33,6 +34,17 @@ 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 timeout(t:FiniteDuration)(f: FiniteDuration => T): FoldSelectorBuilder[T] =
+        macro SelectorBuilder.timeoutImpl[T,FoldSelectorBuilder[T]]
+
+   @inline
+   def timeoutWithFlowTerminationAsync(t:FiniteDuration,
+                       f: (ExecutionContext, FlowTermination[T], FiniteDuration) => Future[T] ): this.type =
+        withTimeout(t){ sk => Some(f(ec,sk.flowTermination,t) map Function.const(sk)  ) }
+                              
+
+
+
   def idle(body:T): FoldSelectorBuilder[T] =
          macro SelectorBuilder.idleImpl[T,FoldSelectorBuilder[T]]
 

src/main/scala/gopher/channels/OnceSelectorBuilder.scala

@@ -6,6 +6,7 @@ import scala.reflect.api._
 import gopher._
 import gopher.util._
 import scala.concurrent._
+import scala.concurrent.duration._
 import scala.annotation.unchecked._
 
 
@@ -35,6 +36,17 @@ trait OnceSelectorBuilder[T] extends SelectorBuilder[T@uncheckedVariance]
    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(x => Done(x,cw.flowTermination)) ) } )
 
+   def timeout(t:FiniteDuration)(f: FiniteDuration => T): OnceSelectorBuilder[T] =
+        macro SelectorBuilder.timeoutImpl[T,OnceSelectorBuilder[T]]
+
+
+   @inline
+   def timeoutWithFlowTerminationAsync(t:FiniteDuration,
+             f: (ExecutionContext, FlowTermination[T], FiniteDuration) => Future[T] ): this.type =
+        withTimeout(t){ sk => Some(f(ec,sk.flowTermination,t).map(x => Done(x,sk.flowTermination)) ) }
+
+
+
    def idle(body: T): OnceSelectorBuilder[T] = 
         macro SelectorBuilder.idleImpl[T,OnceSelectorBuilder[T]]
 

src/main/scala/gopher/channels/SelectorBuilder.scala

@@ -197,13 +197,14 @@ object SelectorBuilder
      import c.universe._
      f.tree match {
        case Function(valdefs, body) =>
-               SelectorBuilder.buildAsyncCall[T,S](c)(valdefs,body,
+               val r = SelectorBuilder.buildAsyncCall[T,S](c)(valdefs,body,
                    { (nvaldefs, nbody) =>
                       q"""${c.prefix}.timeoutWithFlowTerminationAsync(${t},
                                     ${Function(nvaldefs,nbody)}
                           )
                        """
                    })
+              r
       case _ => c.abort(c.enclosingPosition,"second argument of timeout must have shape Function(x,y)")
      }
    }

src/test/scala/gopher/channels/SelectTimeoutSuite.scala

@@ -71,6 +71,41 @@ class SelectTimeoutSuite extends FunSuite
      assert(s > 2)
    }
 
+   test("timeout in select.fold")  {
+     import gopherApi._
+     val ch1 = makeChannel[Int](10)
+     val f = select.afold(0) { (state,sl) => 
+                sl match {
+                   case x: ch1.read => state+1
+                   case x: select.timeout if (x == 100.milliseconds)  => 
+                                                     select.exit(state+10)
+                }
+             } 
+     ch1.awrite(1)
+     val x = Await.result(f, 10 seconds)
+     assert(x==11) 
+   }
+
+   test("timeout in select.once")  {
+     import gopherApi._
+     implicit val timeout = Timeout(100 milliseconds)
+     val ch1 = makeChannel[Int](10)
+     var x = 0
+     val f = go {
+       for(s <- select.once)  {
+        s match {
+            case y: ch1.read => info("ch1 readed")
+                                x=1
+            case y: select.timeout => 
+                                info("ch2 readed")
+                                x=10
+        }
+       }
+     }
+     Await.ready(f, 10 seconds)
+     assert(x==10) 
+     
+   }
 
    lazy val gopherApi = CommonTestObjects.gopherApi