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}