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