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 static org.apache.activemq.broker.region.cursors.AbstractStoreCursor.gotToTheStore; 020import static org.apache.activemq.transaction.Transaction.IN_USE_STATE; 021 022import java.io.IOException; 023import java.util.ArrayList; 024import java.util.Collection; 025import java.util.Collections; 026import java.util.Comparator; 027import java.util.HashSet; 028import java.util.Iterator; 029import java.util.LinkedHashMap; 030import java.util.LinkedHashSet; 031import java.util.LinkedList; 032import java.util.List; 033import java.util.Map; 034import java.util.Set; 035import java.util.concurrent.CancellationException; 036import java.util.concurrent.ConcurrentLinkedQueue; 037import java.util.concurrent.CountDownLatch; 038import java.util.concurrent.DelayQueue; 039import java.util.concurrent.Delayed; 040import java.util.concurrent.ExecutorService; 041import java.util.concurrent.TimeUnit; 042import java.util.concurrent.atomic.AtomicInteger; 043import java.util.concurrent.atomic.AtomicLong; 044import java.util.concurrent.locks.Lock; 045import java.util.concurrent.locks.ReentrantLock; 046import java.util.concurrent.locks.ReentrantReadWriteLock; 047 048import javax.jms.InvalidSelectorException; 049import javax.jms.JMSException; 050import javax.jms.ResourceAllocationException; 051 052import org.apache.activemq.broker.BrokerService; 053import org.apache.activemq.broker.BrokerStoppedException; 054import org.apache.activemq.broker.ConnectionContext; 055import org.apache.activemq.broker.ProducerBrokerExchange; 056import org.apache.activemq.broker.region.cursors.OrderedPendingList; 057import org.apache.activemq.broker.region.cursors.PendingList; 058import org.apache.activemq.broker.region.cursors.PendingMessageCursor; 059import org.apache.activemq.broker.region.cursors.PrioritizedPendingList; 060import org.apache.activemq.broker.region.cursors.QueueDispatchPendingList; 061import org.apache.activemq.broker.region.cursors.StoreQueueCursor; 062import org.apache.activemq.broker.region.cursors.VMPendingMessageCursor; 063import org.apache.activemq.broker.region.group.CachedMessageGroupMapFactory; 064import org.apache.activemq.broker.region.group.MessageGroupMap; 065import org.apache.activemq.broker.region.group.MessageGroupMapFactory; 066import org.apache.activemq.broker.region.policy.DeadLetterStrategy; 067import org.apache.activemq.broker.region.policy.DispatchPolicy; 068import org.apache.activemq.broker.region.policy.RoundRobinDispatchPolicy; 069import org.apache.activemq.broker.util.InsertionCountList; 070import org.apache.activemq.command.ActiveMQDestination; 071import org.apache.activemq.command.ConsumerId; 072import org.apache.activemq.command.ExceptionResponse; 073import org.apache.activemq.command.Message; 074import org.apache.activemq.command.MessageAck; 075import org.apache.activemq.command.MessageDispatchNotification; 076import org.apache.activemq.command.MessageId; 077import org.apache.activemq.command.ProducerAck; 078import org.apache.activemq.command.ProducerInfo; 079import org.apache.activemq.command.RemoveInfo; 080import org.apache.activemq.command.Response; 081import org.apache.activemq.filter.BooleanExpression; 082import org.apache.activemq.filter.MessageEvaluationContext; 083import org.apache.activemq.filter.NonCachedMessageEvaluationContext; 084import org.apache.activemq.selector.SelectorParser; 085import org.apache.activemq.state.ProducerState; 086import org.apache.activemq.store.IndexListener; 087import org.apache.activemq.store.ListenableFuture; 088import org.apache.activemq.store.MessageRecoveryListener; 089import org.apache.activemq.store.MessageStore; 090import org.apache.activemq.thread.Task; 091import org.apache.activemq.thread.TaskRunner; 092import org.apache.activemq.thread.TaskRunnerFactory; 093import org.apache.activemq.transaction.Synchronization; 094import org.apache.activemq.usage.Usage; 095import org.apache.activemq.usage.UsageListener; 096import org.apache.activemq.util.BrokerSupport; 097import org.apache.activemq.util.ThreadPoolUtils; 098import org.slf4j.Logger; 099import org.slf4j.LoggerFactory; 100import org.slf4j.MDC; 101 102/** 103 * The Queue is a List of MessageEntry objects that are dispatched to matching 104 * subscriptions. 105 */ 106public class Queue extends BaseDestination implements Task, UsageListener, IndexListener { 107 protected static final Logger LOG = LoggerFactory.getLogger(Queue.class); 108 protected final TaskRunnerFactory taskFactory; 109 protected TaskRunner taskRunner; 110 private final ReentrantReadWriteLock consumersLock = new ReentrantReadWriteLock(); 111 protected final List<Subscription> consumers = new ArrayList<Subscription>(50); 112 private final ReentrantReadWriteLock messagesLock = new ReentrantReadWriteLock(); 113 protected PendingMessageCursor messages; 114 private final ReentrantReadWriteLock pagedInMessagesLock = new ReentrantReadWriteLock(); 115 private final PendingList pagedInMessages = new OrderedPendingList(); 116 // Messages that are paged in but have not yet been targeted at a subscription 117 private final ReentrantReadWriteLock pagedInPendingDispatchLock = new ReentrantReadWriteLock(); 118 protected QueueDispatchPendingList dispatchPendingList = new QueueDispatchPendingList(); 119 private AtomicInteger pendingSends = new AtomicInteger(0); 120 private MessageGroupMap messageGroupOwners; 121 private DispatchPolicy dispatchPolicy = new RoundRobinDispatchPolicy(); 122 private MessageGroupMapFactory messageGroupMapFactory = new CachedMessageGroupMapFactory(); 123 final Lock sendLock = new ReentrantLock(); 124 private ExecutorService executor; 125 private final Map<MessageId, Runnable> messagesWaitingForSpace = new LinkedHashMap<MessageId, Runnable>(); 126 private boolean useConsumerPriority = true; 127 private boolean strictOrderDispatch = false; 128 private final QueueDispatchSelector dispatchSelector; 129 private boolean optimizedDispatch = false; 130 private boolean iterationRunning = false; 131 private boolean firstConsumer = false; 132 private int timeBeforeDispatchStarts = 0; 133 private int consumersBeforeDispatchStarts = 0; 134 private CountDownLatch consumersBeforeStartsLatch; 135 private final AtomicLong pendingWakeups = new AtomicLong(); 136 private boolean allConsumersExclusiveByDefault = false; 137 138 private volatile boolean resetNeeded; 139 140 private final Runnable sendMessagesWaitingForSpaceTask = new Runnable() { 141 @Override 142 public void run() { 143 asyncWakeup(); 144 } 145 }; 146 private final Runnable expireMessagesTask = new Runnable() { 147 @Override 148 public void run() { 149 expireMessages(); 150 } 151 }; 152 153 private final Object iteratingMutex = new Object(); 154 155 // gate on enabling cursor cache to ensure no outstanding sync 156 // send before async sends resume 157 public boolean singlePendingSend() { 158 return pendingSends.get() <= 1; 159 } 160 161 class TimeoutMessage implements Delayed { 162 163 Message message; 164 ConnectionContext context; 165 long trigger; 166 167 public TimeoutMessage(Message message, ConnectionContext context, long delay) { 168 this.message = message; 169 this.context = context; 170 this.trigger = System.currentTimeMillis() + delay; 171 } 172 173 @Override 174 public long getDelay(TimeUnit unit) { 175 long n = trigger - System.currentTimeMillis(); 176 return unit.convert(n, TimeUnit.MILLISECONDS); 177 } 178 179 @Override 180 public int compareTo(Delayed delayed) { 181 long other = ((TimeoutMessage) delayed).trigger; 182 int returnValue; 183 if (this.trigger < other) { 184 returnValue = -1; 185 } else if (this.trigger > other) { 186 returnValue = 1; 187 } else { 188 returnValue = 0; 189 } 190 return returnValue; 191 } 192 } 193 194 DelayQueue<TimeoutMessage> flowControlTimeoutMessages = new DelayQueue<TimeoutMessage>(); 195 196 class FlowControlTimeoutTask extends Thread { 197 198 @Override 199 public void run() { 200 TimeoutMessage timeout; 201 try { 202 while (true) { 203 timeout = flowControlTimeoutMessages.take(); 204 if (timeout != null) { 205 synchronized (messagesWaitingForSpace) { 206 if (messagesWaitingForSpace.remove(timeout.message.getMessageId()) != null) { 207 ExceptionResponse response = new ExceptionResponse( 208 new ResourceAllocationException( 209 "Usage Manager Memory Limit reached. Stopping producer (" 210 + timeout.message.getProducerId() 211 + ") to prevent flooding " 212 + getActiveMQDestination().getQualifiedName() 213 + "." 214 + " See http://activemq.apache.org/producer-flow-control.html for more info")); 215 response.setCorrelationId(timeout.message.getCommandId()); 216 timeout.context.getConnection().dispatchAsync(response); 217 } 218 } 219 } 220 } 221 } catch (InterruptedException e) { 222 LOG.debug(getName() + "Producer Flow Control Timeout Task is stopping"); 223 } 224 } 225 } 226 227 private final FlowControlTimeoutTask flowControlTimeoutTask = new FlowControlTimeoutTask(); 228 229 private final Comparator<Subscription> orderedCompare = new Comparator<Subscription>() { 230 231 @Override 232 public int compare(Subscription s1, Subscription s2) { 233 // We want the list sorted in descending order 234 int val = s2.getConsumerInfo().getPriority() - s1.getConsumerInfo().getPriority(); 235 if (val == 0 && messageGroupOwners != null) { 236 // then ascending order of assigned message groups to favour less loaded consumers 237 // Long.compare in jdk7 238 long x = s1.getConsumerInfo().getAssignedGroupCount(destination); 239 long y = s2.getConsumerInfo().getAssignedGroupCount(destination); 240 val = (x < y) ? -1 : ((x == y) ? 0 : 1); 241 } 242 return val; 243 } 244 }; 245 246 public Queue(BrokerService brokerService, final ActiveMQDestination destination, MessageStore store, 247 DestinationStatistics parentStats, TaskRunnerFactory taskFactory) throws Exception { 248 super(brokerService, store, destination, parentStats); 249 this.taskFactory = taskFactory; 250 this.dispatchSelector = new QueueDispatchSelector(destination); 251 if (store != null) { 252 store.registerIndexListener(this); 253 } 254 } 255 256 @Override 257 public List<Subscription> getConsumers() { 258 consumersLock.readLock().lock(); 259 try { 260 return new ArrayList<Subscription>(consumers); 261 } finally { 262 consumersLock.readLock().unlock(); 263 } 264 } 265 266 // make the queue easily visible in the debugger from its task runner 267 // threads 268 final class QueueThread extends Thread { 269 final Queue queue; 270 271 public QueueThread(Runnable runnable, String name, Queue queue) { 272 super(runnable, name); 273 this.queue = queue; 274 } 275 } 276 277 class BatchMessageRecoveryListener implements MessageRecoveryListener { 278 final LinkedList<Message> toExpire = new LinkedList<Message>(); 279 final double totalMessageCount; 280 int recoveredAccumulator = 0; 281 int currentBatchCount; 282 283 BatchMessageRecoveryListener(int totalMessageCount) { 284 this.totalMessageCount = totalMessageCount; 285 currentBatchCount = recoveredAccumulator; 286 } 287 288 @Override 289 public boolean recoverMessage(Message message) { 290 recoveredAccumulator++; 291 if ((recoveredAccumulator % 10000) == 0) { 292 LOG.info("cursor for {} has recovered {} messages. {}% complete", new Object[]{ getActiveMQDestination().getQualifiedName(), recoveredAccumulator, new Integer((int) (recoveredAccumulator * 100 / totalMessageCount))}); 293 } 294 // Message could have expired while it was being 295 // loaded.. 296 message.setRegionDestination(Queue.this); 297 if (message.isExpired() && broker.isExpired(message)) { 298 toExpire.add(message); 299 return true; 300 } 301 if (hasSpace()) { 302 messagesLock.writeLock().lock(); 303 try { 304 try { 305 messages.addMessageLast(message); 306 } catch (Exception e) { 307 LOG.error("Failed to add message to cursor", e); 308 } 309 } finally { 310 messagesLock.writeLock().unlock(); 311 } 312 destinationStatistics.getMessages().increment(); 313 return true; 314 } 315 return false; 316 } 317 318 @Override 319 public boolean recoverMessageReference(MessageId messageReference) throws Exception { 320 throw new RuntimeException("Should not be called."); 321 } 322 323 @Override 324 public boolean hasSpace() { 325 return true; 326 } 327 328 @Override 329 public boolean isDuplicate(MessageId id) { 330 return false; 331 } 332 333 public void reset() { 334 currentBatchCount = recoveredAccumulator; 335 } 336 337 public void processExpired() { 338 for (Message message: toExpire) { 339 messageExpired(createConnectionContext(), createMessageReference(message)); 340 // drop message will decrement so counter 341 // balance here 342 destinationStatistics.getMessages().increment(); 343 } 344 toExpire.clear(); 345 } 346 347 public boolean done() { 348 return currentBatchCount == recoveredAccumulator; 349 } 350 } 351 352 @Override 353 public void setPrioritizedMessages(boolean prioritizedMessages) { 354 super.setPrioritizedMessages(prioritizedMessages); 355 dispatchPendingList.setPrioritizedMessages(prioritizedMessages); 356 } 357 358 @Override 359 public void initialize() throws Exception { 360 361 if (this.messages == null) { 362 if (destination.isTemporary() || broker == null || store == null) { 363 this.messages = new VMPendingMessageCursor(isPrioritizedMessages()); 364 } else { 365 this.messages = new StoreQueueCursor(broker, this); 366 } 367 } 368 369 // If a VMPendingMessageCursor don't use the default Producer System 370 // Usage 371 // since it turns into a shared blocking queue which can lead to a 372 // network deadlock. 373 // If we are cursoring to disk..it's not and issue because it does not 374 // block due 375 // to large disk sizes. 376 if (messages instanceof VMPendingMessageCursor) { 377 this.systemUsage = brokerService.getSystemUsage(); 378 memoryUsage.setParent(systemUsage.getMemoryUsage()); 379 } 380 381 this.taskRunner = taskFactory.createTaskRunner(this, "Queue:" + destination.getPhysicalName()); 382 383 super.initialize(); 384 if (store != null) { 385 // Restore the persistent messages. 386 messages.setSystemUsage(systemUsage); 387 messages.setEnableAudit(isEnableAudit()); 388 messages.setMaxAuditDepth(getMaxAuditDepth()); 389 messages.setMaxProducersToAudit(getMaxProducersToAudit()); 390 messages.setUseCache(isUseCache()); 391 messages.setMemoryUsageHighWaterMark(getCursorMemoryHighWaterMark()); 392 store.start(); 393 final int messageCount = store.getMessageCount(); 394 if (messageCount > 0 && messages.isRecoveryRequired()) { 395 BatchMessageRecoveryListener listener = new BatchMessageRecoveryListener(messageCount); 396 do { 397 listener.reset(); 398 store.recoverNextMessages(getMaxPageSize(), listener); 399 listener.processExpired(); 400 } while (!listener.done()); 401 } else { 402 destinationStatistics.getMessages().add(messageCount); 403 } 404 } 405 } 406 407 /* 408 * Holder for subscription that needs attention on next iterate browser 409 * needs access to existing messages in the queue that have already been 410 * dispatched 411 */ 412 class BrowserDispatch { 413 QueueBrowserSubscription browser; 414 415 public BrowserDispatch(QueueBrowserSubscription browserSubscription) { 416 browser = browserSubscription; 417 browser.incrementQueueRef(); 418 } 419 420 public QueueBrowserSubscription getBrowser() { 421 return browser; 422 } 423 } 424 425 ConcurrentLinkedQueue<BrowserDispatch> browserDispatches = new ConcurrentLinkedQueue<BrowserDispatch>(); 426 427 @Override 428 public void addSubscription(ConnectionContext context, Subscription sub) throws Exception { 429 LOG.debug("{} add sub: {}, dequeues: {}, dispatched: {}, inflight: {}", new Object[]{ getActiveMQDestination().getQualifiedName(), sub, getDestinationStatistics().getDequeues().getCount(), getDestinationStatistics().getDispatched().getCount(), getDestinationStatistics().getInflight().getCount() }); 430 431 super.addSubscription(context, sub); 432 // synchronize with dispatch method so that no new messages are sent 433 // while setting up a subscription. avoid out of order messages, 434 // duplicates, etc. 435 pagedInPendingDispatchLock.writeLock().lock(); 436 try { 437 438 sub.add(context, this); 439 440 // needs to be synchronized - so no contention with dispatching 441 // consumersLock. 442 consumersLock.writeLock().lock(); 443 try { 444 // set a flag if this is a first consumer 445 if (consumers.size() == 0) { 446 firstConsumer = true; 447 if (consumersBeforeDispatchStarts != 0) { 448 consumersBeforeStartsLatch = new CountDownLatch(consumersBeforeDispatchStarts - 1); 449 } 450 } else { 451 if (consumersBeforeStartsLatch != null) { 452 consumersBeforeStartsLatch.countDown(); 453 } 454 } 455 456 addToConsumerList(sub); 457 if (sub.getConsumerInfo().isExclusive() || isAllConsumersExclusiveByDefault()) { 458 Subscription exclusiveConsumer = dispatchSelector.getExclusiveConsumer(); 459 if (exclusiveConsumer == null) { 460 exclusiveConsumer = sub; 461 } else if (sub.getConsumerInfo().getPriority() == Byte.MAX_VALUE || 462 sub.getConsumerInfo().getPriority() > exclusiveConsumer.getConsumerInfo().getPriority()) { 463 exclusiveConsumer = sub; 464 } 465 dispatchSelector.setExclusiveConsumer(exclusiveConsumer); 466 } 467 } finally { 468 consumersLock.writeLock().unlock(); 469 } 470 471 if (sub instanceof QueueBrowserSubscription) { 472 // tee up for dispatch in next iterate 473 QueueBrowserSubscription browserSubscription = (QueueBrowserSubscription) sub; 474 BrowserDispatch browserDispatch = new BrowserDispatch(browserSubscription); 475 browserDispatches.add(browserDispatch); 476 } 477 478 if (!this.optimizedDispatch) { 479 wakeup(); 480 } 481 } finally { 482 pagedInPendingDispatchLock.writeLock().unlock(); 483 } 484 if (this.optimizedDispatch) { 485 // Outside of dispatchLock() to maintain the lock hierarchy of 486 // iteratingMutex -> dispatchLock. - see 487 // https://issues.apache.org/activemq/browse/AMQ-1878 488 wakeup(); 489 } 490 } 491 492 @Override 493 public void removeSubscription(ConnectionContext context, Subscription sub, long lastDeliveredSequenceId) 494 throws Exception { 495 super.removeSubscription(context, sub, lastDeliveredSequenceId); 496 // synchronize with dispatch method so that no new messages are sent 497 // while removing up a subscription. 498 pagedInPendingDispatchLock.writeLock().lock(); 499 try { 500 LOG.debug("{} remove sub: {}, lastDeliveredSeqId: {}, dequeues: {}, dispatched: {}, inflight: {}, groups: {}", new Object[]{ 501 getActiveMQDestination().getQualifiedName(), 502 sub, 503 lastDeliveredSequenceId, 504 getDestinationStatistics().getDequeues().getCount(), 505 getDestinationStatistics().getDispatched().getCount(), 506 getDestinationStatistics().getInflight().getCount(), 507 sub.getConsumerInfo().getAssignedGroupCount(destination) 508 }); 509 consumersLock.writeLock().lock(); 510 try { 511 removeFromConsumerList(sub); 512 if (sub.getConsumerInfo().isExclusive()) { 513 Subscription exclusiveConsumer = dispatchSelector.getExclusiveConsumer(); 514 if (exclusiveConsumer == sub) { 515 exclusiveConsumer = null; 516 for (Subscription s : consumers) { 517 if (s.getConsumerInfo().isExclusive() 518 && (exclusiveConsumer == null || s.getConsumerInfo().getPriority() > exclusiveConsumer 519 .getConsumerInfo().getPriority())) { 520 exclusiveConsumer = s; 521 522 } 523 } 524 dispatchSelector.setExclusiveConsumer(exclusiveConsumer); 525 } 526 } else if (isAllConsumersExclusiveByDefault()) { 527 Subscription exclusiveConsumer = null; 528 for (Subscription s : consumers) { 529 if (exclusiveConsumer == null 530 || s.getConsumerInfo().getPriority() > exclusiveConsumer 531 .getConsumerInfo().getPriority()) { 532 exclusiveConsumer = s; 533 } 534 } 535 dispatchSelector.setExclusiveConsumer(exclusiveConsumer); 536 } 537 ConsumerId consumerId = sub.getConsumerInfo().getConsumerId(); 538 getMessageGroupOwners().removeConsumer(consumerId); 539 540 // redeliver inflight messages 541 542 boolean markAsRedelivered = false; 543 MessageReference lastDeliveredRef = null; 544 List<MessageReference> unAckedMessages = sub.remove(context, this); 545 546 // locate last redelivered in unconsumed list (list in delivery rather than seq order) 547 if (lastDeliveredSequenceId > RemoveInfo.LAST_DELIVERED_UNSET) { 548 for (MessageReference ref : unAckedMessages) { 549 if (ref.getMessageId().getBrokerSequenceId() == lastDeliveredSequenceId) { 550 lastDeliveredRef = ref; 551 markAsRedelivered = true; 552 LOG.debug("found lastDeliveredSeqID: {}, message reference: {}", lastDeliveredSequenceId, ref.getMessageId()); 553 break; 554 } 555 } 556 } 557 558 for (Iterator<MessageReference> unackedListIterator = unAckedMessages.iterator(); unackedListIterator.hasNext(); ) { 559 MessageReference ref = unackedListIterator.next(); 560 // AMQ-5107: don't resend if the broker is shutting down 561 if ( this.brokerService.isStopping() ) { 562 break; 563 } 564 QueueMessageReference qmr = (QueueMessageReference) ref; 565 if (qmr.getLockOwner() == sub) { 566 qmr.unlock(); 567 568 // have no delivery information 569 if (lastDeliveredSequenceId == RemoveInfo.LAST_DELIVERED_UNKNOWN) { 570 qmr.incrementRedeliveryCounter(); 571 } else { 572 if (markAsRedelivered) { 573 qmr.incrementRedeliveryCounter(); 574 } 575 if (ref == lastDeliveredRef) { 576 // all that follow were not redelivered 577 markAsRedelivered = false; 578 } 579 } 580 } 581 if (qmr.isDropped()) { 582 unackedListIterator.remove(); 583 } 584 } 585 dispatchPendingList.addForRedelivery(unAckedMessages, strictOrderDispatch && consumers.isEmpty()); 586 if (sub instanceof QueueBrowserSubscription) { 587 ((QueueBrowserSubscription)sub).decrementQueueRef(); 588 browserDispatches.remove(sub); 589 } 590 // AMQ-5107: don't resend if the broker is shutting down 591 if (dispatchPendingList.hasRedeliveries() && (! this.brokerService.isStopping())) { 592 doDispatch(new OrderedPendingList()); 593 } 594 } finally { 595 consumersLock.writeLock().unlock(); 596 } 597 if (!this.optimizedDispatch) { 598 wakeup(); 599 } 600 } finally { 601 pagedInPendingDispatchLock.writeLock().unlock(); 602 } 603 if (this.optimizedDispatch) { 604 // Outside of dispatchLock() to maintain the lock hierarchy of 605 // iteratingMutex -> dispatchLock. - see 606 // https://issues.apache.org/activemq/browse/AMQ-1878 607 wakeup(); 608 } 609 } 610 611 @Override 612 public void send(final ProducerBrokerExchange producerExchange, final Message message) throws Exception { 613 final ConnectionContext context = producerExchange.getConnectionContext(); 614 // There is delay between the client sending it and it arriving at the 615 // destination.. it may have expired. 616 message.setRegionDestination(this); 617 ProducerState state = producerExchange.getProducerState(); 618 if (state == null) { 619 LOG.warn("Send failed for: {}, missing producer state for: {}", message, producerExchange); 620 throw new JMSException("Cannot send message to " + getActiveMQDestination() + " with invalid (null) producer state"); 621 } 622 final ProducerInfo producerInfo = producerExchange.getProducerState().getInfo(); 623 final boolean sendProducerAck = !message.isResponseRequired() && producerInfo.getWindowSize() > 0 624 && !context.isInRecoveryMode(); 625 if (message.isExpired()) { 626 // message not stored - or added to stats yet - so chuck here 627 broker.getRoot().messageExpired(context, message, null); 628 if (sendProducerAck) { 629 ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize()); 630 context.getConnection().dispatchAsync(ack); 631 } 632 return; 633 } 634 if (memoryUsage.isFull()) { 635 isFull(context, memoryUsage); 636 fastProducer(context, producerInfo); 637 if (isProducerFlowControl() && context.isProducerFlowControl()) { 638 if (isFlowControlLogRequired()) { 639 LOG.info("Usage Manager Memory Limit ({}) reached on {}, size {}. Producers will be throttled to the rate at which messages are removed from this destination to prevent flooding it. See http://activemq.apache.org/producer-flow-control.html for more info.", 640 memoryUsage.getLimit(), getActiveMQDestination().getQualifiedName(), destinationStatistics.getMessages().getCount()); 641 642 } 643 if (!context.isNetworkConnection() && systemUsage.isSendFailIfNoSpace()) { 644 throw new ResourceAllocationException("Usage Manager Memory Limit reached. Stopping producer (" 645 + message.getProducerId() + ") to prevent flooding " 646 + getActiveMQDestination().getQualifiedName() + "." 647 + " See http://activemq.apache.org/producer-flow-control.html for more info"); 648 } 649 650 // We can avoid blocking due to low usage if the producer is 651 // sending 652 // a sync message or if it is using a producer window 653 if (producerInfo.getWindowSize() > 0 || message.isResponseRequired()) { 654 // copy the exchange state since the context will be 655 // modified while we are waiting 656 // for space. 657 final ProducerBrokerExchange producerExchangeCopy = producerExchange.copy(); 658 synchronized (messagesWaitingForSpace) { 659 // Start flow control timeout task 660 // Prevent trying to start it multiple times 661 if (!flowControlTimeoutTask.isAlive()) { 662 flowControlTimeoutTask.setName(getName()+" Producer Flow Control Timeout Task"); 663 flowControlTimeoutTask.start(); 664 } 665 messagesWaitingForSpace.put(message.getMessageId(), new Runnable() { 666 @Override 667 public void run() { 668 669 try { 670 // While waiting for space to free up... the 671 // transaction may be done 672 if (message.isInTransaction()) { 673 if (context.getTransaction().getState() > IN_USE_STATE) { 674 throw new JMSException("Send transaction completed while waiting for space"); 675 } 676 } 677 678 // the message may have expired. 679 if (message.isExpired()) { 680 LOG.error("message expired waiting for space"); 681 broker.messageExpired(context, message, null); 682 destinationStatistics.getExpired().increment(); 683 } else { 684 doMessageSend(producerExchangeCopy, message); 685 } 686 687 if (sendProducerAck) { 688 ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message 689 .getSize()); 690 context.getConnection().dispatchAsync(ack); 691 } else { 692 Response response = new Response(); 693 response.setCorrelationId(message.getCommandId()); 694 context.getConnection().dispatchAsync(response); 695 } 696 697 } catch (Exception e) { 698 if (!sendProducerAck && !context.isInRecoveryMode() && !brokerService.isStopping()) { 699 ExceptionResponse response = new ExceptionResponse(e); 700 response.setCorrelationId(message.getCommandId()); 701 context.getConnection().dispatchAsync(response); 702 } else { 703 LOG.debug("unexpected exception on deferred send of: {}", message, e); 704 } 705 } finally { 706 getDestinationStatistics().getBlockedSends().decrement(); 707 producerExchangeCopy.blockingOnFlowControl(false); 708 } 709 } 710 }); 711 712 getDestinationStatistics().getBlockedSends().increment(); 713 producerExchange.blockingOnFlowControl(true); 714 if (!context.isNetworkConnection() && systemUsage.getSendFailIfNoSpaceAfterTimeout() != 0) { 715 flowControlTimeoutMessages.add(new TimeoutMessage(message, context, systemUsage 716 .getSendFailIfNoSpaceAfterTimeout())); 717 } 718 719 registerCallbackForNotFullNotification(); 720 context.setDontSendReponse(true); 721 return; 722 } 723 724 } else { 725 726 if (memoryUsage.isFull()) { 727 waitForSpace(context, producerExchange, memoryUsage, "Usage Manager Memory Limit reached. Producer (" 728 + message.getProducerId() + ") stopped to prevent flooding " 729 + getActiveMQDestination().getQualifiedName() + "." 730 + " See http://activemq.apache.org/producer-flow-control.html for more info"); 731 } 732 733 // The usage manager could have delayed us by the time 734 // we unblock the message could have expired.. 735 if (message.isExpired()) { 736 LOG.debug("Expired message: {}", message); 737 broker.getRoot().messageExpired(context, message, null); 738 return; 739 } 740 } 741 } 742 } 743 doMessageSend(producerExchange, message); 744 if (sendProducerAck) { 745 ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize()); 746 context.getConnection().dispatchAsync(ack); 747 } 748 } 749 750 private void registerCallbackForNotFullNotification() { 751 // If the usage manager is not full, then the task will not 752 // get called.. 753 if (!memoryUsage.notifyCallbackWhenNotFull(sendMessagesWaitingForSpaceTask)) { 754 // so call it directly here. 755 sendMessagesWaitingForSpaceTask.run(); 756 } 757 } 758 759 private final LinkedList<MessageContext> indexOrderedCursorUpdates = new LinkedList<>(); 760 761 @Override 762 public void onAdd(MessageContext messageContext) { 763 synchronized (indexOrderedCursorUpdates) { 764 indexOrderedCursorUpdates.addLast(messageContext); 765 } 766 } 767 768 private void doPendingCursorAdditions() throws Exception { 769 LinkedList<MessageContext> orderedUpdates = new LinkedList<>(); 770 sendLock.lockInterruptibly(); 771 try { 772 synchronized (indexOrderedCursorUpdates) { 773 MessageContext candidate = indexOrderedCursorUpdates.peek(); 774 while (candidate != null && candidate.message.getMessageId().getFutureOrSequenceLong() != null) { 775 candidate = indexOrderedCursorUpdates.removeFirst(); 776 // check for duplicate adds suppressed by the store 777 if (candidate.message.getMessageId().getFutureOrSequenceLong() instanceof Long && ((Long)candidate.message.getMessageId().getFutureOrSequenceLong()).compareTo(-1l) == 0) { 778 LOG.warn("{} messageStore indicated duplicate add attempt for {}, suppressing duplicate dispatch", this, candidate.message.getMessageId()); 779 } else { 780 orderedUpdates.add(candidate); 781 } 782 candidate = indexOrderedCursorUpdates.peek(); 783 } 784 } 785 messagesLock.writeLock().lock(); 786 try { 787 for (MessageContext messageContext : orderedUpdates) { 788 if (!messages.addMessageLast(messageContext.message)) { 789 // cursor suppressed a duplicate 790 messageContext.duplicate = true; 791 } 792 if (messageContext.onCompletion != null) { 793 messageContext.onCompletion.run(); 794 } 795 } 796 } finally { 797 messagesLock.writeLock().unlock(); 798 } 799 } finally { 800 sendLock.unlock(); 801 } 802 for (MessageContext messageContext : orderedUpdates) { 803 if (!messageContext.duplicate) { 804 messageSent(messageContext.context, messageContext.message); 805 } 806 } 807 orderedUpdates.clear(); 808 } 809 810 final class CursorAddSync extends Synchronization { 811 812 private final MessageContext messageContext; 813 814 CursorAddSync(MessageContext messageContext) { 815 this.messageContext = messageContext; 816 this.messageContext.message.incrementReferenceCount(); 817 } 818 819 @Override 820 public void afterCommit() throws Exception { 821 if (store != null && messageContext.message.isPersistent()) { 822 doPendingCursorAdditions(); 823 } else { 824 cursorAdd(messageContext.message); 825 messageSent(messageContext.context, messageContext.message); 826 } 827 messageContext.message.decrementReferenceCount(); 828 } 829 830 @Override 831 public void afterRollback() throws Exception { 832 messageContext.message.decrementReferenceCount(); 833 } 834 } 835 836 void doMessageSend(final ProducerBrokerExchange producerExchange, final Message message) throws IOException, 837 Exception { 838 final ConnectionContext context = producerExchange.getConnectionContext(); 839 ListenableFuture<Object> result = null; 840 841 producerExchange.incrementSend(); 842 pendingSends.incrementAndGet(); 843 do { 844 checkUsage(context, producerExchange, message); 845 message.getMessageId().setBrokerSequenceId(getDestinationSequenceId()); 846 if (store != null && message.isPersistent()) { 847 message.getMessageId().setFutureOrSequenceLong(null); 848 try { 849 //AMQ-6133 - don't store async if using persistJMSRedelivered 850 //This flag causes a sync update later on dispatch which can cause a race 851 //condition if the original add is processed after the update, which can cause 852 //a duplicate message to be stored 853 if (messages.isCacheEnabled() && !isPersistJMSRedelivered()) { 854 result = store.asyncAddQueueMessage(context, message, isOptimizeStorage()); 855 result.addListener(new PendingMarshalUsageTracker(message)); 856 } else { 857 store.addMessage(context, message); 858 } 859 } catch (Exception e) { 860 // we may have a store in inconsistent state, so reset the cursor 861 // before restarting normal broker operations 862 resetNeeded = true; 863 pendingSends.decrementAndGet(); 864 throw e; 865 } 866 } 867 868 //Clear the unmarshalled state if the message is marshalled 869 //Persistent messages will always be marshalled but non-persistent may not be 870 //Specially non-persistent messages over the VM transport won't be 871 if (isReduceMemoryFootprint() && message.isMarshalled()) { 872 message.clearUnMarshalledState(); 873 } 874 if(tryOrderedCursorAdd(message, context)) { 875 break; 876 } 877 } while (started.get()); 878 879 if (result != null && message.isResponseRequired() && !result.isCancelled()) { 880 try { 881 result.get(); 882 } catch (CancellationException e) { 883 // ignore - the task has been cancelled if the message 884 // has already been deleted 885 } 886 } 887 } 888 889 private boolean tryOrderedCursorAdd(Message message, ConnectionContext context) throws Exception { 890 boolean result = true; 891 892 if (context.isInTransaction()) { 893 context.getTransaction().addSynchronization(new CursorAddSync(new MessageContext(context, message, null))); 894 } else if (store != null && message.isPersistent()) { 895 doPendingCursorAdditions(); 896 } else { 897 // no ordering issue with non persistent messages 898 result = tryCursorAdd(message); 899 messageSent(context, message); 900 } 901 902 return result; 903 } 904 905 private void checkUsage(ConnectionContext context,ProducerBrokerExchange producerBrokerExchange, Message message) throws ResourceAllocationException, IOException, InterruptedException { 906 if (message.isPersistent()) { 907 if (store != null && systemUsage.getStoreUsage().isFull(getStoreUsageHighWaterMark())) { 908 final String logMessage = "Persistent store is Full, " + getStoreUsageHighWaterMark() + "% of " 909 + systemUsage.getStoreUsage().getLimit() + ". Stopping producer (" 910 + message.getProducerId() + ") to prevent flooding " 911 + getActiveMQDestination().getQualifiedName() + "." 912 + " See http://activemq.apache.org/producer-flow-control.html for more info"; 913 914 waitForSpace(context, producerBrokerExchange, systemUsage.getStoreUsage(), getStoreUsageHighWaterMark(), logMessage); 915 } 916 } else if (messages.getSystemUsage() != null && systemUsage.getTempUsage().isFull()) { 917 final String logMessage = "Temp Store is Full (" 918 + systemUsage.getTempUsage().getPercentUsage() + "% of " + systemUsage.getTempUsage().getLimit() 919 +"). Stopping producer (" + message.getProducerId() 920 + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "." 921 + " See http://activemq.apache.org/producer-flow-control.html for more info"; 922 923 waitForSpace(context, producerBrokerExchange, messages.getSystemUsage().getTempUsage(), logMessage); 924 } 925 } 926 927 private void expireMessages() { 928 LOG.debug("{} expiring messages ..", getActiveMQDestination().getQualifiedName()); 929 930 // just track the insertion count 931 List<Message> browsedMessages = new InsertionCountList<Message>(); 932 doBrowse(browsedMessages, this.getMaxExpirePageSize()); 933 asyncWakeup(); 934 LOG.debug("{} expiring messages done.", getActiveMQDestination().getQualifiedName()); 935 } 936 937 @Override 938 public void gc() { 939 } 940 941 @Override 942 public void acknowledge(ConnectionContext context, Subscription sub, MessageAck ack, MessageReference node) 943 throws IOException { 944 messageConsumed(context, node); 945 if (store != null && node.isPersistent()) { 946 store.removeAsyncMessage(context, convertToNonRangedAck(ack, node)); 947 } 948 } 949 950 Message loadMessage(MessageId messageId) throws IOException { 951 Message msg = null; 952 if (store != null) { // can be null for a temp q 953 msg = store.getMessage(messageId); 954 if (msg != null) { 955 msg.setRegionDestination(this); 956 } 957 } 958 return msg; 959 } 960 961 public long getPendingMessageSize() { 962 messagesLock.readLock().lock(); 963 try{ 964 return messages.messageSize(); 965 } finally { 966 messagesLock.readLock().unlock(); 967 } 968 } 969 970 public long getPendingMessageCount() { 971 return this.destinationStatistics.getMessages().getCount(); 972 } 973 974 @Override 975 public String toString() { 976 return destination.getQualifiedName() + ", subscriptions=" + consumers.size() 977 + ", memory=" + memoryUsage.getPercentUsage() + "%, size=" + destinationStatistics.getMessages().getCount() + ", pending=" 978 + indexOrderedCursorUpdates.size(); 979 } 980 981 @Override 982 public void start() throws Exception { 983 if (started.compareAndSet(false, true)) { 984 if (memoryUsage != null) { 985 memoryUsage.start(); 986 } 987 if (systemUsage.getStoreUsage() != null) { 988 systemUsage.getStoreUsage().start(); 989 } 990 systemUsage.getMemoryUsage().addUsageListener(this); 991 messages.start(); 992 if (getExpireMessagesPeriod() > 0) { 993 scheduler.executePeriodically(expireMessagesTask, getExpireMessagesPeriod()); 994 } 995 doPageIn(false); 996 } 997 } 998 999 @Override 1000 public void stop() throws Exception { 1001 if (started.compareAndSet(true, false)) { 1002 if (taskRunner != null) { 1003 taskRunner.shutdown(); 1004 } 1005 if (this.executor != null) { 1006 ThreadPoolUtils.shutdownNow(executor); 1007 executor = null; 1008 } 1009 1010 scheduler.cancel(expireMessagesTask); 1011 1012 if (flowControlTimeoutTask.isAlive()) { 1013 flowControlTimeoutTask.interrupt(); 1014 } 1015 1016 if (messages != null) { 1017 messages.stop(); 1018 } 1019 1020 for (MessageReference messageReference : pagedInMessages.values()) { 1021 messageReference.decrementReferenceCount(); 1022 } 1023 pagedInMessages.clear(); 1024 1025 systemUsage.getMemoryUsage().removeUsageListener(this); 1026 if (memoryUsage != null) { 1027 memoryUsage.stop(); 1028 } 1029 if (systemUsage.getStoreUsage() != null) { 1030 systemUsage.getStoreUsage().stop(); 1031 } 1032 if (store != null) { 1033 store.stop(); 1034 } 1035 } 1036 } 1037 1038 // Properties 1039 // ------------------------------------------------------------------------- 1040 @Override 1041 public ActiveMQDestination getActiveMQDestination() { 1042 return destination; 1043 } 1044 1045 public MessageGroupMap getMessageGroupOwners() { 1046 if (messageGroupOwners == null) { 1047 messageGroupOwners = getMessageGroupMapFactory().createMessageGroupMap(); 1048 messageGroupOwners.setDestination(this); 1049 } 1050 return messageGroupOwners; 1051 } 1052 1053 public DispatchPolicy getDispatchPolicy() { 1054 return dispatchPolicy; 1055 } 1056 1057 public void setDispatchPolicy(DispatchPolicy dispatchPolicy) { 1058 this.dispatchPolicy = dispatchPolicy; 1059 } 1060 1061 public MessageGroupMapFactory getMessageGroupMapFactory() { 1062 return messageGroupMapFactory; 1063 } 1064 1065 public void setMessageGroupMapFactory(MessageGroupMapFactory messageGroupMapFactory) { 1066 this.messageGroupMapFactory = messageGroupMapFactory; 1067 } 1068 1069 public PendingMessageCursor getMessages() { 1070 return this.messages; 1071 } 1072 1073 public void setMessages(PendingMessageCursor messages) { 1074 this.messages = messages; 1075 } 1076 1077 public boolean isUseConsumerPriority() { 1078 return useConsumerPriority; 1079 } 1080 1081 public void setUseConsumerPriority(boolean useConsumerPriority) { 1082 this.useConsumerPriority = useConsumerPriority; 1083 } 1084 1085 public boolean isStrictOrderDispatch() { 1086 return strictOrderDispatch; 1087 } 1088 1089 public void setStrictOrderDispatch(boolean strictOrderDispatch) { 1090 this.strictOrderDispatch = strictOrderDispatch; 1091 } 1092 1093 public boolean isOptimizedDispatch() { 1094 return optimizedDispatch; 1095 } 1096 1097 public void setOptimizedDispatch(boolean optimizedDispatch) { 1098 this.optimizedDispatch = optimizedDispatch; 1099 } 1100 1101 public int getTimeBeforeDispatchStarts() { 1102 return timeBeforeDispatchStarts; 1103 } 1104 1105 public void setTimeBeforeDispatchStarts(int timeBeforeDispatchStarts) { 1106 this.timeBeforeDispatchStarts = timeBeforeDispatchStarts; 1107 } 1108 1109 public int getConsumersBeforeDispatchStarts() { 1110 return consumersBeforeDispatchStarts; 1111 } 1112 1113 public void setConsumersBeforeDispatchStarts(int consumersBeforeDispatchStarts) { 1114 this.consumersBeforeDispatchStarts = consumersBeforeDispatchStarts; 1115 } 1116 1117 public void setAllConsumersExclusiveByDefault(boolean allConsumersExclusiveByDefault) { 1118 this.allConsumersExclusiveByDefault = allConsumersExclusiveByDefault; 1119 } 1120 1121 public boolean isAllConsumersExclusiveByDefault() { 1122 return allConsumersExclusiveByDefault; 1123 } 1124 1125 public boolean isResetNeeded() { 1126 return resetNeeded; 1127 } 1128 1129 // Implementation methods 1130 // ------------------------------------------------------------------------- 1131 private QueueMessageReference createMessageReference(Message message) { 1132 QueueMessageReference result = new IndirectMessageReference(message); 1133 return result; 1134 } 1135 1136 @Override 1137 public Message[] browse() { 1138 List<Message> browseList = new ArrayList<Message>(); 1139 doBrowse(browseList, getMaxBrowsePageSize()); 1140 return browseList.toArray(new Message[browseList.size()]); 1141 } 1142 1143 public void doBrowse(List<Message> browseList, int max) { 1144 final ConnectionContext connectionContext = createConnectionContext(); 1145 try { 1146 int maxPageInAttempts = 1; 1147 if (max > 0) { 1148 messagesLock.readLock().lock(); 1149 try { 1150 maxPageInAttempts += (messages.size() / max); 1151 } finally { 1152 messagesLock.readLock().unlock(); 1153 } 1154 while (shouldPageInMoreForBrowse(max) && maxPageInAttempts-- > 0) { 1155 pageInMessages(!memoryUsage.isFull(110), max); 1156 } 1157 } 1158 doBrowseList(browseList, max, dispatchPendingList, pagedInPendingDispatchLock, connectionContext, "redeliveredWaitingDispatch+pagedInPendingDispatch"); 1159 doBrowseList(browseList, max, pagedInMessages, pagedInMessagesLock, connectionContext, "pagedInMessages"); 1160 1161 // we need a store iterator to walk messages on disk, independent of the cursor which is tracking 1162 // the next message batch 1163 } catch (BrokerStoppedException ignored) { 1164 } catch (Exception e) { 1165 LOG.error("Problem retrieving message for browse", e); 1166 } 1167 } 1168 1169 protected void doBrowseList(List<Message> browseList, int max, PendingList list, ReentrantReadWriteLock lock, ConnectionContext connectionContext, String name) throws Exception { 1170 List<MessageReference> toExpire = new ArrayList<MessageReference>(); 1171 lock.readLock().lock(); 1172 try { 1173 addAll(list.values(), browseList, max, toExpire); 1174 } finally { 1175 lock.readLock().unlock(); 1176 } 1177 for (MessageReference ref : toExpire) { 1178 if (broker.isExpired(ref)) { 1179 LOG.debug("expiring from {}: {}", name, ref); 1180 messageExpired(connectionContext, ref); 1181 } else { 1182 lock.writeLock().lock(); 1183 try { 1184 list.remove(ref); 1185 } finally { 1186 lock.writeLock().unlock(); 1187 } 1188 ref.decrementReferenceCount(); 1189 } 1190 } 1191 } 1192 1193 private boolean shouldPageInMoreForBrowse(int max) { 1194 int alreadyPagedIn = 0; 1195 pagedInMessagesLock.readLock().lock(); 1196 try { 1197 alreadyPagedIn = pagedInMessages.size(); 1198 } finally { 1199 pagedInMessagesLock.readLock().unlock(); 1200 } 1201 int messagesInQueue = alreadyPagedIn; 1202 messagesLock.readLock().lock(); 1203 try { 1204 messagesInQueue += messages.size(); 1205 } finally { 1206 messagesLock.readLock().unlock(); 1207 } 1208 1209 LOG.trace("max {}, alreadyPagedIn {}, messagesCount {}, memoryUsage {}%", new Object[]{max, alreadyPagedIn, messagesInQueue, memoryUsage.getPercentUsage()}); 1210 return (alreadyPagedIn < max) 1211 && (alreadyPagedIn < messagesInQueue) 1212 && messages.hasSpace(); 1213 } 1214 1215 private void addAll(Collection<? extends MessageReference> refs, List<Message> l, int max, 1216 List<MessageReference> toExpire) throws Exception { 1217 for (Iterator<? extends MessageReference> i = refs.iterator(); i.hasNext() && l.size() < max;) { 1218 QueueMessageReference ref = (QueueMessageReference) i.next(); 1219 if (ref.isExpired() && (ref.getLockOwner() == null)) { 1220 toExpire.add(ref); 1221 } else if (l.contains(ref.getMessage()) == false) { 1222 l.add(ref.getMessage()); 1223 } 1224 } 1225 } 1226 1227 public QueueMessageReference getMessage(String id) { 1228 MessageId msgId = new MessageId(id); 1229 pagedInMessagesLock.readLock().lock(); 1230 try { 1231 QueueMessageReference ref = (QueueMessageReference)this.pagedInMessages.get(msgId); 1232 if (ref != null) { 1233 return ref; 1234 } 1235 } finally { 1236 pagedInMessagesLock.readLock().unlock(); 1237 } 1238 messagesLock.writeLock().lock(); 1239 try{ 1240 try { 1241 messages.reset(); 1242 while (messages.hasNext()) { 1243 MessageReference mr = messages.next(); 1244 QueueMessageReference qmr = createMessageReference(mr.getMessage()); 1245 qmr.decrementReferenceCount(); 1246 messages.rollback(qmr.getMessageId()); 1247 if (msgId.equals(qmr.getMessageId())) { 1248 return qmr; 1249 } 1250 } 1251 } finally { 1252 messages.release(); 1253 } 1254 }finally { 1255 messagesLock.writeLock().unlock(); 1256 } 1257 return null; 1258 } 1259 1260 public void purge() throws Exception { 1261 ConnectionContext c = createConnectionContext(); 1262 List<MessageReference> list = null; 1263 try { 1264 sendLock.lock(); 1265 long originalMessageCount = this.destinationStatistics.getMessages().getCount(); 1266 do { 1267 doPageIn(true, false, getMaxPageSize()); // signal no expiry processing needed. 1268 pagedInMessagesLock.readLock().lock(); 1269 try { 1270 list = new ArrayList<MessageReference>(pagedInMessages.values()); 1271 }finally { 1272 pagedInMessagesLock.readLock().unlock(); 1273 } 1274 1275 for (MessageReference ref : list) { 1276 try { 1277 QueueMessageReference r = (QueueMessageReference) ref; 1278 removeMessage(c, r); 1279 } catch (IOException e) { 1280 } 1281 } 1282 // don't spin/hang if stats are out and there is nothing left in the 1283 // store 1284 } while (!list.isEmpty() && this.destinationStatistics.getMessages().getCount() > 0); 1285 1286 if (getMessages().getMessageAudit() != null) { 1287 getMessages().getMessageAudit().clear(); 1288 } 1289 1290 if (this.destinationStatistics.getMessages().getCount() > 0) { 1291 LOG.warn("{} after purge of {} messages, message count stats report: {}", getActiveMQDestination().getQualifiedName(), originalMessageCount, this.destinationStatistics.getMessages().getCount()); 1292 } 1293 } finally { 1294 sendLock.unlock(); 1295 } 1296 } 1297 1298 @Override 1299 public void clearPendingMessages() { 1300 messagesLock.writeLock().lock(); 1301 try { 1302 if (resetNeeded) { 1303 messages.gc(); 1304 messages.reset(); 1305 resetNeeded = false; 1306 } else { 1307 messages.rebase(); 1308 } 1309 asyncWakeup(); 1310 } finally { 1311 messagesLock.writeLock().unlock(); 1312 } 1313 } 1314 1315 /** 1316 * Removes the message matching the given messageId 1317 */ 1318 public boolean removeMessage(String messageId) throws Exception { 1319 return removeMatchingMessages(createMessageIdFilter(messageId), 1) > 0; 1320 } 1321 1322 /** 1323 * Removes the messages matching the given selector 1324 * 1325 * @return the number of messages removed 1326 */ 1327 public int removeMatchingMessages(String selector) throws Exception { 1328 return removeMatchingMessages(selector, -1); 1329 } 1330 1331 /** 1332 * Removes the messages matching the given selector up to the maximum number 1333 * of matched messages 1334 * 1335 * @return the number of messages removed 1336 */ 1337 public int removeMatchingMessages(String selector, int maximumMessages) throws Exception { 1338 return removeMatchingMessages(createSelectorFilter(selector), maximumMessages); 1339 } 1340 1341 /** 1342 * Removes the messages matching the given filter up to the maximum number 1343 * of matched messages 1344 * 1345 * @return the number of messages removed 1346 */ 1347 public int removeMatchingMessages(MessageReferenceFilter filter, int maximumMessages) throws Exception { 1348 int movedCounter = 0; 1349 Set<MessageReference> set = new LinkedHashSet<MessageReference>(); 1350 ConnectionContext context = createConnectionContext(); 1351 do { 1352 doPageIn(true); 1353 pagedInMessagesLock.readLock().lock(); 1354 try { 1355 set.addAll(pagedInMessages.values()); 1356 } finally { 1357 pagedInMessagesLock.readLock().unlock(); 1358 } 1359 List<MessageReference> list = new ArrayList<MessageReference>(set); 1360 for (MessageReference ref : list) { 1361 IndirectMessageReference r = (IndirectMessageReference) ref; 1362 if (filter.evaluate(context, r)) { 1363 1364 removeMessage(context, r); 1365 set.remove(r); 1366 if (++movedCounter >= maximumMessages && maximumMessages > 0) { 1367 return movedCounter; 1368 } 1369 } 1370 } 1371 } while (set.size() < this.destinationStatistics.getMessages().getCount()); 1372 return movedCounter; 1373 } 1374 1375 /** 1376 * Copies the message matching the given messageId 1377 */ 1378 public boolean copyMessageTo(ConnectionContext context, String messageId, ActiveMQDestination dest) 1379 throws Exception { 1380 return copyMatchingMessages(context, createMessageIdFilter(messageId), dest, 1) > 0; 1381 } 1382 1383 /** 1384 * Copies the messages matching the given selector 1385 * 1386 * @return the number of messages copied 1387 */ 1388 public int copyMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest) 1389 throws Exception { 1390 return copyMatchingMessagesTo(context, selector, dest, -1); 1391 } 1392 1393 /** 1394 * Copies the messages matching the given selector up to the maximum number 1395 * of matched messages 1396 * 1397 * @return the number of messages copied 1398 */ 1399 public int copyMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest, 1400 int maximumMessages) throws Exception { 1401 return copyMatchingMessages(context, createSelectorFilter(selector), dest, maximumMessages); 1402 } 1403 1404 /** 1405 * Copies the messages matching the given filter up to the maximum number of 1406 * matched messages 1407 * 1408 * @return the number of messages copied 1409 */ 1410 public int copyMatchingMessages(ConnectionContext context, MessageReferenceFilter filter, ActiveMQDestination dest, 1411 int maximumMessages) throws Exception { 1412 1413 if (destination.equals(dest)) { 1414 return 0; 1415 } 1416 1417 int movedCounter = 0; 1418 int count = 0; 1419 Set<MessageReference> set = new LinkedHashSet<MessageReference>(); 1420 do { 1421 int oldMaxSize = getMaxPageSize(); 1422 setMaxPageSize((int) this.destinationStatistics.getMessages().getCount()); 1423 doPageIn(true); 1424 setMaxPageSize(oldMaxSize); 1425 pagedInMessagesLock.readLock().lock(); 1426 try { 1427 set.addAll(pagedInMessages.values()); 1428 } finally { 1429 pagedInMessagesLock.readLock().unlock(); 1430 } 1431 List<MessageReference> list = new ArrayList<MessageReference>(set); 1432 for (MessageReference ref : list) { 1433 IndirectMessageReference r = (IndirectMessageReference) ref; 1434 if (filter.evaluate(context, r)) { 1435 1436 r.incrementReferenceCount(); 1437 try { 1438 Message m = r.getMessage(); 1439 BrokerSupport.resend(context, m, dest); 1440 if (++movedCounter >= maximumMessages && maximumMessages > 0) { 1441 return movedCounter; 1442 } 1443 } finally { 1444 r.decrementReferenceCount(); 1445 } 1446 } 1447 count++; 1448 } 1449 } while (count < this.destinationStatistics.getMessages().getCount()); 1450 return movedCounter; 1451 } 1452 1453 /** 1454 * Move a message 1455 * 1456 * @param context 1457 * connection context 1458 * @param m 1459 * QueueMessageReference 1460 * @param dest 1461 * ActiveMQDestination 1462 * @throws Exception 1463 */ 1464 public boolean moveMessageTo(ConnectionContext context, QueueMessageReference m, ActiveMQDestination dest) throws Exception { 1465 Set<Destination> destsToPause = regionBroker.getDestinations(dest); 1466 try { 1467 for (Destination d: destsToPause) { 1468 if (d instanceof Queue) { 1469 ((Queue)d).pauseDispatch(); 1470 } 1471 } 1472 BrokerSupport.resend(context, m.getMessage(), dest); 1473 removeMessage(context, m); 1474 messagesLock.writeLock().lock(); 1475 try { 1476 messages.rollback(m.getMessageId()); 1477 if (isDLQ()) { 1478 DeadLetterStrategy strategy = getDeadLetterStrategy(); 1479 strategy.rollback(m.getMessage()); 1480 } 1481 } finally { 1482 messagesLock.writeLock().unlock(); 1483 } 1484 } finally { 1485 for (Destination d: destsToPause) { 1486 if (d instanceof Queue) { 1487 ((Queue)d).resumeDispatch(); 1488 } 1489 } 1490 } 1491 1492 return true; 1493 } 1494 1495 /** 1496 * Moves the message matching the given messageId 1497 */ 1498 public boolean moveMessageTo(ConnectionContext context, String messageId, ActiveMQDestination dest) 1499 throws Exception { 1500 return moveMatchingMessagesTo(context, createMessageIdFilter(messageId), dest, 1) > 0; 1501 } 1502 1503 /** 1504 * Moves the messages matching the given selector 1505 * 1506 * @return the number of messages removed 1507 */ 1508 public int moveMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest) 1509 throws Exception { 1510 return moveMatchingMessagesTo(context, selector, dest, Integer.MAX_VALUE); 1511 } 1512 1513 /** 1514 * Moves the messages matching the given selector up to the maximum number 1515 * of matched messages 1516 */ 1517 public int moveMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest, 1518 int maximumMessages) throws Exception { 1519 return moveMatchingMessagesTo(context, createSelectorFilter(selector), dest, maximumMessages); 1520 } 1521 1522 /** 1523 * Moves the messages matching the given filter up to the maximum number of 1524 * matched messages 1525 */ 1526 public int moveMatchingMessagesTo(ConnectionContext context, MessageReferenceFilter filter, 1527 ActiveMQDestination dest, int maximumMessages) throws Exception { 1528 1529 if (destination.equals(dest)) { 1530 return 0; 1531 } 1532 1533 int movedCounter = 0; 1534 Set<MessageReference> set = new LinkedHashSet<MessageReference>(); 1535 do { 1536 doPageIn(true); 1537 pagedInMessagesLock.readLock().lock(); 1538 try { 1539 set.addAll(pagedInMessages.values()); 1540 } finally { 1541 pagedInMessagesLock.readLock().unlock(); 1542 } 1543 List<MessageReference> list = new ArrayList<MessageReference>(set); 1544 for (MessageReference ref : list) { 1545 if (filter.evaluate(context, ref)) { 1546 // We should only move messages that can be locked. 1547 moveMessageTo(context, (QueueMessageReference)ref, dest); 1548 set.remove(ref); 1549 if (++movedCounter >= maximumMessages && maximumMessages > 0) { 1550 return movedCounter; 1551 } 1552 } 1553 } 1554 } while (set.size() < this.destinationStatistics.getMessages().getCount() && set.size() < maximumMessages); 1555 return movedCounter; 1556 } 1557 1558 public int retryMessages(ConnectionContext context, int maximumMessages) throws Exception { 1559 if (!isDLQ()) { 1560 throw new Exception("Retry of message is only possible on Dead Letter Queues!"); 1561 } 1562 int restoredCounter = 0; 1563 // ensure we deal with a snapshot to avoid potential duplicates in the event of messages 1564 // getting immediate dlq'ed 1565 long numberOfRetryAttemptsToCheckAllMessagesOnce = this.destinationStatistics.getMessages().getCount(); 1566 Set<MessageReference> set = new LinkedHashSet<MessageReference>(); 1567 do { 1568 doPageIn(true); 1569 pagedInMessagesLock.readLock().lock(); 1570 try { 1571 set.addAll(pagedInMessages.values()); 1572 } finally { 1573 pagedInMessagesLock.readLock().unlock(); 1574 } 1575 List<MessageReference> list = new ArrayList<MessageReference>(set); 1576 for (MessageReference ref : list) { 1577 numberOfRetryAttemptsToCheckAllMessagesOnce--; 1578 if (ref.getMessage().getOriginalDestination() != null) { 1579 1580 moveMessageTo(context, (QueueMessageReference)ref, ref.getMessage().getOriginalDestination()); 1581 set.remove(ref); 1582 if (++restoredCounter >= maximumMessages && maximumMessages > 0) { 1583 return restoredCounter; 1584 } 1585 } 1586 } 1587 } while (numberOfRetryAttemptsToCheckAllMessagesOnce > 0 && set.size() < this.destinationStatistics.getMessages().getCount()); 1588 return restoredCounter; 1589 } 1590 1591 /** 1592 * @return true if we would like to iterate again 1593 * @see org.apache.activemq.thread.Task#iterate() 1594 */ 1595 @Override 1596 public boolean iterate() { 1597 MDC.put("activemq.destination", getName()); 1598 boolean pageInMoreMessages = false; 1599 synchronized (iteratingMutex) { 1600 1601 // If optimize dispatch is on or this is a slave this method could be called recursively 1602 // we set this state value to short-circuit wakeup in those cases to avoid that as it 1603 // could lead to errors. 1604 iterationRunning = true; 1605 1606 // do early to allow dispatch of these waiting messages 1607 synchronized (messagesWaitingForSpace) { 1608 Iterator<Runnable> it = messagesWaitingForSpace.values().iterator(); 1609 while (it.hasNext()) { 1610 if (!memoryUsage.isFull()) { 1611 Runnable op = it.next(); 1612 it.remove(); 1613 op.run(); 1614 } else { 1615 registerCallbackForNotFullNotification(); 1616 break; 1617 } 1618 } 1619 } 1620 1621 if (firstConsumer) { 1622 firstConsumer = false; 1623 try { 1624 if (consumersBeforeDispatchStarts > 0) { 1625 int timeout = 1000; // wait one second by default if 1626 // consumer count isn't reached 1627 if (timeBeforeDispatchStarts > 0) { 1628 timeout = timeBeforeDispatchStarts; 1629 } 1630 if (consumersBeforeStartsLatch.await(timeout, TimeUnit.MILLISECONDS)) { 1631 LOG.debug("{} consumers subscribed. Starting dispatch.", consumers.size()); 1632 } else { 1633 LOG.debug("{} ms elapsed and {} consumers subscribed. Starting dispatch.", timeout, consumers.size()); 1634 } 1635 } 1636 if (timeBeforeDispatchStarts > 0 && consumersBeforeDispatchStarts <= 0) { 1637 iteratingMutex.wait(timeBeforeDispatchStarts); 1638 LOG.debug("{} ms elapsed. Starting dispatch.", timeBeforeDispatchStarts); 1639 } 1640 } catch (Exception e) { 1641 LOG.error(e.toString()); 1642 } 1643 } 1644 1645 messagesLock.readLock().lock(); 1646 try{ 1647 pageInMoreMessages |= !messages.isEmpty(); 1648 } finally { 1649 messagesLock.readLock().unlock(); 1650 } 1651 1652 pagedInPendingDispatchLock.readLock().lock(); 1653 try { 1654 pageInMoreMessages |= !dispatchPendingList.isEmpty(); 1655 } finally { 1656 pagedInPendingDispatchLock.readLock().unlock(); 1657 } 1658 1659 boolean hasBrowsers = !browserDispatches.isEmpty(); 1660 1661 if (pageInMoreMessages || hasBrowsers || !dispatchPendingList.hasRedeliveries()) { 1662 try { 1663 pageInMessages(hasBrowsers && getMaxBrowsePageSize() > 0, getMaxPageSize()); 1664 } catch (Throwable e) { 1665 LOG.error("Failed to page in more queue messages ", e); 1666 } 1667 } 1668 1669 if (hasBrowsers) { 1670 PendingList messagesInMemory = isPrioritizedMessages() ? 1671 new PrioritizedPendingList() : new OrderedPendingList(); 1672 pagedInMessagesLock.readLock().lock(); 1673 try { 1674 messagesInMemory.addAll(pagedInMessages); 1675 } finally { 1676 pagedInMessagesLock.readLock().unlock(); 1677 } 1678 1679 Iterator<BrowserDispatch> browsers = browserDispatches.iterator(); 1680 while (browsers.hasNext()) { 1681 BrowserDispatch browserDispatch = browsers.next(); 1682 try { 1683 MessageEvaluationContext msgContext = new NonCachedMessageEvaluationContext(); 1684 msgContext.setDestination(destination); 1685 1686 QueueBrowserSubscription browser = browserDispatch.getBrowser(); 1687 1688 LOG.debug("dispatch to browser: {}, already dispatched/paged count: {}", browser, messagesInMemory.size()); 1689 boolean added = false; 1690 for (MessageReference node : messagesInMemory) { 1691 if (!((QueueMessageReference)node).isAcked() && !browser.isDuplicate(node.getMessageId()) && !browser.atMax()) { 1692 msgContext.setMessageReference(node); 1693 if (browser.matches(node, msgContext)) { 1694 browser.add(node); 1695 added = true; 1696 } 1697 } 1698 } 1699 // are we done browsing? no new messages paged 1700 if (!added || browser.atMax()) { 1701 browser.decrementQueueRef(); 1702 browserDispatches.remove(browserDispatch); 1703 } else { 1704 wakeup(); 1705 } 1706 } catch (Exception e) { 1707 LOG.warn("exception on dispatch to browser: {}", browserDispatch.getBrowser(), e); 1708 } 1709 } 1710 } 1711 1712 if (pendingWakeups.get() > 0) { 1713 pendingWakeups.decrementAndGet(); 1714 } 1715 MDC.remove("activemq.destination"); 1716 iterationRunning = false; 1717 1718 return pendingWakeups.get() > 0; 1719 } 1720 } 1721 1722 public void pauseDispatch() { 1723 dispatchSelector.pause(); 1724 } 1725 1726 public void resumeDispatch() { 1727 dispatchSelector.resume(); 1728 wakeup(); 1729 } 1730 1731 public boolean isDispatchPaused() { 1732 return dispatchSelector.isPaused(); 1733 } 1734 1735 protected MessageReferenceFilter createMessageIdFilter(final String messageId) { 1736 return new MessageReferenceFilter() { 1737 @Override 1738 public boolean evaluate(ConnectionContext context, MessageReference r) { 1739 return messageId.equals(r.getMessageId().toString()); 1740 } 1741 1742 @Override 1743 public String toString() { 1744 return "MessageIdFilter: " + messageId; 1745 } 1746 }; 1747 } 1748 1749 protected MessageReferenceFilter createSelectorFilter(String selector) throws InvalidSelectorException { 1750 1751 if (selector == null || selector.isEmpty()) { 1752 return new MessageReferenceFilter() { 1753 1754 @Override 1755 public boolean evaluate(ConnectionContext context, MessageReference messageReference) throws JMSException { 1756 return true; 1757 } 1758 }; 1759 } 1760 1761 final BooleanExpression selectorExpression = SelectorParser.parse(selector); 1762 1763 return new MessageReferenceFilter() { 1764 @Override 1765 public boolean evaluate(ConnectionContext context, MessageReference r) throws JMSException { 1766 MessageEvaluationContext messageEvaluationContext = context.getMessageEvaluationContext(); 1767 1768 messageEvaluationContext.setMessageReference(r); 1769 if (messageEvaluationContext.getDestination() == null) { 1770 messageEvaluationContext.setDestination(getActiveMQDestination()); 1771 } 1772 1773 return selectorExpression.matches(messageEvaluationContext); 1774 } 1775 }; 1776 } 1777 1778 protected void removeMessage(ConnectionContext c, QueueMessageReference r) throws IOException { 1779 removeMessage(c, null, r); 1780 pagedInPendingDispatchLock.writeLock().lock(); 1781 try { 1782 dispatchPendingList.remove(r); 1783 } finally { 1784 pagedInPendingDispatchLock.writeLock().unlock(); 1785 } 1786 } 1787 1788 protected void removeMessage(ConnectionContext c, Subscription subs, QueueMessageReference r) throws IOException { 1789 MessageAck ack = new MessageAck(); 1790 ack.setAckType(MessageAck.STANDARD_ACK_TYPE); 1791 ack.setDestination(destination); 1792 ack.setMessageID(r.getMessageId()); 1793 removeMessage(c, subs, r, ack); 1794 } 1795 1796 protected void removeMessage(ConnectionContext context, Subscription sub, final QueueMessageReference reference, 1797 MessageAck ack) throws IOException { 1798 LOG.trace("ack of {} with {}", reference.getMessageId(), ack); 1799 // This sends the ack the the journal.. 1800 if (!ack.isInTransaction()) { 1801 acknowledge(context, sub, ack, reference); 1802 dropMessage(reference); 1803 } else { 1804 try { 1805 acknowledge(context, sub, ack, reference); 1806 } finally { 1807 context.getTransaction().addSynchronization(new Synchronization() { 1808 1809 @Override 1810 public void afterCommit() throws Exception { 1811 dropMessage(reference); 1812 wakeup(); 1813 } 1814 1815 @Override 1816 public void afterRollback() throws Exception { 1817 reference.setAcked(false); 1818 wakeup(); 1819 } 1820 }); 1821 } 1822 } 1823 if (ack.isPoisonAck() || (sub != null && sub.getConsumerInfo().isNetworkSubscription())) { 1824 // message gone to DLQ, is ok to allow redelivery 1825 messagesLock.writeLock().lock(); 1826 try { 1827 messages.rollback(reference.getMessageId()); 1828 } finally { 1829 messagesLock.writeLock().unlock(); 1830 } 1831 if (sub != null && sub.getConsumerInfo().isNetworkSubscription()) { 1832 getDestinationStatistics().getForwards().increment(); 1833 } 1834 } 1835 // after successful store update 1836 reference.setAcked(true); 1837 } 1838 1839 private void dropMessage(QueueMessageReference reference) { 1840 //use dropIfLive so we only process the statistics at most one time 1841 if (reference.dropIfLive()) { 1842 getDestinationStatistics().getDequeues().increment(); 1843 getDestinationStatistics().getMessages().decrement(); 1844 pagedInMessagesLock.writeLock().lock(); 1845 try { 1846 pagedInMessages.remove(reference); 1847 } finally { 1848 pagedInMessagesLock.writeLock().unlock(); 1849 } 1850 } 1851 } 1852 1853 public void messageExpired(ConnectionContext context, MessageReference reference) { 1854 messageExpired(context, null, reference); 1855 } 1856 1857 @Override 1858 public void messageExpired(ConnectionContext context, Subscription subs, MessageReference reference) { 1859 LOG.debug("message expired: {}", reference); 1860 broker.messageExpired(context, reference, subs); 1861 destinationStatistics.getExpired().increment(); 1862 try { 1863 removeMessage(context, subs, (QueueMessageReference) reference); 1864 messagesLock.writeLock().lock(); 1865 try { 1866 messages.rollback(reference.getMessageId()); 1867 } finally { 1868 messagesLock.writeLock().unlock(); 1869 } 1870 } catch (IOException e) { 1871 LOG.error("Failed to remove expired Message from the store ", e); 1872 } 1873 } 1874 1875 private final boolean cursorAdd(final Message msg) throws Exception { 1876 messagesLock.writeLock().lock(); 1877 try { 1878 return messages.addMessageLast(msg); 1879 } finally { 1880 messagesLock.writeLock().unlock(); 1881 } 1882 } 1883 1884 private final boolean tryCursorAdd(final Message msg) throws Exception { 1885 messagesLock.writeLock().lock(); 1886 try { 1887 return messages.tryAddMessageLast(msg, 50); 1888 } finally { 1889 messagesLock.writeLock().unlock(); 1890 } 1891 } 1892 1893 final void messageSent(final ConnectionContext context, final Message msg) throws Exception { 1894 pendingSends.decrementAndGet(); 1895 destinationStatistics.getEnqueues().increment(); 1896 destinationStatistics.getMessages().increment(); 1897 destinationStatistics.getMessageSize().addSize(msg.getSize()); 1898 messageDelivered(context, msg); 1899 consumersLock.readLock().lock(); 1900 try { 1901 if (consumers.isEmpty()) { 1902 onMessageWithNoConsumers(context, msg); 1903 } 1904 }finally { 1905 consumersLock.readLock().unlock(); 1906 } 1907 LOG.debug("{} Message {} sent to {}", new Object[]{ broker.getBrokerName(), msg.getMessageId(), this.destination }); 1908 wakeup(); 1909 } 1910 1911 @Override 1912 public void wakeup() { 1913 if (optimizedDispatch && !iterationRunning) { 1914 iterate(); 1915 pendingWakeups.incrementAndGet(); 1916 } else { 1917 asyncWakeup(); 1918 } 1919 } 1920 1921 private void asyncWakeup() { 1922 try { 1923 pendingWakeups.incrementAndGet(); 1924 this.taskRunner.wakeup(); 1925 } catch (InterruptedException e) { 1926 LOG.warn("Async task runner failed to wakeup ", e); 1927 } 1928 } 1929 1930 private void doPageIn(boolean force) throws Exception { 1931 doPageIn(force, true, getMaxPageSize()); 1932 } 1933 1934 private void doPageIn(boolean force, boolean processExpired, int maxPageSize) throws Exception { 1935 PendingList newlyPaged = doPageInForDispatch(force, processExpired, maxPageSize); 1936 pagedInPendingDispatchLock.writeLock().lock(); 1937 try { 1938 if (dispatchPendingList.isEmpty()) { 1939 dispatchPendingList.addAll(newlyPaged); 1940 1941 } else { 1942 for (MessageReference qmr : newlyPaged) { 1943 if (!dispatchPendingList.contains(qmr)) { 1944 dispatchPendingList.addMessageLast(qmr); 1945 } 1946 } 1947 } 1948 } finally { 1949 pagedInPendingDispatchLock.writeLock().unlock(); 1950 } 1951 } 1952 1953 private PendingList doPageInForDispatch(boolean force, boolean processExpired, int maxPageSize) throws Exception { 1954 List<QueueMessageReference> result = null; 1955 PendingList resultList = null; 1956 1957 int toPageIn = maxPageSize; 1958 messagesLock.readLock().lock(); 1959 try { 1960 toPageIn = Math.min(toPageIn, messages.size()); 1961 } finally { 1962 messagesLock.readLock().unlock(); 1963 } 1964 int pagedInPendingSize = 0; 1965 pagedInPendingDispatchLock.readLock().lock(); 1966 try { 1967 pagedInPendingSize = dispatchPendingList.size(); 1968 } finally { 1969 pagedInPendingDispatchLock.readLock().unlock(); 1970 } 1971 if (isLazyDispatch() && !force) { 1972 // Only page in the minimum number of messages which can be 1973 // dispatched immediately. 1974 toPageIn = Math.min(toPageIn, getConsumerMessageCountBeforeFull()); 1975 } 1976 1977 if (LOG.isDebugEnabled()) { 1978 LOG.debug("{} toPageIn: {}, force:{}, Inflight: {}, pagedInMessages.size {}, pagedInPendingDispatch.size {}, enqueueCount: {}, dequeueCount: {}, memUsage:{}, maxPageSize:{}", 1979 new Object[]{ 1980 this, 1981 toPageIn, 1982 force, 1983 destinationStatistics.getInflight().getCount(), 1984 pagedInMessages.size(), 1985 pagedInPendingSize, 1986 destinationStatistics.getEnqueues().getCount(), 1987 destinationStatistics.getDequeues().getCount(), 1988 getMemoryUsage().getUsage(), 1989 maxPageSize 1990 }); 1991 } 1992 1993 if (toPageIn > 0 && (force || (haveRealConsumer() && pagedInPendingSize < maxPageSize))) { 1994 int count = 0; 1995 result = new ArrayList<QueueMessageReference>(toPageIn); 1996 messagesLock.writeLock().lock(); 1997 try { 1998 try { 1999 messages.setMaxBatchSize(toPageIn); 2000 messages.reset(); 2001 while (count < toPageIn && messages.hasNext()) { 2002 MessageReference node = messages.next(); 2003 messages.remove(); 2004 2005 QueueMessageReference ref = createMessageReference(node.getMessage()); 2006 if (processExpired && ref.isExpired()) { 2007 if (broker.isExpired(ref)) { 2008 messageExpired(createConnectionContext(), ref); 2009 } else { 2010 ref.decrementReferenceCount(); 2011 } 2012 } else { 2013 result.add(ref); 2014 count++; 2015 } 2016 } 2017 } finally { 2018 messages.release(); 2019 } 2020 } finally { 2021 messagesLock.writeLock().unlock(); 2022 } 2023 // Only add new messages, not already pagedIn to avoid multiple 2024 // dispatch attempts 2025 pagedInMessagesLock.writeLock().lock(); 2026 try { 2027 if(isPrioritizedMessages()) { 2028 resultList = new PrioritizedPendingList(); 2029 } else { 2030 resultList = new OrderedPendingList(); 2031 } 2032 for (QueueMessageReference ref : result) { 2033 if (!pagedInMessages.contains(ref)) { 2034 pagedInMessages.addMessageLast(ref); 2035 resultList.addMessageLast(ref); 2036 } else { 2037 ref.decrementReferenceCount(); 2038 // store should have trapped duplicate in it's index, or cursor audit trapped insert 2039 // or producerBrokerExchange suppressed send. 2040 // note: jdbc store will not trap unacked messages as a duplicate b/c it gives each message a unique sequence id 2041 LOG.warn("{}, duplicate message {} - {} from cursor, is cursor audit disabled or too constrained? Redirecting to dlq", this, ref.getMessageId(), ref.getMessage().getMessageId().getFutureOrSequenceLong()); 2042 if (store != null) { 2043 ConnectionContext connectionContext = createConnectionContext(); 2044 dropMessage(ref); 2045 if (gotToTheStore(ref.getMessage())) { 2046 LOG.debug("Duplicate message {} from cursor, removing from store", this, ref.getMessage()); 2047 store.removeMessage(connectionContext, new MessageAck(ref.getMessage(), MessageAck.POSION_ACK_TYPE, 1)); 2048 } 2049 broker.getRoot().sendToDeadLetterQueue(connectionContext, ref.getMessage(), null, new Throwable("duplicate paged in from cursor for " + destination)); 2050 } 2051 } 2052 } 2053 } finally { 2054 pagedInMessagesLock.writeLock().unlock(); 2055 } 2056 } else { 2057 // Avoid return null list, if condition is not validated 2058 resultList = new OrderedPendingList(); 2059 } 2060 2061 return resultList; 2062 } 2063 2064 private final boolean haveRealConsumer() { 2065 return consumers.size() - browserDispatches.size() > 0; 2066 } 2067 2068 private void doDispatch(PendingList list) throws Exception { 2069 boolean doWakeUp = false; 2070 2071 pagedInPendingDispatchLock.writeLock().lock(); 2072 try { 2073 if (isPrioritizedMessages() && !dispatchPendingList.isEmpty() && list != null && !list.isEmpty()) { 2074 // merge all to select priority order 2075 for (MessageReference qmr : list) { 2076 if (!dispatchPendingList.contains(qmr)) { 2077 dispatchPendingList.addMessageLast(qmr); 2078 } 2079 } 2080 list = null; 2081 } 2082 2083 doActualDispatch(dispatchPendingList); 2084 // and now see if we can dispatch the new stuff.. and append to the pending 2085 // list anything that does not actually get dispatched. 2086 if (list != null && !list.isEmpty()) { 2087 if (dispatchPendingList.isEmpty()) { 2088 dispatchPendingList.addAll(doActualDispatch(list)); 2089 } else { 2090 for (MessageReference qmr : list) { 2091 if (!dispatchPendingList.contains(qmr)) { 2092 dispatchPendingList.addMessageLast(qmr); 2093 } 2094 } 2095 doWakeUp = true; 2096 } 2097 } 2098 } finally { 2099 pagedInPendingDispatchLock.writeLock().unlock(); 2100 } 2101 2102 if (doWakeUp) { 2103 // avoid lock order contention 2104 asyncWakeup(); 2105 } 2106 } 2107 2108 /** 2109 * @return list of messages that could get dispatched to consumers if they 2110 * were not full. 2111 */ 2112 private PendingList doActualDispatch(PendingList list) throws Exception { 2113 List<Subscription> consumers; 2114 consumersLock.readLock().lock(); 2115 2116 try { 2117 if (this.consumers.isEmpty()) { 2118 // slave dispatch happens in processDispatchNotification 2119 return list; 2120 } 2121 consumers = new ArrayList<Subscription>(this.consumers); 2122 } finally { 2123 consumersLock.readLock().unlock(); 2124 } 2125 2126 Set<Subscription> fullConsumers = new HashSet<Subscription>(this.consumers.size()); 2127 2128 for (Iterator<MessageReference> iterator = list.iterator(); iterator.hasNext();) { 2129 2130 MessageReference node = iterator.next(); 2131 Subscription target = null; 2132 for (Subscription s : consumers) { 2133 if (s instanceof QueueBrowserSubscription) { 2134 continue; 2135 } 2136 if (!fullConsumers.contains(s)) { 2137 if (!s.isFull()) { 2138 if (dispatchSelector.canSelect(s, node) && assignMessageGroup(s, (QueueMessageReference)node) && !((QueueMessageReference) node).isAcked() ) { 2139 // Dispatch it. 2140 s.add(node); 2141 LOG.trace("assigned {} to consumer {}", node.getMessageId(), s.getConsumerInfo().getConsumerId()); 2142 iterator.remove(); 2143 target = s; 2144 break; 2145 } 2146 } else { 2147 // no further dispatch of list to a full consumer to 2148 // avoid out of order message receipt 2149 fullConsumers.add(s); 2150 LOG.trace("Subscription full {}", s); 2151 } 2152 } 2153 } 2154 2155 if (target == null && node.isDropped()) { 2156 iterator.remove(); 2157 } 2158 2159 // return if there are no consumers or all consumers are full 2160 if (target == null && consumers.size() == fullConsumers.size()) { 2161 return list; 2162 } 2163 2164 // If it got dispatched, rotate the consumer list to get round robin 2165 // distribution. 2166 if (target != null && !strictOrderDispatch && consumers.size() > 1 2167 && !dispatchSelector.isExclusiveConsumer(target)) { 2168 consumersLock.writeLock().lock(); 2169 try { 2170 if (removeFromConsumerList(target)) { 2171 addToConsumerList(target); 2172 consumers = new ArrayList<Subscription>(this.consumers); 2173 } 2174 } finally { 2175 consumersLock.writeLock().unlock(); 2176 } 2177 } 2178 } 2179 2180 return list; 2181 } 2182 2183 protected boolean assignMessageGroup(Subscription subscription, QueueMessageReference node) throws Exception { 2184 boolean result = true; 2185 // Keep message groups together. 2186 String groupId = node.getGroupID(); 2187 int sequence = node.getGroupSequence(); 2188 if (groupId != null) { 2189 2190 MessageGroupMap messageGroupOwners = getMessageGroupOwners(); 2191 // If we can own the first, then no-one else should own the 2192 // rest. 2193 if (sequence == 1) { 2194 assignGroup(subscription, messageGroupOwners, node, groupId); 2195 } else { 2196 2197 // Make sure that the previous owner is still valid, we may 2198 // need to become the new owner. 2199 ConsumerId groupOwner; 2200 2201 groupOwner = messageGroupOwners.get(groupId); 2202 if (groupOwner == null) { 2203 assignGroup(subscription, messageGroupOwners, node, groupId); 2204 } else { 2205 if (groupOwner.equals(subscription.getConsumerInfo().getConsumerId())) { 2206 // A group sequence < 1 is an end of group signal. 2207 if (sequence < 0) { 2208 messageGroupOwners.removeGroup(groupId); 2209 subscription.getConsumerInfo().decrementAssignedGroupCount(destination); 2210 } 2211 } else { 2212 result = false; 2213 } 2214 } 2215 } 2216 } 2217 2218 return result; 2219 } 2220 2221 protected void assignGroup(Subscription subs, MessageGroupMap messageGroupOwners, MessageReference n, String groupId) throws IOException { 2222 messageGroupOwners.put(groupId, subs.getConsumerInfo().getConsumerId()); 2223 Message message = n.getMessage(); 2224 message.setJMSXGroupFirstForConsumer(true); 2225 subs.getConsumerInfo().incrementAssignedGroupCount(destination); 2226 } 2227 2228 protected void pageInMessages(boolean force, int maxPageSize) throws Exception { 2229 doDispatch(doPageInForDispatch(force, true, maxPageSize)); 2230 } 2231 2232 private void addToConsumerList(Subscription sub) { 2233 if (useConsumerPriority) { 2234 consumers.add(sub); 2235 Collections.sort(consumers, orderedCompare); 2236 } else { 2237 consumers.add(sub); 2238 } 2239 } 2240 2241 private boolean removeFromConsumerList(Subscription sub) { 2242 return consumers.remove(sub); 2243 } 2244 2245 private int getConsumerMessageCountBeforeFull() throws Exception { 2246 int total = 0; 2247 consumersLock.readLock().lock(); 2248 try { 2249 for (Subscription s : consumers) { 2250 if (s.isBrowser()) { 2251 continue; 2252 } 2253 int countBeforeFull = s.countBeforeFull(); 2254 total += countBeforeFull; 2255 } 2256 } finally { 2257 consumersLock.readLock().unlock(); 2258 } 2259 return total; 2260 } 2261 2262 /* 2263 * In slave mode, dispatch is ignored till we get this notification as the 2264 * dispatch process is non deterministic between master and slave. On a 2265 * notification, the actual dispatch to the subscription (as chosen by the 2266 * master) is completed. (non-Javadoc) 2267 * @see 2268 * org.apache.activemq.broker.region.BaseDestination#processDispatchNotification 2269 * (org.apache.activemq.command.MessageDispatchNotification) 2270 */ 2271 @Override 2272 public void processDispatchNotification(MessageDispatchNotification messageDispatchNotification) throws Exception { 2273 // do dispatch 2274 Subscription sub = getMatchingSubscription(messageDispatchNotification); 2275 if (sub != null) { 2276 MessageReference message = getMatchingMessage(messageDispatchNotification); 2277 sub.add(message); 2278 sub.processMessageDispatchNotification(messageDispatchNotification); 2279 } 2280 } 2281 2282 private QueueMessageReference getMatchingMessage(MessageDispatchNotification messageDispatchNotification) 2283 throws Exception { 2284 QueueMessageReference message = null; 2285 MessageId messageId = messageDispatchNotification.getMessageId(); 2286 2287 pagedInPendingDispatchLock.writeLock().lock(); 2288 try { 2289 for (MessageReference ref : dispatchPendingList) { 2290 if (messageId.equals(ref.getMessageId())) { 2291 message = (QueueMessageReference)ref; 2292 dispatchPendingList.remove(ref); 2293 break; 2294 } 2295 } 2296 } finally { 2297 pagedInPendingDispatchLock.writeLock().unlock(); 2298 } 2299 2300 if (message == null) { 2301 pagedInMessagesLock.readLock().lock(); 2302 try { 2303 message = (QueueMessageReference)pagedInMessages.get(messageId); 2304 } finally { 2305 pagedInMessagesLock.readLock().unlock(); 2306 } 2307 } 2308 2309 if (message == null) { 2310 messagesLock.writeLock().lock(); 2311 try { 2312 try { 2313 messages.setMaxBatchSize(getMaxPageSize()); 2314 messages.reset(); 2315 while (messages.hasNext()) { 2316 MessageReference node = messages.next(); 2317 messages.remove(); 2318 if (messageId.equals(node.getMessageId())) { 2319 message = this.createMessageReference(node.getMessage()); 2320 break; 2321 } 2322 } 2323 } finally { 2324 messages.release(); 2325 } 2326 } finally { 2327 messagesLock.writeLock().unlock(); 2328 } 2329 } 2330 2331 if (message == null) { 2332 Message msg = loadMessage(messageId); 2333 if (msg != null) { 2334 message = this.createMessageReference(msg); 2335 } 2336 } 2337 2338 if (message == null) { 2339 throw new JMSException("Slave broker out of sync with master - Message: " 2340 + messageDispatchNotification.getMessageId() + " on " 2341 + messageDispatchNotification.getDestination() + " does not exist among pending(" 2342 + dispatchPendingList.size() + ") for subscription: " 2343 + messageDispatchNotification.getConsumerId()); 2344 } 2345 return message; 2346 } 2347 2348 /** 2349 * Find a consumer that matches the id in the message dispatch notification 2350 * 2351 * @param messageDispatchNotification 2352 * @return sub or null if the subscription has been removed before dispatch 2353 * @throws JMSException 2354 */ 2355 private Subscription getMatchingSubscription(MessageDispatchNotification messageDispatchNotification) 2356 throws JMSException { 2357 Subscription sub = null; 2358 consumersLock.readLock().lock(); 2359 try { 2360 for (Subscription s : consumers) { 2361 if (messageDispatchNotification.getConsumerId().equals(s.getConsumerInfo().getConsumerId())) { 2362 sub = s; 2363 break; 2364 } 2365 } 2366 } finally { 2367 consumersLock.readLock().unlock(); 2368 } 2369 return sub; 2370 } 2371 2372 @Override 2373 public void onUsageChanged(@SuppressWarnings("rawtypes") Usage usage, int oldPercentUsage, int newPercentUsage) { 2374 if (oldPercentUsage > newPercentUsage) { 2375 asyncWakeup(); 2376 } 2377 } 2378 2379 @Override 2380 protected Logger getLog() { 2381 return LOG; 2382 } 2383 2384 protected boolean isOptimizeStorage(){ 2385 boolean result = false; 2386 if (isDoOptimzeMessageStorage()){ 2387 consumersLock.readLock().lock(); 2388 try{ 2389 if (consumers.isEmpty()==false){ 2390 result = true; 2391 for (Subscription s : consumers) { 2392 if (s.getPrefetchSize()==0){ 2393 result = false; 2394 break; 2395 } 2396 if (s.isSlowConsumer()){ 2397 result = false; 2398 break; 2399 } 2400 if (s.getInFlightUsage() > getOptimizeMessageStoreInFlightLimit()){ 2401 result = false; 2402 break; 2403 } 2404 } 2405 } 2406 } finally { 2407 consumersLock.readLock().unlock(); 2408 } 2409 } 2410 return result; 2411 } 2412}