added EffectedChannel

Author: rssh

src/main/scala/gopher/GopherAPI.scala

@@ -41,7 +41,7 @@ class GopherAPI(as: ActorSystem, es: ExecutionContext)
    *  channel.awrite(1 to 100)
    *}}}
    */
-  def makeChannel[A](capacity: Int = 0) =
+  def makeChannel[A](capacity: Int = 0): Channel[A] =
     {
      require(capacity >= 0)
      val nextId = newChannelId
@@ -54,12 +54,14 @@ class GopherAPI(as: ActorSystem, es: ExecutionContext)
      new ActorBackedChannel[A](futureChannelRef, this)
     }
 
-  def makeEffectedInput[A](in: Input[A], threadingPolicy: ThreadingPolicy = ThreadingPolicy.Single) =
+  def makeEffectedInput[A](in: Input[A], threadingPolicy: ThreadingPolicy = ThreadingPolicy.Single): EffectedInput[A] =
      EffectedInput(in,threadingPolicy)
 
   def makeEffectedOutput[A](out: Output[A], threadingPolicy: ThreadingPolicy = ThreadingPolicy.Single) =
      EffectedOutput(out,threadingPolicy)
 
+  def makeEffectedChannel[A](ch: Channel[A], threadingPolicy: ThreadingPolicy = ThreadingPolicy.Single) =
+     EffectedChannel(ch,threadingPolicy)
 
   /**
    * Represent Scala future as channel from which we can read one value.
@@ -72,6 +74,7 @@ class GopherAPI(as: ActorSystem, es: ExecutionContext)
    */
   def iterableInput[A](iterable:Iterable[A]): Input[A] = Input.asInput(iterable, this)
 
+
   /**
    * create and start instance of transputer with given recovery policy.
    *@see gopher.Transputer

src/main/scala/gopher/channels/ActorBackedChannel.scala

@@ -11,7 +11,7 @@ import scala.language.postfixOps
 import scala.reflect.macros.blackbox.Context
 import scala.reflect.api._
 
-class ActorBackedChannel[A](futureChannelRef: Future[ActorRef], override val api: GopherAPI) extends Channel[A](api)
+class ActorBackedChannel[A](futureChannelRef: Future[ActorRef], override val api: GopherAPI) extends Channel[A]
 {
 
   def  cbread[B](f: ContRead[A,B] => Option[ContRead.In[A] => Future[Continuated[B]]], flwt: FlowTermination[B] ): Unit = 

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

@@ -11,7 +11,7 @@ import scala.language.postfixOps
 import scala.reflect.macros.blackbox.Context
 import scala.reflect.api._
 
-abstract class Channel[A](override val api: GopherAPI) extends InputOutput[A]
+trait Channel[A] extends InputOutput[A]
 {
 
 

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

@@ -0,0 +1,68 @@
+package gopher.channels
+
+
+import gopher._
+import gopher.util._
+import scala.concurrent._
+
+trait EffectedChannel[A] extends Channel[A] with Effected[Channel[A]]
+{
+   def asInput(): EffectedInput[A]
+   def asOutput(): EffectedOutput[A]
+}
+
+
+object EffectedChannel
+{
+   def apply[A](in: Channel[A], policy: ThreadingPolicy): EffectedChannel[A] =
+     policy match {
+       case ThreadingPolicy.Single => new SinglethreadedEffectedChannel(in)
+       case ThreadingPolicy.Multi => new MultithreadedEffectedChannel(in)
+     }
+}
+
+
+class SinglethreadedEffectedChannel[A](ch:Channel[A]) extends SinglethreadedEffected[Channel[A]](ch)
+                                                          with EffectedChannel[A]
+{
+
+  def cbread[B](f: ContRead[A,B] => Option[ContRead.In[A] => Future[Continuated[B]]],ft: FlowTermination[B]): Unit = v.cbread(f,ft)
+
+  def  cbwrite[B](f: ContWrite[A,B] => Option[
+                   (A,Future[Continuated[B]])
+                  ],
+                  ft: FlowTermination[B]): Unit = v.cbwrite(f,ft)
+
+  def close() = v.close()
+
+  def asInput() = api.makeEffectedInput(v, ThreadingPolicy.Single)
+
+  def asOutput() = api.makeEffectedOutput(v, ThreadingPolicy.Single)
+
+  def api: GopherAPI = v.api
+
+}
+
+class MultithreadedEffectedChannel[A](ch:Channel[A]) extends MultithreadedEffected[Channel[A]](ch)
+                                                         with EffectedChannel[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  cbwrite[B](f: ContWrite[A,B] => Option[
+                   (A,Future[Continuated[B]])
+                  ],
+                  ft: FlowTermination[B]): Unit = v.get().cbwrite(f,ft)
+
+  def close() = v.get().close()
+
+  def asInput() = api.makeEffectedInput(v.get(), ThreadingPolicy.Multi)
+
+  def asOutput() = api.makeEffectedOutput(v.get(), ThreadingPolicy.Multi)
+
+
+  def api: GopherAPI = v.get().api
+
+}
+
+

src/test/scala/example/BroadcasterSuite.scala

@@ -25,7 +25,7 @@ class Broadcaster[A]
     val sendc:     Channel[A] = makeChannel()
     val quitc:     Channel[Boolean] = makeChannel()
 
-    select.afold(makeChannel[Message[A]]()) { (last,s) =>
+    val process = select.afold(makeChannel[Message[A]]()) { (last,s) =>
          s match {
            case v: sendc.read @unchecked =>
                     val next = makeChannel[Message[A]]()
@@ -40,25 +40,6 @@ class Broadcaster[A]
      }
 
 
-/*
-    go {
-      var last = makeChannel[Message[A]]()
-      for (s <- select.forever) {
-        s match {
-           case v: sendc.read @ unchecked =>
-                val c = makeChannel[Message[A]](1)
-                val b = ValueMessage(c,v)
-                last <~ b
-                last = c
-           case r: listenc.read @ unchecked =>
-                 r <~ last
-           case q: quitc.read =>
-                last write EndMessage
-                implicitly[FlowTermination[Unit]].doExit(())
-        }
-      }
-    }
-*/
 
     def alisten(): Future[Receiver[A]] = go {
       val c = makeChannel[Channel[Message[A]]]()