implemented timeoit for inputSelectorBuilder

Author: rssh

src/main/scala/gopher/channels/InputSelectorBuilder.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._
 
 
@@ -54,6 +55,16 @@ class InputSelectorBuilder[T](override val api: GopherAPI) extends SelectorBuild
        withIdle{ sk => Some(f(ec,sk.flowTermination) flatMap(x => 
                             proxy.awrite(x)) map(Function.const(sk)) ) }
 
+   def timeout(t:FiniteDuration)(f: FiniteDuration => T): InputSelectorBuilder[T] =
+        macro SelectorBuilder.timeoutImpl[T,InputSelectorBuilder[T]]
+
+   @inline
+   def timeoutWithFlowTerminationAsync(t:FiniteDuration, 
+                       f: (ExecutionContext, FlowTermination[T], FiniteDuration) => Future[T] ): this.type =
+        withTimeout(t){ sk => Some(f(ec,sk.flowTermination,t) flatMap( x =>
+                              proxy.awrite(x)) map(Function.const(sk)) ) }
+
+
    def foreach(f:Any=>T):T = 
         macro SelectorBuilder.foreachImpl[T]
 

src/main/scala/gopher/channels/Selector.scala

@@ -32,7 +32,7 @@ class Selector[A](api: GopherAPI) extends PromiseFlowTermination[A]
    idleWaiters add makeLocked(Skip(f,this))
   }
 
-  def addTimeoutSkip(f: Skip[A] => Option[Future[Continuated[A]]], timeout: FiniteDuration):Unit =
+  def addTimeout(timeout:FiniteDuration, f: Skip[A] => Option[Future[Continuated[A]]]):Unit =
   {
    if (!timeoutRecord.isDefined) {
      timeoutRecord.lastNOperations = nOperations.get
@@ -46,6 +46,9 @@ class Selector[A](api: GopherAPI) extends PromiseFlowTermination[A]
   def run:Future[A] =
   {
     sendWaits()
+    if (timeoutRecord.isDefined) {
+       scheduleTimeout()
+    }
     api.idleDetector put this
     future
   }

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

@@ -6,11 +6,14 @@ import scala.reflect.api._
 import gopher._
 import gopher.util._
 import scala.concurrent._
+import scala.concurrent.duration._
 import scala.annotation.unchecked._
 
 trait SelectorBuilder[A]
 {
 
+   type timeout = FiniteDuration
+
    def api: GopherAPI
 
    def onRead[E](ch:Input[E])(arg: ReadSelectorArgument[E,A]): this.type =
@@ -52,6 +55,12 @@ trait SelectorBuilder[A]
      this
    }
 
+   @inline
+   def withTimeout(t:FiniteDuration)(f: Skip[A] => Option[Future[Continuated[A]]]):this.type =
+   {
+     selector.addTimeout(t,f)
+     this
+   }
 
    def go: Future[A] = selector.run
 
@@ -183,6 +192,23 @@ object SelectorBuilder
                    })
    }
 
+   def timeoutImpl[T:c.WeakTypeTag,S](c:Context)(t:c.Expr[FiniteDuration])(f:c.Expr[FiniteDuration=>T]):c.Expr[S] = 
+   {
+     import c.universe._
+     f.tree match {
+       case Function(valdefs, body) =>
+               SelectorBuilder.buildAsyncCall[T,S](c)(valdefs,body,
+                   { (nvaldefs, nbody) =>
+                      q"""${c.prefix}.timeoutWithFlowTerminationAsync(${t},
+                                    ${Function(nvaldefs,nbody)}
+                          )
+                       """
+                   })
+      case _ => c.abort(c.enclosingPosition,"second argument of timeout must have shape Function(x,y)")
+     }
+   }
+
+
    def foreachImpl[T](c:Context)(f:c.Expr[Any=>T]):c.Expr[T] =
    {
      import c.universe._
@@ -365,7 +391,6 @@ object SelectorBuilder
                                                       atPos(caseDef.pat.pos)(q"implicitly[akka.util.Timeout]")
                                                     }
                                    val timeout = q"${builderName}.timeout(${expression})(${param} => ${body} )"
-                                   System.err.println(s"generate timeout statement: ${timeout}")
                                    atPos(caseDef.pat.pos)(timeout)
                        case _ =>
                          if (caseDef.guard.isEmpty) {

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

@@ -336,20 +336,20 @@ class MacroSelectSuite extends FunSuite
    }
 
    test("select with constant timeout")  {
-     pending
-   /*
+     //pending
      import gopherApi._
      val ch1 = makeChannel[Int](10)
      val r = select.amap {
        case x:ch1.read =>
-                  System.err.println(s"readed ${x}")
+                  //System.err.println(s"readed ${x}")
                   x
-       case x:select.timeout if (x==500.milliseconds) =>
-                 System.err.println(s"timeout 500 ms")
+       case y:select.timeout if (y==500.milliseconds) =>
+                 //System.err.println(s"timeout ${y}")
                  -1
      }
      val f1 = ch1.awrite(1)
-    */
+     val x = Await.result(r.aread, 10 seconds)
+     assert(x==1)
    }
 
    lazy val gopherApi = CommonTestObjects.gopherApi