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