re-introduce optimize dispatch for queues

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@634588 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java

@@ -78,6 +78,7 @@ import org.apache.commons.logging.LogFactory;
  */
 public class Queue extends BaseDestination implements Task {
     protected final Log log;
+    protected TaskRunnerFactory taskFactory;
     protected TaskRunner taskRunner;    
     protected final List<Subscription> consumers = new ArrayList<Subscription>(50);
     protected PendingMessageCursor messages;
@@ -93,6 +94,7 @@ public class Queue extends BaseDestination implements Task {
     private boolean useConsumerPriority=true;
     private boolean strictOrderDispatch=false;
     private QueueDispatchSelector  dispatchSelector;
+    private boolean optimizedDispatch=false;
     private final Runnable sendMessagesWaitingForSpaceTask = new Runnable() {
         public void run() {
             wakeup();
@@ -109,11 +111,9 @@ public class Queue extends BaseDestination implements Task {
     public Queue(BrokerService brokerService, final ActiveMQDestination destination, MessageStore store,DestinationStatistics parentStats,
                  TaskRunnerFactory taskFactory) throws Exception {
         super(brokerService, store, destination, parentStats);
-        
+        this.taskFactory=taskFactory;       
-        
         this.log = LogFactory.getLog(getClass().getName() + "." + destination.getPhysicalName());
         this.dispatchSelector=new QueueDispatchSelector(destination);
-       
     }
         
     public void initialize() throws Exception {
@@ -133,7 +133,9 @@ public class Queue extends BaseDestination implements Task {
             memoryUsage.setParent(systemUsage.getMemoryUsage());
         }
         
-       
+        if (isOptimizedDispatch()) {
+            this.taskRunner = taskFactory.createTaskRunner(this, "TempQueue:  " + destination.getPhysicalName());
+        }else {
             this.executor =  Executors.newSingleThreadExecutor(new ThreadFactory() {
                 public Thread newThread(Runnable runnable) {
                     Thread thread = new Thread(runnable, "QueueThread:"+destination);
@@ -144,6 +146,7 @@ public class Queue extends BaseDestination implements Task {
             });
                
             this.taskRunner = new DeterministicTaskRunner(this.executor,this);
+        }
         super.initialize();
         if (store != null) {
             // Restore the persistent messages.
@@ -584,6 +587,15 @@ public class Queue extends BaseDestination implements Task {
         this.strictOrderDispatch = strictOrderDispatch;
     }
     
+
+    public boolean isOptimizedDispatch() {
+        return optimizedDispatch;
+    }
+
+    public void setOptimizedDispatch(boolean optimizedDispatch) {
+        this.optimizedDispatch = optimizedDispatch;
+    }
+
     // Implementation methods
     // -------------------------------------------------------------------------
     private QueueMessageReference createMessageReference(Message message) {
@@ -956,12 +968,16 @@ public class Queue extends BaseDestination implements Task {
     }
     
     protected void wakeup() {
+        if (optimizedDispatch) {
+            iterate();
+        }else {
             try {
                 taskRunner.wakeup();
             } catch (InterruptedException e) {
                 log.warn("Task Runner failed to wakeup ", e);
             }
         }
+    }
 
     private List<QueueMessageReference> doPageIn(boolean force) throws Exception {
         List<QueueMessageReference> result = null;
@@ -1075,4 +1091,5 @@ public class Queue extends BaseDestination implements Task {
     private void removeFromConsumerList(Subscription sub) {
         consumers.remove(sub);
     }
+
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/TempQueue.java

@@ -32,7 +32,7 @@ import org.apache.activemq.thread.TaskRunnerFactory;
  */
 public class TempQueue extends Queue{
     private final ActiveMQTempDestination tempDest;
-    private TaskRunnerFactory taskFactory;
+   
     
     /**
      * @param brokerService
@@ -48,7 +48,6 @@ public class TempQueue extends Queue{
             throws Exception {
         super(brokerService, destination, store, parentStats, taskFactory);
         this.tempDest = (ActiveMQTempDestination) destination;
-        this.taskFactory=taskFactory;
     }
     
     public void initialize() throws Exception {

activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java

@@ -86,6 +86,7 @@ public class PolicyEntry extends DestinationMapEntry {
         queue.setMinimumMessageSize((int) getMinimumMessageSize());
         queue.setUseConsumerPriority(isUseConsumerPriority());
         queue.setStrictOrderDispatch(isStrictOrderDispatch());
+        queue.setOptimizedDispatch(isOptimizedDispatch());
     }
 
     public void configure(Topic topic) {