add map for selector

Author: rssh

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

@@ -65,6 +65,8 @@ trait ForeverSelectorBuilder extends SelectorBuilder[Unit]
    def foreach(f:Any=>Unit):Unit = 
         macro SelectorBuilder.foreachImpl[Unit]
 
+   
+
    /**
     * provide syntax for running select loop as async operation.
     *
@@ -77,6 +79,25 @@ trait ForeverSelectorBuilder extends SelectorBuilder[Unit]
    def apply(f: PartialFunction[Any,Unit]): Future[Unit] =
         macro SelectorBuilder.applyImpl[Unit]
 
+
+   def inputBuilder[B]() = new InputSelectorBuilder[B](api) 
+
+   /**
+    * provide syntax for creating output channels.
+    *{{{
+    *
+    *  val multiplexed = for(s <- gopherApi.select.forever) yield 
+    *       s match {
+    *          case x: channelA => s"A:${x}"
+    *          case x: channelB => s"B:${x}"
+    *       }
+    *
+    *}}}
+    **/
+    def map[B](f:Any=>B):Input[B] = macro SelectorBuilder.mapImpl[B]
+ 
+    def input[B](f:PartialFunction[Any,B]):Input[B] = ???
+
 }
 
 

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

@@ -0,0 +1,73 @@
+package gopher.channels
+
+import scala.language.experimental.macros
+import scala.reflect.macros.whitebox.Context
+import scala.reflect.api._
+import gopher._
+import gopher.util._
+import scala.concurrent._
+import scala.annotation.unchecked._
+
+
+/**
+ * Builder for 'input' selector. Can be obtained as `gopherApi.select.input`.
+ * or map of forever selector.
+ *
+ * 
+ */
+class InputSelectorBuilder[T](override val api: GopherAPI) extends SelectorBuilder[T@uncheckedVariance]
+                                                 with Input[T]
+{
+
+   val proxy = api.makeChannel[T]()
+
+   def reading[A](ch: Input[A])(f: A=>T): InputSelectorBuilder[T] =
+        macro SelectorBuilder.readingImpl[A,T,InputSelectorBuilder[T]] 
+
+   @inline
+   def readingWithFlowTerminationAsync[A](ch: Input[A], f: (ExecutionContext, FlowTermination[T], A) => Future[T] ): InputSelectorBuilder[T] =
+   {
+      def normalized(_cont:ContRead[A,T]):Option[ContRead.In[A]=>Future[Continuated[T]]] =
+              Some(ContRead.liftIn(_cont)(a=>f(ec,selector,a) flatMap {
+                    proxy.awrite(_)
+                  } map Function.const(ContRead(normalized,ch,selector))))
+      withReader[A](ch,normalized)
+   }
+
+   /**
+    * write x to channel if possible
+    */
+   def writing[A](ch: Output[A], x: A)(f: A=>T): InputSelectorBuilder[T] = 
+        macro SelectorBuilder.writingImpl[A,T,InputSelectorBuilder[T]]
+ 
+   @inline
+   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) flatMap {
+                                  x=>proxy.awrite(x)
+                          } map Function.const(cw)) })
+
+   def idle(body: T): InputSelectorBuilder[T] = 
+        macro SelectorBuilder.idleImpl[T,InputSelectorBuilder[T]]
+
+   @inline
+   def idleWithFlowTerminationAsync(f: (ExecutionContext, FlowTermination[T]) => Future[T] ): this.type =
+       withIdle{ sk => Some(f(ec,sk.flowTermination) flatMap(x => 
+                            proxy.awrite(x)) map(Function.const(sk)) ) }
+
+   def foreach(f:Any=>T):T = 
+        macro SelectorBuilder.foreachImpl[T]
+
+   def apply(f: PartialFunction[Any,T]): Future[T] =
+        macro SelectorBuilder.applyImpl[T]
+
+   // input methods
+   def  cbread[B](f:
+            ContRead[T,B]=>Option[
+                    ContRead.In[T]=>Future[Continuated[B]]
+            ],
+            ft: FlowTermination[B]): Unit = proxy.cbread(f,ft)
+  
+
+}
+
+

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

@@ -423,6 +423,40 @@ object SelectorBuilder
     q"${builderName}.idle(${caseDef.body})"
    }
 
+   def mapImpl[T](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)
+       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))
+   }
+
+   def mapBuildMatch(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}"""
+   }
+
+   def inputImpl[T](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)
+        case _ => c.abort(f.tree.pos,"expected partial function with syntax case ... =>, have ${MacroUtil.shortString(f.tree)}");
+     }
+     c.Expr[Input[T]](c.untypecheck(builder))
+   }
+
 }