001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq;
018
019import java.io.IOException;
020import java.util.ArrayList;
021import java.util.Collections;
022import java.util.HashMap;
023import java.util.Iterator;
024import java.util.LinkedList;
025import java.util.List;
026import java.util.Map;
027import java.util.Map.Entry;
028import java.util.concurrent.ExecutorService;
029import java.util.concurrent.Executors;
030import java.util.concurrent.atomic.AtomicBoolean;
031import java.util.concurrent.atomic.AtomicInteger;
032import java.util.concurrent.atomic.AtomicReference;
033
034import javax.jms.IllegalStateException;
035import javax.jms.InvalidDestinationException;
036import javax.jms.JMSException;
037import javax.jms.Message;
038import javax.jms.MessageConsumer;
039import javax.jms.MessageListener;
040import javax.jms.TransactionRolledBackException;
041
042import org.apache.activemq.blob.BlobDownloader;
043import org.apache.activemq.command.ActiveMQBlobMessage;
044import org.apache.activemq.command.ActiveMQDestination;
045import org.apache.activemq.command.ActiveMQMessage;
046import org.apache.activemq.command.ActiveMQTempDestination;
047import org.apache.activemq.command.CommandTypes;
048import org.apache.activemq.command.ConsumerId;
049import org.apache.activemq.command.ConsumerInfo;
050import org.apache.activemq.command.MessageAck;
051import org.apache.activemq.command.MessageDispatch;
052import org.apache.activemq.command.MessageId;
053import org.apache.activemq.command.MessagePull;
054import org.apache.activemq.command.RemoveInfo;
055import org.apache.activemq.command.TransactionId;
056import org.apache.activemq.management.JMSConsumerStatsImpl;
057import org.apache.activemq.management.StatsCapable;
058import org.apache.activemq.management.StatsImpl;
059import org.apache.activemq.selector.SelectorParser;
060import org.apache.activemq.transaction.Synchronization;
061import org.apache.activemq.util.Callback;
062import org.apache.activemq.util.IntrospectionSupport;
063import org.apache.activemq.util.JMSExceptionSupport;
064import org.apache.activemq.util.ThreadPoolUtils;
065import org.slf4j.Logger;
066import org.slf4j.LoggerFactory;
067
068/**
069 * A client uses a <CODE>MessageConsumer</CODE> object to receive messages
070 * from a destination. A <CODE> MessageConsumer</CODE> object is created by
071 * passing a <CODE>Destination</CODE> object to a message-consumer creation
072 * method supplied by a session.
073 * <P>
074 * <CODE>MessageConsumer</CODE> is the parent interface for all message
075 * consumers.
076 * <P>
077 * A message consumer can be created with a message selector. A message selector
078 * allows the client to restrict the messages delivered to the message consumer
079 * to those that match the selector.
080 * <P>
081 * A client may either synchronously receive a message consumer's messages or
082 * have the consumer asynchronously deliver them as they arrive.
083 * <P>
084 * For synchronous receipt, a client can request the next message from a message
085 * consumer using one of its <CODE> receive</CODE> methods. There are several
086 * variations of <CODE>receive</CODE> that allow a client to poll or wait for
087 * the next message.
088 * <P>
089 * For asynchronous delivery, a client can register a
090 * <CODE>MessageListener</CODE> object with a message consumer. As messages
091 * arrive at the message consumer, it delivers them by calling the
092 * <CODE>MessageListener</CODE>'s<CODE>
093 * onMessage</CODE> method.
094 * <P>
095 * It is a client programming error for a <CODE>MessageListener</CODE> to
096 * throw an exception.
097 *
098 *
099 * @see javax.jms.MessageConsumer
100 * @see javax.jms.QueueReceiver
101 * @see javax.jms.TopicSubscriber
102 * @see javax.jms.Session
103 */
104public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsCapable, ActiveMQDispatcher {
105
106    @SuppressWarnings("serial")
107    class PreviouslyDeliveredMap<K, V> extends HashMap<K, V> {
108        final TransactionId transactionId;
109        public PreviouslyDeliveredMap(TransactionId transactionId) {
110            this.transactionId = transactionId;
111        }
112    }
113
114    private static final Logger LOG = LoggerFactory.getLogger(ActiveMQMessageConsumer.class);
115    protected final ActiveMQSession session;
116    protected final ConsumerInfo info;
117
118    // These are the messages waiting to be delivered to the client
119    protected final MessageDispatchChannel unconsumedMessages;
120
121    // The are the messages that were delivered to the consumer but that have
122    // not been acknowledged. It's kept in reverse order since we
123    // Always walk list in reverse order.
124    protected final LinkedList<MessageDispatch> deliveredMessages = new LinkedList<MessageDispatch>();
125    // track duplicate deliveries in a transaction such that the tx integrity can be validated
126    private PreviouslyDeliveredMap<MessageId, Boolean> previouslyDeliveredMessages;
127    private int deliveredCounter;
128    private int additionalWindowSize;
129    private long redeliveryDelay;
130    private int ackCounter;
131    private int dispatchedCount;
132    private final AtomicReference<MessageListener> messageListener = new AtomicReference<MessageListener>();
133    private final JMSConsumerStatsImpl stats;
134
135    private final String selector;
136    private boolean synchronizationRegistered;
137    private final AtomicBoolean started = new AtomicBoolean(false);
138
139    private MessageAvailableListener availableListener;
140
141    private RedeliveryPolicy redeliveryPolicy;
142    private boolean optimizeAcknowledge;
143    private final AtomicBoolean deliveryingAcknowledgements = new AtomicBoolean();
144    private ExecutorService executorService;
145    private MessageTransformer transformer;
146    private boolean clearDeliveredList;
147    AtomicInteger inProgressClearRequiredFlag = new AtomicInteger(0);
148
149    private MessageAck pendingAck;
150    private long lastDeliveredSequenceId = -1;
151
152    private IOException failureError;
153
154    private long optimizeAckTimestamp = System.currentTimeMillis();
155    private long optimizeAcknowledgeTimeOut = 0;
156    private long optimizedAckScheduledAckInterval = 0;
157    private Runnable optimizedAckTask;
158    private long failoverRedeliveryWaitPeriod = 0;
159    private boolean transactedIndividualAck = false;
160    private boolean nonBlockingRedelivery = false;
161    private boolean consumerExpiryCheckEnabled = true;
162
163    /**
164     * Create a MessageConsumer
165     *
166     * @param session
167     * @param dest
168     * @param name
169     * @param selector
170     * @param prefetch
171     * @param maximumPendingMessageCount
172     * @param noLocal
173     * @param browser
174     * @param dispatchAsync
175     * @param messageListener
176     * @throws JMSException
177     */
178    public ActiveMQMessageConsumer(ActiveMQSession session, ConsumerId consumerId, ActiveMQDestination dest,
179            String name, String selector, int prefetch,
180            int maximumPendingMessageCount, boolean noLocal, boolean browser,
181            boolean dispatchAsync, MessageListener messageListener) throws JMSException {
182        if (dest == null) {
183            throw new InvalidDestinationException("Don't understand null destinations");
184        } else if (dest.getPhysicalName() == null) {
185            throw new InvalidDestinationException("The destination object was not given a physical name.");
186        } else if (dest.isTemporary()) {
187            String physicalName = dest.getPhysicalName();
188
189            if (physicalName == null) {
190                throw new IllegalArgumentException("Physical name of Destination should be valid: " + dest);
191            }
192
193            String connectionID = session.connection.getConnectionInfo().getConnectionId().getValue();
194
195            if (physicalName.indexOf(connectionID) < 0) {
196                throw new InvalidDestinationException("Cannot use a Temporary destination from another Connection");
197            }
198
199            if (session.connection.isDeleted(dest)) {
200                throw new InvalidDestinationException("Cannot use a Temporary destination that has been deleted");
201            }
202            if (prefetch < 0) {
203                throw new JMSException("Cannot have a prefetch size less than zero");
204            }
205        }
206        if (session.connection.isMessagePrioritySupported()) {
207            this.unconsumedMessages = new SimplePriorityMessageDispatchChannel();
208        }else {
209            this.unconsumedMessages = new FifoMessageDispatchChannel();
210        }
211
212        this.session = session;
213        this.redeliveryPolicy = session.connection.getRedeliveryPolicyMap().getEntryFor(dest);
214        setTransformer(session.getTransformer());
215
216        this.info = new ConsumerInfo(consumerId);
217        this.info.setExclusive(this.session.connection.isExclusiveConsumer());
218        this.info.setClientId(this.session.connection.getClientID());
219        this.info.setSubscriptionName(name);
220        this.info.setPrefetchSize(prefetch);
221        this.info.setCurrentPrefetchSize(prefetch);
222        this.info.setMaximumPendingMessageLimit(maximumPendingMessageCount);
223        this.info.setNoLocal(noLocal);
224        this.info.setDispatchAsync(dispatchAsync);
225        this.info.setRetroactive(this.session.connection.isUseRetroactiveConsumer());
226        this.info.setSelector(null);
227
228        // Allows the options on the destination to configure the consumerInfo
229        if (dest.getOptions() != null) {
230            Map<String, Object> options = IntrospectionSupport.extractProperties(
231                new HashMap<String, Object>(dest.getOptions()), "consumer.");
232            IntrospectionSupport.setProperties(this.info, options);
233            if (options.size() > 0) {
234                String msg = "There are " + options.size()
235                    + " consumer options that couldn't be set on the consumer."
236                    + " Check the options are spelled correctly."
237                    + " Unknown parameters=[" + options + "]."
238                    + " This consumer cannot be started.";
239                LOG.warn(msg);
240                throw new ConfigurationException(msg);
241            }
242        }
243
244        this.info.setDestination(dest);
245        this.info.setBrowser(browser);
246        if (selector != null && selector.trim().length() != 0) {
247            // Validate the selector
248            SelectorParser.parse(selector);
249            this.info.setSelector(selector);
250            this.selector = selector;
251        } else if (info.getSelector() != null) {
252            // Validate the selector
253            SelectorParser.parse(this.info.getSelector());
254            this.selector = this.info.getSelector();
255        } else {
256            this.selector = null;
257        }
258
259        this.stats = new JMSConsumerStatsImpl(session.getSessionStats(), dest);
260        this.optimizeAcknowledge = session.connection.isOptimizeAcknowledge() && session.isAutoAcknowledge()
261                                   && !info.isBrowser();
262        if (this.optimizeAcknowledge) {
263            this.optimizeAcknowledgeTimeOut = session.connection.getOptimizeAcknowledgeTimeOut();
264            setOptimizedAckScheduledAckInterval(session.connection.getOptimizedAckScheduledAckInterval());
265        }
266
267        this.info.setOptimizedAcknowledge(this.optimizeAcknowledge);
268        this.failoverRedeliveryWaitPeriod = session.connection.getConsumerFailoverRedeliveryWaitPeriod();
269        this.nonBlockingRedelivery = session.connection.isNonBlockingRedelivery();
270        this.transactedIndividualAck = session.connection.isTransactedIndividualAck()
271                        || this.nonBlockingRedelivery
272                        || session.connection.isMessagePrioritySupported();
273        this.consumerExpiryCheckEnabled = session.connection.isConsumerExpiryCheckEnabled();
274        if (messageListener != null) {
275            setMessageListener(messageListener);
276        }
277        try {
278            this.session.addConsumer(this);
279            this.session.syncSendPacket(info);
280        } catch (JMSException e) {
281            this.session.removeConsumer(this);
282            throw e;
283        }
284
285        if (session.connection.isStarted()) {
286            start();
287        }
288    }
289
290    private boolean isAutoAcknowledgeEach() {
291        return session.isAutoAcknowledge() || ( session.isDupsOkAcknowledge() && getDestination().isQueue() );
292    }
293
294    private boolean isAutoAcknowledgeBatch() {
295        return session.isDupsOkAcknowledge() && !getDestination().isQueue() ;
296    }
297
298    @Override
299    public StatsImpl getStats() {
300        return stats;
301    }
302
303    public JMSConsumerStatsImpl getConsumerStats() {
304        return stats;
305    }
306
307    public RedeliveryPolicy getRedeliveryPolicy() {
308        return redeliveryPolicy;
309    }
310
311    /**
312     * Sets the redelivery policy used when messages are redelivered
313     */
314    public void setRedeliveryPolicy(RedeliveryPolicy redeliveryPolicy) {
315        this.redeliveryPolicy = redeliveryPolicy;
316    }
317
318    public MessageTransformer getTransformer() {
319        return transformer;
320    }
321
322    /**
323     * Sets the transformer used to transform messages before they are sent on
324     * to the JMS bus
325     */
326    public void setTransformer(MessageTransformer transformer) {
327        this.transformer = transformer;
328    }
329
330    /**
331     * @return Returns the value.
332     */
333    public ConsumerId getConsumerId() {
334        return info.getConsumerId();
335    }
336
337    /**
338     * @return the consumer name - used for durable consumers
339     */
340    public String getConsumerName() {
341        return this.info.getSubscriptionName();
342    }
343
344    /**
345     * @return true if this consumer does not accept locally produced messages
346     */
347    protected boolean isNoLocal() {
348        return info.isNoLocal();
349    }
350
351    /**
352     * Retrieve is a browser
353     *
354     * @return true if a browser
355     */
356    protected boolean isBrowser() {
357        return info.isBrowser();
358    }
359
360    /**
361     * @return ActiveMQDestination
362     */
363    protected ActiveMQDestination getDestination() {
364        return info.getDestination();
365    }
366
367    /**
368     * @return Returns the prefetchNumber.
369     */
370    public int getPrefetchNumber() {
371        return info.getPrefetchSize();
372    }
373
374    /**
375     * @return true if this is a durable topic subscriber
376     */
377    public boolean isDurableSubscriber() {
378        return info.getSubscriptionName() != null && info.getDestination().isTopic();
379    }
380
381    /**
382     * Gets this message consumer's message selector expression.
383     *
384     * @return this message consumer's message selector, or null if no message
385     *         selector exists for the message consumer (that is, if the message
386     *         selector was not set or was set to null or the empty string)
387     * @throws JMSException if the JMS provider fails to receive the next
388     *                 message due to some internal error.
389     */
390    @Override
391    public String getMessageSelector() throws JMSException {
392        checkClosed();
393        return selector;
394    }
395
396    /**
397     * Gets the message consumer's <CODE>MessageListener</CODE>.
398     *
399     * @return the listener for the message consumer, or null if no listener is
400     *         set
401     * @throws JMSException if the JMS provider fails to get the message
402     *                 listener due to some internal error.
403     * @see javax.jms.MessageConsumer#setMessageListener(javax.jms.MessageListener)
404     */
405    @Override
406    public MessageListener getMessageListener() throws JMSException {
407        checkClosed();
408        return this.messageListener.get();
409    }
410
411    /**
412     * Sets the message consumer's <CODE>MessageListener</CODE>.
413     * <P>
414     * Setting the message listener to null is the equivalent of unsetting the
415     * message listener for the message consumer.
416     * <P>
417     * The effect of calling <CODE>MessageConsumer.setMessageListener</CODE>
418     * while messages are being consumed by an existing listener or the consumer
419     * is being used to consume messages synchronously is undefined.
420     *
421     * @param listener the listener to which the messages are to be delivered
422     * @throws JMSException if the JMS provider fails to receive the next
423     *                 message due to some internal error.
424     * @see javax.jms.MessageConsumer#getMessageListener
425     */
426    @Override
427    public void setMessageListener(MessageListener listener) throws JMSException {
428        checkClosed();
429        if (info.getPrefetchSize() == 0) {
430            throw new JMSException("Illegal prefetch size of zero. This setting is not supported for asynchronous consumers please set a value of at least 1");
431        }
432        if (listener != null) {
433            boolean wasRunning = session.isRunning();
434            if (wasRunning) {
435                session.stop();
436            }
437
438            this.messageListener.set(listener);
439            session.redispatch(this, unconsumedMessages);
440
441            if (wasRunning) {
442                session.start();
443            }
444        } else {
445            this.messageListener.set(null);
446        }
447    }
448
449    @Override
450    public MessageAvailableListener getAvailableListener() {
451        return availableListener;
452    }
453
454    /**
455     * Sets the listener used to notify synchronous consumers that there is a
456     * message available so that the {@link MessageConsumer#receiveNoWait()} can
457     * be called.
458     */
459    @Override
460    public void setAvailableListener(MessageAvailableListener availableListener) {
461        this.availableListener = availableListener;
462    }
463
464    /**
465     * Used to get an enqueued message from the unconsumedMessages list. The
466     * amount of time this method blocks is based on the timeout value. - if
467     * timeout==-1 then it blocks until a message is received. - if timeout==0
468     * then it it tries to not block at all, it returns a message if it is
469     * available - if timeout>0 then it blocks up to timeout amount of time.
470     * Expired messages will consumed by this method.
471     *
472     * @throws JMSException
473     * @return null if we timeout or if the consumer is closed.
474     */
475    private MessageDispatch dequeue(long timeout) throws JMSException {
476        try {
477            long deadline = 0;
478            if (timeout > 0) {
479                deadline = System.currentTimeMillis() + timeout;
480            }
481            while (true) {
482                MessageDispatch md = unconsumedMessages.dequeue(timeout);
483                if (md == null) {
484                    if (timeout > 0 && !unconsumedMessages.isClosed()) {
485                        timeout = Math.max(deadline - System.currentTimeMillis(), 0);
486                    } else {
487                        if (failureError != null) {
488                            throw JMSExceptionSupport.create(failureError);
489                        } else {
490                            return null;
491                        }
492                    }
493                } else if (md.getMessage() == null) {
494                    return null;
495                } else if (consumeExpiredMessage(md)) {
496                    LOG.debug("{} received expired message: {}", getConsumerId(), md);
497                    beforeMessageIsConsumed(md);
498                    afterMessageIsConsumed(md, true);
499                    if (timeout > 0) {
500                        timeout = Math.max(deadline - System.currentTimeMillis(), 0);
501                    }
502                    sendPullCommand(timeout);
503                } else if (redeliveryExceeded(md)) {
504                    LOG.debug("{} received with excessive redelivered: {}", getConsumerId(), md);
505                    posionAck(md, "dispatch to " + getConsumerId() + " exceeds redelivery policy limit:" + redeliveryPolicy);
506                    if (timeout > 0) {
507                        timeout = Math.max(deadline - System.currentTimeMillis(), 0);
508                    }
509                    sendPullCommand(timeout);
510                } else {
511                    if (LOG.isTraceEnabled()) {
512                        LOG.trace(getConsumerId() + " received message: " + md);
513                    }
514                    return md;
515                }
516            }
517        } catch (InterruptedException e) {
518            Thread.currentThread().interrupt();
519            throw JMSExceptionSupport.create(e);
520        }
521    }
522
523    private boolean consumeExpiredMessage(MessageDispatch dispatch) {
524        if (dispatch.getMessage().isExpired()) {
525            return !isBrowser() && isConsumerExpiryCheckEnabled();
526        }
527
528        return false;
529    }
530
531    private void posionAck(MessageDispatch md, String cause) throws JMSException {
532        MessageAck posionAck = new MessageAck(md, MessageAck.POSION_ACK_TYPE, 1);
533        posionAck.setFirstMessageId(md.getMessage().getMessageId());
534        posionAck.setPoisonCause(new Throwable(cause));
535        session.sendAck(posionAck);
536    }
537
538    private boolean redeliveryExceeded(MessageDispatch md) {
539        try {
540            return session.getTransacted()
541                    && redeliveryPolicy != null
542                    && redeliveryPolicy.getMaximumRedeliveries() != RedeliveryPolicy.NO_MAXIMUM_REDELIVERIES
543                    && md.getRedeliveryCounter() > redeliveryPolicy.getMaximumRedeliveries()
544                    // redeliveryCounter > x expected after resend via brokerRedeliveryPlugin
545                    && md.getMessage().getProperty("redeliveryDelay") == null;
546        } catch (Exception ignored) {
547            return false;
548        }
549    }
550
551    /**
552     * Receives the next message produced for this message consumer.
553     * <P>
554     * This call blocks indefinitely until a message is produced or until this
555     * message consumer is closed.
556     * <P>
557     * If this <CODE>receive</CODE> is done within a transaction, the consumer
558     * retains the message until the transaction commits.
559     *
560     * @return the next message produced for this message consumer, or null if
561     *         this message consumer is concurrently closed
562     */
563    @Override
564    public Message receive() throws JMSException {
565        checkClosed();
566        checkMessageListener();
567
568        sendPullCommand(0);
569        MessageDispatch md = dequeue(-1);
570        if (md == null) {
571            return null;
572        }
573
574        beforeMessageIsConsumed(md);
575        afterMessageIsConsumed(md, false);
576
577        return createActiveMQMessage(md);
578    }
579
580    /**
581     * @param md
582     *      the MessageDispatch that arrived from the Broker.
583     *
584     * @return an ActiveMQMessage initialized from the Message in the dispatch.
585     */
586    private ActiveMQMessage createActiveMQMessage(final MessageDispatch md) throws JMSException {
587        ActiveMQMessage m = (ActiveMQMessage)md.getMessage().copy();
588        if (m.getDataStructureType()==CommandTypes.ACTIVEMQ_BLOB_MESSAGE) {
589            ((ActiveMQBlobMessage)m).setBlobDownloader(new BlobDownloader(session.getBlobTransferPolicy()));
590        }
591        if (transformer != null) {
592            Message transformedMessage = transformer.consumerTransform(session, this, m);
593            if (transformedMessage != null) {
594                m = ActiveMQMessageTransformation.transformMessage(transformedMessage, session.connection);
595            }
596        }
597        if (session.isClientAcknowledge()) {
598            m.setAcknowledgeCallback(new Callback() {
599                @Override
600                public void execute() throws Exception {
601                    session.checkClosed();
602                    session.acknowledge();
603                }
604            });
605        } else if (session.isIndividualAcknowledge()) {
606            m.setAcknowledgeCallback(new Callback() {
607                @Override
608                public void execute() throws Exception {
609                    session.checkClosed();
610                    acknowledge(md);
611                }
612            });
613        }
614        return m;
615    }
616
617    /**
618     * Receives the next message that arrives within the specified timeout
619     * interval.
620     * <P>
621     * This call blocks until a message arrives, the timeout expires, or this
622     * message consumer is closed. A <CODE>timeout</CODE> of zero never
623     * expires, and the call blocks indefinitely.
624     *
625     * @param timeout the timeout value (in milliseconds), a time out of zero
626     *                never expires.
627     * @return the next message produced for this message consumer, or null if
628     *         the timeout expires or this message consumer is concurrently
629     *         closed
630     */
631    @Override
632    public Message receive(long timeout) throws JMSException {
633        checkClosed();
634        checkMessageListener();
635        if (timeout == 0) {
636            return this.receive();
637        }
638
639        sendPullCommand(timeout);
640        while (timeout > 0) {
641
642            MessageDispatch md;
643            if (info.getPrefetchSize() == 0) {
644                md = dequeue(-1); // We let the broker let us know when we timeout.
645            } else {
646                md = dequeue(timeout);
647            }
648
649            if (md == null) {
650                return null;
651            }
652
653            beforeMessageIsConsumed(md);
654            afterMessageIsConsumed(md, false);
655            return createActiveMQMessage(md);
656        }
657        return null;
658    }
659
660    /**
661     * Receives the next message if one is immediately available.
662     *
663     * @return the next message produced for this message consumer, or null if
664     *         one is not available
665     * @throws JMSException if the JMS provider fails to receive the next
666     *                 message due to some internal error.
667     */
668    @Override
669    public Message receiveNoWait() throws JMSException {
670        checkClosed();
671        checkMessageListener();
672        sendPullCommand(-1);
673
674        MessageDispatch md;
675        if (info.getPrefetchSize() == 0) {
676            md = dequeue(-1); // We let the broker let us know when we
677            // timeout.
678        } else {
679            md = dequeue(0);
680        }
681
682        if (md == null) {
683            return null;
684        }
685
686        beforeMessageIsConsumed(md);
687        afterMessageIsConsumed(md, false);
688        return createActiveMQMessage(md);
689    }
690
691    /**
692     * Closes the message consumer.
693     * <P>
694     * Since a provider may allocate some resources on behalf of a <CODE>
695     * MessageConsumer</CODE>
696     * outside the Java virtual machine, clients should close them when they are
697     * not needed. Relying on garbage collection to eventually reclaim these
698     * resources may not be timely enough.
699     * <P>
700     * This call blocks until a <CODE>receive</CODE> or message listener in
701     * progress has completed. A blocked message consumer <CODE>receive </CODE>
702     * call returns null when this message consumer is closed.
703     *
704     * @throws JMSException if the JMS provider fails to close the consumer due
705     *                 to some internal error.
706     */
707    @Override
708    public void close() throws JMSException {
709        if (!unconsumedMessages.isClosed()) {
710            if (!deliveredMessages.isEmpty() && session.getTransactionContext().isInTransaction()) {
711                session.getTransactionContext().addSynchronization(new Synchronization() {
712                    @Override
713                    public void afterCommit() throws Exception {
714                        doClose();
715                    }
716
717                    @Override
718                    public void afterRollback() throws Exception {
719                        doClose();
720                    }
721                });
722            } else {
723                doClose();
724            }
725        }
726    }
727
728    void doClose() throws JMSException {
729        dispose();
730        RemoveInfo removeCommand = info.createRemoveCommand();
731        LOG.debug("remove: {}, lastDeliveredSequenceId: {}", getConsumerId(), lastDeliveredSequenceId);
732        removeCommand.setLastDeliveredSequenceId(lastDeliveredSequenceId);
733        this.session.asyncSendPacket(removeCommand);
734    }
735
736    void inProgressClearRequired() {
737        inProgressClearRequiredFlag.incrementAndGet();
738        // deal with delivered messages async to avoid lock contention with in progress acks
739        clearDeliveredList = true;
740        // force a rollback if we will be acking in a transaction after/during failover
741        // bc acks are async they may not get there reliably on reconnect and the consumer
742        // may not be aware of the reconnect in a timely fashion if in onMessage
743        if (!deliveredMessages.isEmpty() && session.getTransactionContext().isInTransaction()) {
744            session.getTransactionContext().setRollbackOnly(true);
745        }
746    }
747
748    void clearMessagesInProgress() {
749        if (inProgressClearRequiredFlag.get() > 0) {
750            synchronized (unconsumedMessages.getMutex()) {
751                if (inProgressClearRequiredFlag.get() > 0) {
752                    LOG.debug("{} clearing unconsumed list ({}) on transport interrupt", getConsumerId(), unconsumedMessages.size());
753                    // ensure unconsumed are rolledback up front as they may get redelivered to another consumer
754                    List<MessageDispatch> list = unconsumedMessages.removeAll();
755                    if (!this.info.isBrowser()) {
756                        for (MessageDispatch old : list) {
757                            session.connection.rollbackDuplicate(this, old.getMessage());
758                        }
759                    }
760                    // allow dispatch on this connection to resume
761                    session.connection.transportInterruptionProcessingComplete();
762                    inProgressClearRequiredFlag.decrementAndGet();
763
764                    // Wake up any blockers and allow them to recheck state.
765                    unconsumedMessages.getMutex().notifyAll();
766                }
767            }
768        }
769        clearDeliveredList();
770    }
771
772    void deliverAcks() {
773        MessageAck ack = null;
774        if (deliveryingAcknowledgements.compareAndSet(false, true)) {
775            if (isAutoAcknowledgeEach()) {
776                synchronized(deliveredMessages) {
777                    ack = makeAckForAllDeliveredMessages(MessageAck.STANDARD_ACK_TYPE);
778                    if (ack != null) {
779                        deliveredMessages.clear();
780                        ackCounter = 0;
781                    } else {
782                        ack = pendingAck;
783                        pendingAck = null;
784                    }
785                }
786            } else if (pendingAck != null && pendingAck.isStandardAck()) {
787                ack = pendingAck;
788                pendingAck = null;
789            }
790            if (ack != null) {
791                final MessageAck ackToSend = ack;
792
793                if (executorService == null) {
794                    executorService = Executors.newSingleThreadExecutor();
795                }
796                executorService.submit(new Runnable() {
797                    @Override
798                    public void run() {
799                        try {
800                            session.sendAck(ackToSend,true);
801                        } catch (JMSException e) {
802                            LOG.error(getConsumerId() + " failed to delivered acknowledgements", e);
803                        } finally {
804                            deliveryingAcknowledgements.set(false);
805                        }
806                    }
807                });
808            } else {
809                deliveryingAcknowledgements.set(false);
810            }
811        }
812    }
813
814    public void dispose() throws JMSException {
815        if (!unconsumedMessages.isClosed()) {
816
817            // Do we have any acks we need to send out before closing?
818            // Ack any delivered messages now.
819            if (!session.getTransacted()) {
820                deliverAcks();
821                if (isAutoAcknowledgeBatch()) {
822                    acknowledge();
823                }
824            }
825            if (executorService != null) {
826                ThreadPoolUtils.shutdownGraceful(executorService, 60000L);
827                executorService = null;
828            }
829            if (optimizedAckTask != null) {
830                this.session.connection.getScheduler().cancel(optimizedAckTask);
831                optimizedAckTask = null;
832            }
833
834            if (session.isClientAcknowledge() || session.isIndividualAcknowledge()) {
835                if (!this.info.isBrowser()) {
836                    // rollback duplicates that aren't acknowledged
837                    List<MessageDispatch> tmp = null;
838                    synchronized (this.deliveredMessages) {
839                        tmp = new ArrayList<MessageDispatch>(this.deliveredMessages);
840                    }
841                    for (MessageDispatch old : tmp) {
842                        this.session.connection.rollbackDuplicate(this, old.getMessage());
843                    }
844                    tmp.clear();
845                }
846            }
847            if (!session.isTransacted()) {
848                synchronized(deliveredMessages) {
849                    deliveredMessages.clear();
850                }
851            }
852            unconsumedMessages.close();
853            this.session.removeConsumer(this);
854            List<MessageDispatch> list = unconsumedMessages.removeAll();
855            if (!this.info.isBrowser()) {
856                for (MessageDispatch old : list) {
857                    // ensure we don't filter this as a duplicate
858                    LOG.debug("on close, rollback duplicate: {}", old.getMessage().getMessageId());
859                    session.connection.rollbackDuplicate(this, old.getMessage());
860                }
861            }
862        }
863    }
864
865    /**
866     * @throws IllegalStateException
867     */
868    protected void checkClosed() throws IllegalStateException {
869        if (unconsumedMessages.isClosed()) {
870            throw new IllegalStateException("The Consumer is closed");
871        }
872    }
873
874    /**
875     * If we have a zero prefetch specified then send a pull command to the
876     * broker to pull a message we are about to receive
877     */
878    protected void sendPullCommand(long timeout) throws JMSException {
879        clearDeliveredList();
880        if (info.getCurrentPrefetchSize() == 0 && unconsumedMessages.isEmpty()) {
881            MessagePull messagePull = new MessagePull();
882            messagePull.configure(info);
883            messagePull.setTimeout(timeout);
884            session.asyncSendPacket(messagePull);
885        }
886    }
887
888    protected void checkMessageListener() throws JMSException {
889        session.checkMessageListener();
890    }
891
892    protected void setOptimizeAcknowledge(boolean value) {
893        if (optimizeAcknowledge && !value) {
894            deliverAcks();
895        }
896        optimizeAcknowledge = value;
897    }
898
899    protected void setPrefetchSize(int prefetch) {
900        deliverAcks();
901        this.info.setCurrentPrefetchSize(prefetch);
902    }
903
904    private void beforeMessageIsConsumed(MessageDispatch md) throws JMSException {
905        md.setDeliverySequenceId(session.getNextDeliveryId());
906        lastDeliveredSequenceId = md.getMessage().getMessageId().getBrokerSequenceId();
907        if (!isAutoAcknowledgeBatch()) {
908            synchronized(deliveredMessages) {
909                deliveredMessages.addFirst(md);
910            }
911            if (session.getTransacted()) {
912                if (transactedIndividualAck) {
913                    immediateIndividualTransactedAck(md);
914                } else {
915                    ackLater(md, MessageAck.DELIVERED_ACK_TYPE);
916                }
917            }
918        }
919    }
920
921    private void immediateIndividualTransactedAck(MessageDispatch md) throws JMSException {
922        // acks accumulate on the broker pending transaction completion to indicate
923        // delivery status
924        registerSync();
925        MessageAck ack = new MessageAck(md, MessageAck.INDIVIDUAL_ACK_TYPE, 1);
926        ack.setTransactionId(session.getTransactionContext().getTransactionId());
927        session.syncSendPacket(ack);
928    }
929
930    private void afterMessageIsConsumed(MessageDispatch md, boolean messageExpired) throws JMSException {
931        if (unconsumedMessages.isClosed()) {
932            return;
933        }
934        if (messageExpired) {
935            acknowledge(md, MessageAck.EXPIRED_ACK_TYPE);
936            stats.getExpiredMessageCount().increment();
937        } else {
938            stats.onMessage();
939            if (session.getTransacted()) {
940                // Do nothing.
941            } else if (isAutoAcknowledgeEach()) {
942                if (deliveryingAcknowledgements.compareAndSet(false, true)) {
943                    synchronized (deliveredMessages) {
944                        if (!deliveredMessages.isEmpty()) {
945                            if (optimizeAcknowledge) {
946                                ackCounter++;
947
948                                // AMQ-3956 evaluate both expired and normal msgs as
949                                // otherwise consumer may get stalled
950                                if (ackCounter + deliveredCounter >= (info.getPrefetchSize() * .65) || (optimizeAcknowledgeTimeOut > 0 && System.currentTimeMillis() >= (optimizeAckTimestamp + optimizeAcknowledgeTimeOut))) {
951                                    MessageAck ack = makeAckForAllDeliveredMessages(MessageAck.STANDARD_ACK_TYPE);
952                                    if (ack != null) {
953                                        deliveredMessages.clear();
954                                        ackCounter = 0;
955                                        session.sendAck(ack);
956                                        optimizeAckTimestamp = System.currentTimeMillis();
957                                    }
958                                    // AMQ-3956 - as further optimization send
959                                    // ack for expired msgs when there are any.
960                                    // This resets the deliveredCounter to 0 so that
961                                    // we won't sent standard acks with every msg just
962                                    // because the deliveredCounter just below
963                                    // 0.5 * prefetch as used in ackLater()
964                                    if (pendingAck != null && deliveredCounter > 0) {
965                                        session.sendAck(pendingAck);
966                                        pendingAck = null;
967                                        deliveredCounter = 0;
968                                    }
969                                }
970                            } else {
971                                MessageAck ack = makeAckForAllDeliveredMessages(MessageAck.STANDARD_ACK_TYPE);
972                                if (ack!=null) {
973                                    deliveredMessages.clear();
974                                    session.sendAck(ack);
975                                }
976                            }
977                        }
978                    }
979                    deliveryingAcknowledgements.set(false);
980                }
981            } else if (isAutoAcknowledgeBatch()) {
982                ackLater(md, MessageAck.STANDARD_ACK_TYPE);
983            } else if (session.isClientAcknowledge()||session.isIndividualAcknowledge()) {
984                boolean messageUnackedByConsumer = false;
985                synchronized (deliveredMessages) {
986                    messageUnackedByConsumer = deliveredMessages.contains(md);
987                }
988                if (messageUnackedByConsumer) {
989                    ackLater(md, MessageAck.DELIVERED_ACK_TYPE);
990                }
991            }
992            else {
993                throw new IllegalStateException("Invalid session state.");
994            }
995        }
996    }
997
998    /**
999     * Creates a MessageAck for all messages contained in deliveredMessages.
1000     * Caller should hold the lock for deliveredMessages.
1001     *
1002     * @param type Ack-Type (i.e. MessageAck.STANDARD_ACK_TYPE)
1003     * @return <code>null</code> if nothing to ack.
1004     */
1005    private MessageAck makeAckForAllDeliveredMessages(byte type) {
1006        synchronized (deliveredMessages) {
1007            if (deliveredMessages.isEmpty()) {
1008                return null;
1009            }
1010
1011            MessageDispatch md = deliveredMessages.getFirst();
1012            MessageAck ack = new MessageAck(md, type, deliveredMessages.size());
1013            ack.setFirstMessageId(deliveredMessages.getLast().getMessage().getMessageId());
1014            return ack;
1015        }
1016    }
1017
1018    private void ackLater(MessageDispatch md, byte ackType) throws JMSException {
1019
1020        // Don't acknowledge now, but we may need to let the broker know the
1021        // consumer got the message to expand the pre-fetch window
1022        if (session.getTransacted()) {
1023            registerSync();
1024        }
1025
1026        deliveredCounter++;
1027
1028        MessageAck oldPendingAck = pendingAck;
1029        pendingAck = new MessageAck(md, ackType, deliveredCounter);
1030        pendingAck.setTransactionId(session.getTransactionContext().getTransactionId());
1031        if( oldPendingAck==null ) {
1032            pendingAck.setFirstMessageId(pendingAck.getLastMessageId());
1033        } else if ( oldPendingAck.getAckType() == pendingAck.getAckType() ) {
1034            pendingAck.setFirstMessageId(oldPendingAck.getFirstMessageId());
1035        } else {
1036            // old pending ack being superseded by ack of another type, if is is not a delivered
1037            // ack and hence important, send it now so it is not lost.
1038            if (!oldPendingAck.isDeliveredAck()) {
1039                LOG.debug("Sending old pending ack {}, new pending: {}", oldPendingAck, pendingAck);
1040                session.sendAck(oldPendingAck);
1041            } else {
1042                LOG.debug("dropping old pending ack {}, new pending: {}", oldPendingAck, pendingAck);
1043            }
1044        }
1045        // AMQ-3956 evaluate both expired and normal msgs as
1046        // otherwise consumer may get stalled
1047        if ((0.5 * info.getPrefetchSize()) <= (deliveredCounter + ackCounter - additionalWindowSize)) {
1048            LOG.debug("ackLater: sending: {}", pendingAck);
1049            session.sendAck(pendingAck);
1050            pendingAck=null;
1051            deliveredCounter = 0;
1052            additionalWindowSize = 0;
1053        }
1054    }
1055
1056    private void registerSync() throws JMSException {
1057        session.doStartTransaction();
1058        if (!synchronizationRegistered) {
1059            synchronizationRegistered = true;
1060            session.getTransactionContext().addSynchronization(new Synchronization() {
1061                @Override
1062                public void beforeEnd() throws Exception {
1063                    if (transactedIndividualAck) {
1064                        clearDeliveredList();
1065                        waitForRedeliveries();
1066                        synchronized(deliveredMessages) {
1067                            rollbackOnFailedRecoveryRedelivery();
1068                        }
1069                    } else {
1070                        acknowledge();
1071                    }
1072                    synchronizationRegistered = false;
1073                }
1074
1075                @Override
1076                public void afterCommit() throws Exception {
1077                    commit();
1078                    synchronizationRegistered = false;
1079                }
1080
1081                @Override
1082                public void afterRollback() throws Exception {
1083                    rollback();
1084                    synchronizationRegistered = false;
1085                }
1086            });
1087        }
1088    }
1089
1090    /**
1091     * Acknowledge all the messages that have been delivered to the client up to
1092     * this point.
1093     *
1094     * @throws JMSException
1095     */
1096    public void acknowledge() throws JMSException {
1097        clearDeliveredList();
1098        waitForRedeliveries();
1099        synchronized(deliveredMessages) {
1100            // Acknowledge all messages so far.
1101            MessageAck ack = makeAckForAllDeliveredMessages(MessageAck.STANDARD_ACK_TYPE);
1102            if (ack == null) {
1103                return; // no msgs
1104            }
1105
1106            if (session.getTransacted()) {
1107                rollbackOnFailedRecoveryRedelivery();
1108                session.doStartTransaction();
1109                ack.setTransactionId(session.getTransactionContext().getTransactionId());
1110            }
1111
1112            pendingAck = null;
1113            session.sendAck(ack);
1114
1115            // Adjust the counters
1116            deliveredCounter = Math.max(0, deliveredCounter - deliveredMessages.size());
1117            additionalWindowSize = Math.max(0, additionalWindowSize - deliveredMessages.size());
1118
1119            if (!session.getTransacted()) {
1120                deliveredMessages.clear();
1121            }
1122        }
1123    }
1124
1125    private void waitForRedeliveries() {
1126        if (failoverRedeliveryWaitPeriod > 0 && previouslyDeliveredMessages != null) {
1127            long expiry = System.currentTimeMillis() + failoverRedeliveryWaitPeriod;
1128            int numberNotReplayed;
1129            do {
1130                numberNotReplayed = 0;
1131                synchronized(deliveredMessages) {
1132                    if (previouslyDeliveredMessages != null) {
1133                        for (Entry<MessageId, Boolean> entry: previouslyDeliveredMessages.entrySet()) {
1134                            if (!entry.getValue()) {
1135                                numberNotReplayed++;
1136                            }
1137                        }
1138                    }
1139                }
1140                if (numberNotReplayed > 0) {
1141                    LOG.info("waiting for redelivery of {} in transaction: {}, to consumer: {}",
1142                             numberNotReplayed, this.getConsumerId(), previouslyDeliveredMessages.transactionId);
1143                    try {
1144                        Thread.sleep(Math.max(500, failoverRedeliveryWaitPeriod/4));
1145                    } catch (InterruptedException outOfhere) {
1146                        break;
1147                    }
1148                }
1149            } while (numberNotReplayed > 0 && expiry < System.currentTimeMillis());
1150        }
1151    }
1152
1153    /*
1154     * called with deliveredMessages locked
1155     */
1156    private void rollbackOnFailedRecoveryRedelivery() throws JMSException {
1157        if (previouslyDeliveredMessages != null) {
1158            // if any previously delivered messages was not re-delivered, transaction is invalid and must rollback
1159            // as messages have been dispatched else where.
1160            int numberNotReplayed = 0;
1161            for (Entry<MessageId, Boolean> entry: previouslyDeliveredMessages.entrySet()) {
1162                if (!entry.getValue()) {
1163                    numberNotReplayed++;
1164                    LOG.debug("previously delivered message has not been replayed in transaction: {}, messageId: {}",
1165                              previouslyDeliveredMessages.transactionId, entry.getKey());
1166                }
1167            }
1168            if (numberNotReplayed > 0) {
1169                String message = "rolling back transaction ("
1170                    + previouslyDeliveredMessages.transactionId + ") post failover recovery. " + numberNotReplayed
1171                    + " previously delivered message(s) not replayed to consumer: " + this.getConsumerId();
1172                LOG.warn(message);
1173                throw new TransactionRolledBackException(message);
1174            }
1175        }
1176    }
1177
1178    void acknowledge(MessageDispatch md) throws JMSException {
1179        acknowledge(md, MessageAck.INDIVIDUAL_ACK_TYPE);
1180    }
1181
1182    void acknowledge(MessageDispatch md, byte ackType) throws JMSException {
1183        MessageAck ack = new MessageAck(md, ackType, 1);
1184        if (ack.isExpiredAck()) {
1185            ack.setFirstMessageId(ack.getLastMessageId());
1186        }
1187        session.sendAck(ack);
1188        synchronized(deliveredMessages){
1189            deliveredMessages.remove(md);
1190        }
1191    }
1192
1193    public void commit() throws JMSException {
1194        synchronized (deliveredMessages) {
1195            deliveredMessages.clear();
1196            clearPreviouslyDelivered();
1197        }
1198        redeliveryDelay = 0;
1199    }
1200
1201    public void rollback() throws JMSException {
1202        clearDeliveredList();
1203        synchronized (unconsumedMessages.getMutex()) {
1204            if (optimizeAcknowledge) {
1205                // remove messages read but not acked at the broker yet through
1206                // optimizeAcknowledge
1207                if (!this.info.isBrowser()) {
1208                    synchronized(deliveredMessages) {
1209                        for (int i = 0; (i < deliveredMessages.size()) && (i < ackCounter); i++) {
1210                            // ensure we don't filter this as a duplicate
1211                            MessageDispatch md = deliveredMessages.removeLast();
1212                            session.connection.rollbackDuplicate(this, md.getMessage());
1213                        }
1214                    }
1215                }
1216            }
1217            synchronized(deliveredMessages) {
1218                rollbackPreviouslyDeliveredAndNotRedelivered();
1219                if (deliveredMessages.isEmpty()) {
1220                    return;
1221                }
1222
1223                // use initial delay for first redelivery
1224                MessageDispatch lastMd = deliveredMessages.getFirst();
1225                final int currentRedeliveryCount = lastMd.getMessage().getRedeliveryCounter();
1226                if (currentRedeliveryCount > 0) {
1227                    redeliveryDelay = redeliveryPolicy.getNextRedeliveryDelay(redeliveryDelay);
1228                } else {
1229                    redeliveryDelay = redeliveryPolicy.getInitialRedeliveryDelay();
1230                }
1231                MessageId firstMsgId = deliveredMessages.getLast().getMessage().getMessageId();
1232
1233                for (Iterator<MessageDispatch> iter = deliveredMessages.iterator(); iter.hasNext();) {
1234                    MessageDispatch md = iter.next();
1235                    md.getMessage().onMessageRolledBack();
1236                    // ensure we don't filter this as a duplicate
1237                    session.connection.rollbackDuplicate(this, md.getMessage());
1238                }
1239
1240                if (redeliveryPolicy.getMaximumRedeliveries() != RedeliveryPolicy.NO_MAXIMUM_REDELIVERIES
1241                    && lastMd.getMessage().getRedeliveryCounter() > redeliveryPolicy.getMaximumRedeliveries()) {
1242                    // We need to NACK the messages so that they get sent to the
1243                    // DLQ.
1244                    // Acknowledge the last message.
1245
1246                    MessageAck ack = new MessageAck(lastMd, MessageAck.POSION_ACK_TYPE, deliveredMessages.size());
1247                    ack.setFirstMessageId(firstMsgId);
1248                    ack.setPoisonCause(new Throwable("Exceeded redelivery policy limit:" + redeliveryPolicy
1249                            + ", cause:" + lastMd.getRollbackCause(), lastMd.getRollbackCause()));
1250                    session.sendAck(ack,true);
1251                    // Adjust the window size.
1252                    additionalWindowSize = Math.max(0, additionalWindowSize - deliveredMessages.size());
1253                    redeliveryDelay = 0;
1254
1255                    deliveredCounter -= deliveredMessages.size();
1256                    deliveredMessages.clear();
1257
1258                } else {
1259
1260                    // only redelivery_ack after first delivery
1261                    if (currentRedeliveryCount > 0) {
1262                        MessageAck ack = new MessageAck(lastMd, MessageAck.REDELIVERED_ACK_TYPE, deliveredMessages.size());
1263                        ack.setFirstMessageId(firstMsgId);
1264                        session.sendAck(ack,true);
1265                    }
1266
1267                    // stop the delivery of messages.
1268                    if (nonBlockingRedelivery) {
1269                        if (!unconsumedMessages.isClosed()) {
1270
1271                            final LinkedList<MessageDispatch> pendingRedeliveries =
1272                                new LinkedList<MessageDispatch>(deliveredMessages);
1273
1274                            Collections.reverse(pendingRedeliveries);
1275
1276                            deliveredCounter -= deliveredMessages.size();
1277                            deliveredMessages.clear();
1278
1279                            // Start up the delivery again a little later.
1280                            session.getScheduler().executeAfterDelay(new Runnable() {
1281                                @Override
1282                                public void run() {
1283                                    try {
1284                                        if (!unconsumedMessages.isClosed()) {
1285                                            for(MessageDispatch dispatch : pendingRedeliveries) {
1286                                                session.dispatch(dispatch);
1287                                            }
1288                                        }
1289                                    } catch (Exception e) {
1290                                        session.connection.onAsyncException(e);
1291                                    }
1292                                }
1293                            }, redeliveryDelay);
1294                        }
1295
1296                    } else {
1297                        unconsumedMessages.stop();
1298
1299                        for (MessageDispatch md : deliveredMessages) {
1300                            unconsumedMessages.enqueueFirst(md);
1301                        }
1302
1303                        deliveredCounter -= deliveredMessages.size();
1304                        deliveredMessages.clear();
1305
1306                        if (redeliveryDelay > 0 && !unconsumedMessages.isClosed()) {
1307                            // Start up the delivery again a little later.
1308                            session.getScheduler().executeAfterDelay(new Runnable() {
1309                                @Override
1310                                public void run() {
1311                                    try {
1312                                        if (started.get()) {
1313                                            start();
1314                                        }
1315                                    } catch (JMSException e) {
1316                                        session.connection.onAsyncException(e);
1317                                    }
1318                                }
1319                            }, redeliveryDelay);
1320                        } else {
1321                            start();
1322                        }
1323                    }
1324                }
1325            }
1326        }
1327        if (messageListener.get() != null) {
1328            session.redispatch(this, unconsumedMessages);
1329        }
1330    }
1331
1332    /*
1333     * called with unconsumedMessages && deliveredMessages locked
1334     * remove any message not re-delivered as they can't be replayed to this
1335     * consumer on rollback
1336     */
1337    private void rollbackPreviouslyDeliveredAndNotRedelivered() {
1338        if (previouslyDeliveredMessages != null) {
1339            for (Entry<MessageId, Boolean> entry: previouslyDeliveredMessages.entrySet()) {
1340                if (!entry.getValue()) {
1341                    LOG.trace("rollback non redelivered: {}" + entry.getKey());
1342                    removeFromDeliveredMessages(entry.getKey());
1343                }
1344            }
1345            clearPreviouslyDelivered();
1346        }
1347    }
1348
1349    /*
1350     * called with deliveredMessages locked
1351     */
1352    private void removeFromDeliveredMessages(MessageId key) {
1353        Iterator<MessageDispatch> iterator = deliveredMessages.iterator();
1354        while (iterator.hasNext()) {
1355            MessageDispatch candidate = iterator.next();
1356            if (key.equals(candidate.getMessage().getMessageId())) {
1357                session.connection.rollbackDuplicate(this, candidate.getMessage());
1358                iterator.remove();
1359                break;
1360            }
1361        }
1362    }
1363
1364    /*
1365     * called with deliveredMessages locked
1366     */
1367    private void clearPreviouslyDelivered() {
1368        if (previouslyDeliveredMessages != null) {
1369            previouslyDeliveredMessages.clear();
1370            previouslyDeliveredMessages = null;
1371        }
1372    }
1373
1374    @Override
1375    public void dispatch(MessageDispatch md) {
1376        MessageListener listener = this.messageListener.get();
1377        try {
1378            clearMessagesInProgress();
1379            clearDeliveredList();
1380            synchronized (unconsumedMessages.getMutex()) {
1381                if (!unconsumedMessages.isClosed()) {
1382                    if (this.info.isBrowser() || !session.connection.isDuplicate(this, md.getMessage())) {
1383                        if (listener != null && unconsumedMessages.isRunning()) {
1384                            if (redeliveryExceeded(md)) {
1385                                posionAck(md, "dispatch to " + getConsumerId() + " exceeds redelivery policy limit:" + redeliveryPolicy);
1386                                return;
1387                            }
1388                            ActiveMQMessage message = createActiveMQMessage(md);
1389                            beforeMessageIsConsumed(md);
1390                            try {
1391                                boolean expired = isConsumerExpiryCheckEnabled() && message.isExpired();
1392                                if (!expired) {
1393                                    listener.onMessage(message);
1394                                }
1395                                afterMessageIsConsumed(md, expired);
1396                            } catch (RuntimeException e) {
1397                                LOG.error("{} Exception while processing message: {}", getConsumerId(), md.getMessage().getMessageId(), e);
1398                                md.setRollbackCause(e);
1399                                if (isAutoAcknowledgeBatch() || isAutoAcknowledgeEach() || session.isIndividualAcknowledge()) {
1400                                    // schedual redelivery and possible dlq processing
1401                                    rollback();
1402                                } else {
1403                                    // Transacted or Client ack: Deliver the next message.
1404                                    afterMessageIsConsumed(md, false);
1405                                }
1406                            }
1407                        } else {
1408                            if (!unconsumedMessages.isRunning()) {
1409                                // delayed redelivery, ensure it can be re delivered
1410                                session.connection.rollbackDuplicate(this, md.getMessage());
1411                            }
1412                            unconsumedMessages.enqueue(md);
1413                            if (availableListener != null) {
1414                                availableListener.onMessageAvailable(this);
1415                            }
1416                        }
1417                    } else {
1418                        // deal with duplicate delivery
1419                        ConsumerId consumerWithPendingTransaction;
1420                        if (redeliveryExpectedInCurrentTransaction(md, true)) {
1421                            LOG.debug("{} tracking transacted redelivery {}", getConsumerId(), md.getMessage());
1422                            if (transactedIndividualAck) {
1423                                immediateIndividualTransactedAck(md);
1424                            } else {
1425                                session.sendAck(new MessageAck(md, MessageAck.DELIVERED_ACK_TYPE, 1));
1426                            }
1427                        } else if ((consumerWithPendingTransaction = redeliveryPendingInCompetingTransaction(md)) != null) {
1428                            LOG.warn("{} delivering duplicate {}, pending transaction completion on {} will rollback", getConsumerId(), md.getMessage(), consumerWithPendingTransaction);
1429                            session.getConnection().rollbackDuplicate(this, md.getMessage());
1430                            dispatch(md);
1431                        } else {
1432                            LOG.warn("{} suppressing duplicate delivery on connection, poison acking: {}", getConsumerId(), md);
1433                            posionAck(md, "Suppressing duplicate delivery on connection, consumer " + getConsumerId());
1434                        }
1435                    }
1436                }
1437            }
1438            if (++dispatchedCount % 1000 == 0) {
1439                dispatchedCount = 0;
1440                Thread.yield();
1441            }
1442        } catch (Exception e) {
1443            session.connection.onClientInternalException(e);
1444        }
1445    }
1446
1447    private boolean redeliveryExpectedInCurrentTransaction(MessageDispatch md, boolean markReceipt) {
1448        if (session.isTransacted()) {
1449            synchronized (deliveredMessages) {
1450                if (previouslyDeliveredMessages != null) {
1451                    if (previouslyDeliveredMessages.containsKey(md.getMessage().getMessageId())) {
1452                        if (markReceipt) {
1453                            previouslyDeliveredMessages.put(md.getMessage().getMessageId(), true);
1454                        }
1455                        return true;
1456                    }
1457                }
1458            }
1459        }
1460        return false;
1461    }
1462
1463    private ConsumerId redeliveryPendingInCompetingTransaction(MessageDispatch md) {
1464        for (ActiveMQSession activeMQSession: session.connection.getSessions()) {
1465            for (ActiveMQMessageConsumer activeMQMessageConsumer : activeMQSession.consumers) {
1466                if (activeMQMessageConsumer.redeliveryExpectedInCurrentTransaction(md, false)) {
1467                    return activeMQMessageConsumer.getConsumerId();
1468                }
1469            }
1470        }
1471        return null;
1472    }
1473
1474    // async (on next call) clear or track delivered as they may be flagged as duplicates if they arrive again
1475    private void clearDeliveredList() {
1476        if (clearDeliveredList) {
1477            synchronized (deliveredMessages) {
1478                if (clearDeliveredList) {
1479                    if (!deliveredMessages.isEmpty()) {
1480                        if (session.isTransacted()) {
1481
1482                            if (previouslyDeliveredMessages == null) {
1483                                previouslyDeliveredMessages = new PreviouslyDeliveredMap<MessageId, Boolean>(session.getTransactionContext().getTransactionId());
1484                            }
1485                            for (MessageDispatch delivered : deliveredMessages) {
1486                                previouslyDeliveredMessages.put(delivered.getMessage().getMessageId(), false);
1487                            }
1488                            LOG.debug("{} tracking existing transacted {} delivered list ({}) on transport interrupt",
1489                                      getConsumerId(), previouslyDeliveredMessages.transactionId, deliveredMessages.size());
1490                        } else {
1491                            if (session.isClientAcknowledge()) {
1492                                LOG.debug("{} rolling back delivered list ({}) on transport interrupt", getConsumerId(), deliveredMessages.size());
1493                                // allow redelivery
1494                                if (!this.info.isBrowser()) {
1495                                    for (MessageDispatch md: deliveredMessages) {
1496                                        this.session.connection.rollbackDuplicate(this, md.getMessage());
1497                                    }
1498                                }
1499                            }
1500                            LOG.debug("{} clearing delivered list ({}) on transport interrupt", getConsumerId(), deliveredMessages.size());
1501                            deliveredMessages.clear();
1502                            pendingAck = null;
1503                        }
1504                    }
1505                    clearDeliveredList = false;
1506                }
1507            }
1508        }
1509    }
1510
1511    public int getMessageSize() {
1512        return unconsumedMessages.size();
1513    }
1514
1515    public void start() throws JMSException {
1516        if (unconsumedMessages.isClosed()) {
1517            return;
1518        }
1519        started.set(true);
1520        unconsumedMessages.start();
1521        session.executor.wakeup();
1522    }
1523
1524    public void stop() {
1525        started.set(false);
1526        unconsumedMessages.stop();
1527    }
1528
1529    @Override
1530    public String toString() {
1531        return "ActiveMQMessageConsumer { value=" + info.getConsumerId() + ", started=" + started.get()
1532               + " }";
1533    }
1534
1535    /**
1536     * Delivers a message to the message listener.
1537     *
1538     * @return true if another execution is needed.
1539     *
1540     * @throws JMSException
1541     */
1542    public boolean iterate() {
1543        MessageListener listener = this.messageListener.get();
1544        if (listener != null) {
1545            MessageDispatch md = unconsumedMessages.dequeueNoWait();
1546            if (md != null) {
1547                dispatch(md);
1548                return true;
1549            }
1550        }
1551        return false;
1552    }
1553
1554    public boolean isInUse(ActiveMQTempDestination destination) {
1555        return info.getDestination().equals(destination);
1556    }
1557
1558    public long getLastDeliveredSequenceId() {
1559        return lastDeliveredSequenceId;
1560    }
1561
1562    public IOException getFailureError() {
1563        return failureError;
1564    }
1565
1566    public void setFailureError(IOException failureError) {
1567        this.failureError = failureError;
1568    }
1569
1570    /**
1571     * @return the optimizedAckScheduledAckInterval
1572     */
1573    public long getOptimizedAckScheduledAckInterval() {
1574        return optimizedAckScheduledAckInterval;
1575    }
1576
1577    /**
1578     * @param optimizedAckScheduledAckInterval the optimizedAckScheduledAckInterval to set
1579     */
1580    public void setOptimizedAckScheduledAckInterval(long optimizedAckScheduledAckInterval) throws JMSException {
1581        this.optimizedAckScheduledAckInterval = optimizedAckScheduledAckInterval;
1582
1583        if (this.optimizedAckTask != null) {
1584            try {
1585                this.session.connection.getScheduler().cancel(optimizedAckTask);
1586            } catch (JMSException e) {
1587                LOG.debug("Caught exception while cancelling old optimized ack task", e);
1588                throw e;
1589            }
1590            this.optimizedAckTask = null;
1591        }
1592
1593        // Should we periodically send out all outstanding acks.
1594        if (this.optimizeAcknowledge && this.optimizedAckScheduledAckInterval > 0) {
1595            this.optimizedAckTask = new Runnable() {
1596
1597                @Override
1598                public void run() {
1599                    try {
1600                        if (optimizeAcknowledge && !unconsumedMessages.isClosed()) {
1601                            LOG.info("Consumer:{} is performing scheduled delivery of outstanding optimized Acks", info.getConsumerId());
1602                            deliverAcks();
1603                        }
1604                    } catch (Exception e) {
1605                        LOG.debug("Optimized Ack Task caught exception during ack", e);
1606                    }
1607                }
1608            };
1609
1610            try {
1611                this.session.connection.getScheduler().executePeriodically(optimizedAckTask, optimizedAckScheduledAckInterval);
1612            } catch (JMSException e) {
1613                LOG.debug("Caught exception while scheduling new optimized ack task", e);
1614                throw e;
1615            }
1616        }
1617    }
1618
1619    public boolean hasMessageListener() {
1620        return messageListener.get() != null;
1621    }
1622
1623    public boolean isConsumerExpiryCheckEnabled() {
1624        return consumerExpiryCheckEnabled;
1625    }
1626
1627    public void setConsumerExpiryCheckEnabled(boolean consumerExpiryCheckEnabled) {
1628        this.consumerExpiryCheckEnabled = consumerExpiryCheckEnabled;
1629    }
1630}