implemented map over selector

Author: rssh

notes/0.99.7.markdown

@@ -8,5 +8,6 @@
 - akka 2.4.2, scala-2.11.8
 - channels by defaults now unbuffered. (as in Go)
 - implemented fold over input
+- implemented map/amap over selector
 
 

src/main/scala/gopher/Transputer.scala

@@ -205,7 +205,7 @@ trait Transputer
 
 
  /**
-  * called when transducer is choose resume durign recovery.
+  * called when transducer is choose to resume durign recovery.
   */
  protected def onResume() { }
 
@@ -391,7 +391,7 @@ trait SelectTransputer extends ForeverSelectorBuilder with Transputer
  private[gopher] override def beforeResume() : Unit =
  {
    super.beforeResume()
-   selector = new Selector[Unit](api)
+   //selector.clear()
    selectorInit()
  }
 

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

@@ -96,7 +96,8 @@ trait ForeverSelectorBuilder extends SelectorBuilder[Unit]
     **/
     def map[B](f:Any=>B):Input[B] = macro SelectorBuilder.mapImpl[B]
 
-    def input[B](f:PartialFunction[Any,B]):Input[B] = ???
+    def input[B](f:PartialFunction[Any,B]):Input[B] = 
+                                    macro SelectorBuilder.inputImpl[B]
 
 }
 

src/main/scala/gopher/channels/InputSelectorBuilder.scala

@@ -67,6 +67,18 @@ class InputSelectorBuilder[T](override val api: GopherAPI) extends SelectorBuild
             ],
             ft: FlowTermination[B]): Unit = proxy.cbread(f,ft)
 
+   def started: InputSelectorBuilder[T] = { go; this }
+  
+   // 
+   override val selector = new Selector[T](api) {
+          override def doExit(a: T): T =
+          {
+            proxy.awrite(a) onComplete {
+              _ => proxy.close()
+            }
+            super.doExit(a)
+          }
+   }
 
 }
 

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

@@ -60,7 +60,7 @@ trait SelectorBuilder[A]
 
    implicit def ec: ExecutionContext = api.executionContext
 
-   private[gopher] var selector=new Selector[A](api)
+   private[gopher] val selector=new Selector[A](api)
 
    // used for reading from future
    @inline
@@ -423,38 +423,38 @@ object SelectorBuilder
     q"${builderName}.idle(${caseDef.body})"
    }
 
-   def mapImpl[T](c:Context)(f:c.Expr[Any=>T]):c.Expr[Input[T]] =
+   def mapImpl[T:c.WeakTypeTag](c:Context)(f:c.Expr[Any=>T]):c.Expr[Input[T]] =
    {
      import c.universe._
      val builder = f.tree match {
        case Function(forvals,Match(choice,cases)) => 
-                                foreachBuildMatch(c)(cases)
+                                mapBuildMatch[T](c)(cases)
        case Function(a,b) =>
             c.abort(f.tree.pos, "match expected in gopher select map, have: ${MacroUtil.shortString(b)} ");
        case _ =>
             c.abort(f.tree.pos, "match expected in gopher select map, have: ${MacroUtil.shortString(f.tree)}");
 
      }
-     c.Expr[Input[T]](c.untypecheck(builder))
+     c.Expr[Input[T]](c.untypecheck(q"${builder}.started"))
    }
 
-   def mapBuildMatch(c:Context)(cases:List[c.universe.CaseDef]):c.Tree =
+   def mapBuildMatch[T:c.WeakTypeTag](c:Context)(cases:List[c.universe.CaseDef]):c.Tree =
    {
      import c.universe._
      val bn = TermName(c.freshName)
      val calls = transformSelectMatch(c)(bn,cases)
-     q"""..${q"val ${bn} = ${c.prefix}.inputBuilder()" :: calls}"""
+     q"""..${q"val ${bn} = ${c.prefix}.inputBuilder[${weakTypeOf[T]}]()" :: calls}"""
    }
 
-   def inputImpl[T](c:Context)(f:c.Expr[PartialFunction[Any,T]]):c.Expr[Input[T]] = 
+   def inputImpl[T:c.WeakTypeTag](c:Context)(f:c.Expr[PartialFunction[Any,T]]):c.Expr[Input[T]] = 
    {
      import c.universe._
      val builder = f.tree match {
         case q"{case ..$cases}" =>
-                         mapBuildMatch(c)(cases)
+                         mapBuildMatch[T](c)(cases)
         case _ => c.abort(f.tree.pos,"expected partial function with syntax case ... =>, have ${MacroUtil.shortString(f.tree)}");
      }
-     c.Expr[Input[T]](c.untypecheck(builder))
+     c.Expr[Input[T]](c.untypecheck(q"${builder}.started"))
    }
 
 }

src/main/scala/gopher/channels/SelectorFactory.scala

@@ -39,6 +39,8 @@ class SelectFactory(val api: GopherAPI)
    */
   def once[T]: OnceSelectorBuilder[T] = new OnceSelectorBuilder[T] with SelectFactoryApi {}
 
+  def inputBuilder[T]() = new InputSelectorBuilder[T](api)
+
   /**
    * generic selector builder
    */
@@ -48,6 +50,14 @@ class SelectFactory(val api: GopherAPI)
 
   def fold[S](s:S)(op:(S,Any)=>S):S = macro FoldSelectorBuilderImpl.fold[S]
 
+  def map[B](f:Any=>B):Input[B] = macro SelectorBuilder.mapImpl[B]
+
+  def input[B](f:PartialFunction[Any,B]):Input[B] =
+                                    macro SelectorBuilder.inputImpl[B]
+
+  def amap[B](f:PartialFunction[Any,B]):Input[B] =
+                                    macro SelectorBuilder.inputImpl[B]
+
 
 }
 

src/test/scala/gopher/channels/CloseSuite.scala renamed to src/test/scala/gopher/channels/ChannelCloseSuite.scala

@@ -11,7 +11,7 @@ import gopher.tags._
 import scala.async.Async._
 import scala.concurrent.ExecutionContext.Implicits.global
 
-class CloseSuite extends FunSuite 
+class ChannelCloseSuite extends FunSuite 
 {
 
 

src/test/scala/gopher/channels/FlowTerminationSuite.scala

@@ -6,6 +6,7 @@ import scala.concurrent._
 import scala.concurrent.duration._
 import gopher._
 
+import scala.concurrent.ExecutionContext.Implicits.global
 
 class FlowTerminationSuite extends FunSuite 
 {
@@ -22,39 +23,36 @@ class FlowTerminationSuite extends FunSuite
    }
 
 
-/*
-  
    test("select with queue type") {
+     import gopherApi._
 
-     val channel = make[Int](100)
+     val channel = makeChannel[Int](100)
 
-     val producer = Future {
-       for( i <- 1 to 1000) {
-         channel <~ i 
-       }       
-     }
+     val producer = channel.awriteAll(1 to 1000)
 
      var sum = 0;
      val consumer = Future {
-       val sc = new SelectorContext()
-       sc.addInputAction(channel, 
-            (i: channel.OutputElement) => { sum = sum + i; 
-                          if (i == 1000) {
-                            sc.shutdown()
-                          }
-                          Promise successful true future 
-                        }
-       )
-       Await.ready(sc.go, 1000.second)
+       val sc = new Selector[Unit](gopherApi)
+       def f(self: ContRead[Int,Unit]):Option[ContRead.In[Int]=>Future[Continuated[Unit]]] =
+       {
+         Some {
+           case ContRead.Value(a) => sum = sum + a
+                                     if (a == 1000) sc.doExit(())
+                                     Future successful self
+           case ContRead.Failure(e) => Future failed e
+           case _     =>
+                                     Future successful self
+         }
+       }
+       sc.addReader(channel,f)
+       Await.ready(sc.run, 10.second)
      }
 
-    
-     Await.ready(consumer, 1000.second)
+     Await.ready(consumer, 10.second)
 
    }
 
-  */ 
 
-  def gopherApi = CommonTestObjects.gopherApi
+  val gopherApi = CommonTestObjects.gopherApi
 
 }

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

@@ -260,6 +260,31 @@ class MacroSelectSuite extends FunSuite
      assert(n2i>3)
    }
 
+   test("amap over selector")  {
+     import gopherApi._
+     val ch1 = makeChannel[Int](10)
+     val ch2 = makeChannel[Int](10)
+     val quit = Promise[Boolean]()
+     val out = select.amap {
+         case x:ch1.read => x*2
+         case x:ch2.read => 
+                         //System.err.println(s"received:${x}")
+                         x*3
+         case q:Boolean if (q==quit.future.read) => 
+                         //System.err.println("received quit")
+                         CurrentFlowTermination.exit(1)
+     }
+     ch1.awriteAll(1 to 10)
+     ch2.awriteAll(100 to 110)
+     val f = out.afold(0){ 
+               case (s,x) => //System.err.println(s"in afold ${x}")
+                             s+x }
+     Thread.sleep(1000)
+     quit success true
+     val x = Await.result(f, 10 seconds)
+     assert(x > 3000)
+   }
+
    test("generic channel make")  {
      val ch1 = gopherApi.make[Channel[Int]]()
      val ch2 = gopherApi.make[Channel[Int]](1)
@@ -271,6 +296,35 @@ class MacroSelectSuite extends FunSuite
      assert(x==1)
    }
 
+   test("input afold") {
+     import gopherApi._
+     val ch1 = makeChannel[Int]()
+     ch1.awriteAll(1 to 10) map { _ => ch1.close() }
+     val f = ch1.afold(0){ case (s,x) => s+x }
+     val x = Await.result(f, 10 seconds)
+     assert(x==55)
+   }
+
+   test("map over selector")  {
+     import gopherApi._
+     val ch1 = gopherApi.make[Channel[Int]]()
+     val ch2 = gopherApi.make[Channel[Int]](1)
+     val f1 = ch1.awrite(1)
+     val f2 = ch2.awrite(2)
+     val chs = for(s <- select) yield {
+                s match {
+                  case x:ch1.read => x*3
+                  case x:ch2.read => x*5
+                }
+              }
+     val fs1 = chs.aread
+     val fs2 = chs.aread
+     val s1 = Await.result(fs1, 1 second)
+     val s2 = Await.result(fs2, 1 second)
+     assert(s1==3 || s1==10)
+   }
+
+
    lazy val gopherApi = CommonTestObjects.gopherApi
 
 }