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