svn commit: r1405940 - /activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Sink.scala

[ch...@apache.org]

Author: chirino
Date: Mon Nov  5 20:32:11 2012
New Revision: 1405940

URL: http://svn.apache.org/viewvc?rev=1405940&view=rev
Log:
Added a FullSink (a sink which is always full) and you can now poll sink items out of a SessionMux

Modified:
    activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Sink.scala

Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Sink.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Sink.scala?rev=1405940&r1=1405939&r2=1405940&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Sink.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Sink.scala Mon Nov  5 20:32:11 2012
@@ -74,6 +74,13 @@ abstract class Sink[T] {
   }
 }
 
+case class FullSink[T]() extends Sink[T] {
+  def refiller:Task = null
+  def refiller_=(value:Task) = {}
+  def full = true
+  def offer(value: T) = false
+}
+
 case class BlackHoleSink[T]() extends Sink[T] {
   var refiller:Task = null
   def full = false
@@ -427,26 +434,35 @@ class SessionSinkMux[T](val downstream:S
 
   def drain_overflow:Unit = {
     while( !overflowed_sessions.isEmpty) {
-      val session = overflowed_sessions.getHead.session
       if( !downstream.full ) {
-        val value = session.overflow.removeFirst()
-        val accepted = downstream.offer((session, value))
+        val accepted = downstream.offer(poll)
         assert(accepted)
-        if( session.stall_counter > 0 ) {
-          schedual_rebalance
-        }
-        if( session.overflow.isEmpty ) {
-          session.overflow_node.unlink()
-        } else {
-          // to fairly consume values from all sessions.
-          overflowed_sessions.rotate()
-        }
       } else {
         return
       }
     }
   }
 
+  def poll:(Session[T], T) = {
+    consumer_queue.assertExecuting()
+    if(overflowed_sessions.isEmpty) {
+      null
+    } else {
+      val session = overflowed_sessions.getHead.session
+      val value = session.overflow.removeFirst()
+      if( session.stall_counter > 0 ) {
+        schedual_rebalance
+      }
+      if( session.overflow.isEmpty ) {
+        session.overflow_node.unlink()
+      } else {
+        // to fairly consume values from all sessions.
+        overflowed_sessions.rotate()
+      }
+      (session, value)
+    }
+  }
+
   def delivered(session:Session[Delivery], size:Int) = {
     session.credit(1, size)
   }