svn commit: r1042876 - in /activemq/activemq-apollo/trunk: apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Queue.scala apollo-store/src/main/scala/org/apache/activemq/apollo/store/DelayingStoreSupport.scala

[ch...@apache.org]

Author: chirino
Date: Tue Dec  7 00:25:06 2010
New Revision: 1042876

URL: http://svn.apache.org/viewvc?rev=1042876&view=rev
Log:
fix sync-peristent bug.

Modified:
    activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Queue.scala
    activemq/activemq-apollo/trunk/apollo-store/src/main/scala/org/apache/activemq/apollo/store/DelayingStoreSupport.scala

Modified: activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Queue.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Queue.scala?rev=1042876&r1=1042875&r2=1042876&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Queue.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-broker/src/main/scala/org/apache/activemq/apollo/broker/Queue.scala Tue Dec  7 00:25:06 2010
@@ -122,7 +122,7 @@ class Queue(val host: VirtualHost, var i
     tune_persistent = host.store !=null && config.persistent.getOrElse(true)
     tune_flush_to_store = tune_persistent && config.flush_to_store.getOrElse(true)
     tune_flush_range_size = config.flush_range_size.getOrElse(10000)
-    tune_consumer_buffer = config.consumer_buffer.getOrElse(1024*128)
+    tune_consumer_buffer = config.consumer_buffer.getOrElse(32*1024)
   }
   configure(config)
 
@@ -365,15 +365,17 @@ class Queue(val host: VirtualHost, var i
         // from the entry list.
         val next = cur.getNext
 
-        if( cur.hasSubs || cur.is_prefetched ) {
+        if( cur.prefetch_flags!=0 ) {
           distance_from_sub = 0
         } else {
           distance_from_sub += 1
           if( cur.can_combine_with_prev ) {
             cur.getPrevious.as_flushed_range.combineNext
+            combine_counter += 1
           } else {
             if( cur.is_flushed && distance_from_sub > tune_flush_range_size ) {
               cur.flush_range
+              combine_counter += 1
             }
           }
 

Modified: activemq/activemq-apollo/trunk/apollo-store/src/main/scala/org/apache/activemq/apollo/store/DelayingStoreSupport.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-store/src/main/scala/org/apache/activemq/apollo/store/DelayingStoreSupport.scala?rev=1042876&r1=1042875&r2=1042876&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-store/src/main/scala/org/apache/activemq/apollo/store/DelayingStoreSupport.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-store/src/main/scala/org/apache/activemq/apollo/store/DelayingStoreSupport.scala Tue Dec  7 00:25:06 2010
@@ -167,6 +167,8 @@ trait DelayingStoreSupport extends Store
     if( action == null ) {
       cb
     } else {
+      // TODO: protect against this causing a 2nd flush.
+      delayedUOWs.put(action.uow.uow_id, action.uow)
       action.uow.onComplete(^{ cb })
       flush(action.uow.uow_id)
     }