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