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