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