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