001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq.broker.region;
018
019import java.io.IOException;
020import java.util.AbstractList;
021import java.util.ArrayList;
022import java.util.Collection;
023import java.util.Collections;
024import java.util.Comparator;
025import java.util.HashSet;
026import java.util.Iterator;
027import java.util.LinkedHashMap;
028import java.util.LinkedList;
029import java.util.List;
030import java.util.Map;
031import java.util.Set;
032import java.util.concurrent.CancellationException;
033import java.util.concurrent.CopyOnWriteArraySet;
034import java.util.concurrent.CountDownLatch;
035import java.util.concurrent.DelayQueue;
036import java.util.concurrent.Delayed;
037import java.util.concurrent.ExecutorService;
038import java.util.concurrent.Future;
039import java.util.concurrent.TimeUnit;
040import java.util.concurrent.atomic.AtomicLong;
041import java.util.concurrent.locks.Lock;
042import java.util.concurrent.locks.ReentrantLock;
043import java.util.concurrent.locks.ReentrantReadWriteLock;
044import javax.jms.InvalidSelectorException;
045import javax.jms.JMSException;
046import javax.jms.ResourceAllocationException;
047import org.apache.activemq.broker.BrokerService;
048import org.apache.activemq.broker.ConnectionContext;
049import org.apache.activemq.broker.ProducerBrokerExchange;
050import org.apache.activemq.broker.region.cursors.PendingMessageCursor;
051import org.apache.activemq.broker.region.cursors.StoreQueueCursor;
052import org.apache.activemq.broker.region.cursors.VMPendingMessageCursor;
053import org.apache.activemq.broker.region.group.MessageGroupHashBucketFactory;
054import org.apache.activemq.broker.region.group.MessageGroupMap;
055import org.apache.activemq.broker.region.group.MessageGroupMapFactory;
056import org.apache.activemq.broker.region.policy.DispatchPolicy;
057import org.apache.activemq.broker.region.policy.RoundRobinDispatchPolicy;
058import org.apache.activemq.command.*;
059import org.apache.activemq.filter.BooleanExpression;
060import org.apache.activemq.filter.MessageEvaluationContext;
061import org.apache.activemq.filter.NonCachedMessageEvaluationContext;
062import org.apache.activemq.security.SecurityContext;
063import org.apache.activemq.selector.SelectorParser;
064import org.apache.activemq.store.MessageRecoveryListener;
065import org.apache.activemq.store.MessageStore;
066import org.apache.activemq.thread.Scheduler;
067import org.apache.activemq.thread.Task;
068import org.apache.activemq.thread.TaskRunner;
069import org.apache.activemq.thread.TaskRunnerFactory;
070import org.apache.activemq.transaction.Synchronization;
071import org.apache.activemq.usage.Usage;
072import org.apache.activemq.usage.UsageListener;
073import org.apache.activemq.util.BrokerSupport;
074import org.slf4j.Logger;
075import org.slf4j.LoggerFactory;
076import org.slf4j.MDC;
077
078/**
079 * The Queue is a List of MessageEntry objects that are dispatched to matching
080 * subscriptions.
081 * 
082 * 
083 */
084public class Queue extends BaseDestination implements Task, UsageListener {
085    protected static final Logger LOG = LoggerFactory.getLogger(Queue.class);
086    protected final TaskRunnerFactory taskFactory;
087    protected TaskRunner taskRunner;
088    private final ReentrantReadWriteLock consumersLock = new ReentrantReadWriteLock();
089    protected final List<Subscription> consumers = new ArrayList<Subscription>(50);
090    private final ReentrantReadWriteLock messagesLock = new ReentrantReadWriteLock();
091    protected PendingMessageCursor messages;
092    private final ReentrantReadWriteLock pagedInMessagesLock = new ReentrantReadWriteLock();
093    private final LinkedHashMap<MessageId, QueueMessageReference> pagedInMessages = new LinkedHashMap<MessageId, QueueMessageReference>();
094    // Messages that are paged in but have not yet been targeted at a
095    // subscription
096    private final ReentrantReadWriteLock pagedInPendingDispatchLock = new ReentrantReadWriteLock();
097    private List<QueueMessageReference> pagedInPendingDispatch = new ArrayList<QueueMessageReference>(100);
098    private List<QueueMessageReference> redeliveredWaitingDispatch = new ArrayList<QueueMessageReference>();
099    private MessageGroupMap messageGroupOwners;
100    private DispatchPolicy dispatchPolicy = new RoundRobinDispatchPolicy();
101    private MessageGroupMapFactory messageGroupMapFactory = new MessageGroupHashBucketFactory();
102    final Lock sendLock = new ReentrantLock();
103    private ExecutorService executor;
104    protected final Map<MessageId, Runnable> messagesWaitingForSpace = Collections
105            .synchronizedMap(new LinkedHashMap<MessageId, Runnable>());
106    private boolean useConsumerPriority = true;
107    private boolean strictOrderDispatch = false;
108    private final QueueDispatchSelector dispatchSelector;
109    private boolean optimizedDispatch = false;
110    private boolean firstConsumer = false;
111    private int timeBeforeDispatchStarts = 0;
112    private int consumersBeforeDispatchStarts = 0;
113    private CountDownLatch consumersBeforeStartsLatch;
114    private final AtomicLong pendingWakeups = new AtomicLong();
115    private boolean allConsumersExclusiveByDefault = false;
116    
117    private final Runnable sendMessagesWaitingForSpaceTask = new Runnable() {
118        public void run() {
119            asyncWakeup();
120        }
121    };
122    private final Runnable expireMessagesTask = new Runnable() {
123        public void run() {
124            expireMessages();
125        }
126    };
127
128    private final Object iteratingMutex = new Object() {
129    };
130    private final Scheduler scheduler;
131
132    class TimeoutMessage implements Delayed {
133
134        Message message;
135        ConnectionContext context;
136        long trigger;
137
138        public TimeoutMessage(Message message, ConnectionContext context, long delay) {
139            this.message = message;
140            this.context = context;
141            this.trigger = System.currentTimeMillis() + delay;
142        }
143
144        public long getDelay(TimeUnit unit) {
145            long n = trigger - System.currentTimeMillis();
146            return unit.convert(n, TimeUnit.MILLISECONDS);
147        }
148
149        public int compareTo(Delayed delayed) {
150            long other = ((TimeoutMessage) delayed).trigger;
151            int returnValue;
152            if (this.trigger < other) {
153                returnValue = -1;
154            } else if (this.trigger > other) {
155                returnValue = 1;
156            } else {
157                returnValue = 0;
158            }
159            return returnValue;
160        }
161
162    }
163
164    DelayQueue<TimeoutMessage> flowControlTimeoutMessages = new DelayQueue<TimeoutMessage>();
165
166    class FlowControlTimeoutTask extends Thread {
167
168        @Override
169        public void run() {
170            TimeoutMessage timeout;
171            try {
172                while (true) {
173                    timeout = flowControlTimeoutMessages.take();
174                    if (timeout != null) {
175                        synchronized (messagesWaitingForSpace) {
176                            if (messagesWaitingForSpace.remove(timeout.message.getMessageId()) != null) {
177                                ExceptionResponse response = new ExceptionResponse(
178                                        new ResourceAllocationException(
179                                                "Usage Manager Memory Limit reached. Stopping producer ("
180                                                        + timeout.message.getProducerId()
181                                                        + ") to prevent flooding "
182                                                        + getActiveMQDestination().getQualifiedName()
183                                                        + "."
184                                                        + " See http://activemq.apache.org/producer-flow-control.html for more info"));
185                                response.setCorrelationId(timeout.message.getCommandId());
186                                timeout.context.getConnection().dispatchAsync(response);
187                            }
188                        }
189                    }
190                }
191            } catch (InterruptedException e) {
192                if (LOG.isDebugEnabled()) {
193                    LOG.debug(getName() + "Producer Flow Control Timeout Task is stopping");
194                }
195            }
196        }
197    };
198
199    private final FlowControlTimeoutTask flowControlTimeoutTask = new FlowControlTimeoutTask();
200
201    private static final Comparator<Subscription> orderedCompare = new Comparator<Subscription>() {
202
203        public int compare(Subscription s1, Subscription s2) {
204            // We want the list sorted in descending order
205            return s2.getConsumerInfo().getPriority() - s1.getConsumerInfo().getPriority();
206        }
207    };
208
209    public Queue(BrokerService brokerService, final ActiveMQDestination destination, MessageStore store,
210            DestinationStatistics parentStats, TaskRunnerFactory taskFactory) throws Exception {
211        super(brokerService, store, destination, parentStats);
212        this.taskFactory = taskFactory;
213        this.dispatchSelector = new QueueDispatchSelector(destination);
214        this.scheduler = brokerService.getBroker().getScheduler();
215    }
216
217    public List<Subscription> getConsumers() {
218        consumersLock.readLock().lock();
219        try {
220            return new ArrayList<Subscription>(consumers);
221        }finally {
222            consumersLock.readLock().unlock();
223        }
224    }
225
226    // make the queue easily visible in the debugger from its task runner
227    // threads
228    final class QueueThread extends Thread {
229        final Queue queue;
230
231        public QueueThread(Runnable runnable, String name, Queue queue) {
232            super(runnable, name);
233            this.queue = queue;
234        }
235    }
236
237    @Override
238    public void initialize() throws Exception {
239        if (this.messages == null) {
240            if (destination.isTemporary() || broker == null || store == null) {
241                this.messages = new VMPendingMessageCursor(isPrioritizedMessages());
242            } else {
243                this.messages = new StoreQueueCursor(broker, this);
244            }
245        }
246        // If a VMPendingMessageCursor don't use the default Producer System
247        // Usage
248        // since it turns into a shared blocking queue which can lead to a
249        // network deadlock.
250        // If we are cursoring to disk..it's not and issue because it does not
251        // block due
252        // to large disk sizes.
253        if (messages instanceof VMPendingMessageCursor) {
254            this.systemUsage = brokerService.getSystemUsage();
255            memoryUsage.setParent(systemUsage.getMemoryUsage());
256        }
257
258        this.taskRunner = taskFactory.createTaskRunner(this, "Queue:" + destination.getPhysicalName());
259
260        super.initialize();
261        if (store != null) {
262            // Restore the persistent messages.
263            messages.setSystemUsage(systemUsage);
264            messages.setEnableAudit(isEnableAudit());
265            messages.setMaxAuditDepth(getMaxAuditDepth());
266            messages.setMaxProducersToAudit(getMaxProducersToAudit());
267            messages.setUseCache(isUseCache());
268            messages.setMemoryUsageHighWaterMark(getCursorMemoryHighWaterMark());
269            if (messages.isRecoveryRequired()) {
270                store.recover(new MessageRecoveryListener() {
271                    double totalMessageCount = store.getMessageCount();
272                    int recoveredMessageCount = 0;
273
274                    public boolean recoverMessage(Message message) {
275                        // Message could have expired while it was being
276                        // loaded..
277                        if ((++recoveredMessageCount % 50000) == 0) {
278                            LOG.info("cursor for " + getActiveMQDestination().getQualifiedName() + " has recovered "
279                                    + recoveredMessageCount + " messages. " +
280                                    (int)(recoveredMessageCount*100/totalMessageCount) + "% complete");
281                        }
282                        if (message.isExpired()) {
283                            if (broker.isExpired(message)) {
284                                messageExpired(createConnectionContext(), createMessageReference(message));
285                                // drop message will decrement so counter
286                                // balance here
287                                destinationStatistics.getMessages().increment();
288                            }
289                            return true;
290                        }
291                        if (hasSpace()) {
292                            message.setRegionDestination(Queue.this);
293                            messagesLock.writeLock().lock();
294                            try{
295                                try {
296                                    messages.addMessageLast(message);
297                                } catch (Exception e) {
298                                    LOG.error("Failed to add message to cursor", e);
299                                }
300                            }finally {
301                                messagesLock.writeLock().unlock();
302                            }
303                            destinationStatistics.getMessages().increment();
304                            return true;
305                        }
306                        return false;
307                    }
308
309                    public boolean recoverMessageReference(MessageId messageReference) throws Exception {
310                        throw new RuntimeException("Should not be called.");
311                    }
312
313                    public boolean hasSpace() {
314                        return true;
315                    }
316
317                    public boolean isDuplicate(MessageId id) {
318                        return false;
319                    }
320                });
321            } else {
322                int messageCount = store.getMessageCount();
323                destinationStatistics.getMessages().setCount(messageCount);
324            }
325        }
326    }
327
328    /*
329     * Holder for subscription that needs attention on next iterate browser
330     * needs access to existing messages in the queue that have already been
331     * dispatched
332     */
333    class BrowserDispatch {
334        QueueBrowserSubscription browser;
335
336        public BrowserDispatch(QueueBrowserSubscription browserSubscription) {
337            browser = browserSubscription;
338            browser.incrementQueueRef();
339        }
340
341        void done() {
342            try {
343                browser.decrementQueueRef();
344            } catch (Exception e) {
345                LOG.warn("decrement ref on browser: " + browser, e);
346            }
347        }
348
349        public QueueBrowserSubscription getBrowser() {
350            return browser;
351        }
352    }
353
354    LinkedList<BrowserDispatch> browserDispatches = new LinkedList<BrowserDispatch>();
355
356    public void addSubscription(ConnectionContext context, Subscription sub) throws Exception {
357        if (LOG.isDebugEnabled()) {
358            LOG.debug(getActiveMQDestination().getQualifiedName() + " add sub: " + sub + ", dequeues: "
359                    + getDestinationStatistics().getDequeues().getCount() + ", dispatched: "
360                    + getDestinationStatistics().getDispatched().getCount() + ", inflight: "
361                    + getDestinationStatistics().getInflight().getCount());
362        }
363
364        super.addSubscription(context, sub);
365        // synchronize with dispatch method so that no new messages are sent
366        // while setting up a subscription. avoid out of order messages,
367        // duplicates, etc.
368        pagedInPendingDispatchLock.writeLock().lock();
369        try {
370
371            sub.add(context, this);
372           
373            // needs to be synchronized - so no contention with dispatching
374           // consumersLock.
375            consumersLock.writeLock().lock();
376            try {
377
378                // set a flag if this is a first consumer
379                if (consumers.size() == 0) {
380                    firstConsumer = true;
381                    if (consumersBeforeDispatchStarts != 0) {
382                        consumersBeforeStartsLatch = new CountDownLatch(consumersBeforeDispatchStarts - 1);
383                    }
384                } else {
385                    if (consumersBeforeStartsLatch != null) {
386                        consumersBeforeStartsLatch.countDown();
387                    }
388                }
389
390                addToConsumerList(sub);
391                if (sub.getConsumerInfo().isExclusive() || isAllConsumersExclusiveByDefault()) {
392                    Subscription exclusiveConsumer = dispatchSelector.getExclusiveConsumer();
393                    if (exclusiveConsumer == null) {
394                        exclusiveConsumer = sub;
395                    } else if (sub.getConsumerInfo().getPriority() == Byte.MAX_VALUE ||
396                        sub.getConsumerInfo().getPriority() > exclusiveConsumer.getConsumerInfo().getPriority()) {
397                        exclusiveConsumer = sub;
398                    }
399                    dispatchSelector.setExclusiveConsumer(exclusiveConsumer);
400                }
401            }finally {
402                consumersLock.writeLock().unlock();
403            }
404
405            if (sub instanceof QueueBrowserSubscription) {
406                // tee up for dispatch in next iterate
407                QueueBrowserSubscription browserSubscription = (QueueBrowserSubscription) sub;
408                pagedInMessagesLock.readLock().lock();
409                try{
410                    BrowserDispatch browserDispatch = new BrowserDispatch(browserSubscription);
411                    browserDispatches.addLast(browserDispatch);
412                }finally {
413                    pagedInMessagesLock.readLock().unlock();
414                }
415            }
416
417            if (!(this.optimizedDispatch || isSlave())) {
418                wakeup();
419            }
420        }finally {
421            pagedInPendingDispatchLock.writeLock().unlock();
422        }
423        if (this.optimizedDispatch || isSlave()) {
424            // Outside of dispatchLock() to maintain the lock hierarchy of
425            // iteratingMutex -> dispatchLock. - see
426            // https://issues.apache.org/activemq/browse/AMQ-1878
427            wakeup();
428        }
429    }
430
431    public void removeSubscription(ConnectionContext context, Subscription sub, long lastDeiveredSequenceId)
432            throws Exception {
433        super.removeSubscription(context, sub, lastDeiveredSequenceId);
434        // synchronize with dispatch method so that no new messages are sent
435        // while removing up a subscription.
436        pagedInPendingDispatchLock.writeLock().lock();
437        try {
438            if (LOG.isDebugEnabled()) {
439                LOG.debug(getActiveMQDestination().getQualifiedName() + " remove sub: " + sub + ", lastDeliveredSeqId: " + lastDeiveredSequenceId + ", dequeues: "
440                        + getDestinationStatistics().getDequeues().getCount() + ", dispatched: "
441                        + getDestinationStatistics().getDispatched().getCount() + ", inflight: "
442                        + getDestinationStatistics().getInflight().getCount());
443            }
444            consumersLock.writeLock().lock();
445            try {
446                removeFromConsumerList(sub);
447                if (sub.getConsumerInfo().isExclusive()) {
448                    Subscription exclusiveConsumer = dispatchSelector.getExclusiveConsumer();
449                    if (exclusiveConsumer == sub) {
450                        exclusiveConsumer = null;
451                        for (Subscription s : consumers) {
452                            if (s.getConsumerInfo().isExclusive()
453                                    && (exclusiveConsumer == null || s.getConsumerInfo().getPriority() > exclusiveConsumer
454                                            .getConsumerInfo().getPriority())) {
455                                exclusiveConsumer = s;
456
457                            }
458                        }
459                        dispatchSelector.setExclusiveConsumer(exclusiveConsumer);
460                    }
461                } else if (isAllConsumersExclusiveByDefault()) {
462                    Subscription exclusiveConsumer = null;
463                    for (Subscription s : consumers) {
464                        if (exclusiveConsumer == null 
465                                || s.getConsumerInfo().getPriority() > exclusiveConsumer
466                                .getConsumerInfo().getPriority()) {
467                            exclusiveConsumer = s;
468                                }
469                    }
470                    dispatchSelector.setExclusiveConsumer(exclusiveConsumer);
471                }
472                ConsumerId consumerId = sub.getConsumerInfo().getConsumerId();
473                getMessageGroupOwners().removeConsumer(consumerId);
474
475                // redeliver inflight messages
476
477                boolean markAsRedelivered = false;
478                MessageReference lastDeliveredRef = null;
479                List<MessageReference> unAckedMessages = sub.remove(context, this);
480
481                // locate last redelivered in unconsumed list (list in delivery rather than seq order)
482                if (lastDeiveredSequenceId != 0) {
483                    for (MessageReference ref : unAckedMessages) {
484                        if (ref.getMessageId().getBrokerSequenceId() == lastDeiveredSequenceId) {
485                            lastDeliveredRef = ref;
486                            markAsRedelivered = true;
487                            LOG.debug("found lastDeliveredSeqID: " + lastDeiveredSequenceId + ", message reference: " + ref.getMessageId());
488                            break;
489                        }
490                    }
491                }
492                for (MessageReference ref : unAckedMessages) {
493                    QueueMessageReference qmr = (QueueMessageReference) ref;
494                    if (qmr.getLockOwner() == sub) {
495                        qmr.unlock();
496
497                        // have no delivery information
498                        if (lastDeiveredSequenceId == 0) {
499                            qmr.incrementRedeliveryCounter();
500                        } else {
501                            if (markAsRedelivered) {
502                                qmr.incrementRedeliveryCounter();
503                            }
504                            if (ref == lastDeliveredRef) {
505                                // all that follow were not redelivered
506                                markAsRedelivered = false;
507                            }
508                        }
509                    }
510                    redeliveredWaitingDispatch.add(qmr);
511                }
512                if (!redeliveredWaitingDispatch.isEmpty()) {
513                    doDispatch(new ArrayList<QueueMessageReference>());
514                }
515            }finally {
516                consumersLock.writeLock().unlock();
517            }
518            if (!(this.optimizedDispatch || isSlave())) {
519                wakeup();
520            }
521        }finally {
522            pagedInPendingDispatchLock.writeLock().unlock();
523        }
524        if (this.optimizedDispatch || isSlave()) {
525            // Outside of dispatchLock() to maintain the lock hierarchy of
526            // iteratingMutex -> dispatchLock. - see
527            // https://issues.apache.org/activemq/browse/AMQ-1878
528            wakeup();
529        }
530    }
531
532    public void send(final ProducerBrokerExchange producerExchange, final Message message) throws Exception {
533        final ConnectionContext context = producerExchange.getConnectionContext();
534        // There is delay between the client sending it and it arriving at the
535        // destination.. it may have expired.
536        message.setRegionDestination(this);
537        final ProducerInfo producerInfo = producerExchange.getProducerState().getInfo();
538        final boolean sendProducerAck = !message.isResponseRequired() && producerInfo.getWindowSize() > 0
539                && !context.isInRecoveryMode();
540        if (message.isExpired()) {
541            // message not stored - or added to stats yet - so chuck here
542            broker.getRoot().messageExpired(context, message, null);
543            if (sendProducerAck) {
544                ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize());
545                context.getConnection().dispatchAsync(ack);
546            }
547            return;
548        }
549        if (memoryUsage.isFull()) {
550            isFull(context, memoryUsage);
551            fastProducer(context, producerInfo);
552            if (isProducerFlowControl() && context.isProducerFlowControl()) {
553                if (warnOnProducerFlowControl) {
554                    warnOnProducerFlowControl = false;
555                    LOG
556                            .info("Usage Manager Memory Limit ("
557                                    + memoryUsage.getLimit()
558                                    + ") reached on "
559                                    + getActiveMQDestination().getQualifiedName()
560                                    + ". Producers will be throttled to the rate at which messages are removed from this destination to prevent flooding it."
561                                    + " See http://activemq.apache.org/producer-flow-control.html for more info");
562                }
563
564                if (systemUsage.isSendFailIfNoSpace()) {
565                    throw new ResourceAllocationException("Usage Manager Memory Limit reached. Stopping producer ("
566                            + message.getProducerId() + ") to prevent flooding "
567                            + getActiveMQDestination().getQualifiedName() + "."
568                            + " See http://activemq.apache.org/producer-flow-control.html for more info");
569                }
570
571                // We can avoid blocking due to low usage if the producer is
572                // sending
573                // a sync message or if it is using a producer window
574                if (producerInfo.getWindowSize() > 0 || message.isResponseRequired()) {
575                    // copy the exchange state since the context will be
576                    // modified while we are waiting
577                    // for space.
578                    final ProducerBrokerExchange producerExchangeCopy = producerExchange.copy();
579                    synchronized (messagesWaitingForSpace) {
580                     // Start flow control timeout task
581                        // Prevent trying to start it multiple times
582                        if (!flowControlTimeoutTask.isAlive()) {
583                            flowControlTimeoutTask.setName(getName()+" Producer Flow Control Timeout Task");
584                            flowControlTimeoutTask.start();
585                        }
586                        messagesWaitingForSpace.put(message.getMessageId(), new Runnable() {
587                            public void run() {
588
589                                try {
590                                    // While waiting for space to free up... the
591                                    // message may have expired.
592                                    if (message.isExpired()) {
593                                        LOG.error("expired waiting for space..");
594                                        broker.messageExpired(context, message, null);
595                                        destinationStatistics.getExpired().increment();
596                                    } else {
597                                        doMessageSend(producerExchangeCopy, message);
598                                    }
599
600                                    if (sendProducerAck) {
601                                        ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message
602                                                .getSize());
603                                        context.getConnection().dispatchAsync(ack);
604                                    } else {
605                                        Response response = new Response();
606                                        response.setCorrelationId(message.getCommandId());
607                                        context.getConnection().dispatchAsync(response);
608                                    }
609
610                                } catch (Exception e) {
611                                    if (!sendProducerAck && !context.isInRecoveryMode()) {
612                                        ExceptionResponse response = new ExceptionResponse(e);
613                                        response.setCorrelationId(message.getCommandId());
614                                        context.getConnection().dispatchAsync(response);
615                                    } else {
616                                        LOG.debug("unexpected exception on deferred send of :" + message, e);
617                                    }
618                                }
619                            }
620                        });
621
622                        if (systemUsage.getSendFailIfNoSpaceAfterTimeout() != 0) {
623                            flowControlTimeoutMessages.add(new TimeoutMessage(message, context, systemUsage
624                                    .getSendFailIfNoSpaceAfterTimeout()));
625                        }
626
627                        registerCallbackForNotFullNotification();
628                        context.setDontSendReponse(true);
629                        return;
630                    }
631
632                } else {
633
634                    if (memoryUsage.isFull()) {
635                        waitForSpace(context, memoryUsage, "Usage Manager Memory Limit is full. Producer ("
636                                + message.getProducerId() + ") stopped to prevent flooding "
637                                + getActiveMQDestination().getQualifiedName() + "."
638                                + " See http://activemq.apache.org/producer-flow-control.html for more info");
639                    }
640
641                    // The usage manager could have delayed us by the time
642                    // we unblock the message could have expired..
643                    if (message.isExpired()) {
644                        if (LOG.isDebugEnabled()) {
645                            LOG.debug("Expired message: " + message);
646                        }
647                        broker.getRoot().messageExpired(context, message, null);
648                        return;
649                    }
650                }
651            }
652        }
653        doMessageSend(producerExchange, message);
654        if (sendProducerAck) {
655            ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize());
656            context.getConnection().dispatchAsync(ack);
657        }
658    }
659
660    private void registerCallbackForNotFullNotification() {
661        // If the usage manager is not full, then the task will not
662        // get called..
663        if (!memoryUsage.notifyCallbackWhenNotFull(sendMessagesWaitingForSpaceTask)) {
664            // so call it directly here.
665            sendMessagesWaitingForSpaceTask.run();
666        }
667    }
668
669    void doMessageSend(final ProducerBrokerExchange producerExchange, final Message message) throws IOException,
670            Exception {
671        final ConnectionContext context = producerExchange.getConnectionContext();
672        Future<Object> result = null;
673        
674        checkUsage(context, message);
675        sendLock.lockInterruptibly();
676        try {
677            if (store != null && message.isPersistent()) {        
678                message.getMessageId().setBrokerSequenceId(getDestinationSequenceId());
679                if (messages.isCacheEnabled()) {
680                    result = store.asyncAddQueueMessage(context, message);
681                } else {
682                    store.addMessage(context, message);
683                }
684                if (isReduceMemoryFootprint()) {
685                    message.clearMarshalledState();
686                }
687            }
688            if (context.isInTransaction()) {
689                // If this is a transacted message.. increase the usage now so that
690                // a big TX does not blow up
691                // our memory. This increment is decremented once the tx finishes..
692                message.incrementReferenceCount();
693            
694                context.getTransaction().addSynchronization(new Synchronization() {
695                    @Override
696                    public void afterCommit() throws Exception {
697                        sendLock.lockInterruptibly();
698                        try {
699                            // It could take while before we receive the commit
700                            // op, by that time the message could have expired..
701                            if (broker.isExpired(message)) {
702                                broker.messageExpired(context, message, null);
703                                destinationStatistics.getExpired().increment();
704                                return;
705                            }
706                            sendMessage(message);
707                        } finally {
708                            sendLock.unlock();
709                            message.decrementReferenceCount();
710                        }
711                        messageSent(context, message);
712                    }
713                    @Override
714                    public void afterRollback() throws Exception {
715                        message.decrementReferenceCount();
716                    }
717                });
718            } else {
719                // Add to the pending list, this takes care of incrementing the
720                // usage manager.
721                sendMessage(message);
722            }
723        } finally {
724            sendLock.unlock();
725        }
726        if (!context.isInTransaction()) {
727            messageSent(context, message);
728        }
729        if (result != null && !result.isCancelled()) {
730            try {
731                result.get();
732            } catch (CancellationException e) {
733                // ignore - the task has been cancelled if the message
734                // has already been deleted
735            }
736        }
737    }
738
739    private void checkUsage(ConnectionContext context, Message message) throws ResourceAllocationException, IOException, InterruptedException {
740        if (message.isPersistent()) {
741            if (store != null && systemUsage.getStoreUsage().isFull(getStoreUsageHighWaterMark())) {
742                final String logMessage = "Usage Manager Store is Full, " + getStoreUsageHighWaterMark() + "% of "
743                    + systemUsage.getStoreUsage().getLimit() + ". Stopping producer ("
744                    + message.getProducerId() + ") to prevent flooding "
745                    + getActiveMQDestination().getQualifiedName() + "."
746                    + " See http://activemq.apache.org/producer-flow-control.html for more info";
747
748                waitForSpace(context, systemUsage.getStoreUsage(), getStoreUsageHighWaterMark(), logMessage);
749            }
750        } else if (messages.getSystemUsage() != null && systemUsage.getTempUsage().isFull()) {
751            final String logMessage = "Usage Manager Temp Store is Full ("
752                    + systemUsage.getTempUsage().getPercentUsage() + "% of " + systemUsage.getTempUsage().getLimit() 
753                    +"). Stopping producer (" + message.getProducerId()
754                + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "."
755                + " See http://activemq.apache.org/producer-flow-control.html for more info";
756            
757            waitForSpace(context, messages.getSystemUsage().getTempUsage(), logMessage);
758        }
759    }
760
761    private void expireMessages() {
762        if (LOG.isDebugEnabled()) {
763            LOG.debug(getActiveMQDestination().getQualifiedName() + " expiring messages ..");
764        }
765
766        // just track the insertion count
767        List<Message> browsedMessages = new AbstractList<Message>() {
768            int size = 0;
769
770            @Override
771            public void add(int index, Message element) {
772                size++;
773            }
774
775            @Override
776            public int size() {
777                return size;
778            }
779
780            @Override
781            public Message get(int index) {
782                return null;
783            }
784        };
785        doBrowse(browsedMessages, this.getMaxExpirePageSize());
786        asyncWakeup();
787        if (LOG.isDebugEnabled()) {
788            LOG.debug(getActiveMQDestination().getQualifiedName() + " expiring messages done.");
789        }
790    }
791
792    public void gc() {
793    }
794
795    public void acknowledge(ConnectionContext context, Subscription sub, MessageAck ack, MessageReference node)
796            throws IOException {
797        messageConsumed(context, node);
798        if (store != null && node.isPersistent()) {
799            // the original ack may be a ranged ack, but we are trying to delete
800            // a specific
801            // message store here so we need to convert to a non ranged ack.
802            if (ack.getMessageCount() > 0) {
803                // Dup the ack
804                MessageAck a = new MessageAck();
805                ack.copy(a);
806                ack = a;
807                // Convert to non-ranged.
808                ack.setFirstMessageId(node.getMessageId());
809                ack.setLastMessageId(node.getMessageId());
810                ack.setMessageCount(1);
811            }
812
813            store.removeAsyncMessage(context, ack);
814        }
815    }
816
817    Message loadMessage(MessageId messageId) throws IOException {
818        Message msg = null;
819        if (store != null) { // can be null for a temp q
820            msg = store.getMessage(messageId);
821            if (msg != null) {
822                msg.setRegionDestination(this);
823            }
824        }
825        return msg;
826    }
827
828    @Override
829    public String toString() {
830        int size = 0;
831        messagesLock.readLock().lock();
832        try{
833            size = messages.size();
834        }finally {
835            messagesLock.readLock().unlock();
836        }
837        return "Queue: destination=" + destination.getPhysicalName() + ", subscriptions=" + consumers.size()
838                + ", memory=" + memoryUsage.getPercentUsage() + "%, size=" + size + ", in flight groups="
839                + messageGroupOwners;
840    }
841
842    public void start() throws Exception {
843        if (memoryUsage != null) {
844            memoryUsage.start();
845        }
846        if (systemUsage.getStoreUsage() != null) {
847            systemUsage.getStoreUsage().start();
848        }
849        systemUsage.getMemoryUsage().addUsageListener(this);
850        messages.start();
851        if (getExpireMessagesPeriod() > 0) {
852            scheduler.schedualPeriodically(expireMessagesTask, getExpireMessagesPeriod());
853        }
854        doPageIn(false);
855    }
856
857    public void stop() throws Exception {
858        if (taskRunner != null) {
859            taskRunner.shutdown();
860        }
861        if (this.executor != null) {
862            this.executor.shutdownNow();
863        }
864
865        scheduler.cancel(expireMessagesTask);
866
867        if (flowControlTimeoutTask.isAlive()) {
868            flowControlTimeoutTask.interrupt();
869        }
870
871        if (messages != null) {
872            messages.stop();
873        }
874
875        systemUsage.getMemoryUsage().removeUsageListener(this);
876        if (memoryUsage != null) {
877            memoryUsage.stop();
878        }
879        if (store != null) {
880            store.stop();
881        }
882    }
883
884    // Properties
885    // -------------------------------------------------------------------------
886    @Override
887    public ActiveMQDestination getActiveMQDestination() {
888        return destination;
889    }
890
891    public MessageGroupMap getMessageGroupOwners() {
892        if (messageGroupOwners == null) {
893            messageGroupOwners = getMessageGroupMapFactory().createMessageGroupMap();
894        }
895        return messageGroupOwners;
896    }
897
898    public DispatchPolicy getDispatchPolicy() {
899        return dispatchPolicy;
900    }
901
902    public void setDispatchPolicy(DispatchPolicy dispatchPolicy) {
903        this.dispatchPolicy = dispatchPolicy;
904    }
905
906    public MessageGroupMapFactory getMessageGroupMapFactory() {
907        return messageGroupMapFactory;
908    }
909
910    public void setMessageGroupMapFactory(MessageGroupMapFactory messageGroupMapFactory) {
911        this.messageGroupMapFactory = messageGroupMapFactory;
912    }
913
914    public PendingMessageCursor getMessages() {
915        return this.messages;
916    }
917
918    public void setMessages(PendingMessageCursor messages) {
919        this.messages = messages;
920    }
921
922    public boolean isUseConsumerPriority() {
923        return useConsumerPriority;
924    }
925
926    public void setUseConsumerPriority(boolean useConsumerPriority) {
927        this.useConsumerPriority = useConsumerPriority;
928    }
929
930    public boolean isStrictOrderDispatch() {
931        return strictOrderDispatch;
932    }
933
934    public void setStrictOrderDispatch(boolean strictOrderDispatch) {
935        this.strictOrderDispatch = strictOrderDispatch;
936    }
937
938    public boolean isOptimizedDispatch() {
939        return optimizedDispatch;
940    }
941
942    public void setOptimizedDispatch(boolean optimizedDispatch) {
943        this.optimizedDispatch = optimizedDispatch;
944    }
945
946    public int getTimeBeforeDispatchStarts() {
947        return timeBeforeDispatchStarts;
948    }
949
950    public void setTimeBeforeDispatchStarts(int timeBeforeDispatchStarts) {
951        this.timeBeforeDispatchStarts = timeBeforeDispatchStarts;
952    }
953
954    public int getConsumersBeforeDispatchStarts() {
955        return consumersBeforeDispatchStarts;
956    }
957
958    public void setConsumersBeforeDispatchStarts(int consumersBeforeDispatchStarts) {
959        this.consumersBeforeDispatchStarts = consumersBeforeDispatchStarts;
960    }
961
962    public void setAllConsumersExclusiveByDefault(boolean allConsumersExclusiveByDefault) {
963        this.allConsumersExclusiveByDefault = allConsumersExclusiveByDefault;
964    }
965
966    public boolean isAllConsumersExclusiveByDefault() {
967        return allConsumersExclusiveByDefault;
968    }
969
970
971    // Implementation methods
972    // -------------------------------------------------------------------------
973    private QueueMessageReference createMessageReference(Message message) {
974        QueueMessageReference result = new IndirectMessageReference(message);
975        return result;
976    }
977
978    public Message[] browse() {
979        List<Message> browseList = new ArrayList<Message>();
980        doBrowse(browseList, getMaxBrowsePageSize());
981        return browseList.toArray(new Message[browseList.size()]);
982    }
983
984    public void doBrowse(List<Message> browseList, int max) {
985        final ConnectionContext connectionContext = createConnectionContext();
986        try {
987            pageInMessages(false);
988            List<MessageReference> toExpire = new ArrayList<MessageReference>();
989
990            pagedInPendingDispatchLock.writeLock().lock();
991            try {
992                addAll(pagedInPendingDispatch, browseList, max, toExpire);
993                for (MessageReference ref : toExpire) {
994                    pagedInPendingDispatch.remove(ref);
995                    if (broker.isExpired(ref)) {
996                        LOG.debug("expiring from pagedInPending: " + ref);
997                        messageExpired(connectionContext, ref);
998                    }
999                }
1000            } finally {
1001                pagedInPendingDispatchLock.writeLock().unlock();
1002            }
1003            toExpire.clear();
1004            pagedInMessagesLock.readLock().lock();
1005            try {
1006                addAll(pagedInMessages.values(), browseList, max, toExpire);
1007            } finally {
1008                pagedInMessagesLock.readLock().unlock();
1009            }
1010            for (MessageReference ref : toExpire) {
1011                if (broker.isExpired(ref)) {
1012                    LOG.debug("expiring from pagedInMessages: " + ref);
1013                    messageExpired(connectionContext, ref);
1014                } else {
1015                    pagedInMessagesLock.writeLock().lock();
1016                    try {
1017                        pagedInMessages.remove(ref.getMessageId());
1018                    } finally {
1019                        pagedInMessagesLock.writeLock().unlock();
1020                    }
1021                }
1022            }
1023
1024            if (browseList.size() < getMaxBrowsePageSize()) {
1025                messagesLock.writeLock().lock();
1026                try {
1027                    try {
1028                        messages.reset();
1029                        while (messages.hasNext() && browseList.size() < max) {
1030                            MessageReference node = messages.next();
1031                            if (node.isExpired()) {
1032                                if (broker.isExpired(node)) {
1033                                    LOG.debug("expiring from messages: " + node);
1034                                    messageExpired(connectionContext, createMessageReference(node.getMessage()));
1035                                }
1036                                messages.remove();
1037                            } else {
1038                                messages.rollback(node.getMessageId());
1039                                if (browseList.contains(node.getMessage()) == false) {
1040                                    browseList.add(node.getMessage());
1041                                }
1042                            }
1043                            node.decrementReferenceCount();
1044                        }
1045                    } finally {
1046                        messages.release();
1047                    }
1048                } finally {
1049                    messagesLock.writeLock().unlock();
1050                }
1051            }
1052
1053        } catch (Exception e) {
1054            LOG.error("Problem retrieving message for browse", e);
1055        }
1056    }
1057
1058    private void addAll(Collection<QueueMessageReference> refs, List<Message> l, int maxBrowsePageSize,
1059            List<MessageReference> toExpire) throws Exception {
1060        for (Iterator<QueueMessageReference> i = refs.iterator(); i.hasNext() && l.size() < getMaxBrowsePageSize();) {
1061            QueueMessageReference ref = i.next();
1062            if (ref.isExpired()) {
1063                toExpire.add(ref);
1064            } else if (l.contains(ref.getMessage()) == false) {
1065                l.add(ref.getMessage());
1066            }
1067        }
1068    }
1069
1070    public QueueMessageReference getMessage(String id) {
1071        MessageId msgId = new MessageId(id);
1072        pagedInMessagesLock.readLock().lock();
1073        try{
1074            QueueMessageReference ref = this.pagedInMessages.get(msgId);
1075            if (ref != null) {
1076                return ref;
1077            }
1078        }finally {
1079            pagedInMessagesLock.readLock().unlock();
1080        }
1081        messagesLock.readLock().lock();
1082        try{
1083            try {
1084                messages.reset();
1085                while (messages.hasNext()) {
1086                    MessageReference mr = messages.next();
1087                    QueueMessageReference qmr = createMessageReference(mr.getMessage());
1088                    qmr.decrementReferenceCount();
1089                    messages.rollback(qmr.getMessageId());
1090                    if (msgId.equals(qmr.getMessageId())) {
1091                        return qmr;
1092                    }
1093                }
1094            } finally {
1095                messages.release();
1096            }
1097        }finally {
1098            messagesLock.readLock().unlock();
1099        }
1100        return null;
1101    }
1102
1103    public void purge() throws Exception {
1104        ConnectionContext c = createConnectionContext();
1105        List<MessageReference> list = null;
1106        do {
1107            doPageIn(true);
1108            pagedInMessagesLock.readLock().lock();
1109            try {
1110                list = new ArrayList<MessageReference>(pagedInMessages.values());
1111            }finally {
1112                pagedInMessagesLock.readLock().unlock();
1113            }
1114
1115            for (MessageReference ref : list) {
1116                try {
1117                    QueueMessageReference r = (QueueMessageReference) ref;
1118                    removeMessage(c, r);
1119                } catch (IOException e) {
1120                }
1121            }
1122            // don't spin/hang if stats are out and there is nothing left in the
1123            // store
1124        } while (!list.isEmpty() && this.destinationStatistics.getMessages().getCount() > 0);
1125        if (this.destinationStatistics.getMessages().getCount() > 0) {
1126            LOG.warn(getActiveMQDestination().getQualifiedName()
1127                    + " after purge complete, message count stats report: "
1128                    + this.destinationStatistics.getMessages().getCount());
1129        }
1130        gc();
1131        this.destinationStatistics.getMessages().setCount(0);
1132        getMessages().clear();
1133    }
1134
1135    /**
1136     * Removes the message matching the given messageId
1137     */
1138    public boolean removeMessage(String messageId) throws Exception {
1139        return removeMatchingMessages(createMessageIdFilter(messageId), 1) > 0;
1140    }
1141
1142    /**
1143     * Removes the messages matching the given selector
1144     * 
1145     * @return the number of messages removed
1146     */
1147    public int removeMatchingMessages(String selector) throws Exception {
1148        return removeMatchingMessages(selector, -1);
1149    }
1150
1151    /**
1152     * Removes the messages matching the given selector up to the maximum number
1153     * of matched messages
1154     * 
1155     * @return the number of messages removed
1156     */
1157    public int removeMatchingMessages(String selector, int maximumMessages) throws Exception {
1158        return removeMatchingMessages(createSelectorFilter(selector), maximumMessages);
1159    }
1160
1161    /**
1162     * Removes the messages matching the given filter up to the maximum number
1163     * of matched messages
1164     * 
1165     * @return the number of messages removed
1166     */
1167    public int removeMatchingMessages(MessageReferenceFilter filter, int maximumMessages) throws Exception {
1168        int movedCounter = 0;
1169        Set<MessageReference> set = new CopyOnWriteArraySet<MessageReference>();
1170        ConnectionContext context = createConnectionContext();
1171        do {
1172            doPageIn(true);
1173            pagedInMessagesLock.readLock().lock();
1174            try{
1175                set.addAll(pagedInMessages.values());
1176            }finally {
1177                pagedInMessagesLock.readLock().unlock();
1178            }
1179            List<MessageReference> list = new ArrayList<MessageReference>(set);
1180            for (MessageReference ref : list) {
1181                IndirectMessageReference r = (IndirectMessageReference) ref;
1182                if (filter.evaluate(context, r)) {
1183
1184                    removeMessage(context, r);
1185                    set.remove(r);
1186                    if (++movedCounter >= maximumMessages && maximumMessages > 0) {
1187                        return movedCounter;
1188                    }
1189                }
1190            }
1191        } while (set.size() < this.destinationStatistics.getMessages().getCount());
1192        return movedCounter;
1193    }
1194
1195    /**
1196     * Copies the message matching the given messageId
1197     */
1198    public boolean copyMessageTo(ConnectionContext context, String messageId, ActiveMQDestination dest)
1199            throws Exception {
1200        return copyMatchingMessages(context, createMessageIdFilter(messageId), dest, 1) > 0;
1201    }
1202
1203    /**
1204     * Copies the messages matching the given selector
1205     * 
1206     * @return the number of messages copied
1207     */
1208    public int copyMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest)
1209            throws Exception {
1210        return copyMatchingMessagesTo(context, selector, dest, -1);
1211    }
1212
1213    /**
1214     * Copies the messages matching the given selector up to the maximum number
1215     * of matched messages
1216     * 
1217     * @return the number of messages copied
1218     */
1219    public int copyMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest,
1220            int maximumMessages) throws Exception {
1221        return copyMatchingMessages(context, createSelectorFilter(selector), dest, maximumMessages);
1222    }
1223
1224    /**
1225     * Copies the messages matching the given filter up to the maximum number of
1226     * matched messages
1227     * 
1228     * @return the number of messages copied
1229     */
1230    public int copyMatchingMessages(ConnectionContext context, MessageReferenceFilter filter, ActiveMQDestination dest,
1231            int maximumMessages) throws Exception {
1232        int movedCounter = 0;
1233        int count = 0;
1234        Set<MessageReference> set = new CopyOnWriteArraySet<MessageReference>();
1235        do {
1236            int oldMaxSize = getMaxPageSize();
1237            setMaxPageSize((int) this.destinationStatistics.getMessages().getCount());
1238            doPageIn(true);
1239            setMaxPageSize(oldMaxSize);
1240            pagedInMessagesLock.readLock().lock();
1241            try {
1242                set.addAll(pagedInMessages.values());
1243            }finally {
1244                pagedInMessagesLock.readLock().unlock();
1245            }
1246            List<MessageReference> list = new ArrayList<MessageReference>(set);
1247            for (MessageReference ref : list) {
1248                IndirectMessageReference r = (IndirectMessageReference) ref;
1249                if (filter.evaluate(context, r)) {
1250
1251                    r.incrementReferenceCount();
1252                    try {
1253                        Message m = r.getMessage();
1254                        BrokerSupport.resend(context, m, dest);
1255                        if (++movedCounter >= maximumMessages && maximumMessages > 0) {
1256                            return movedCounter;
1257                        }
1258                    } finally {
1259                        r.decrementReferenceCount();
1260                    }
1261                }
1262                count++;
1263            }
1264        } while (count < this.destinationStatistics.getMessages().getCount());
1265        return movedCounter;
1266    }
1267
1268    /**
1269     * Move a message
1270     *
1271     * @param context
1272     *            connection context
1273     * @param m
1274     *            QueueMessageReference
1275     * @param dest
1276     *            ActiveMQDestination
1277     * @throws Exception
1278     */
1279    public boolean moveMessageTo(ConnectionContext context, QueueMessageReference m, ActiveMQDestination dest) throws Exception {
1280        BrokerSupport.resend(context, m.getMessage(), dest);
1281        removeMessage(context, m);
1282        messagesLock.writeLock().lock();
1283        try{
1284            messages.rollback(m.getMessageId());
1285        }finally {
1286            messagesLock.writeLock().unlock();
1287        }
1288        return true;
1289    }
1290
1291    /**
1292     * Moves the message matching the given messageId
1293     */
1294    public boolean moveMessageTo(ConnectionContext context, String messageId, ActiveMQDestination dest)
1295            throws Exception {
1296        return moveMatchingMessagesTo(context, createMessageIdFilter(messageId), dest, 1) > 0;
1297    }
1298
1299    /**
1300     * Moves the messages matching the given selector
1301     * 
1302     * @return the number of messages removed
1303     */
1304    public int moveMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest)
1305            throws Exception {
1306        return moveMatchingMessagesTo(context, selector, dest, Integer.MAX_VALUE);
1307    }
1308
1309    /**
1310     * Moves the messages matching the given selector up to the maximum number
1311     * of matched messages
1312     */
1313    public int moveMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest,
1314            int maximumMessages) throws Exception {
1315        return moveMatchingMessagesTo(context, createSelectorFilter(selector), dest, maximumMessages);
1316    }
1317
1318    /**
1319     * Moves the messages matching the given filter up to the maximum number of
1320     * matched messages
1321     */
1322    public int moveMatchingMessagesTo(ConnectionContext context, MessageReferenceFilter filter,
1323            ActiveMQDestination dest, int maximumMessages) throws Exception {
1324        int movedCounter = 0;
1325        Set<QueueMessageReference> set = new CopyOnWriteArraySet<QueueMessageReference>();
1326        do {
1327            doPageIn(true);
1328            pagedInMessagesLock.readLock().lock();
1329            try{
1330                set.addAll(pagedInMessages.values());
1331            }finally {
1332                pagedInMessagesLock.readLock().unlock();
1333            }
1334            List<QueueMessageReference> list = new ArrayList<QueueMessageReference>(set);
1335            for (QueueMessageReference ref : list) {
1336                if (filter.evaluate(context, ref)) {
1337                    // We should only move messages that can be locked.
1338                    moveMessageTo(context, ref, dest);
1339                    set.remove(ref);
1340                    if (++movedCounter >= maximumMessages && maximumMessages > 0) {
1341                        return movedCounter;
1342                    }
1343                }
1344            }
1345        } while (set.size() < this.destinationStatistics.getMessages().getCount() && set.size() < maximumMessages);
1346        return movedCounter;
1347    }
1348
1349    BrowserDispatch getNextBrowserDispatch() {
1350        pagedInMessagesLock.readLock().lock();
1351        try{
1352            if (browserDispatches.isEmpty()) {
1353                return null;
1354            }
1355            return browserDispatches.removeFirst();
1356        }finally {
1357            pagedInMessagesLock.readLock().unlock();
1358        }
1359
1360    }
1361
1362    /**
1363     * @return true if we would like to iterate again
1364     * @see org.apache.activemq.thread.Task#iterate()
1365     */
1366    public boolean iterate() {
1367        MDC.put("activemq.destination", getName());
1368        boolean pageInMoreMessages = false;
1369        synchronized (iteratingMutex) {
1370
1371            // do early to allow dispatch of these waiting messages
1372            synchronized (messagesWaitingForSpace) {
1373                Iterator<Runnable> it = messagesWaitingForSpace.values().iterator();
1374                while (it.hasNext()) {
1375                    if (!memoryUsage.isFull()) {
1376                        Runnable op = it.next();
1377                        it.remove();
1378                        op.run();
1379                    } else {
1380                        registerCallbackForNotFullNotification();
1381                        break;
1382                    }
1383                }
1384            }
1385
1386            if (firstConsumer) {
1387                firstConsumer = false;
1388                try {
1389                    if (consumersBeforeDispatchStarts > 0) {
1390                        int timeout = 1000; // wait one second by default if
1391                                            // consumer count isn't reached
1392                        if (timeBeforeDispatchStarts > 0) {
1393                            timeout = timeBeforeDispatchStarts;
1394                        }
1395                        if (consumersBeforeStartsLatch.await(timeout, TimeUnit.MILLISECONDS)) {
1396                            if (LOG.isDebugEnabled()) {
1397                                LOG.debug(consumers.size() + " consumers subscribed. Starting dispatch.");
1398                            }
1399                        } else {
1400                            if (LOG.isDebugEnabled()) {
1401                                LOG.debug(timeout + " ms elapsed and " + consumers.size()
1402                                        + " consumers subscribed. Starting dispatch.");
1403                            }
1404                        }
1405                    }
1406                    if (timeBeforeDispatchStarts > 0 && consumersBeforeDispatchStarts <= 0) {
1407                        iteratingMutex.wait(timeBeforeDispatchStarts);
1408                        if (LOG.isDebugEnabled()) {
1409                            LOG.debug(timeBeforeDispatchStarts + " ms elapsed. Starting dispatch.");
1410                        }
1411                    }
1412                } catch (Exception e) {
1413                    LOG.error(e.toString());
1414                }
1415            }
1416
1417            BrowserDispatch pendingBrowserDispatch = getNextBrowserDispatch();
1418
1419            messagesLock.readLock().lock();
1420            try{
1421                pageInMoreMessages |= !messages.isEmpty();
1422            }finally {
1423                messagesLock.readLock().unlock();
1424            }
1425
1426            pagedInPendingDispatchLock.readLock().lock();
1427            try {
1428                pageInMoreMessages |= !pagedInPendingDispatch.isEmpty();
1429            }finally {
1430                pagedInPendingDispatchLock.readLock().unlock();
1431            }
1432
1433            // Perhaps we should page always into the pagedInPendingDispatch
1434            // list if
1435            // !messages.isEmpty(), and then if
1436            // !pagedInPendingDispatch.isEmpty()
1437            // then we do a dispatch.
1438            if (pageInMoreMessages || pendingBrowserDispatch != null || !redeliveredWaitingDispatch.isEmpty()) {
1439                try {
1440                    pageInMessages(pendingBrowserDispatch != null);
1441
1442                } catch (Throwable e) {
1443                    LOG.error("Failed to page in more queue messages ", e);
1444                }
1445            }
1446
1447            if (pendingBrowserDispatch != null) {
1448                ArrayList<QueueMessageReference> alreadyDispatchedMessages = null;
1449                pagedInMessagesLock.readLock().lock();
1450                try{
1451                    alreadyDispatchedMessages = new ArrayList<QueueMessageReference>(pagedInMessages.values());
1452                }finally {
1453                    pagedInMessagesLock.readLock().unlock();
1454                }
1455                if (LOG.isDebugEnabled()) {
1456                    LOG.debug("dispatch to browser: " + pendingBrowserDispatch.getBrowser()
1457                            + ", already dispatched/paged count: " + alreadyDispatchedMessages.size());
1458                }
1459                do {
1460                    try {
1461                        MessageEvaluationContext msgContext = new NonCachedMessageEvaluationContext();
1462                        msgContext.setDestination(destination);
1463
1464                        QueueBrowserSubscription browser = pendingBrowserDispatch.getBrowser();
1465                        for (QueueMessageReference node : alreadyDispatchedMessages) {
1466                            if (!node.isAcked()) {
1467                                msgContext.setMessageReference(node);
1468                                if (browser.matches(node, msgContext)) {
1469                                    browser.add(node);
1470                                }
1471                            }
1472                        }
1473                        pendingBrowserDispatch.done();
1474                    } catch (Exception e) {
1475                        LOG.warn("exception on dispatch to browser: " + pendingBrowserDispatch.getBrowser(), e);
1476                    }
1477
1478                } while ((pendingBrowserDispatch = getNextBrowserDispatch()) != null);
1479            }
1480
1481            if (pendingWakeups.get() > 0) {
1482                pendingWakeups.decrementAndGet();
1483            }
1484            MDC.remove("activemq.destination");
1485            return pendingWakeups.get() > 0;
1486        }
1487    }
1488
1489    protected MessageReferenceFilter createMessageIdFilter(final String messageId) {
1490        return new MessageReferenceFilter() {
1491            public boolean evaluate(ConnectionContext context, MessageReference r) {
1492                return messageId.equals(r.getMessageId().toString());
1493            }
1494
1495            @Override
1496            public String toString() {
1497                return "MessageIdFilter: " + messageId;
1498            }
1499        };
1500    }
1501
1502    protected MessageReferenceFilter createSelectorFilter(String selector) throws InvalidSelectorException {
1503        final BooleanExpression selectorExpression = SelectorParser.parse(selector);
1504
1505        return new MessageReferenceFilter() {
1506            public boolean evaluate(ConnectionContext context, MessageReference r) throws JMSException {
1507                MessageEvaluationContext messageEvaluationContext = context.getMessageEvaluationContext();
1508
1509                messageEvaluationContext.setMessageReference(r);
1510                if (messageEvaluationContext.getDestination() == null) {
1511                    messageEvaluationContext.setDestination(getActiveMQDestination());
1512                }
1513
1514                return selectorExpression.matches(messageEvaluationContext);
1515            }
1516        };
1517    }
1518
1519    protected void removeMessage(ConnectionContext c, QueueMessageReference r) throws IOException {
1520        removeMessage(c, null, r);
1521        pagedInPendingDispatchLock.writeLock().lock();
1522        try {
1523            pagedInPendingDispatch.remove(r);
1524        } finally {
1525            pagedInPendingDispatchLock.writeLock().unlock();
1526        }
1527
1528    }
1529
1530    protected void removeMessage(ConnectionContext c, Subscription subs, QueueMessageReference r) throws IOException {
1531        MessageAck ack = new MessageAck();
1532        ack.setAckType(MessageAck.STANDARD_ACK_TYPE);
1533        ack.setDestination(destination);
1534        ack.setMessageID(r.getMessageId());
1535        removeMessage(c, subs, r, ack);
1536    }
1537
1538    protected void removeMessage(ConnectionContext context, Subscription sub, final QueueMessageReference reference,
1539            MessageAck ack) throws IOException {
1540        reference.setAcked(true);
1541        // This sends the ack the the journal..
1542        if (!ack.isInTransaction()) {
1543            acknowledge(context, sub, ack, reference);
1544            getDestinationStatistics().getDequeues().increment();
1545            dropMessage(reference);
1546        } else {
1547            try {
1548                acknowledge(context, sub, ack, reference);
1549            } finally {
1550                context.getTransaction().addSynchronization(new Synchronization() {
1551
1552                    @Override
1553                    public void afterCommit() throws Exception {
1554                        getDestinationStatistics().getDequeues().increment();
1555                        dropMessage(reference);
1556                        wakeup();
1557                    }
1558
1559                    @Override
1560                    public void afterRollback() throws Exception {
1561                        reference.setAcked(false);
1562                    }
1563                });
1564            }
1565        }
1566        if (ack.isPoisonAck()) {
1567            // message gone to DLQ, is ok to allow redelivery
1568            messagesLock.writeLock().lock();
1569            try{
1570                messages.rollback(reference.getMessageId());
1571            }finally {
1572                messagesLock.writeLock().unlock();
1573            }
1574        }
1575
1576    }
1577
1578    private void dropMessage(QueueMessageReference reference) {
1579        reference.drop();
1580        destinationStatistics.getMessages().decrement();
1581        pagedInMessagesLock.writeLock().lock();
1582        try{
1583            pagedInMessages.remove(reference.getMessageId());
1584        }finally {
1585            pagedInMessagesLock.writeLock().unlock();
1586        }
1587    }
1588
1589    public void messageExpired(ConnectionContext context, MessageReference reference) {
1590        messageExpired(context, null, reference);
1591    }
1592
1593    public void messageExpired(ConnectionContext context, Subscription subs, MessageReference reference) {
1594        if (LOG.isDebugEnabled()) {
1595            LOG.debug("message expired: " + reference);
1596        }
1597        broker.messageExpired(context, reference, subs);
1598        destinationStatistics.getExpired().increment();
1599        try {
1600            removeMessage(context, subs, (QueueMessageReference) reference);
1601        } catch (IOException e) {
1602            LOG.error("Failed to remove expired Message from the store ", e);
1603        }
1604    }
1605
1606    protected ConnectionContext createConnectionContext() {
1607        ConnectionContext answer = new ConnectionContext(new NonCachedMessageEvaluationContext());
1608        answer.setBroker(this.broker);
1609        answer.getMessageEvaluationContext().setDestination(getActiveMQDestination());
1610        answer.setSecurityContext(SecurityContext.BROKER_SECURITY_CONTEXT);
1611        return answer;
1612    }
1613
1614    final void sendMessage(final Message msg) throws Exception {
1615        messagesLock.writeLock().lock();
1616        try{
1617            messages.addMessageLast(msg);
1618        }finally {
1619            messagesLock.writeLock().unlock();
1620        }
1621    }
1622    
1623    final void messageSent(final ConnectionContext context, final Message msg) throws Exception {     
1624        destinationStatistics.getEnqueues().increment();
1625        destinationStatistics.getMessages().increment();
1626        messageDelivered(context, msg);
1627        consumersLock.readLock().lock();
1628        try {
1629            if (consumers.isEmpty()) {
1630                onMessageWithNoConsumers(context, msg);
1631            }
1632        }finally {
1633            consumersLock.readLock().unlock();
1634        }
1635        if (LOG.isTraceEnabled()) {
1636            LOG.trace("Message " + msg.getMessageId() + " sent to " + this.destination);
1637        }
1638        wakeup();
1639    }
1640
1641    public void wakeup() {
1642        if (optimizedDispatch || isSlave()) {
1643            iterate();
1644            pendingWakeups.incrementAndGet();
1645        } else {
1646            asyncWakeup();
1647        }
1648    }
1649
1650    private void asyncWakeup() {
1651        try {
1652            pendingWakeups.incrementAndGet();
1653            this.taskRunner.wakeup();
1654        } catch (InterruptedException e) {
1655            LOG.warn("Async task tunner failed to wakeup ", e);
1656        }
1657    }
1658
1659    private boolean isSlave() {
1660        return broker.getBrokerService().isSlave();
1661    }
1662
1663    private void doPageIn(boolean force) throws Exception {
1664        List<QueueMessageReference> newlyPaged = doPageInForDispatch(force);
1665        pagedInPendingDispatchLock.writeLock().lock();
1666        try {
1667            if (pagedInPendingDispatch.isEmpty()) {
1668                pagedInPendingDispatch.addAll(newlyPaged);
1669            } else {
1670                for (QueueMessageReference qmr : newlyPaged) {
1671                    if (!pagedInPendingDispatch.contains(qmr)) {
1672                        pagedInPendingDispatch.add(qmr);
1673                    }
1674                }
1675            }
1676        } finally {
1677            pagedInPendingDispatchLock.writeLock().unlock();
1678        }
1679    }
1680
1681    private List<QueueMessageReference> doPageInForDispatch(boolean force) throws Exception {
1682        List<QueueMessageReference> result = null;
1683        List<QueueMessageReference> resultList = null;
1684
1685        int toPageIn = Math.min(getMaxPageSize(), messages.size());
1686        if (LOG.isDebugEnabled()) {
1687            LOG.debug(destination.getPhysicalName() + " toPageIn: " + toPageIn + ", Inflight: "
1688                    + destinationStatistics.getInflight().getCount() + ", pagedInMessages.size "
1689                    + pagedInMessages.size() + ", enqueueCount: " + destinationStatistics.getEnqueues().getCount()
1690                    + ", dequeueCount: " + destinationStatistics.getDequeues().getCount());
1691        }
1692
1693        if (isLazyDispatch() && !force) {
1694            // Only page in the minimum number of messages which can be
1695            // dispatched immediately.
1696            toPageIn = Math.min(getConsumerMessageCountBeforeFull(), toPageIn);
1697        }
1698        int pagedInPendingSize = 0;
1699        pagedInPendingDispatchLock.readLock().lock();
1700        try {
1701            pagedInPendingSize = pagedInPendingDispatch.size();
1702        } finally {
1703            pagedInPendingDispatchLock.readLock().unlock();
1704        }
1705        if (toPageIn > 0 && (force || (!consumers.isEmpty() && pagedInPendingSize < getMaxPageSize()))) {
1706            int count = 0;
1707            result = new ArrayList<QueueMessageReference>(toPageIn);
1708            messagesLock.writeLock().lock();
1709            try {
1710                try {
1711                    messages.setMaxBatchSize(toPageIn);
1712                    messages.reset();
1713                    while (messages.hasNext() && count < toPageIn) {
1714                        MessageReference node = messages.next();
1715                        messages.remove();
1716
1717                        QueueMessageReference ref = createMessageReference(node.getMessage());
1718                        if (ref.isExpired()) {
1719                            if (broker.isExpired(ref)) {
1720                                messageExpired(createConnectionContext(), ref);
1721                            } else {
1722                                ref.decrementReferenceCount();
1723                            }
1724                        } else {
1725                            result.add(ref);
1726                            count++;
1727                        }
1728                    }
1729                } finally {
1730                    messages.release();
1731                }
1732            } finally {
1733                messagesLock.writeLock().unlock();
1734            }
1735            // Only add new messages, not already pagedIn to avoid multiple
1736            // dispatch attempts
1737            pagedInMessagesLock.writeLock().lock();
1738            try {
1739                resultList = new ArrayList<QueueMessageReference>(result.size());
1740                for (QueueMessageReference ref : result) {
1741                    if (!pagedInMessages.containsKey(ref.getMessageId())) {            
1742                        pagedInMessages.put(ref.getMessageId(), ref);
1743                        resultList.add(ref);
1744                    } else {
1745                        ref.decrementReferenceCount();
1746                    }
1747                }
1748            } finally {
1749                pagedInMessagesLock.writeLock().unlock();
1750            }
1751        } else {
1752            // Avoid return null list, if condition is not validated
1753            resultList = new ArrayList<QueueMessageReference>();
1754        }
1755
1756        return resultList;
1757    }
1758
1759    private void doDispatch(List<QueueMessageReference> list) throws Exception {
1760        boolean doWakeUp = false;
1761
1762        pagedInPendingDispatchLock.writeLock().lock();
1763        try {
1764            if (!redeliveredWaitingDispatch.isEmpty()) {
1765                // Try first to dispatch redelivered messages to keep an
1766                // proper order
1767                redeliveredWaitingDispatch = doActualDispatch(redeliveredWaitingDispatch);
1768            }
1769            if (!pagedInPendingDispatch.isEmpty()) {
1770                // Next dispatch anything that had not been
1771                // dispatched before.
1772                pagedInPendingDispatch = doActualDispatch(pagedInPendingDispatch);
1773            }
1774            // and now see if we can dispatch the new stuff.. and append to
1775            // the pending
1776            // list anything that does not actually get dispatched.
1777            if (list != null && !list.isEmpty()) {
1778                if (pagedInPendingDispatch.isEmpty()) {
1779                    pagedInPendingDispatch.addAll(doActualDispatch(list));
1780                } else {
1781                    for (QueueMessageReference qmr : list) {
1782                        if (!pagedInPendingDispatch.contains(qmr)) {
1783                            pagedInPendingDispatch.add(qmr);
1784                        }
1785                    }
1786                    doWakeUp = true;
1787                }
1788            }
1789        } finally {
1790            pagedInPendingDispatchLock.writeLock().unlock();
1791        }
1792
1793        if (doWakeUp) {
1794            // avoid lock order contention
1795            asyncWakeup();
1796        }
1797    }
1798
1799    /**
1800     * @return list of messages that could get dispatched to consumers if they
1801     *         were not full.
1802     */
1803    private List<QueueMessageReference> doActualDispatch(List<QueueMessageReference> list) throws Exception {
1804        List<Subscription> consumers;
1805        consumersLock.writeLock().lock();
1806        try {
1807            if (this.consumers.isEmpty() || isSlave()) {
1808                // slave dispatch happens in processDispatchNotification
1809                return list;
1810            }
1811            consumers = new ArrayList<Subscription>(this.consumers);
1812        }finally {
1813            consumersLock.writeLock().unlock();
1814        }
1815
1816        List<QueueMessageReference> rc = new ArrayList<QueueMessageReference>(list.size());
1817        Set<Subscription> fullConsumers = new HashSet<Subscription>(this.consumers.size());
1818
1819        for (MessageReference node : list) {
1820            Subscription target = null;
1821            int interestCount = 0;
1822            for (Subscription s : consumers) {
1823                if (s instanceof QueueBrowserSubscription) {
1824                    interestCount++;
1825                    continue;
1826                }
1827                if (dispatchSelector.canSelect(s, node)) {
1828                    if (!fullConsumers.contains(s)) {
1829                        if (!s.isFull()) {
1830                            if (assignMessageGroup(s, (QueueMessageReference)node)) {
1831                                // Dispatch it.
1832                                s.add(node);
1833                                target = s;
1834                                break;
1835                            }
1836                        } else {
1837                            // no further dispatch of list to a full consumer to
1838                            // avoid out of order message receipt
1839                            fullConsumers.add(s);
1840                        }
1841                    }
1842                    interestCount++;
1843                } else {
1844                    // makes sure it gets dispatched again
1845                    if (!node.isDropped() && !((QueueMessageReference) node).isAcked()
1846                            && (!node.isDropped() || s.getConsumerInfo().isBrowser())) {
1847                        interestCount++;
1848                    }
1849                }
1850            }
1851
1852            if ((target == null && interestCount > 0) || consumers.size() == 0) {
1853                // This means all subs were full or that there are no
1854                // consumers...
1855                rc.add((QueueMessageReference) node);
1856            }
1857
1858            // If it got dispatched, rotate the consumer list to get round robin
1859            // distribution.
1860            if (target != null && !strictOrderDispatch && consumers.size() > 1
1861                    && !dispatchSelector.isExclusiveConsumer(target)) {
1862                consumersLock.writeLock().lock();
1863                try {
1864                    if (removeFromConsumerList(target)) {
1865                        addToConsumerList(target);
1866                        consumers = new ArrayList<Subscription>(this.consumers);
1867                    }
1868                }finally {
1869                    consumersLock.writeLock().unlock();
1870                }
1871            }
1872        }
1873
1874        return rc;
1875    }
1876
1877    protected boolean assignMessageGroup(Subscription subscription, QueueMessageReference node) throws Exception {
1878        //QueueMessageReference node = (QueueMessageReference) m;
1879        boolean result = true;
1880        // Keep message groups together.
1881        String groupId = node.getGroupID();
1882        int sequence = node.getGroupSequence();
1883        if (groupId != null) {
1884            //MessageGroupMap messageGroupOwners = ((Queue) node
1885            //        .getRegionDestination()).getMessageGroupOwners();
1886
1887            MessageGroupMap messageGroupOwners = getMessageGroupOwners();
1888            // If we can own the first, then no-one else should own the
1889            // rest.
1890            if (sequence == 1) {
1891                assignGroup(subscription, messageGroupOwners, node, groupId);
1892            } else {
1893
1894                // Make sure that the previous owner is still valid, we may
1895                // need to become the new owner.
1896                ConsumerId groupOwner;
1897
1898                groupOwner = messageGroupOwners.get(groupId);
1899                if (groupOwner == null) {
1900                    assignGroup(subscription, messageGroupOwners, node, groupId);
1901                } else {
1902                    if (groupOwner.equals(subscription.getConsumerInfo().getConsumerId())) {
1903                        // A group sequence < 1 is an end of group signal.
1904                        if (sequence < 0) {
1905                            messageGroupOwners.removeGroup(groupId);
1906                        }
1907                    } else {
1908                        result = false;
1909                    }
1910                }
1911            }
1912        }
1913
1914        return result;
1915
1916    }
1917
1918    protected void assignGroup(Subscription subs, MessageGroupMap messageGroupOwners, MessageReference n, String groupId) throws IOException {
1919        messageGroupOwners.put(groupId, subs.getConsumerInfo().getConsumerId());
1920        Message message = n.getMessage();
1921        if (message instanceof ActiveMQMessage) {
1922            ActiveMQMessage activeMessage = (ActiveMQMessage) message;
1923            try {
1924                activeMessage.setBooleanProperty("JMSXGroupFirstForConsumer", true, false);
1925            } catch (JMSException e) {
1926                LOG.warn("Failed to set boolean header: " + e, e);
1927            }
1928        }
1929    }
1930
1931    protected void pageInMessages(boolean force) throws Exception {
1932        doDispatch(doPageInForDispatch(force));
1933    }
1934
1935    private void addToConsumerList(Subscription sub) {
1936        if (useConsumerPriority) {
1937            consumers.add(sub);
1938            Collections.sort(consumers, orderedCompare);
1939        } else {
1940            consumers.add(sub);
1941        }
1942    }
1943
1944    private boolean removeFromConsumerList(Subscription sub) {
1945        return consumers.remove(sub);
1946    }
1947
1948    private int getConsumerMessageCountBeforeFull() throws Exception {
1949        int total = 0;
1950        boolean zeroPrefetch = false;
1951        consumersLock.readLock().lock();
1952        try{
1953            for (Subscription s : consumers) {
1954                zeroPrefetch |= s.getPrefetchSize() == 0;
1955                int countBeforeFull = s.countBeforeFull();
1956                total += countBeforeFull;
1957            }
1958        }finally {
1959            consumersLock.readLock().unlock();
1960        }
1961        if (total == 0 && zeroPrefetch) {
1962            total = 1;
1963        }
1964        return total;
1965    }
1966
1967    /*
1968     * In slave mode, dispatch is ignored till we get this notification as the
1969     * dispatch process is non deterministic between master and slave. On a
1970     * notification, the actual dispatch to the subscription (as chosen by the
1971     * master) is completed. (non-Javadoc)
1972     * @see
1973     * org.apache.activemq.broker.region.BaseDestination#processDispatchNotification
1974     * (org.apache.activemq.command.MessageDispatchNotification)
1975     */
1976    @Override
1977    public void processDispatchNotification(MessageDispatchNotification messageDispatchNotification) throws Exception {
1978        // do dispatch
1979        Subscription sub = getMatchingSubscription(messageDispatchNotification);
1980        if (sub != null) {
1981            MessageReference message = getMatchingMessage(messageDispatchNotification);
1982            sub.add(message);
1983            sub.processMessageDispatchNotification(messageDispatchNotification);
1984        }
1985    }
1986
1987    private QueueMessageReference getMatchingMessage(MessageDispatchNotification messageDispatchNotification)
1988            throws Exception {
1989        QueueMessageReference message = null;
1990        MessageId messageId = messageDispatchNotification.getMessageId();
1991
1992        pagedInPendingDispatchLock.writeLock().lock();
1993        try {
1994            for (QueueMessageReference ref : pagedInPendingDispatch) {
1995                if (messageId.equals(ref.getMessageId())) {
1996                    message = ref;
1997                    pagedInPendingDispatch.remove(ref);
1998                    break;
1999                }
2000            }
2001        } finally {
2002            pagedInPendingDispatchLock.writeLock().unlock();
2003        }
2004
2005        if (message == null) {
2006            pagedInMessagesLock.readLock().lock();
2007            try {
2008                message = pagedInMessages.get(messageId);
2009            } finally {
2010                pagedInMessagesLock.readLock().unlock();
2011            }
2012        }
2013
2014        if (message == null) {
2015            messagesLock.writeLock().lock();
2016            try {
2017                try {
2018                    messages.setMaxBatchSize(getMaxPageSize());
2019                    messages.reset();
2020                    while (messages.hasNext()) {
2021                        MessageReference node = messages.next();
2022                        messages.remove();
2023                        if (messageId.equals(node.getMessageId())) {
2024                            message = this.createMessageReference(node.getMessage());
2025                            break;
2026                        }
2027                    }
2028                } finally {
2029                    messages.release();
2030                }
2031            } finally {
2032                messagesLock.writeLock().unlock();
2033            }
2034        }
2035
2036        if (message == null) {
2037            Message msg = loadMessage(messageId);
2038            if (msg != null) {
2039                message = this.createMessageReference(msg);
2040            }
2041        }
2042
2043        if (message == null) {
2044            throw new JMSException("Slave broker out of sync with master - Message: "
2045                    + messageDispatchNotification.getMessageId() + " on "
2046                    + messageDispatchNotification.getDestination() + " does not exist among pending("
2047                    + pagedInPendingDispatch.size() + ") for subscription: "
2048                    + messageDispatchNotification.getConsumerId());
2049        }
2050        return message;
2051    }
2052
2053    /**
2054     * Find a consumer that matches the id in the message dispatch notification
2055     * 
2056     * @param messageDispatchNotification
2057     * @return sub or null if the subscription has been removed before dispatch
2058     * @throws JMSException
2059     */
2060    private Subscription getMatchingSubscription(MessageDispatchNotification messageDispatchNotification)
2061            throws JMSException {
2062        Subscription sub = null;
2063        consumersLock.readLock().lock();
2064        try {
2065            for (Subscription s : consumers) {
2066                if (messageDispatchNotification.getConsumerId().equals(s.getConsumerInfo().getConsumerId())) {
2067                    sub = s;
2068                    break;
2069                }
2070            }
2071        }finally {
2072            consumersLock.readLock().unlock();
2073        }
2074        return sub;
2075    }
2076
2077    public void onUsageChanged(Usage usage, int oldPercentUsage, int newPercentUsage) {
2078        if (oldPercentUsage > newPercentUsage) {
2079            asyncWakeup();
2080        }
2081    }
2082
2083    @Override
2084    protected Logger getLog() {
2085        return LOG;
2086    }
2087}