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 getSubscriptionStatistics().getDispatched().increment(); 236 node.decrementReferenceCount(); 237 if (broker.isExpired(node)) { 238 ((Destination) node.getRegionDestination()).getDestinationStatistics().getExpired().increment(); 239 broker.messageExpired(getContext(), node, this); 240 } 241 break; 242 } 243 } 244 } finally { 245 matched.release(); 246 } 247 } 248 249 @Override 250 public void processMessageDispatchNotification(MessageDispatchNotification mdn) { 251 synchronized (matchedListMutex) { 252 try { 253 matched.reset(); 254 while (matched.hasNext()) { 255 MessageReference node = matched.next(); 256 node.decrementReferenceCount(); 257 if (node.getMessageId().equals(mdn.getMessageId())) { 258 synchronized(dispatchLock) { 259 matched.remove(); 260 getSubscriptionStatistics().getDispatched().increment(); 261 dispatched.add(node); 262 getSubscriptionStatistics().getInflightMessageSize().addSize(node.getSize()); 263 node.decrementReferenceCount(); 264 } 265 break; 266 } 267 } 268 } finally { 269 matched.release(); 270 } 271 } 272 } 273 274 @Override 275 public synchronized void acknowledge(final ConnectionContext context, final MessageAck ack) throws Exception { 276 super.acknowledge(context, ack); 277 278 if (ack.isStandardAck()) { 279 updateStatsOnAck(context, ack); 280 } else if (ack.isPoisonAck()) { 281 if (ack.isInTransaction()) { 282 throw new JMSException("Poison ack cannot be transacted: " + ack); 283 } 284 updateStatsOnAck(context, ack); 285 if (getPrefetchSize() != 0) { 286 decrementPrefetchExtension(ack.getMessageCount()); 287 } 288 } else if (ack.isIndividualAck()) { 289 updateStatsOnAck(context, ack); 290 if (getPrefetchSize() != 0 && ack.isInTransaction()) { 291 incrementPrefetchExtension(ack.getMessageCount()); 292 } 293 } else if (ack.isExpiredAck()) { 294 updateStatsOnAck(ack); 295 if (getPrefetchSize() != 0) { 296 incrementPrefetchExtension(ack.getMessageCount()); 297 } 298 } else if (ack.isDeliveredAck()) { 299 // Message was delivered but not acknowledged: update pre-fetch counters. 300 if (getPrefetchSize() != 0) { 301 incrementPrefetchExtension(ack.getMessageCount()); 302 } 303 } else if (ack.isRedeliveredAck()) { 304 // No processing for redelivered needed 305 return; 306 } else { 307 throw new JMSException("Invalid acknowledgment: " + ack); 308 } 309 310 dispatchMatched(); 311 } 312 313 private void updateStatsOnAck(final ConnectionContext context, final MessageAck ack) { 314 if (context.isInTransaction()) { 315 context.getTransaction().addSynchronization(new Synchronization() { 316 317 @Override 318 public void beforeEnd() { 319 if (getPrefetchSize() != 0) { 320 decrementPrefetchExtension(ack.getMessageCount()); 321 } 322 } 323 324 @Override 325 public void afterCommit() throws Exception { 326 updateStatsOnAck(ack); 327 dispatchMatched(); 328 } 329 }); 330 } else { 331 updateStatsOnAck(ack); 332 } 333 } 334 335 @Override 336 public Response pullMessage(ConnectionContext context, final MessagePull pull) throws Exception { 337 338 // The slave should not deliver pull messages. 339 if (getPrefetchSize() == 0) { 340 341 final long currentDispatchedCount = getSubscriptionStatistics().getDispatched().getCount(); 342 prefetchExtension.set(pull.getQuantity()); 343 dispatchMatched(); 344 345 // If there was nothing dispatched.. we may need to setup a timeout. 346 if (currentDispatchedCount == getSubscriptionStatistics().getDispatched().getCount() || pull.isAlwaysSignalDone()) { 347 348 // immediate timeout used by receiveNoWait() 349 if (pull.getTimeout() == -1) { 350 // Send a NULL message to signal nothing pending. 351 dispatch(null); 352 prefetchExtension.set(0); 353 } 354 355 if (pull.getTimeout() > 0) { 356 scheduler.executeAfterDelay(new Runnable() { 357 358 @Override 359 public void run() { 360 pullTimeout(currentDispatchedCount, pull.isAlwaysSignalDone()); 361 } 362 }, pull.getTimeout()); 363 } 364 } 365 } 366 return null; 367 } 368 369 /** 370 * Occurs when a pull times out. If nothing has been dispatched since the 371 * timeout was setup, then send the NULL message. 372 */ 373 private final void pullTimeout(long currentDispatchedCount, boolean alwaysSendDone) { 374 synchronized (matchedListMutex) { 375 if (currentDispatchedCount == getSubscriptionStatistics().getDispatched().getCount() || alwaysSendDone) { 376 try { 377 dispatch(null); 378 } catch (Exception e) { 379 context.getConnection().serviceException(e); 380 } finally { 381 prefetchExtension.set(0); 382 } 383 } 384 } 385 } 386 387 /** 388 * Update the statistics on message ack. 389 * @param ack 390 */ 391 private void updateStatsOnAck(final MessageAck ack) { 392 synchronized(dispatchLock) { 393 boolean inAckRange = false; 394 List<MessageReference> removeList = new ArrayList<MessageReference>(); 395 for (final MessageReference node : dispatched) { 396 MessageId messageId = node.getMessageId(); 397 if (ack.getFirstMessageId() == null 398 || ack.getFirstMessageId().equals(messageId)) { 399 inAckRange = true; 400 } 401 if (inAckRange) { 402 removeList.add(node); 403 if (ack.getLastMessageId().equals(messageId)) { 404 break; 405 } 406 } 407 } 408 409 for (final MessageReference node : removeList) { 410 dispatched.remove(node); 411 getSubscriptionStatistics().getInflightMessageSize().addSize(-node.getSize()); 412 getSubscriptionStatistics().getDequeues().increment(); 413 ((Destination)node.getRegionDestination()).getDestinationStatistics().getDequeues().increment(); 414 ((Destination)node.getRegionDestination()).getDestinationStatistics().getInflight().decrement(); 415 if (info.isNetworkSubscription()) { 416 ((Destination)node.getRegionDestination()).getDestinationStatistics().getForwards().add(ack.getMessageCount()); 417 } 418 if (ack.isExpiredAck()) { 419 destination.getDestinationStatistics().getExpired().add(ack.getMessageCount()); 420 } 421 } 422 } 423 } 424 425 private void incrementPrefetchExtension(int amount) { 426 while (true) { 427 int currentExtension = prefetchExtension.get(); 428 int newExtension = Math.max(0, currentExtension + amount); 429 if (prefetchExtension.compareAndSet(currentExtension, newExtension)) { 430 break; 431 } 432 } 433 } 434 435 private void decrementPrefetchExtension(int amount) { 436 while (true) { 437 int currentExtension = prefetchExtension.get(); 438 int newExtension = Math.max(0, currentExtension - amount); 439 if (prefetchExtension.compareAndSet(currentExtension, newExtension)) { 440 break; 441 } 442 } 443 } 444 445 @Override 446 public int countBeforeFull() { 447 return getPrefetchSize() == 0 ? prefetchExtension.get() : info.getPrefetchSize() + prefetchExtension.get() - getDispatchedQueueSize(); 448 } 449 450 @Override 451 public int getPendingQueueSize() { 452 return matched(); 453 } 454 455 @Override 456 public long getPendingMessageSize() { 457 synchronized (matchedListMutex) { 458 return matched.messageSize(); 459 } 460 } 461 462 @Override 463 public int getDispatchedQueueSize() { 464 return (int)(getSubscriptionStatistics().getDispatched().getCount() - 465 getSubscriptionStatistics().getDequeues().getCount()); 466 } 467 468 public int getMaximumPendingMessages() { 469 return maximumPendingMessages; 470 } 471 472 @Override 473 public long getDispatchedCounter() { 474 return getSubscriptionStatistics().getDispatched().getCount(); 475 } 476 477 @Override 478 public long getEnqueueCounter() { 479 return getSubscriptionStatistics().getEnqueues().getCount(); 480 } 481 482 @Override 483 public long getDequeueCounter() { 484 return getSubscriptionStatistics().getDequeues().getCount(); 485 } 486 487 /** 488 * @return the number of messages discarded due to being a slow consumer 489 */ 490 public int discarded() { 491 synchronized (matchedListMutex) { 492 return discarded; 493 } 494 } 495 496 /** 497 * @return the number of matched messages (messages targeted for the 498 * subscription but not yet able to be dispatched due to the 499 * prefetch buffer being full). 500 */ 501 public int matched() { 502 synchronized (matchedListMutex) { 503 return matched.size(); 504 } 505 } 506 507 /** 508 * Sets the maximum number of pending messages that can be matched against 509 * this consumer before old messages are discarded. 510 */ 511 public void setMaximumPendingMessages(int maximumPendingMessages) { 512 this.maximumPendingMessages = maximumPendingMessages; 513 } 514 515 public MessageEvictionStrategy getMessageEvictionStrategy() { 516 return messageEvictionStrategy; 517 } 518 519 /** 520 * Sets the eviction strategy used to decide which message to evict when the 521 * slow consumer needs to discard messages 522 */ 523 public void setMessageEvictionStrategy(MessageEvictionStrategy messageEvictionStrategy) { 524 this.messageEvictionStrategy = messageEvictionStrategy; 525 } 526 527 public int getMaxProducersToAudit() { 528 return maxProducersToAudit; 529 } 530 531 public synchronized void setMaxProducersToAudit(int maxProducersToAudit) { 532 this.maxProducersToAudit = maxProducersToAudit; 533 if (audit != null) { 534 audit.setMaximumNumberOfProducersToTrack(maxProducersToAudit); 535 } 536 } 537 538 public int getMaxAuditDepth() { 539 return maxAuditDepth; 540 } 541 542 public synchronized void setMaxAuditDepth(int maxAuditDepth) { 543 this.maxAuditDepth = maxAuditDepth; 544 if (audit != null) { 545 audit.setAuditDepth(maxAuditDepth); 546 } 547 } 548 549 public boolean isEnableAudit() { 550 return enableAudit; 551 } 552 553 public synchronized void setEnableAudit(boolean enableAudit) { 554 this.enableAudit = enableAudit; 555 if (enableAudit && audit == null) { 556 audit = new ActiveMQMessageAudit(maxAuditDepth,maxProducersToAudit); 557 } 558 } 559 560 // Implementation methods 561 // ------------------------------------------------------------------------- 562 @Override 563 public boolean isFull() { 564 return getPrefetchSize() == 0 ? prefetchExtension.get() == 0 : getDispatchedQueueSize() - prefetchExtension.get() >= info.getPrefetchSize(); 565 } 566 567 @Override 568 public int getInFlightSize() { 569 return getDispatchedQueueSize(); 570 } 571 572 /** 573 * @return true when 60% or more room is left for dispatching messages 574 */ 575 @Override 576 public boolean isLowWaterMark() { 577 return (getDispatchedQueueSize() - prefetchExtension.get()) <= (info.getPrefetchSize() * .4); 578 } 579 580 /** 581 * @return true when 10% or less room is left for dispatching messages 582 */ 583 @Override 584 public boolean isHighWaterMark() { 585 return (getDispatchedQueueSize() - prefetchExtension.get()) >= (info.getPrefetchSize() * .9); 586 } 587 588 /** 589 * @param memoryUsageHighWaterMark the memoryUsageHighWaterMark to set 590 */ 591 public void setMemoryUsageHighWaterMark(int memoryUsageHighWaterMark) { 592 this.memoryUsageHighWaterMark = memoryUsageHighWaterMark; 593 } 594 595 /** 596 * @return the memoryUsageHighWaterMark 597 */ 598 public int getMemoryUsageHighWaterMark() { 599 return this.memoryUsageHighWaterMark; 600 } 601 602 /** 603 * @return the usageManager 604 */ 605 public SystemUsage getUsageManager() { 606 return this.usageManager; 607 } 608 609 /** 610 * @return the matched 611 */ 612 public PendingMessageCursor getMatched() { 613 return this.matched; 614 } 615 616 /** 617 * @param matched the matched to set 618 */ 619 public void setMatched(PendingMessageCursor matched) { 620 this.matched = matched; 621 } 622 623 /** 624 * inform the MessageConsumer on the client to change it's prefetch 625 * 626 * @param newPrefetch 627 */ 628 @Override 629 public void updateConsumerPrefetch(int newPrefetch) { 630 if (context != null && context.getConnection() != null && context.getConnection().isManageable()) { 631 ConsumerControl cc = new ConsumerControl(); 632 cc.setConsumerId(info.getConsumerId()); 633 cc.setPrefetch(newPrefetch); 634 context.getConnection().dispatchAsync(cc); 635 } 636 } 637 638 private void dispatchMatched() throws IOException { 639 synchronized (matchedListMutex) { 640 if (!matched.isEmpty() && !isFull()) { 641 try { 642 matched.reset(); 643 644 while (matched.hasNext() && !isFull()) { 645 MessageReference message = matched.next(); 646 message.decrementReferenceCount(); 647 matched.remove(); 648 // Message may have been sitting in the matched list a while 649 // waiting for the consumer to ak the message. 650 if (message.isExpired()) { 651 discard(message); 652 continue; // just drop it. 653 } 654 dispatch(message); 655 } 656 } finally { 657 matched.release(); 658 } 659 } 660 } 661 } 662 663 private void dispatch(final MessageReference node) throws IOException { 664 Message message = node != null ? node.getMessage() : null; 665 if (node != null) { 666 node.incrementReferenceCount(); 667 } 668 // Make sure we can dispatch a message. 669 MessageDispatch md = new MessageDispatch(); 670 md.setMessage(message); 671 md.setConsumerId(info.getConsumerId()); 672 if (node != null) { 673 md.setDestination(((Destination)node.getRegionDestination()).getActiveMQDestination()); 674 synchronized(dispatchLock) { 675 getSubscriptionStatistics().getDispatched().increment(); 676 dispatched.add(node); 677 getSubscriptionStatistics().getInflightMessageSize().addSize(node.getSize()); 678 } 679 680 // Keep track if this subscription is receiving messages from a single destination. 681 if (singleDestination) { 682 if (destination == null) { 683 destination = (Destination)node.getRegionDestination(); 684 } else { 685 if (destination != node.getRegionDestination()) { 686 singleDestination = false; 687 } 688 } 689 } 690 691 if (getPrefetchSize() == 0) { 692 decrementPrefetchExtension(1); 693 } 694 } 695 696 if (info.isDispatchAsync()) { 697 if (node != null) { 698 md.setTransmitCallback(new TransmitCallback() { 699 700 @Override 701 public void onSuccess() { 702 Destination regionDestination = (Destination) node.getRegionDestination(); 703 regionDestination.getDestinationStatistics().getDispatched().increment(); 704 regionDestination.getDestinationStatistics().getInflight().increment(); 705 node.decrementReferenceCount(); 706 } 707 708 @Override 709 public void onFailure() { 710 Destination regionDestination = (Destination) node.getRegionDestination(); 711 regionDestination.getDestinationStatistics().getDispatched().increment(); 712 regionDestination.getDestinationStatistics().getInflight().increment(); 713 node.decrementReferenceCount(); 714 } 715 }); 716 } 717 context.getConnection().dispatchAsync(md); 718 } else { 719 context.getConnection().dispatchSync(md); 720 if (node != null) { 721 Destination regionDestination = (Destination) node.getRegionDestination(); 722 regionDestination.getDestinationStatistics().getDispatched().increment(); 723 regionDestination.getDestinationStatistics().getInflight().increment(); 724 node.decrementReferenceCount(); 725 } 726 } 727 } 728 729 private void discard(MessageReference message) { 730 discarding = true; 731 try { 732 message.decrementReferenceCount(); 733 matched.remove(message); 734 discarded++; 735 if (destination != null) { 736 destination.getDestinationStatistics().getDequeues().increment(); 737 } 738 LOG.debug("{}, discarding message {}", this, message); 739 Destination dest = (Destination) message.getRegionDestination(); 740 if (dest != null) { 741 dest.messageDiscarded(getContext(), this, message); 742 } 743 broker.getRoot().sendToDeadLetterQueue(getContext(), message, this, new Throwable("TopicSubDiscard. ID:" + info.getConsumerId())); 744 } finally { 745 discarding = false; 746 } 747 } 748 749 @Override 750 public String toString() { 751 return "TopicSubscription:" + " consumer=" + info.getConsumerId() + ", destinations=" + destinations.size() + ", dispatched=" + getDispatchedQueueSize() + ", delivered=" 752 + getDequeueCounter() + ", matched=" + matched() + ", discarded=" + discarded() + ", prefetchExtension=" + prefetchExtension.get(); 753 } 754 755 @Override 756 public void destroy() { 757 this.active=false; 758 synchronized (matchedListMutex) { 759 try { 760 matched.destroy(); 761 } catch (Exception e) { 762 LOG.warn("Failed to destroy cursor", e); 763 } 764 } 765 setSlowConsumer(false); 766 synchronized(dispatchLock) { 767 dispatched.clear(); 768 } 769 } 770 771 @Override 772 public int getPrefetchSize() { 773 return info.getPrefetchSize(); 774 } 775 776 @Override 777 public void setPrefetchSize(int newSize) { 778 info.setPrefetchSize(newSize); 779 try { 780 dispatchMatched(); 781 } catch(Exception e) { 782 LOG.trace("Caught exception on dispatch after prefetch size change."); 783 } 784 } 785}