lock dispatching (again) whilst adding a consumer

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

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

@@ -129,9 +129,8 @@ public abstract class PrefetchSubscription extends AbstractSubscription {
         synchronized (pendingLock) {
             enqueueCounter++;
             pending.addMessageLast(node);    
-            dispatchPending();
         }
-       
+        dispatchPending();
     }
 
     public void processMessageDispatchNotification(MessageDispatchNotification mdn) throws Exception {

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

@@ -22,6 +22,7 @@ import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
 
 import javax.jms.InvalidSelectorException;
 import javax.jms.JMSException;
@@ -83,6 +84,7 @@ public class Queue extends BaseDestination implements Task {
     private final Object sendLock = new Object();
     private final TaskRunner taskRunner;    
     private final LinkedList<Runnable> messagesWaitingForSpace = new LinkedList<Runnable>();
+    private final ReentrantLock dispatchLock = new ReentrantLock();
     private final Runnable sendMessagesWaitingForSpaceTask = new Runnable() {
         public void run() {
             wakeup();
@@ -98,7 +100,6 @@ public class Queue extends BaseDestination implements Task {
         } else {
             this.messages = new StoreQueueCursor(broker,this);
         }
-
         this.taskRunner = taskFactory.createTaskRunner(this, "Queue  " + destination.getPhysicalName());
         this.log = LogFactory.getLog(getClass().getName() + "." + destination.getPhysicalName());
     }
@@ -173,6 +174,8 @@ public class Queue extends BaseDestination implements Task {
     }
 
     public void addSubscription(ConnectionContext context, Subscription sub) throws Exception {
+        dispatchLock.lock();
+        try {
             sub.add(context, this);
             destinationStatistics.getConsumers().increment();
             MessageEvaluationContext msgContext = new MessageEvaluationContext();
@@ -194,9 +197,10 @@ public class Queue extends BaseDestination implements Task {
                 }
             }
 
-        // we hold the lock on the dispatchValue - so lets build the paged in
+            // we hold the lock on the dispatchValue - so lets build the paged
+            // in
             // list directly;
-        buildList(false);
+            doPageIn(false);
 
             // synchronize with dispatch method so that no new messages are sent
             // while
@@ -208,9 +212,10 @@ public class Queue extends BaseDestination implements Task {
             synchronized (pagedInMessages) {
                 // Add all the matching messages in the queue to the
                 // subscription.
-            for (Iterator<MessageReference> i = pagedInMessages.values().iterator(); i
+                for (Iterator<MessageReference> i = pagedInMessages.values()
-                    .hasNext();) {
+                        .iterator(); i.hasNext();) {
-                QueueMessageReference node = (QueueMessageReference) i.next();
+                    QueueMessageReference node = (QueueMessageReference) i
+                            .next();
                     if (node.isDropped()
                             || (!sub.getConsumerInfo().isBrowser() && node
                                     .getLockOwner() != null)) {
@@ -226,7 +231,9 @@ public class Queue extends BaseDestination implements Task {
                     }
                 }
             }
-
+        } finally {
+            dispatchLock.unlock();
+        }
     }
 
     public void removeSubscription(ConnectionContext context, Subscription sub)
@@ -956,25 +963,19 @@ public class Queue extends BaseDestination implements Task {
         wakeup();
     }
     
-    final synchronized void wakeup() {
+    final void wakeup() {
         try {
             taskRunner.wakeup();
-            
         } catch (InterruptedException e) {
             log.warn("Task Runner failed to wakeup ", e);
         }
     }
 
-    
     private List<MessageReference> doPageIn(boolean force) throws Exception {
         List<MessageReference> result = null;
-    		result  = buildList(force);
+        dispatchLock.lock();
-    	return  result;
+        try {
-    }
-
-    private List<MessageReference> buildList(boolean force) throws Exception {
             final int toPageIn = getMaxPageSize() - pagedInMessages.size();
-        List<MessageReference> result = null;
             if ((force || !consumers.isEmpty()) && toPageIn > 0) {
                 messages.setMaxBatchSize(toPageIn);
                 int count = 0;
@@ -1005,6 +1006,9 @@ public class Queue extends BaseDestination implements Task {
                     }
                 }
             }
+        }finally {
+            dispatchLock.unlock();
+        }
         return result;
     }