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