You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ch...@apache.org on 2013/02/20 16:45:11 UTC
svn commit: r1448257 -
/activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala
Author: chirino
Date: Wed Feb 20 15:45:10 2013
New Revision: 1448257
URL: http://svn.apache.org/r1448257
Log:
Don't cleanup a session route which is currently sending a message.
Modified:
activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala
Modified: activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala?rev=1448257&r1=1448256&r2=1448257&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-stomp/src/main/scala/org/apache/activemq/apollo/stomp/StompProtocolHandler.scala Wed Feb 20 15:45:10 2013
@@ -1169,17 +1169,20 @@ class StompProtocolHandler extends Proto
}
}
+ var routing_items = 0
override def offer(delivery: Delivery): Boolean = {
if( full )
return false
routing_size += delivery.size
+ routing_items += 1
val original_ack = delivery.ack
delivery.ack = (result, uow) => {
dispatch_queue.assertExecuting()
if ( original_ack!=null ) {
original_ack(result, uow)
}
+ routing_items -= 1
routing_size -= delivery.size
if( routing_size==0 && !pending_routing_empty_callbacks.isEmpty) {
val t = pending_routing_empty_callbacks
@@ -1216,7 +1219,7 @@ class StompProtocolHandler extends Proto
import collection.JavaConversions._
val expired = ListBuffer[StompProducerRoute]()
for( route <- producer_routes.values() ) {
- if( (now - route.last_send) > 2000 ) {
+ if( (now - route.last_send) > 2000 && route.routing_items==0 ) {
expired += route
}
}