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