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