added SelectTimeout

Author: rssh

src/main/scala/gopher/channels/ForeverSelectorBuilder.scala

@@ -1,11 +1,12 @@
 package gopher.channels
 
 import scala.language.experimental.macros
-import scala.reflect.macros.whitebox.Context
+import scala.reflect.macros.blackbox.Context
 import scala.reflect.api._
 import gopher._
 import gopher.util._
 import scala.concurrent._
+import scala.concurrent.duration._
 import scala.annotation.unchecked._
 
 
@@ -36,6 +37,14 @@ trait ForeverSelectorBuilder extends SelectorBuilder[Unit]
    def writingWithFlowTerminationAsync[A](ch:Output[A], x: =>A, f: (ExecutionContext, FlowTermination[Unit], A) => Future[Unit] ): ForeverSelectorBuilder =
        withWriter[A](ch,   { cw => Some(x,f(ec,cw.flowTermination, x) map Function.const(cw)) } )
 
+   def timeout(t:FiniteDuration)(f: FiniteDuration => Unit): ForeverSelectorBuilder =
+        macro SelectorBuilder.timeoutImpl[Unit,ForeverSelectorBuilder]
+
+ @inline
+   def timeoutWithFlowTerminationAsync(t:FiniteDuration,
+                       f: (ExecutionContext, FlowTermination[Unit], FiniteDuration) => Future[Unit] ): this.type =
+        withTimeout(t){ sk => Some(f(ec,sk.flowTermination,t) map Function.const(sk)) }
+
 
    def idle(body:Unit): ForeverSelectorBuilder =
          macro SelectorBuilder.idleImpl[Unit,ForeverSelectorBuilder]

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

@@ -388,7 +388,7 @@ object SelectorBuilder
                                    val expression = if (!caseDef.guard.isEmpty) {
                                                       parseGuardInSelectorCaseDef(c)(termName,caseDef.guard)
                                                     } else {
-                                                      atPos(caseDef.pat.pos)(q"implicitly[akka.util.Timeout]")
+                                                      atPos(caseDef.pat.pos)(q"implicitly[akka.util.Timeout].duration")
                                                     }
                                    val timeout = q"${builderName}.timeout(${expression})(${param} => ${body} )"
                                    atPos(caseDef.pat.pos)(timeout)

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

@@ -335,37 +335,6 @@ class MacroSelectSuite extends FunSuite
      assert(x2.isInstanceOf[ChannelClosedException])
    }
 
-   test("select with constant timeout which not fire")  {
-     //pending
-     import gopherApi._
-     val ch1 = makeChannel[Int](10)
-     val r = select.amap {
-       case x:ch1.read =>
-                  //System.err.println(s"readed ${x}")
-                  x
-       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)
-   }
-
-   test("select with constant timeout which fire")  {
-     import gopherApi._
-     val ch1 = makeChannel[Int](10)
-     val r = select.amap {
-       case x:ch1.read =>
-                  //System.err.println(s"readed ${x}")
-                  x
-       case x:select.timeout if (x==500.milliseconds) =>
-                 //System.err.println(s"timeout ${x}")
-                 -1
-     }
-     val x = Await.result(r.aread, 10 seconds)
-     assert(x == -1)
-   }
 
    lazy val gopherApi = CommonTestObjects.gopherApi
 

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

@@ -0,0 +1,77 @@
+package gopher.channels
+
+import gopher._
+import gopher.channels._
+import gopher.tags._
+
+import org.scalatest._
+
+import scala.language._
+import scala.concurrent._
+import scala.concurrent.duration._
+
+import akka.util.Timeout
+
+class SelectTimeoutSuite extends FunSuite 
+{
+
+   import scala.concurrent.ExecutionContext.Implicits.global
+
+  
+   test("select with constant timeout which not fire")  {
+     //pending
+     import gopherApi._
+     val ch1 = makeChannel[Int](10)
+     val r = select.amap {
+       case x:ch1.read =>
+                  //System.err.println(s"readed ${x}")
+                  x
+       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)
+   }
+
+   test("select with constant timeout which fire")  {
+     import gopherApi._
+     val ch1 = makeChannel[Int](10)
+     val r = select.amap {
+       case x:ch1.read =>
+                  //System.err.println(s"readed ${x}")
+                  x
+       case x:select.timeout if (x==500.milliseconds) =>
+                 //System.err.println(s"timeout ${x}")
+                 -1
+     }
+     val x = Await.result(r.aread, 10 seconds)
+     assert(x == -1)
+   }
+
+   test("timeout in select.forever")  {
+     import gopherApi._
+     val ch1 = makeChannel[Int](10)
+     val ch2 = makeChannel[Int]()
+     val chS = makeChannel[String](10)
+     var s = 0
+     implicit val timeout = Timeout(100 milliseconds)
+     val f = select.forever{
+               case x: ch1.read => 
+                             chS.write("1") 
+               case x: ch2.read => 
+                             chS.write("2") 
+               case x:select.timeout =>
+                             s += 1
+                             chS.write("t") 
+                             if (s > 2) select.exit(())
+             }
+     val x = Await.result(f, 10 seconds)
+     assert(s > 2)
+   }
+
+
+   lazy val gopherApi = CommonTestObjects.gopherApi
+   
+}