prepare infrastructure for use channels as part of state in select fold.

Author: rssh

build.sbt

@@ -18,7 +18,7 @@ libraryDependencies += "org.scala-lang.modules" %% "scala-async" % "0.9.6-RC2"
 
 libraryDependencies += "org.scalatest" %% "scalatest" % "2.2.6" % "test"
 
-libraryDependencies += "com.typesafe.akka" %% "akka-actor" % "2.4.7"
+libraryDependencies += "com.typesafe.akka" %% "akka-actor" % "2.4.8"
 
 //TODO: enable after 1.0
 //libraryDependencies += "com.typesafe.akka" %% "akka-stream-experimental" % "0.9"

src/main/scala/gopher/channels/Channel.scala

@@ -14,8 +14,34 @@ import scala.reflect.api._
 trait Channel[A] extends InputOutput[A]
 {
 
+   thisChannel =>
+
    def close(): Unit
 
+   // override some operations
+   
+   override def filter(p:A=>Boolean): Channel[A] =
+   {
+    val filteredInput = super.filter(p)
+    new Channel[A] {
+         def cbread[B](f:
+              ContRead[A,B]=>Option[
+                    ContRead.In[A]=>Future[Continuated[B]]
+            ],
+            ft: FlowTermination[B]): Unit = filteredInput.cbread(f,ft)
+
+         def  cbwrite[B](f: ContWrite[A,B] => Option[
+                   (A,Future[Continuated[B]])
+                  ],
+                  ft: FlowTermination[B]): Unit =
+            thisChannel.cbwrite(f,ft)  // TODO: optimize by filteredOutput.cbwrite() ?
+
+        def api = thisChannel.api
+
+        def close() = thisChannel.close()
+    }
+   }
+
 }
 
 object Channel

src/main/scala/gopher/channels/Continuated.scala

@@ -80,6 +80,10 @@ object ContRead
                                             Option[In[A] => Future[Continuated[B]]] =
          prev.function(prev) map (f1 => liftInValue(prev) { v => fn(v,f1) } )
 
+   type Aux[A,B] = ContRead[A,B]{ type El=A
+                    type S=B
+                    type F = ContRead[A,B]=>Option[ContRead.In[A]=>Future[Continuated[B]]] 
+                   }
 }
 
 

src/main/scala/gopher/channels/EffectedChannel.scala

@@ -41,6 +41,8 @@ class SinglethreadedEffectedChannel[A](ch:Channel[A]) extends SinglethreadedEffe
 
   def api: GopherAPI = v.api
 
+  override def filter(p:A=>Boolean):Channel[A] = new SinglethreadedEffectedChannel(v.filter(p))
+
 }
 
 class MultithreadedEffectedChannel[A](ch:Channel[A]) extends MultithreadedEffected[Channel[A]](ch)

src/main/scala/gopher/channels/EffectedInput.scala

@@ -4,7 +4,20 @@ import gopher._
 import gopher.util._
 import scala.concurrent._
 
+import scala.collection.mutable.{HashSet => MutableHashSet}
+import java.util.concurrent.{ConcurrentHashMap=>JavaConcurrentHashMap}
+
 trait EffectedInput[A] extends Input[A] with Effected[Input[A]]
+{
+
+  def cbread[B](f: ContRead[A,B] => Option[ContRead.In[A] => Future[Continuated[B]]],ft: FlowTermination[B]): Unit = {
+    val sv = current
+    sv.cbread((cr:ContRead[A,B]) => if (sv==current) f(cr.copy(channel=this)) else None, ft)
+  }
+
+  def api: GopherAPI = current.api
+
+}
 
 
 object EffectedInput
@@ -18,20 +31,6 @@ object EffectedInput
 
 class SinglethreadedEffectedInput[A](in:Input[A]) extends SinglethreadedEffected[Input[A]](in) 
                                                           with EffectedInput[A]
-{
-
-  def cbread[B](f: ContRead[A,B] => Option[ContRead.In[A] => Future[Continuated[B]]],ft: FlowTermination[B]): Unit = v.cbread(f,ft)
-
-  def api: GopherAPI = v.api
-
-}
 
 class MultithreadedEffectedInput[A](in:Input[A]) extends MultithreadedEffected[Input[A]](in) 
                                                          with EffectedInput[A]
-{
-
-  def cbread[B](f: ContRead[A,B] => Option[ContRead.In[A] => Future[Continuated[B]]],ft: FlowTermination[B]): Unit = v.get().cbread(f,ft)
-
-  def api: GopherAPI = v.get().api
-
-}

src/main/scala/gopher/channels/EffectedOutput.scala

@@ -5,6 +5,19 @@ import gopher.util._
 import scala.concurrent._
 
 trait EffectedOutput[A] extends Effected[Output[A]] with Output[A]
+{
+
+  def  cbwrite[B](f: ContWrite[A,B] => Option[
+                   (A,Future[Continuated[B]])
+                  ],
+                  ft: FlowTermination[B]): Unit = {
+    val sv = current
+    sv.cbwrite[B](cw => if (current eq sv) f(cw.copy(channel=this)) else None,ft)
+  }
+
+  def api: GopherAPI = current.api
+
+}
 
 object EffectedOutput
 {
@@ -17,31 +30,9 @@ object EffectedOutput
 
 class SinglethreadedEffectedOutput[A](out:Output[A]) extends SinglethreadedEffected[Output[A]](out)
                                                               with EffectedOutput[A]
-{
-
-  def  cbwrite[B](f: ContWrite[A,B] => Option[
-                   (A,Future[Continuated[B]])
-                  ],
-                  ft: FlowTermination[B]): Unit = v.cbwrite(f,ft)
-
-
-  def api: GopherAPI = v.api
-
-}
 
 class MultithreadedEffectedOutput[A](out:Output[A]) extends MultithreadedEffected[Output[A]](out)
                                                               with EffectedOutput[A]
-{
-
-  def  cbwrite[B](f: ContWrite[A,B] => Option[
-                   (A,Future[Continuated[B]])
-                  ],
-                  ft: FlowTermination[B]): Unit = v.get().cbwrite(f,ft)
-
-
-  def api: GopherAPI = v.get().api
-
-}