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