refactor to avoid code duplication between buffered and unbuffered channels.

rssh · rssh · commit 0d8d203d6cd6 · 2016-03-09T00:42:11.000+02:00
diff --git a/src/main/scala/gopher/channels/BufferedChannelActor.scala b/src/main/scala/gopher/channels/BufferedChannelActor.scala
@@ -10,14 +10,14 @@ import gopher._
 /**
  * ChannelActor - actor, which leave
  */
-class BufferedChannelActor[A](id:Long, capacity:Int, api: GopherAPI) extends Actor
+class BufferedChannelActor[A](id:Long, capacity:Int, api: GopherAPI) extends ChannelActor[A](id,api)
 {
 
-  def receive = {
-    case cw@ContWrite(_,_,ft) =>
-            val cwa = cw.asInstanceOf[ContWrite[A,_]]
+
+  protected[this] def onContWrite(cwa: gopher.channels.ContWrite[A, _]): Unit = 
+  {
             if (closed) {
-               ft.throwIfNotCompleted(new ChannelClosedException())
+               cwa.flowTermination.throwIfNotCompleted(new ChannelClosedException())
             } else {
               if (nElements==capacity) {
                writers = writers :+ cwa
@@ -28,8 +28,10 @@ class BufferedChannelActor[A](id:Long, capacity:Int, api: GopherAPI) extends Act
                }
               }
             }
-    case cr@ContRead(_,_,ft) =>
-            val cra = cr.asInstanceOf[ContRead[A,_]]
+  }
+
+  protected[this] def onContRead(cra: gopher.channels.ContRead[A, _]): Unit =
+  {
             if (nElements==0) {
                if (closed) {
                  processReaderClosed(cra)
@@ -46,22 +48,11 @@ class BufferedChannelActor[A](id:Long, capacity:Int, api: GopherAPI) extends Act
                  }
                }
             }
-     case ccr@ClosedChannelRead(_) =>
-            self ! ccr.cont
-            sender ! ChannelCloseProcessed(nElements)
-     case ChannelClose =>
-            closed=true
-            stopIfEmpty
-     case ChannelRefDecrement =>
-            nRefs -= 1
-            if (nRefs == 0) {
-               stopAll
-            }
-     case ChannelRefIncrement =>
-            nRefs += 1
-     case GracefullChannelStop =>
-            context.stop(self)
-  }
+   }
+
+
+  protected[this] def getNElements(): Int = nElements;
+
 
   def processReaders() : Boolean =
   {
@@ -123,27 +114,13 @@ class BufferedChannelActor[A](id:Long, capacity:Int, api: GopherAPI) extends Act
    }
 
 
-  private[this] def stopIfEmpty: Boolean =
+  def stopIfEmpty: Boolean =
   {
    require(closed==true)
    if (nElements == 0) {
-      while(!readers.isEmpty) {
-        val reader = readers.head
-        val c = reader.asInstanceOf[ContRead[A,reader.R]]
-        readers = readers.tail
-        c.function(c) foreach { f1 =>
-            api.continue(f1(ContRead.ChannelClosed), c.flowTermination)
-        }
-      }
+      stopReaders()
    } 
-   while(!writers.isEmpty) {
-      val writer = writers.head
-      val c = writer.asInstanceOf[ContWrite[A,writer.R]]
-      writers = writers.tail
-      c.function(c) foreach {
-         f1 => c.flowTermination.throwIfNotCompleted(new ChannelClosedException())
-      }
-   }
+   stopWriters()
    if (nElements == 0) {
       if (nRefs == 0) {
         // here we leave 'closed' channels in actor-system untile they will be
@@ -155,19 +132,6 @@ class BufferedChannelActor[A](id:Long, capacity:Int, api: GopherAPI) extends Act
       false
   }
 
-  def stopAll: Unit =
-  {
-    if (!closed) {
-       closed=true
-    } 
-    if (!stopIfEmpty) {
-       // stop anyway
-       self ! GracefullChannelStop
-    }
-  }
-
-  private[this] implicit def ec: ExecutionContext = api.executionContext
-
   @inline
   private[this] def elementAt(i:Int): A =
     buffer(i).asInstanceOf[A]
@@ -182,10 +146,5 @@ class BufferedChannelActor[A](id:Long, capacity:Int, api: GopherAPI) extends Act
   var readIndex=0
   var writeIndex=0
   var nElements=0
-  var closed=false
-  var readers = Queue[ContRead[A,_]] ()
-  var writers = Queue[ContWrite[A,_]] ()
-
-  var nRefs = 1
 
 }
diff --git a/src/main/scala/gopher/channels/ChannelActor.scala b/src/main/scala/gopher/channels/ChannelActor.scala
@@ -0,0 +1,97 @@
+package gopher.channels
+
+import akka.actor._
+import scala.language._
+import scala.concurrent._
+import scala.collection.immutable._
+import gopher._
+
+
+/**
+ * ChannelActor - actor, which leave
+ */
+abstract class ChannelActor[A](id:Long, api: GopherAPI) extends Actor
+{
+
+  def receive = {
+    case cw@ContWrite(_,_,ft) => 
+            val cwa = cw.asInstanceOf[ContWrite[A,cw.R]]
+            onContWrite(cwa)
+    case cr@ContRead(_,_,ft) =>
+            val cra = cr.asInstanceOf[ContRead[A,cr.R]]
+            onContRead(cra)
+    case ccr@ClosedChannelRead(_) =>
+            self ! ccr.cont
+            sender ! ChannelCloseProcessed(getNElements())
+    case ChannelClose =>
+            closed=true
+            stopIfEmpty
+    case ChannelRefDecrement =>
+            nRefs -= 1
+            if (nRefs == 0) {
+               stopAll
+            }
+    case ChannelRefIncrement =>
+            nRefs += 1
+    case GracefullChannelStop =>
+            context.stop(self)
+  }
+
+  protected[this] def onContWrite(cw:ContWrite[A,_]):Unit
+
+  protected[this] def onContRead(cr:ContRead[A,_]):Unit
+
+  protected[this] def getNElements():Int
+
+  protected[this] def processReaderClosed[B](reader:ContRead[A,B]): Boolean =
+   reader.function(reader) match {
+       case Some(f1) => api.continue(f1(ContRead.ChannelClosed), reader.flowTermination)
+                        true
+       case None => false
+   }
+
+  protected[this] def stopReaders(): Unit =
+  {
+      while(!readers.isEmpty) {
+        val reader = readers.head
+        val c = reader.asInstanceOf[ContRead[A,reader.R]]
+        readers = readers.tail
+        c.function(c) foreach { f1 =>
+            api.continue(f1(ContRead.ChannelClosed), c.flowTermination)
+        }
+      }
+  }
+
+  protected[this] def stopWriters(): Unit =
+  {
+   while(!writers.isEmpty) {
+      val writer = writers.head
+      val c = writer.asInstanceOf[ContWrite[A,writer.R]]
+      writers = writers.tail
+      c.function(c) foreach {
+         f1 => c.flowTermination.throwIfNotCompleted(new ChannelClosedException())
+      }
+   }
+  }
+
+  def stopIfEmpty: Boolean
+
+  def stopAll: Unit =
+  {
+    if (!closed) {
+       closed=true
+    } 
+    if (!stopIfEmpty) {
+       // stop anyway
+       self ! GracefullChannelStop
+    }
+  }
+
+  protected[this] implicit def ec: ExecutionContext = api.executionContext
+
+  protected[this] var closed=false
+  var readers = Queue[ContRead[A,_]] ()
+  var writers = Queue[ContWrite[A,_]] ()
+  var nRefs = 1
+
+}
diff --git a/src/main/scala/gopher/channels/Continuated.scala b/src/main/scala/gopher/channels/Continuated.scala
@@ -91,6 +91,11 @@ case class ContWrite[A,B](function: ContWrite[A,B] => Option[(A,Future[Continuat
   type El = A
 }
 
+object ContWrite
+{
+   
+}
+
 /**
  * skip (i.e. do some operation not related to reading or writing.)
  */
diff --git a/src/main/scala/gopher/channels/UnbufferedChannelActor.scala b/src/main/scala/gopher/channels/UnbufferedChannelActor.scala
@@ -10,40 +10,26 @@ import gopher._
 /**
  * ChannelActor - actor, which leave
  */
-class UnbufferedChannelActor[A](id:Long, api: GopherAPI) extends Actor
+class UnbufferedChannelActor[A](id:Long, api: GopherAPI) extends ChannelActor[A](id,api)
 {
 
-  def receive = {
-    case cw@ContWrite(_,_,ft) =>
-            val cwa = cw.asInstanceOf[ContWrite[A,_]]
-            if (closed) {
-               ft.throwIfNotCompleted(new ChannelClosedException())
-            } else if (!processReaders(cwa))  {
-               writers = writers :+ cwa
-            }
-    case cr@ContRead(_,_,ft) =>
-            val cra = cr.asInstanceOf[ContRead[A,_]]
+  protected[this] def onContWrite(cw:ContWrite[A,_]):Unit =
+         if (closed) {
+               cw.flowTermination.throwIfNotCompleted(new ChannelClosedException())
+         } else if (!processReaders(cw))  {
+               writers = writers :+ cw
+         }
+         
+
+  protected[this] def onContRead(cr:ContRead[A,_]):Unit =
             if (closed) {
-               processReaderClosed(cra)
-            } else if (!processWriters(cra)) {
-               readers = readers :+ cra;         
+               processReaderClosed(cr)
+            } else if (!processWriters(cr)) {
+               readers = readers :+ cr;         
             }
-     case ccr@ClosedChannelRead(_) =>
-            self ! ccr.cont
-            sender ! ChannelCloseProcessed(0)
-     case ChannelClose =>
-            closed=true
-            stopIfEmpty
-     case ChannelRefDecrement =>
-            nRefs -= 1
-            if (nRefs == 0) {
-               stopAll
-            }
-     case ChannelRefIncrement =>
-            nRefs += 1
-     case GracefullChannelStop =>
-            context.stop(self)
-  }
+
+  protected[this] def getNElements():Int = 0
+
 
   def processReaders(w: ContWrite[A,_]) : Boolean =
   {
@@ -76,13 +62,6 @@ class UnbufferedChannelActor[A](id:Long, api: GopherAPI) extends Actor
               false
    }
 
-  private[this] def processReaderClosed[B](reader:ContRead[A,B]): Boolean =
-   reader.function(reader) match {
-       case Some(f1) => api.continue(f1(ContRead.ChannelClosed), reader.flowTermination)
-                        true
-       case None => false
-   }
-
   def processWriters[C](reader:ContRead[A,C]): Boolean =
   {
     if (writers.isEmpty) {
@@ -121,25 +100,11 @@ class UnbufferedChannelActor[A](id:Long, api: GopherAPI) extends Actor
    }
 
 
-  private[this] def stopIfEmpty: Boolean =
+  def stopIfEmpty: Boolean =
   {
    require(closed==true)
-   while(!readers.isEmpty) {
-      val reader = readers.head
-      val c = reader.asInstanceOf[ContRead[A,reader.R]]
-      readers = readers.tail
-      c.function(c) foreach { f1 =>
-            api.continue(f1(ContRead.ChannelClosed), c.flowTermination)
-      }
-   }
-   while(!writers.isEmpty) {
-      val writer = writers.head
-      val c = writer.asInstanceOf[ContWrite[A,writer.R]]
-      writers = writers.tail
-      c.function(c) foreach {
-         f1 => c.flowTermination.throwIfNotCompleted(new ChannelClosedException())
-      }
-   }
+   stopReaders()
+   stopWriters()
       if (nRefs == 0) {
         // here we leave 'closed' channels in actor-system untile they will be
         // garbage-collected.  TODO: think about actual stop ?
@@ -148,22 +113,8 @@ class UnbufferedChannelActor[A](id:Long, api: GopherAPI) extends Actor
       true
   }
 
-  def stopAll: Unit =
-  {
-    if (!closed) {
-       closed=true
-    } 
-    if (!stopIfEmpty) {
-       // stop anyway
-       self ! GracefullChannelStop
-    }
-  }
 
   private[this] implicit def ec: ExecutionContext = api.executionContext
 
-  var closed=false
-  var readers = Queue[ContRead[A,_]] ()
-  var writers = Queue[ContWrite[A,_]] ()
-  var nRefs = 1
 
 }
