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