001/** 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.activemq.broker.region; 018 019import java.io.IOException; 020import java.util.ArrayList; 021import java.util.LinkedList; 022import java.util.List; 023import java.util.concurrent.atomic.AtomicLong; 024 025import javax.jms.JMSException; 026 027import org.apache.activemq.ActiveMQMessageAudit; 028import org.apache.activemq.broker.Broker; 029import org.apache.activemq.broker.ConnectionContext; 030import org.apache.activemq.broker.region.cursors.FilePendingMessageCursor; 031import org.apache.activemq.broker.region.cursors.PendingMessageCursor; 032import org.apache.activemq.broker.region.cursors.VMPendingMessageCursor; 033import org.apache.activemq.broker.region.policy.MessageEvictionStrategy; 034import org.apache.activemq.broker.region.policy.OldestMessageEvictionStrategy; 035import org.apache.activemq.command.ConsumerControl; 036import org.apache.activemq.command.ConsumerInfo; 037import org.apache.activemq.command.Message; 038import org.apache.activemq.command.MessageAck; 039import org.apache.activemq.command.MessageDispatch; 040import org.apache.activemq.command.MessageDispatchNotification; 041import org.apache.activemq.command.MessageId; 042import org.apache.activemq.command.MessagePull; 043import org.apache.activemq.command.Response; 044import org.apache.activemq.thread.Scheduler; 045import org.apache.activemq.transaction.Synchronization; 046import org.apache.activemq.transport.TransmitCallback; 047import org.apache.activemq.usage.SystemUsage; 048import org.slf4j.Logger; 049import org.slf4j.LoggerFactory; 050 051public class TopicSubscription extends AbstractSubscription { 052 053 private static final Logger LOG = LoggerFactory.getLogger(TopicSubscription.class); 054 private static final AtomicLong CURSOR_NAME_COUNTER = new AtomicLong(0); 055 056 protected PendingMessageCursor matched; 057 protected final SystemUsage usageManager; 058 boolean singleDestination = true; 059 Destination destination; 060 private final Scheduler scheduler; 061 062 private int maximumPendingMessages = -1; 063 private MessageEvictionStrategy messageEvictionStrategy = new OldestMessageEvictionStrategy(); 064 private int discarded; 065 private final Object matchedListMutex = new Object(); 066 private int memoryUsageHighWaterMark = 95; 067 // allow duplicate suppression in a ring network of brokers 068 protected int maxProducersToAudit = 1024; 069 protected int maxAuditDepth = 1000; 070 protected boolean enableAudit = false; 071 protected ActiveMQMessageAudit audit; 072 protected boolean active = false; 073 protected boolean discarding = false; 074 075 //Used for inflight message size calculations 076 protected final Object dispatchLock = new Object(); 077 protected final List<MessageReference> dispatched = new ArrayList<MessageReference>(); 078 079 public TopicSubscription(Broker broker,ConnectionContext context, ConsumerInfo info, SystemUsage usageManager) throws Exception { 080 super(broker, context, info); 081 this.usageManager = usageManager; 082 String matchedName = "TopicSubscription:" + CURSOR_NAME_COUNTER.getAndIncrement() + "[" + info.getConsumerId().toString() + "]"; 083 if (info.getDestination().isTemporary() || broker.getTempDataStore()==null ) { 084 this.matched = new VMPendingMessageCursor(false); 085 } else { 086 this.matched = new FilePendingMessageCursor(broker,matchedName,false); 087 } 088 089 this.scheduler = broker.getScheduler(); 090 } 091 092 public void init() throws Exception { 093 this.matched.setSystemUsage(usageManager); 094 this.matched.setMemoryUsageHighWaterMark(getCursorMemoryHighWaterMark()); 095 this.matched.start(); 096 if (enableAudit) { 097 audit= new ActiveMQMessageAudit(maxAuditDepth, maxProducersToAudit); 098 } 099 this.active=true; 100 } 101 102 @Override 103 public void add(MessageReference node) throws Exception { 104 if (isDuplicate(node)) { 105 return; 106 } 107 // Lets use an indirect reference so that we can associate a unique 108 // locator /w the message. 109 node = new IndirectMessageReference(node.getMessage()); 110 getSubscriptionStatistics().getEnqueues().increment(); 111 synchronized (matchedListMutex) { 112 // if this subscriber is already discarding a message, we don't want to add 113 // any more messages to it as those messages can only be advisories generated in the process, 114 // which can trigger the recursive call loop 115 if (discarding) return; 116 117 if (!isFull() && matched.isEmpty()) { 118 // if maximumPendingMessages is set we will only discard messages which 119 // have not been dispatched (i.e. we allow the prefetch buffer to be filled) 120 dispatch(node); 121 setSlowConsumer(false); 122 } else { 123 if (info.getPrefetchSize() > 1 && matched.size() > info.getPrefetchSize()) { 124 // Slow consumers should log and set their state as such. 125 if (!isSlowConsumer()) { 126 LOG.warn("{}: has twice its prefetch limit pending, without an ack; it appears to be slow", toString()); 127 setSlowConsumer(true); 128 for (Destination dest: destinations) { 129 dest.slowConsumer(getContext(), this); 130 } 131 } 132 } 133 if (maximumPendingMessages != 0) { 134 boolean warnedAboutWait = false; 135 while (active) { 136 while (matched.isFull()) { 137 if (getContext().getStopping().get()) { 138 LOG.warn("{}: stopped waiting for space in pendingMessage cursor for: {}", toString(), node.getMessageId()); 139 getSubscriptionStatistics().getEnqueues().decrement(); 140 return; 141 } 142 if (!warnedAboutWait) { 143 LOG.info("{}: Pending message cursor [{}] is full, temp usag ({}%) or memory usage ({}%) limit reached, blocking message add() pending the release of resources.", 144 new Object[]{ 145 toString(), 146 matched, 147 matched.getSystemUsage().getTempUsage().getPercentUsage(), 148 matched.getSystemUsage().getMemoryUsage().getPercentUsage() 149 }); 150 warnedAboutWait = true; 151 } 152 matchedListMutex.wait(20); 153 } 154 // Temporary storage could be full - so just try to add the message 155 // see https://issues.apache.org/activemq/browse/AMQ-2475 156 if (matched.tryAddMessageLast(node, 10)) { 157 break; 158 } 159 } 160 if (maximumPendingMessages > 0) { 161 // calculate the high water mark from which point we 162 // will eagerly evict expired messages 163 int max = messageEvictionStrategy.getEvictExpiredMessagesHighWatermark(); 164 if (maximumPendingMessages > 0 && maximumPendingMessages < max) { 165 max = maximumPendingMessages; 166 } 167 if (!matched.isEmpty() && matched.size() > max) { 168 removeExpiredMessages(); 169 } 170 // lets discard old messages as we are a slow consumer 171 while (!matched.isEmpty() && matched.size() > maximumPendingMessages) { 172 int pageInSize = matched.size() - maximumPendingMessages; 173 // only page in a 1000 at a time - else we could blow the memory 174 pageInSize = Math.max(1000, pageInSize); 175 LinkedList<MessageReference> list = null; 176 MessageReference[] oldMessages=null; 177 synchronized(matched){ 178 list = matched.pageInList(pageInSize); 179 oldMessages = messageEvictionStrategy.evictMessages(list); 180 for (MessageReference ref : list) { 181 ref.decrementReferenceCount(); 182 } 183 } 184 int messagesToEvict = 0; 185 if (oldMessages != null){ 186 messagesToEvict = oldMessages.length; 187 for (int i = 0; i < messagesToEvict; i++) { 188 MessageReference oldMessage = oldMessages[i]; 189 discard(oldMessage); 190 } 191 } 192 // lets avoid an infinite loop if we are given a bad eviction strategy 193 // for a bad strategy lets just not evict 194 if (messagesToEvict == 0) { 195 LOG.warn("No messages to evict returned for {} from eviction strategy: {} out of {} candidates", new Object[]{ 196 destination, messageEvictionStrategy, list.size() 197 }); 198 break; 199 } 200 } 201 } 202 dispatchMatched(); 203 } 204 } 205 } 206 } 207 208 private boolean isDuplicate(MessageReference node) { 209 boolean duplicate = false; 210 if (enableAudit && audit != null) { 211 duplicate = audit.isDuplicate(node); 212 if (LOG.isDebugEnabled()) { 213 if (duplicate) { 214 LOG.debug("{}, ignoring duplicate add: {}", this, node.getMessageId()); 215 } 216 } 217 } 218 return duplicate; 219 } 220 221 /** 222 * Discard any expired messages from the matched list. Called from a 223 * synchronized block. 224 * 225 * @throws IOException 226 */ 227 protected void removeExpiredMessages() throws IOException { 228 try { 229 matched.reset(); 230 while (matched.hasNext()) { 231 MessageReference node = matched.next(); 232 node.decrementReferenceCount(); 233 if (node.isExpired()) { 234 matched.remove(); 235 node.decrementReferenceCount(); 236 if (broker.isExpired(node)) { 237 ((Destination) node.getRegionDestination()).getDestinationStatistics().getExpired().increment(); 238 broker.messageExpired(getContext(), node, this); 239 } 240 break; 241 } 242 } 243 } finally { 244 matched.release(); 245 } 246 } 247 248 @Override 249 public void processMessageDispatchNotification(MessageDispatchNotification mdn) { 250 synchronized (matchedListMutex) { 251 try { 252 matched.reset(); 253 while (matched.hasNext()) { 254 MessageReference node = matched.next(); 255 node.decrementReferenceCount(); 256 if (node.getMessageId().equals(mdn.getMessageId())) { 257 synchronized(dispatchLock) { 258 matched.remove(); 259 getSubscriptionStatistics().getDispatched().increment(); 260 dispatched.add(node); 261 getSubscriptionStatistics().getInflightMessageSize().addSize(node.getSize()); 262 node.decrementReferenceCount(); 263 } 264 break; 265 } 266 } 267 } finally { 268 matched.release(); 269 } 270 } 271 } 272 273 @Override 274 public synchronized void acknowledge(final ConnectionContext context, final MessageAck ack) throws Exception { 275 super.acknowledge(context, ack); 276 277 if (ack.isStandardAck()) { 278 updateStatsOnAck(context, ack); 279 } else if (ack.isPoisonAck()) { 280 if (ack.isInTransaction()) { 281 throw new JMSException("Poison ack cannot be transacted: " + ack); 282 } 283 updateStatsOnAck(context, ack); 284 if (getPrefetchSize() != 0) { 285 decrementPrefetchExtension(ack.getMessageCount()); 286 } 287 } else if (ack.isIndividualAck()) { 288 updateStatsOnAck(context, ack); 289 if (getPrefetchSize() != 0 && ack.isInTransaction()) { 290 incrementPrefetchExtension(ack.getMessageCount()); 291 } 292 } else if (ack.isExpiredAck()) { 293 updateStatsOnAck(ack); 294 if (getPrefetchSize() != 0) { 295 incrementPrefetchExtension(ack.getMessageCount()); 296 } 297 } else if (ack.isDeliveredAck()) { 298 // Message was delivered but not acknowledged: update pre-fetch counters. 299 if (getPrefetchSize() != 0) { 300 incrementPrefetchExtension(ack.getMessageCount()); 301 } 302 } else if (ack.isRedeliveredAck()) { 303 // No processing for redelivered needed 304 return; 305 } else { 306 throw new JMSException("Invalid acknowledgment: " + ack); 307 } 308 309 dispatchMatched(); 310 } 311 312 private void updateStatsOnAck(final ConnectionContext context, final MessageAck ack) { 313 if (context.isInTransaction()) { 314 context.getTransaction().addSynchronization(new Synchronization() { 315 316 @Override 317 public void beforeEnd() { 318 if (getPrefetchSize() != 0) { 319 decrementPrefetchExtension(ack.getMessageCount()); 320 } 321 } 322 323 @Override 324 public void afterCommit() throws Exception { 325 updateStatsOnAck(ack); 326 dispatchMatched(); 327 } 328 }); 329 } else { 330 updateStatsOnAck(ack); 331 } 332 } 333 334 @Override 335 public Response pullMessage(ConnectionContext context, final MessagePull pull) throws Exception { 336 337 // The slave should not deliver pull messages. 338 if (getPrefetchSize() == 0) { 339 340 final long currentDispatchedCount = getSubscriptionStatistics().getDispatched().getCount(); 341 prefetchExtension.set(pull.getQuantity()); 342 dispatchMatched(); 343 344 // If there was nothing dispatched.. we may need to setup a timeout. 345 if (currentDispatchedCount == getSubscriptionStatistics().getDispatched().getCount() || pull.isAlwaysSignalDone()) { 346 347 // immediate timeout used by receiveNoWait() 348 if (pull.getTimeout() == -1) { 349 // Send a NULL message to signal nothing pending. 350 dispatch(null); 351 prefetchExtension.set(0); 352 } 353 354 if (pull.getTimeout() > 0) { 355 scheduler.executeAfterDelay(new Runnable() { 356 357 @Override 358 public void run() { 359 pullTimeout(currentDispatchedCount, pull.isAlwaysSignalDone()); 360 } 361 }, pull.getTimeout()); 362 } 363 } 364 } 365 return null; 366 } 367 368 /** 369 * Occurs when a pull times out. If nothing has been dispatched since the 370 * timeout was setup, then send the NULL message. 371 */ 372 private final void pullTimeout(long currentDispatchedCount, boolean alwaysSendDone) { 373 synchronized (matchedListMutex) { 374 if (currentDispatchedCount == getSubscriptionStatistics().getDispatched().getCount() || alwaysSendDone) { 375 try { 376 dispatch(null); 377 } catch (Exception e) { 378 context.getConnection().serviceException(e); 379 } finally { 380 prefetchExtension.set(0); 381 } 382 } 383 } 384 } 385 386 /** 387 * Update the statistics on message ack. 388 * @param ack 389 */ 390 private void updateStatsOnAck(final MessageAck ack) { 391 synchronized(dispatchLock) { 392 boolean inAckRange = false; 393 List<MessageReference> removeList = new ArrayList<MessageReference>(); 394 for (final MessageReference node : dispatched) { 395 MessageId messageId = node.getMessageId(); 396 if (ack.getFirstMessageId() == null 397 || ack.getFirstMessageId().equals(messageId)) { 398 inAckRange = true; 399 } 400 if (inAckRange) { 401 removeList.add(node); 402 if (ack.getLastMessageId().equals(messageId)) { 403 break; 404 } 405 } 406 } 407 408 for (final MessageReference node : removeList) { 409 dispatched.remove(node); 410 getSubscriptionStatistics().getInflightMessageSize().addSize(-node.getSize()); 411 getSubscriptionStatistics().getDequeues().increment(); 412 ((Destination)node.getRegionDestination()).getDestinationStatistics().getDequeues().increment(); 413 ((Destination)node.getRegionDestination()).getDestinationStatistics().getInflight().decrement(); 414 if (info.isNetworkSubscription()) { 415 ((Destination)node.getRegionDestination()).getDestinationStatistics().getForwards().add(ack.getMessageCount()); 416 } 417 if (ack.isExpiredAck()) { 418 destination.getDestinationStatistics().getExpired().add(ack.getMessageCount()); 419 } 420 } 421 } 422 } 423 424 private void incrementPrefetchExtension(int amount) { 425 while (true) { 426 int currentExtension = prefetchExtension.get(); 427 int newExtension = Math.max(0, currentExtension + amount); 428 if (prefetchExtension.compareAndSet(currentExtension, newExtension)) { 429 break; 430 } 431 } 432 } 433 434 private void decrementPrefetchExtension(int amount) { 435 while (true) { 436 int currentExtension = prefetchExtension.get(); 437 int newExtension = Math.max(0, currentExtension - amount); 438 if (prefetchExtension.compareAndSet(currentExtension, newExtension)) { 439 break; 440 } 441 } 442 } 443 444 @Override 445 public int countBeforeFull() { 446 return getPrefetchSize() == 0 ? prefetchExtension.get() : info.getPrefetchSize() + prefetchExtension.get() - getDispatchedQueueSize(); 447 } 448 449 @Override 450 public int getPendingQueueSize() { 451 return matched(); 452 } 453 454 @Override 455 public long getPendingMessageSize() { 456 synchronized (matchedListMutex) { 457 return matched.messageSize(); 458 } 459 } 460 461 @Override 462 public int getDispatchedQueueSize() { 463 return (int)(getSubscriptionStatistics().getDispatched().getCount() - 464 getSubscriptionStatistics().getDequeues().getCount()); 465 } 466 467 public int getMaximumPendingMessages() { 468 return maximumPendingMessages; 469 } 470 471 @Override 472 public long getDispatchedCounter() { 473 return getSubscriptionStatistics().getDispatched().getCount(); 474 } 475 476 @Override 477 public long getEnqueueCounter() { 478 return getSubscriptionStatistics().getEnqueues().getCount(); 479 } 480 481 @Override 482 public long getDequeueCounter() { 483 return getSubscriptionStatistics().getDequeues().getCount(); 484 } 485 486 /** 487 * @return the number of messages discarded due to being a slow consumer 488 */ 489 public int discarded() { 490 synchronized (matchedListMutex) { 491 return discarded; 492 } 493 } 494 495 /** 496 * @return the number of matched messages (messages targeted for the 497 * subscription but not yet able to be dispatched due to the 498 * prefetch buffer being full). 499 */ 500 public int matched() { 501 synchronized (matchedListMutex) { 502 return matched.size(); 503 } 504 } 505 506 /** 507 * Sets the maximum number of pending messages that can be matched against 508 * this consumer before old messages are discarded. 509 */ 510 public void setMaximumPendingMessages(int maximumPendingMessages) { 511 this.maximumPendingMessages = maximumPendingMessages; 512 } 513 514 public MessageEvictionStrategy getMessageEvictionStrategy() { 515 return messageEvictionStrategy; 516 } 517 518 /** 519 * Sets the eviction strategy used to decide which message to evict when the 520 * slow consumer needs to discard messages 521 */ 522 public void setMessageEvictionStrategy(MessageEvictionStrategy messageEvictionStrategy) { 523 this.messageEvictionStrategy = messageEvictionStrategy; 524 } 525 526 public int getMaxProducersToAudit() { 527 return maxProducersToAudit; 528 } 529 530 public synchronized void setMaxProducersToAudit(int maxProducersToAudit) { 531 this.maxProducersToAudit = maxProducersToAudit; 532 if (audit != null) { 533 audit.setMaximumNumberOfProducersToTrack(maxProducersToAudit); 534 } 535 } 536 537 public int getMaxAuditDepth() { 538 return maxAuditDepth; 539 } 540 541 public synchronized void setMaxAuditDepth(int maxAuditDepth) { 542 this.maxAuditDepth = maxAuditDepth; 543 if (audit != null) { 544 audit.setAuditDepth(maxAuditDepth); 545 } 546 } 547 548 public boolean isEnableAudit() { 549 return enableAudit; 550 } 551 552 public synchronized void setEnableAudit(boolean enableAudit) { 553 this.enableAudit = enableAudit; 554 if (enableAudit && audit == null) { 555 audit = new ActiveMQMessageAudit(maxAuditDepth,maxProducersToAudit); 556 } 557 } 558 559 // Implementation methods 560 // ------------------------------------------------------------------------- 561 @Override 562 public boolean isFull() { 563 return getPrefetchSize() == 0 ? prefetchExtension.get() == 0 : getDispatchedQueueSize() - prefetchExtension.get() >= info.getPrefetchSize(); 564 } 565 566 @Override 567 public int getInFlightSize() { 568 return getDispatchedQueueSize(); 569 } 570 571 /** 572 * @return true when 60% or more room is left for dispatching messages 573 */ 574 @Override 575 public boolean isLowWaterMark() { 576 return (getDispatchedQueueSize() - prefetchExtension.get()) <= (info.getPrefetchSize() * .4); 577 } 578 579 /** 580 * @return true when 10% or less room is left for dispatching messages 581 */ 582 @Override 583 public boolean isHighWaterMark() { 584 return (getDispatchedQueueSize() - prefetchExtension.get()) >= (info.getPrefetchSize() * .9); 585 } 586 587 /** 588 * @param memoryUsageHighWaterMark the memoryUsageHighWaterMark to set 589 */ 590 public void setMemoryUsageHighWaterMark(int memoryUsageHighWaterMark) { 591 this.memoryUsageHighWaterMark = memoryUsageHighWaterMark; 592 } 593 594 /** 595 * @return the memoryUsageHighWaterMark 596 */ 597 public int getMemoryUsageHighWaterMark() { 598 return this.memoryUsageHighWaterMark; 599 } 600 601 /** 602 * @return the usageManager 603 */ 604 public SystemUsage getUsageManager() { 605 return this.usageManager; 606 } 607 608 /** 609 * @return the matched 610 */ 611 public PendingMessageCursor getMatched() { 612 return this.matched; 613 } 614 615 /** 616 * @param matched the matched to set 617 */ 618 public void setMatched(PendingMessageCursor matched) { 619 this.matched = matched; 620 } 621 622 /** 623 * inform the MessageConsumer on the client to change it's prefetch 624 * 625 * @param newPrefetch 626 */ 627 @Override 628 public void updateConsumerPrefetch(int newPrefetch) { 629 if (context != null && context.getConnection() != null && context.getConnection().isManageable()) { 630 ConsumerControl cc = new ConsumerControl(); 631 cc.setConsumerId(info.getConsumerId()); 632 cc.setPrefetch(newPrefetch); 633 context.getConnection().dispatchAsync(cc); 634 } 635 } 636 637 private void dispatchMatched() throws IOException { 638 synchronized (matchedListMutex) { 639 if (!matched.isEmpty() && !isFull()) { 640 try { 641 matched.reset(); 642 643 while (matched.hasNext() && !isFull()) { 644 MessageReference message = matched.next(); 645 message.decrementReferenceCount(); 646 matched.remove(); 647 // Message may have been sitting in the matched list a while 648 // waiting for the consumer to ak the message. 649 if (message.isExpired()) { 650 discard(message); 651 continue; // just drop it. 652 } 653 dispatch(message); 654 } 655 } finally { 656 matched.release(); 657 } 658 } 659 } 660 } 661 662 private void dispatch(final MessageReference node) throws IOException { 663 Message message = node != null ? node.getMessage() : null; 664 if (node != null) { 665 node.incrementReferenceCount(); 666 } 667 // Make sure we can dispatch a message. 668 MessageDispatch md = new MessageDispatch(); 669 md.setMessage(message); 670 md.setConsumerId(info.getConsumerId()); 671 if (node != null) { 672 md.setDestination(((Destination)node.getRegionDestination()).getActiveMQDestination()); 673 synchronized(dispatchLock) { 674 getSubscriptionStatistics().getDispatched().increment(); 675 dispatched.add(node); 676 getSubscriptionStatistics().getInflightMessageSize().addSize(node.getSize()); 677 } 678 679 // Keep track if this subscription is receiving messages from a single destination. 680 if (singleDestination) { 681 if (destination == null) { 682 destination = (Destination)node.getRegionDestination(); 683 } else { 684 if (destination != node.getRegionDestination()) { 685 singleDestination = false; 686 } 687 } 688 } 689 690 if (getPrefetchSize() == 0) { 691 decrementPrefetchExtension(1); 692 } 693 } 694 695 if (info.isDispatchAsync()) { 696 if (node != null) { 697 md.setTransmitCallback(new TransmitCallback() { 698 699 @Override 700 public void onSuccess() { 701 Destination regionDestination = (Destination) node.getRegionDestination(); 702 regionDestination.getDestinationStatistics().getDispatched().increment(); 703 regionDestination.getDestinationStatistics().getInflight().increment(); 704 node.decrementReferenceCount(); 705 } 706 707 @Override 708 public void onFailure() { 709 Destination regionDestination = (Destination) node.getRegionDestination(); 710 regionDestination.getDestinationStatistics().getDispatched().increment(); 711 regionDestination.getDestinationStatistics().getInflight().increment(); 712 node.decrementReferenceCount(); 713 } 714 }); 715 } 716 context.getConnection().dispatchAsync(md); 717 } else { 718 context.getConnection().dispatchSync(md); 719 if (node != null) { 720 Destination regionDestination = (Destination) node.getRegionDestination(); 721 regionDestination.getDestinationStatistics().getDispatched().increment(); 722 regionDestination.getDestinationStatistics().getInflight().increment(); 723 node.decrementReferenceCount(); 724 } 725 } 726 } 727 728 private void discard(MessageReference message) { 729 discarding = true; 730 try { 731 message.decrementReferenceCount(); 732 matched.remove(message); 733 discarded++; 734 if (destination != null) { 735 destination.getDestinationStatistics().getDequeues().increment(); 736 } 737 LOG.debug("{}, discarding message {}", this, message); 738 Destination dest = (Destination) message.getRegionDestination(); 739 if (dest != null) { 740 dest.messageDiscarded(getContext(), this, message); 741 } 742 broker.getRoot().sendToDeadLetterQueue(getContext(), message, this, new Throwable("TopicSubDiscard. ID:" + info.getConsumerId())); 743 } finally { 744 discarding = false; 745 } 746 } 747 748 @Override 749 public String toString() { 750 return "TopicSubscription:" + " consumer=" + info.getConsumerId() + ", destinations=" + destinations.size() + ", dispatched=" + getDispatchedQueueSize() + ", delivered=" 751 + getDequeueCounter() + ", matched=" + matched() + ", discarded=" + discarded() + ", prefetchExtension=" + prefetchExtension.get(); 752 } 753 754 @Override 755 public void destroy() { 756 this.active=false; 757 synchronized (matchedListMutex) { 758 try { 759 matched.destroy(); 760 } catch (Exception e) { 761 LOG.warn("Failed to destroy cursor", e); 762 } 763 } 764 setSlowConsumer(false); 765 synchronized(dispatchLock) { 766 dispatched.clear(); 767 } 768 } 769 770 @Override 771 public int getPrefetchSize() { 772 return info.getPrefetchSize(); 773 } 774 775 @Override 776 public void setPrefetchSize(int newSize) { 777 info.setPrefetchSize(newSize); 778 try { 779 dispatchMatched(); 780 } catch(Exception e) { 781 LOG.trace("Caught exception on dispatch after prefetch size change."); 782 } 783 } 784}