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 */ 017 package org.apache.activemq.broker.region; 018 019 import java.io.IOException; 020 import java.util.AbstractList; 021 import java.util.ArrayList; 022 import java.util.Collection; 023 import java.util.Collections; 024 import java.util.Comparator; 025 import java.util.HashSet; 026 import java.util.Iterator; 027 import java.util.LinkedHashMap; 028 import java.util.LinkedList; 029 import java.util.List; 030 import java.util.Map; 031 import java.util.Set; 032 import java.util.concurrent.CopyOnWriteArraySet; 033 import java.util.concurrent.CountDownLatch; 034 import java.util.concurrent.DelayQueue; 035 import java.util.concurrent.Delayed; 036 import java.util.concurrent.ExecutorService; 037 import java.util.concurrent.TimeUnit; 038 import java.util.concurrent.atomic.AtomicLong; 039 040 import javax.jms.InvalidSelectorException; 041 import javax.jms.JMSException; 042 import javax.jms.ResourceAllocationException; 043 044 import org.apache.activemq.broker.BrokerService; 045 import org.apache.activemq.broker.ConnectionContext; 046 import org.apache.activemq.broker.ProducerBrokerExchange; 047 import org.apache.activemq.broker.region.cursors.PendingMessageCursor; 048 import org.apache.activemq.broker.region.cursors.StoreQueueCursor; 049 import org.apache.activemq.broker.region.cursors.VMPendingMessageCursor; 050 import org.apache.activemq.broker.region.group.MessageGroupHashBucketFactory; 051 import org.apache.activemq.broker.region.group.MessageGroupMap; 052 import org.apache.activemq.broker.region.group.MessageGroupMapFactory; 053 import org.apache.activemq.broker.region.policy.DispatchPolicy; 054 import org.apache.activemq.broker.region.policy.RoundRobinDispatchPolicy; 055 import org.apache.activemq.command.ActiveMQDestination; 056 import org.apache.activemq.command.ConsumerId; 057 import org.apache.activemq.command.ExceptionResponse; 058 import org.apache.activemq.command.Message; 059 import org.apache.activemq.command.MessageAck; 060 import org.apache.activemq.command.MessageDispatchNotification; 061 import org.apache.activemq.command.MessageId; 062 import org.apache.activemq.command.ProducerAck; 063 import org.apache.activemq.command.ProducerInfo; 064 import org.apache.activemq.command.Response; 065 import org.apache.activemq.filter.BooleanExpression; 066 import org.apache.activemq.filter.MessageEvaluationContext; 067 import org.apache.activemq.filter.NonCachedMessageEvaluationContext; 068 import org.apache.activemq.security.SecurityContext; 069 import org.apache.activemq.selector.SelectorParser; 070 import org.apache.activemq.store.MessageRecoveryListener; 071 import org.apache.activemq.store.MessageStore; 072 import org.apache.activemq.thread.Scheduler; 073 import org.apache.activemq.thread.Task; 074 import org.apache.activemq.thread.TaskRunner; 075 import org.apache.activemq.thread.TaskRunnerFactory; 076 import org.apache.activemq.transaction.Synchronization; 077 import org.apache.activemq.usage.Usage; 078 import org.apache.activemq.usage.UsageListener; 079 import org.apache.activemq.util.BrokerSupport; 080 import org.apache.commons.logging.Log; 081 import org.apache.commons.logging.LogFactory; 082 083 /** 084 * The Queue is a List of MessageEntry objects that are dispatched to matching 085 * subscriptions. 086 * 087 * @version $Revision: 1.28 $ 088 */ 089 public class Queue extends BaseDestination implements Task, UsageListener { 090 protected static final Log LOG = LogFactory.getLog(Queue.class); 091 protected final TaskRunnerFactory taskFactory; 092 protected TaskRunner taskRunner; 093 protected final List<Subscription> consumers = new ArrayList<Subscription>(50); 094 protected PendingMessageCursor messages; 095 private final LinkedHashMap<MessageId, QueueMessageReference> pagedInMessages = new LinkedHashMap<MessageId, QueueMessageReference>(); 096 // Messages that are paged in but have not yet been targeted at a subscription 097 private List<QueueMessageReference> pagedInPendingDispatch = new ArrayList<QueueMessageReference>(100); 098 private MessageGroupMap messageGroupOwners; 099 private DispatchPolicy dispatchPolicy = new RoundRobinDispatchPolicy(); 100 private MessageGroupMapFactory messageGroupMapFactory = new MessageGroupHashBucketFactory(); 101 private final Object sendLock = new Object(); 102 private ExecutorService executor; 103 protected final Map<MessageId, Runnable> messagesWaitingForSpace = Collections.synchronizedMap(new LinkedHashMap<MessageId, Runnable>()); 104 private final Object dispatchMutex = new Object(); 105 private boolean useConsumerPriority = true; 106 private boolean strictOrderDispatch = false; 107 private QueueDispatchSelector dispatchSelector; 108 private boolean optimizedDispatch = false; 109 private boolean firstConsumer = false; 110 private int timeBeforeDispatchStarts = 0; 111 private int consumersBeforeDispatchStarts = 0; 112 private CountDownLatch consumersBeforeStartsLatch; 113 private AtomicLong pendingWakeups = new AtomicLong(); 114 115 private final Runnable sendMessagesWaitingForSpaceTask = new Runnable() { 116 public void run() { 117 asyncWakeup(); 118 } 119 }; 120 private final Runnable expireMessagesTask = new Runnable() { 121 public void run() { 122 expireMessages(); 123 } 124 }; 125 126 private final Object iteratingMutex = new Object() {}; 127 private static final Scheduler scheduler = Scheduler.getInstance(); 128 129 class TimeoutMessage implements Delayed { 130 131 Message message; 132 ConnectionContext context; 133 long trigger; 134 135 public TimeoutMessage(Message message, ConnectionContext context, long delay) { 136 this.message = message; 137 this.context = context; 138 this.trigger = System.currentTimeMillis() + delay; 139 } 140 141 public long getDelay(TimeUnit unit) { 142 long n = trigger - System.currentTimeMillis(); 143 return unit.convert(n, TimeUnit.MILLISECONDS); 144 } 145 146 public int compareTo(Delayed delayed) { 147 long other = ((TimeoutMessage)delayed).trigger; 148 int returnValue; 149 if (this.trigger < other) { 150 returnValue = -1; 151 } else if (this.trigger > other) { 152 returnValue = 1; 153 } else { 154 returnValue = 0; 155 } 156 return returnValue; 157 } 158 159 } 160 161 DelayQueue<TimeoutMessage> flowControlTimeoutMessages = new DelayQueue<TimeoutMessage>(); 162 163 class FlowControlTimeoutTask extends Thread { 164 165 public void run() { 166 TimeoutMessage timeout; 167 try { 168 while (true) { 169 timeout = flowControlTimeoutMessages.take(); 170 if (timeout != null) { 171 synchronized (messagesWaitingForSpace) { 172 if (messagesWaitingForSpace.remove(timeout.message.getMessageId()) != null) { 173 ExceptionResponse response = new ExceptionResponse(new ResourceAllocationException("Usage Manager Memory Limit reached. Stopping producer (" + timeout.message.getProducerId() + ") to prevent flooding " 174 + getActiveMQDestination().getQualifiedName() + "." + " See http://activemq.apache.org/producer-flow-control.html for more info")); 175 response.setCorrelationId(timeout.message.getCommandId()); 176 timeout.context.getConnection().dispatchAsync(response); 177 } 178 } 179 } 180 } 181 } catch (InterruptedException e) { 182 if (LOG.isDebugEnabled()) { 183 LOG.debug("Producer Flow Control Timeout Task is stopping"); 184 } 185 } 186 } 187 }; 188 189 private final FlowControlTimeoutTask flowControlTimeoutTask = new FlowControlTimeoutTask(); 190 191 192 private static final Comparator<Subscription> orderedCompare = new Comparator<Subscription>() { 193 194 public int compare(Subscription s1, Subscription s2) { 195 //We want the list sorted in descending order 196 return s2.getConsumerInfo().getPriority() - s1.getConsumerInfo().getPriority(); 197 } 198 }; 199 200 public Queue(BrokerService brokerService, final ActiveMQDestination destination, MessageStore store, DestinationStatistics parentStats, TaskRunnerFactory taskFactory) throws Exception { 201 super(brokerService, store, destination, parentStats); 202 this.taskFactory = taskFactory; 203 this.dispatchSelector = new QueueDispatchSelector(destination); 204 } 205 206 public List<Subscription> getConsumers() { 207 synchronized (consumers) { 208 return new ArrayList<Subscription>(consumers); 209 } 210 } 211 212 // make the queue easily visible in the debugger from its task runner threads 213 final class QueueThread extends Thread { 214 final Queue queue; 215 216 public QueueThread(Runnable runnable, String name, Queue queue) { 217 super(runnable, name); 218 this.queue = queue; 219 } 220 } 221 222 public void initialize() throws Exception { 223 if (this.messages == null) { 224 if (destination.isTemporary() || broker == null || store == null) { 225 this.messages = new VMPendingMessageCursor(); 226 } else { 227 this.messages = new StoreQueueCursor(broker, this); 228 } 229 } 230 // If a VMPendingMessageCursor don't use the default Producer System Usage 231 // since it turns into a shared blocking queue which can lead to a network deadlock. 232 // If we are cursoring to disk..it's not and issue because it does not block due 233 // to large disk sizes. 234 if (messages instanceof VMPendingMessageCursor) { 235 this.systemUsage = brokerService.getSystemUsage(); 236 memoryUsage.setParent(systemUsage.getMemoryUsage()); 237 } 238 239 this.taskRunner = taskFactory.createTaskRunner(this, "Queue:" + destination.getPhysicalName()); 240 241 super.initialize(); 242 if (store != null) { 243 // Restore the persistent messages. 244 messages.setSystemUsage(systemUsage); 245 messages.setEnableAudit(isEnableAudit()); 246 messages.setMaxAuditDepth(getMaxAuditDepth()); 247 messages.setMaxProducersToAudit(getMaxProducersToAudit()); 248 messages.setUseCache(isUseCache()); 249 messages.setMemoryUsageHighWaterMark(getCursorMemoryHighWaterMark()); 250 if (messages.isRecoveryRequired()) { 251 store.recover(new MessageRecoveryListener() { 252 253 public boolean recoverMessage(Message message) { 254 // Message could have expired while it was being 255 // loaded.. 256 if (message.isExpired()) { 257 if (broker.isExpired(message)) { 258 messageExpired(createConnectionContext(), createMessageReference(message)); 259 // drop message will decrement so counter balance here 260 destinationStatistics.getMessages().increment(); 261 } 262 return true; 263 } 264 if (hasSpace()) { 265 message.setRegionDestination(Queue.this); 266 synchronized (messages) { 267 try { 268 messages.addMessageLast(message); 269 } catch (Exception e) { 270 LOG.fatal("Failed to add message to cursor", e); 271 } 272 } 273 destinationStatistics.getMessages().increment(); 274 return true; 275 } 276 return false; 277 } 278 279 public boolean recoverMessageReference(MessageId messageReference) throws Exception { 280 throw new RuntimeException("Should not be called."); 281 } 282 283 public boolean hasSpace() { 284 return true; 285 } 286 287 public boolean isDuplicate(MessageId id) { 288 return false; 289 } 290 }); 291 } else { 292 int messageCount = store.getMessageCount(); 293 destinationStatistics.getMessages().setCount(messageCount); 294 } 295 } 296 } 297 298 /* 299 * Holder for subscription that needs attention on next iterate 300 * browser needs access to existing messages in the queue that have already been dispatched 301 */ 302 class BrowserDispatch { 303 QueueBrowserSubscription browser; 304 305 public BrowserDispatch(QueueBrowserSubscription browserSubscription) { 306 browser = browserSubscription; 307 browser.incrementQueueRef(); 308 } 309 310 void done() { 311 try { 312 browser.decrementQueueRef(); 313 } catch (Exception e) { 314 LOG.warn("decrement ref on browser: " + browser, e); 315 } 316 } 317 318 public QueueBrowserSubscription getBrowser() { 319 return browser; 320 } 321 } 322 323 LinkedList<BrowserDispatch> browserDispatches = new LinkedList<BrowserDispatch>(); 324 325 public void addSubscription(ConnectionContext context, Subscription sub) throws Exception { 326 // synchronize with dispatch method so that no new messages are sent 327 // while setting up a subscription. avoid out of order messages, 328 // duplicates, etc. 329 synchronized (dispatchMutex) { 330 331 sub.add(context, this); 332 destinationStatistics.getConsumers().increment(); 333 334 // needs to be synchronized - so no contention with dispatching 335 synchronized (consumers) { 336 337 // set a flag if this is a first consumer 338 if (consumers.size() == 0) { 339 firstConsumer = true; 340 if (consumersBeforeDispatchStarts != 0) { 341 consumersBeforeStartsLatch = new CountDownLatch(consumersBeforeDispatchStarts - 1); 342 } 343 } else { 344 if (consumersBeforeStartsLatch != null) { 345 consumersBeforeStartsLatch.countDown(); 346 } 347 } 348 349 addToConsumerList(sub); 350 if (sub.getConsumerInfo().isExclusive()) { 351 Subscription exclusiveConsumer = dispatchSelector.getExclusiveConsumer(); 352 if (exclusiveConsumer == null) { 353 exclusiveConsumer = sub; 354 } else if (sub.getConsumerInfo().getPriority() > exclusiveConsumer.getConsumerInfo().getPriority()) { 355 exclusiveConsumer = sub; 356 } 357 dispatchSelector.setExclusiveConsumer(exclusiveConsumer); 358 } 359 } 360 361 if (sub instanceof QueueBrowserSubscription) { 362 // tee up for dispatch in next iterate 363 QueueBrowserSubscription browserSubscription = (QueueBrowserSubscription) sub; 364 synchronized (pagedInMessages) { 365 BrowserDispatch browserDispatch = new BrowserDispatch(browserSubscription); 366 browserDispatches.addLast(browserDispatch); 367 } 368 } 369 370 if (!(this.optimizedDispatch || isSlave())) { 371 wakeup(); 372 } 373 } 374 if (this.optimizedDispatch || isSlave()) { 375 // Outside of dispatchLock() to maintain the lock hierarchy of 376 // iteratingMutex -> dispatchLock. - see https://issues.apache.org/activemq/browse/AMQ-1878 377 wakeup(); 378 } 379 } 380 381 public void removeSubscription(ConnectionContext context, Subscription sub, long lastDeiveredSequenceId) throws Exception { 382 destinationStatistics.getConsumers().decrement(); 383 // synchronize with dispatch method so that no new messages are sent 384 // while removing up a subscription. 385 synchronized (dispatchMutex) { 386 if (LOG.isDebugEnabled()) { 387 LOG.debug("remove sub: " + sub + ", lastDeliveredSeqId: " + lastDeiveredSequenceId + ", dequeues: " + getDestinationStatistics().getDequeues().getCount() + ", dispatched: " 388 + getDestinationStatistics().getDispatched().getCount() + ", inflight: " + getDestinationStatistics().getInflight().getCount()); 389 } 390 synchronized (consumers) { 391 removeFromConsumerList(sub); 392 if (sub.getConsumerInfo().isExclusive()) { 393 Subscription exclusiveConsumer = dispatchSelector.getExclusiveConsumer(); 394 if (exclusiveConsumer == sub) { 395 exclusiveConsumer = null; 396 for (Subscription s : consumers) { 397 if (s.getConsumerInfo().isExclusive() && (exclusiveConsumer == null || s.getConsumerInfo().getPriority() > exclusiveConsumer.getConsumerInfo().getPriority())) { 398 exclusiveConsumer = s; 399 400 } 401 } 402 dispatchSelector.setExclusiveConsumer(exclusiveConsumer); 403 } 404 } 405 ConsumerId consumerId = sub.getConsumerInfo().getConsumerId(); 406 getMessageGroupOwners().removeConsumer(consumerId); 407 408 // redeliver inflight messages 409 List<QueueMessageReference> list = new ArrayList<QueueMessageReference>(); 410 for (MessageReference ref : sub.remove(context, this)) { 411 QueueMessageReference qmr = (QueueMessageReference) ref; 412 if (qmr.getLockOwner() == sub) { 413 qmr.unlock(); 414 // only increment redelivery if it was delivered or we have no delivery information 415 if (lastDeiveredSequenceId == 0 || qmr.getMessageId().getBrokerSequenceId() <= lastDeiveredSequenceId) { 416 qmr.incrementRedeliveryCounter(); 417 } 418 } 419 list.add(qmr); 420 } 421 422 if (!list.isEmpty()) { 423 doDispatch(list); 424 } 425 } 426 if (!(this.optimizedDispatch || isSlave())) { 427 wakeup(); 428 } 429 } 430 if (this.optimizedDispatch || isSlave()) { 431 // Outside of dispatchLock() to maintain the lock hierarchy of 432 // iteratingMutex -> dispatchLock. - see https://issues.apache.org/activemq/browse/AMQ-1878 433 wakeup(); 434 } 435 } 436 437 public void send(final ProducerBrokerExchange producerExchange, final Message message) throws Exception { 438 final ConnectionContext context = producerExchange.getConnectionContext(); 439 // There is delay between the client sending it and it arriving at the 440 // destination.. it may have expired. 441 message.setRegionDestination(this); 442 final ProducerInfo producerInfo = producerExchange.getProducerState().getInfo(); 443 final boolean sendProducerAck = !message.isResponseRequired() && producerInfo.getWindowSize() > 0 && !context.isInRecoveryMode(); 444 if (message.isExpired()) { 445 //message not stored - or added to stats yet - so chuck here 446 broker.getRoot().messageExpired(context, message); 447 if (sendProducerAck) { 448 ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize()); 449 context.getConnection().dispatchAsync(ack); 450 } 451 return; 452 } 453 if (memoryUsage.isFull()) { 454 isFull(context, memoryUsage); 455 fastProducer(context, producerInfo); 456 if (isProducerFlowControl() && context.isProducerFlowControl()) { 457 if (warnOnProducerFlowControl) { 458 warnOnProducerFlowControl = false; 459 LOG.info("Usage Manager Memory Limit reached on " + getActiveMQDestination().getQualifiedName() 460 + ". Producers will be throttled to the rate at which messages are removed from this destination to prevent flooding it." 461 + " See http://activemq.apache.org/producer-flow-control.html for more info"); 462 } 463 464 if (systemUsage.isSendFailIfNoSpace()) { 465 throw new ResourceAllocationException("Usage Manager Memory Limit reached. Stopping producer (" + message.getProducerId() + ") to prevent flooding " 466 + getActiveMQDestination().getQualifiedName() + "." + " See http://activemq.apache.org/producer-flow-control.html for more info"); 467 } 468 469 // We can avoid blocking due to low usage if the producer is sending 470 // a sync message or if it is using a producer window 471 if (producerInfo.getWindowSize() > 0 || message.isResponseRequired()) { 472 // copy the exchange state since the context will be modified while we are waiting 473 // for space. 474 final ProducerBrokerExchange producerExchangeCopy = producerExchange.copy(); 475 synchronized (messagesWaitingForSpace) { 476 messagesWaitingForSpace.put(message.getMessageId(), new Runnable() { 477 public void run() { 478 479 try { 480 // While waiting for space to free up... the 481 // message may have expired. 482 if (message.isExpired()) { 483 LOG.error("expired waiting for space.."); 484 broker.messageExpired(context, message); 485 destinationStatistics.getExpired().increment(); 486 } else { 487 doMessageSend(producerExchangeCopy, message); 488 } 489 490 if (sendProducerAck) { 491 ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize()); 492 context.getConnection().dispatchAsync(ack); 493 } else { 494 Response response = new Response(); 495 response.setCorrelationId(message.getCommandId()); 496 context.getConnection().dispatchAsync(response); 497 } 498 499 } catch (Exception e) { 500 if (!sendProducerAck && !context.isInRecoveryMode()) { 501 ExceptionResponse response = new ExceptionResponse(e); 502 response.setCorrelationId(message.getCommandId()); 503 context.getConnection().dispatchAsync(response); 504 } else { 505 LOG.debug("unexpected exception on deferred send of :" + message, e); 506 } 507 } 508 } 509 }); 510 511 if (systemUsage.getSendFailIfNoSpaceAfterTimeout() != 0) { 512 flowControlTimeoutMessages.add(new TimeoutMessage(message, context, systemUsage.getSendFailIfNoSpaceAfterTimeout())); 513 } 514 515 registerCallbackForNotFullNotification(); 516 context.setDontSendReponse(true); 517 return; 518 } 519 520 } else { 521 522 if (memoryUsage.isFull()) { 523 waitForSpace(context, memoryUsage, "Usage Manager Memory Limit is full. Producer (" + message.getProducerId() + ") stopped to prevent flooding " 524 + getActiveMQDestination().getQualifiedName() + "." + " See http://activemq.apache.org/producer-flow-control.html for more info"); 525 } 526 527 // The usage manager could have delayed us by the time 528 // we unblock the message could have expired.. 529 if (message.isExpired()) { 530 if (LOG.isDebugEnabled()) { 531 LOG.debug("Expired message: " + message); 532 } 533 broker.getRoot().messageExpired(context, message); 534 return; 535 } 536 } 537 } 538 } 539 doMessageSend(producerExchange, message); 540 if (sendProducerAck) { 541 ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize()); 542 context.getConnection().dispatchAsync(ack); 543 } 544 } 545 546 private void registerCallbackForNotFullNotification() { 547 // If the usage manager is not full, then the task will not 548 // get called.. 549 if (!memoryUsage.notifyCallbackWhenNotFull(sendMessagesWaitingForSpaceTask)) { 550 // so call it directly here. 551 sendMessagesWaitingForSpaceTask.run(); 552 } 553 } 554 555 void doMessageSend(final ProducerBrokerExchange producerExchange, final Message message) throws IOException, Exception { 556 final ConnectionContext context = producerExchange.getConnectionContext(); 557 synchronized (sendLock) { 558 if (store != null && message.isPersistent()) { 559 if (systemUsage.getStoreUsage().isFull()) { 560 561 String logMessage = "Usage Manager Store is Full. Producer (" + message.getProducerId() + ") stopped to prevent flooding " + getActiveMQDestination().getQualifiedName() + "." 562 + " See http://activemq.apache.org/producer-flow-control.html for more info"; 563 564 if (systemUsage.isSendFailIfNoSpace()) { 565 throw new ResourceAllocationException(logMessage); 566 } 567 568 waitForSpace(context, systemUsage.getStoreUsage(), logMessage); 569 } 570 message.getMessageId().setBrokerSequenceId(getDestinationSequenceId()); 571 store.addMessage(context, message); 572 573 } 574 } 575 if (context.isInTransaction()) { 576 // If this is a transacted message.. increase the usage now so that 577 // a big TX does not blow up 578 // our memory. This increment is decremented once the tx finishes.. 579 message.incrementReferenceCount(); 580 context.getTransaction().addSynchronization(new Synchronization() { 581 public void afterCommit() throws Exception { 582 try { 583 // It could take while before we receive the commit 584 // op, by that time the message could have expired.. 585 if (broker.isExpired(message)) { 586 broker.messageExpired(context, message); 587 destinationStatistics.getExpired().increment(); 588 return; 589 } 590 sendMessage(context, message); 591 } finally { 592 message.decrementReferenceCount(); 593 } 594 } 595 596 @Override 597 public void afterRollback() throws Exception { 598 message.decrementReferenceCount(); 599 } 600 }); 601 } else { 602 // Add to the pending list, this takes care of incrementing the 603 // usage manager. 604 sendMessage(context, message); 605 } 606 } 607 608 private void expireMessages() { 609 if (LOG.isDebugEnabled()) { 610 LOG.debug("Expiring messages .."); 611 } 612 613 // just track the insertion count 614 List<Message> browsedMessages = new AbstractList<Message>() { 615 int size = 0; 616 617 @Override 618 public void add(int index, Message element) { 619 size++; 620 } 621 622 @Override 623 public int size() { 624 return size; 625 } 626 627 @Override 628 public Message get(int index) { 629 return null; 630 } 631 }; 632 doBrowse(browsedMessages, this.getMaxExpirePageSize()); 633 asyncWakeup(); 634 } 635 636 public void gc() { 637 } 638 639 public void acknowledge(ConnectionContext context, Subscription sub, MessageAck ack, MessageReference node) throws IOException { 640 messageConsumed(context, node); 641 if (store != null && node.isPersistent()) { 642 // the original ack may be a ranged ack, but we are trying to delete a specific 643 // message store here so we need to convert to a non ranged ack. 644 if (ack.getMessageCount() > 0) { 645 // Dup the ack 646 MessageAck a = new MessageAck(); 647 ack.copy(a); 648 ack = a; 649 // Convert to non-ranged. 650 ack.setFirstMessageId(node.getMessageId()); 651 ack.setLastMessageId(node.getMessageId()); 652 ack.setMessageCount(1); 653 } 654 store.removeMessage(context, ack); 655 } 656 } 657 658 Message loadMessage(MessageId messageId) throws IOException { 659 Message msg = store.getMessage(messageId); 660 if (msg != null) { 661 msg.setRegionDestination(this); 662 } 663 return msg; 664 } 665 666 public String toString() { 667 int size = 0; 668 synchronized (messages) { 669 size = messages.size(); 670 } 671 return "Queue: destination=" + destination.getPhysicalName() + ", subscriptions=" + consumers.size() + ", memory=" + memoryUsage.getPercentUsage() + "%, size=" + size + ", in flight groups=" 672 + messageGroupOwners; 673 } 674 675 public void start() throws Exception { 676 if (memoryUsage != null) { 677 memoryUsage.start(); 678 } 679 systemUsage.getMemoryUsage().addUsageListener(this); 680 messages.start(); 681 if (getExpireMessagesPeriod() > 0) { 682 scheduler.schedualPeriodically(expireMessagesTask, getExpireMessagesPeriod()); 683 } 684 685 flowControlTimeoutTask.setName("Producer Flow Control Timeout Task"); 686 687 // Start flow control timeout task 688 // Prevent trying to start it multiple times 689 if (!flowControlTimeoutTask.isAlive()) { 690 flowControlTimeoutTask.start(); 691 } 692 693 doPageIn(false); 694 } 695 696 public void stop() throws Exception { 697 if (taskRunner != null) { 698 taskRunner.shutdown(); 699 } 700 if (this.executor != null) { 701 this.executor.shutdownNow(); 702 } 703 704 scheduler.cancel(expireMessagesTask); 705 706 if (flowControlTimeoutTask.isAlive()) { 707 flowControlTimeoutTask.interrupt(); 708 } 709 710 if (messages != null) { 711 messages.stop(); 712 } 713 714 systemUsage.getMemoryUsage().removeUsageListener(this); 715 if (memoryUsage != null) { 716 memoryUsage.stop(); 717 } 718 if (store != null) { 719 store.stop(); 720 } 721 } 722 723 // Properties 724 // ------------------------------------------------------------------------- 725 public ActiveMQDestination getActiveMQDestination() { 726 return destination; 727 } 728 729 public MessageGroupMap getMessageGroupOwners() { 730 if (messageGroupOwners == null) { 731 messageGroupOwners = getMessageGroupMapFactory().createMessageGroupMap(); 732 } 733 return messageGroupOwners; 734 } 735 736 public DispatchPolicy getDispatchPolicy() { 737 return dispatchPolicy; 738 } 739 740 public void setDispatchPolicy(DispatchPolicy dispatchPolicy) { 741 this.dispatchPolicy = dispatchPolicy; 742 } 743 744 public MessageGroupMapFactory getMessageGroupMapFactory() { 745 return messageGroupMapFactory; 746 } 747 748 public void setMessageGroupMapFactory(MessageGroupMapFactory messageGroupMapFactory) { 749 this.messageGroupMapFactory = messageGroupMapFactory; 750 } 751 752 public PendingMessageCursor getMessages() { 753 return this.messages; 754 } 755 756 public void setMessages(PendingMessageCursor messages) { 757 this.messages = messages; 758 } 759 760 public boolean isUseConsumerPriority() { 761 return useConsumerPriority; 762 } 763 764 public void setUseConsumerPriority(boolean useConsumerPriority) { 765 this.useConsumerPriority = useConsumerPriority; 766 } 767 768 public boolean isStrictOrderDispatch() { 769 return strictOrderDispatch; 770 } 771 772 public void setStrictOrderDispatch(boolean strictOrderDispatch) { 773 this.strictOrderDispatch = strictOrderDispatch; 774 } 775 776 public boolean isOptimizedDispatch() { 777 return optimizedDispatch; 778 } 779 780 public void setOptimizedDispatch(boolean optimizedDispatch) { 781 this.optimizedDispatch = optimizedDispatch; 782 } 783 784 public int getTimeBeforeDispatchStarts() { 785 return timeBeforeDispatchStarts; 786 } 787 788 public void setTimeBeforeDispatchStarts(int timeBeforeDispatchStarts) { 789 this.timeBeforeDispatchStarts = timeBeforeDispatchStarts; 790 } 791 792 public int getConsumersBeforeDispatchStarts() { 793 return consumersBeforeDispatchStarts; 794 } 795 796 public void setConsumersBeforeDispatchStarts(int consumersBeforeDispatchStarts) { 797 this.consumersBeforeDispatchStarts = consumersBeforeDispatchStarts; 798 } 799 800 // Implementation methods 801 // ------------------------------------------------------------------------- 802 private QueueMessageReference createMessageReference(Message message) { 803 QueueMessageReference result = new IndirectMessageReference(message); 804 return result; 805 } 806 807 public Message[] browse() { 808 List<Message> browseList = new ArrayList<Message>(); 809 doBrowse(browseList, getMaxBrowsePageSize()); 810 return browseList.toArray(new Message[browseList.size()]); 811 } 812 813 public void doBrowse(List<Message> browseList, int max) { 814 final ConnectionContext connectionContext = createConnectionContext(); 815 try { 816 pageInMessages(false); 817 List<MessageReference> toExpire = new ArrayList<MessageReference>(); 818 synchronized (dispatchMutex) { 819 synchronized (pagedInPendingDispatch) { 820 addAll(pagedInPendingDispatch, browseList, max, toExpire); 821 for (MessageReference ref : toExpire) { 822 pagedInPendingDispatch.remove(ref); 823 if (broker.isExpired(ref)) { 824 LOG.debug("expiring from pagedInPending: " + ref); 825 messageExpired(connectionContext, ref); 826 } 827 } 828 } 829 toExpire.clear(); 830 synchronized (pagedInMessages) { 831 addAll(pagedInMessages.values(), browseList, max, toExpire); 832 } 833 for (MessageReference ref : toExpire) { 834 if (broker.isExpired(ref)) { 835 LOG.debug("expiring from pagedInMessages: " + ref); 836 messageExpired(connectionContext, ref); 837 } else { 838 synchronized (pagedInMessages) { 839 pagedInMessages.remove(ref.getMessageId()); 840 } 841 } 842 } 843 844 if (browseList.size() < getMaxBrowsePageSize()) { 845 synchronized (messages) { 846 try { 847 messages.reset(); 848 while (messages.hasNext() && browseList.size() < max) { 849 MessageReference node = messages.next(); 850 if (node.isExpired()) { 851 if (broker.isExpired(node)) { 852 LOG.debug("expiring from messages: " + node); 853 messageExpired(connectionContext, createMessageReference(node.getMessage())); 854 } 855 messages.remove(); 856 } else { 857 messages.rollback(node.getMessageId()); 858 if (browseList.contains(node.getMessage()) == false) { 859 browseList.add(node.getMessage()); 860 } 861 } 862 node.decrementReferenceCount(); 863 } 864 } finally { 865 messages.release(); 866 } 867 } 868 } 869 } 870 } catch (Exception e) { 871 LOG.error("Problem retrieving message for browse", e); 872 } 873 } 874 875 private void addAll(Collection<QueueMessageReference> refs, List<Message> l, int maxBrowsePageSize, List<MessageReference> toExpire) throws Exception { 876 for (Iterator<QueueMessageReference> i = refs.iterator(); i.hasNext() && l.size() < getMaxBrowsePageSize();) { 877 QueueMessageReference ref = i.next(); 878 if (ref.isExpired()) { 879 toExpire.add(ref); 880 } else if (l.contains(ref.getMessage()) == false) { 881 l.add(ref.getMessage()); 882 } 883 } 884 } 885 886 public Message getMessage(String id) { 887 MessageId msgId = new MessageId(id); 888 try { 889 synchronized (pagedInMessages) { 890 QueueMessageReference r = this.pagedInMessages.get(msgId); 891 if (r != null) { 892 return r.getMessage(); 893 } 894 } 895 synchronized (messages) { 896 try { 897 messages.reset(); 898 while (messages.hasNext()) { 899 try { 900 MessageReference r = messages.next(); 901 r.decrementReferenceCount(); 902 messages.rollback(r.getMessageId()); 903 if (msgId.equals(r.getMessageId())) { 904 Message m = r.getMessage(); 905 if (m != null) { 906 return m; 907 } 908 break; 909 } 910 } catch (IOException e) { 911 LOG.error("got an exception retrieving message " + id); 912 } 913 } 914 } finally { 915 messages.release(); 916 } 917 } 918 } catch (IOException e) { 919 LOG.error("got an exception retrieving message " + id); 920 } 921 return null; 922 } 923 924 public void purge() throws Exception { 925 ConnectionContext c = createConnectionContext(); 926 List<MessageReference> list = null; 927 do { 928 doPageIn(true); 929 synchronized (pagedInMessages) { 930 list = new ArrayList<MessageReference>(pagedInMessages.values()); 931 } 932 933 for (MessageReference ref : list) { 934 try { 935 QueueMessageReference r = (QueueMessageReference) ref; 936 removeMessage(c, (IndirectMessageReference) r); 937 } catch (IOException e) { 938 } 939 } 940 941 } while (!pagedInMessages.isEmpty() || this.destinationStatistics.getMessages().getCount() > 0); 942 gc(); 943 this.destinationStatistics.getMessages().setCount(0); 944 getMessages().clear(); 945 } 946 947 /** 948 * Removes the message matching the given messageId 949 */ 950 public boolean removeMessage(String messageId) throws Exception { 951 return removeMatchingMessages(createMessageIdFilter(messageId), 1) > 0; 952 } 953 954 /** 955 * Removes the messages matching the given selector 956 * 957 * @return the number of messages removed 958 */ 959 public int removeMatchingMessages(String selector) throws Exception { 960 return removeMatchingMessages(selector, -1); 961 } 962 963 /** 964 * Removes the messages matching the given selector up to the maximum number 965 * of matched messages 966 * 967 * @return the number of messages removed 968 */ 969 public int removeMatchingMessages(String selector, int maximumMessages) throws Exception { 970 return removeMatchingMessages(createSelectorFilter(selector), maximumMessages); 971 } 972 973 /** 974 * Removes the messages matching the given filter up to the maximum number 975 * of matched messages 976 * 977 * @return the number of messages removed 978 */ 979 public int removeMatchingMessages(MessageReferenceFilter filter, int maximumMessages) throws Exception { 980 int movedCounter = 0; 981 Set<MessageReference> set = new CopyOnWriteArraySet<MessageReference>(); 982 ConnectionContext context = createConnectionContext(); 983 do { 984 doPageIn(true); 985 synchronized (pagedInMessages) { 986 set.addAll(pagedInMessages.values()); 987 } 988 List<MessageReference> list = new ArrayList<MessageReference>(set); 989 for (MessageReference ref : list) { 990 IndirectMessageReference r = (IndirectMessageReference) ref; 991 if (filter.evaluate(context, r)) { 992 993 removeMessage(context, r); 994 set.remove(r); 995 if (++movedCounter >= maximumMessages && maximumMessages > 0) { 996 return movedCounter; 997 } 998 } 999 } 1000 } while (set.size() < this.destinationStatistics.getMessages().getCount()); 1001 return movedCounter; 1002 } 1003 1004 /** 1005 * Copies the message matching the given messageId 1006 */ 1007 public boolean copyMessageTo(ConnectionContext context, String messageId, ActiveMQDestination dest) throws Exception { 1008 return copyMatchingMessages(context, createMessageIdFilter(messageId), dest, 1) > 0; 1009 } 1010 1011 /** 1012 * Copies the messages matching the given selector 1013 * 1014 * @return the number of messages copied 1015 */ 1016 public int copyMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest) throws Exception { 1017 return copyMatchingMessagesTo(context, selector, dest, -1); 1018 } 1019 1020 /** 1021 * Copies the messages matching the given selector up to the maximum number 1022 * of matched messages 1023 * 1024 * @return the number of messages copied 1025 */ 1026 public int copyMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest, int maximumMessages) throws Exception { 1027 return copyMatchingMessages(context, createSelectorFilter(selector), dest, maximumMessages); 1028 } 1029 1030 /** 1031 * Copies the messages matching the given filter up to the maximum number of 1032 * matched messages 1033 * 1034 * @return the number of messages copied 1035 */ 1036 public int copyMatchingMessages(ConnectionContext context, MessageReferenceFilter filter, ActiveMQDestination dest, int maximumMessages) throws Exception { 1037 int movedCounter = 0; 1038 int count = 0; 1039 Set<MessageReference> set = new CopyOnWriteArraySet<MessageReference>(); 1040 do { 1041 int oldMaxSize = getMaxPageSize(); 1042 setMaxPageSize((int) this.destinationStatistics.getMessages().getCount()); 1043 doPageIn(true); 1044 setMaxPageSize(oldMaxSize); 1045 synchronized (pagedInMessages) { 1046 set.addAll(pagedInMessages.values()); 1047 } 1048 List<MessageReference> list = new ArrayList<MessageReference>(set); 1049 for (MessageReference ref : list) { 1050 IndirectMessageReference r = (IndirectMessageReference) ref; 1051 if (filter.evaluate(context, r)) { 1052 1053 r.incrementReferenceCount(); 1054 try { 1055 Message m = r.getMessage(); 1056 BrokerSupport.resend(context, m, dest); 1057 if (++movedCounter >= maximumMessages && maximumMessages > 0) { 1058 return movedCounter; 1059 } 1060 } finally { 1061 r.decrementReferenceCount(); 1062 } 1063 } 1064 count++; 1065 } 1066 } while (count < this.destinationStatistics.getMessages().getCount()); 1067 return movedCounter; 1068 } 1069 1070 /** 1071 * Move a message 1072 * 1073 * @param context connection context 1074 * @param m message 1075 * @param dest ActiveMQDestination 1076 * @throws Exception 1077 */ 1078 public boolean moveMessageTo(ConnectionContext context, Message m, ActiveMQDestination dest) throws Exception { 1079 QueueMessageReference r = createMessageReference(m); 1080 BrokerSupport.resend(context, m, dest); 1081 removeMessage(context, r); 1082 synchronized (messages) { 1083 messages.rollback(r.getMessageId()); 1084 } 1085 return true; 1086 } 1087 1088 /** 1089 * Moves the message matching the given messageId 1090 */ 1091 public boolean moveMessageTo(ConnectionContext context, String messageId, ActiveMQDestination dest) throws Exception { 1092 return moveMatchingMessagesTo(context, createMessageIdFilter(messageId), dest, 1) > 0; 1093 } 1094 1095 /** 1096 * Moves the messages matching the given selector 1097 * 1098 * @return the number of messages removed 1099 */ 1100 public int moveMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest) throws Exception { 1101 return moveMatchingMessagesTo(context, selector, dest, Integer.MAX_VALUE); 1102 } 1103 1104 /** 1105 * Moves the messages matching the given selector up to the maximum number 1106 * of matched messages 1107 */ 1108 public int moveMatchingMessagesTo(ConnectionContext context, String selector, ActiveMQDestination dest, int maximumMessages) throws Exception { 1109 return moveMatchingMessagesTo(context, createSelectorFilter(selector), dest, maximumMessages); 1110 } 1111 1112 /** 1113 * Moves the messages matching the given filter up to the maximum number of 1114 * matched messages 1115 */ 1116 public int moveMatchingMessagesTo(ConnectionContext context, MessageReferenceFilter filter, ActiveMQDestination dest, int maximumMessages) throws Exception { 1117 int movedCounter = 0; 1118 Set<MessageReference> set = new CopyOnWriteArraySet<MessageReference>(); 1119 do { 1120 doPageIn(true); 1121 synchronized (pagedInMessages) { 1122 set.addAll(pagedInMessages.values()); 1123 } 1124 List<MessageReference> list = new ArrayList<MessageReference>(set); 1125 for (MessageReference ref : list) { 1126 IndirectMessageReference r = (IndirectMessageReference) ref; 1127 if (filter.evaluate(context, r)) { 1128 // We should only move messages that can be locked. 1129 moveMessageTo(context, ref.getMessage(), dest); 1130 set.remove(r); 1131 if (++movedCounter >= maximumMessages && maximumMessages > 0) { 1132 return movedCounter; 1133 } 1134 } 1135 } 1136 } while (set.size() < this.destinationStatistics.getMessages().getCount() && set.size() < maximumMessages); 1137 return movedCounter; 1138 } 1139 1140 BrowserDispatch getNextBrowserDispatch() { 1141 synchronized (pagedInMessages) { 1142 if (browserDispatches.isEmpty()) { 1143 return null; 1144 } 1145 return browserDispatches.removeFirst(); 1146 } 1147 1148 } 1149 1150 /** 1151 * @return true if we would like to iterate again 1152 * @see org.apache.activemq.thread.Task#iterate() 1153 */ 1154 public boolean iterate() { 1155 boolean pageInMoreMessages = false; 1156 synchronized (iteratingMutex) { 1157 1158 // do early to allow dispatch of these waiting messages 1159 synchronized (messagesWaitingForSpace) { 1160 Iterator<Runnable> it = messagesWaitingForSpace.values().iterator(); 1161 while (it.hasNext()) { 1162 if (!memoryUsage.isFull()) { 1163 Runnable op = it.next(); 1164 it.remove(); 1165 op.run(); 1166 } else { 1167 registerCallbackForNotFullNotification(); 1168 break; 1169 } 1170 } 1171 } 1172 1173 if (firstConsumer) { 1174 firstConsumer = false; 1175 try { 1176 if (consumersBeforeDispatchStarts > 0) { 1177 int timeout = 1000; // wait one second by default if consumer count isn't reached 1178 if (timeBeforeDispatchStarts > 0) { 1179 timeout = timeBeforeDispatchStarts; 1180 } 1181 if (consumersBeforeStartsLatch.await(timeout, TimeUnit.MILLISECONDS)) { 1182 if (LOG.isDebugEnabled()) { 1183 LOG.debug(consumers.size() + " consumers subscribed. Starting dispatch."); 1184 } 1185 } else { 1186 if (LOG.isDebugEnabled()) { 1187 LOG.debug(timeout + " ms elapsed and " + consumers.size() + " consumers subscribed. Starting dispatch."); 1188 } 1189 } 1190 } 1191 if (timeBeforeDispatchStarts > 0 && consumersBeforeDispatchStarts <= 0) { 1192 iteratingMutex.wait(timeBeforeDispatchStarts); 1193 if (LOG.isDebugEnabled()) { 1194 LOG.debug(timeBeforeDispatchStarts + " ms elapsed. Starting dispatch."); 1195 } 1196 } 1197 } catch (Exception e) { 1198 LOG.error(e); 1199 } 1200 } 1201 1202 BrowserDispatch pendingBrowserDispatch = getNextBrowserDispatch(); 1203 1204 synchronized (messages) { 1205 pageInMoreMessages |= !messages.isEmpty(); 1206 } 1207 1208 // Kinda ugly.. but I think dispatchLock is the only mutex protecting the 1209 // pagedInPendingDispatch variable. 1210 synchronized (dispatchMutex) { 1211 pageInMoreMessages |= !pagedInPendingDispatch.isEmpty(); 1212 } 1213 1214 // Perhaps we should page always into the pagedInPendingDispatch list if 1215 // !messages.isEmpty(), and then if !pagedInPendingDispatch.isEmpty() 1216 // then we do a dispatch. 1217 if (pageInMoreMessages || pendingBrowserDispatch != null) { 1218 try { 1219 pageInMessages(pendingBrowserDispatch != null); 1220 1221 } catch (Throwable e) { 1222 LOG.error("Failed to page in more queue messages ", e); 1223 } 1224 } 1225 1226 if (pendingBrowserDispatch != null) { 1227 ArrayList<QueueMessageReference> alreadyDispatchedMessages = null; 1228 synchronized (pagedInMessages) { 1229 alreadyDispatchedMessages = new ArrayList<QueueMessageReference>(pagedInMessages.values()); 1230 } 1231 if (LOG.isDebugEnabled()) { 1232 LOG.debug("dispatch to browser: " + pendingBrowserDispatch.getBrowser() 1233 + ", already dispatched/paged count: " + alreadyDispatchedMessages.size()); 1234 } 1235 do { 1236 try { 1237 MessageEvaluationContext msgContext = new NonCachedMessageEvaluationContext(); 1238 msgContext.setDestination(destination); 1239 1240 QueueBrowserSubscription browser = pendingBrowserDispatch.getBrowser(); 1241 for (QueueMessageReference node : alreadyDispatchedMessages) { 1242 if (!node.isAcked()) { 1243 msgContext.setMessageReference(node); 1244 if (browser.matches(node, msgContext)) { 1245 browser.add(node); 1246 } 1247 } 1248 } 1249 pendingBrowserDispatch.done(); 1250 } catch (Exception e) { 1251 LOG.warn("exception on dispatch to browser: " + pendingBrowserDispatch.getBrowser(), e); 1252 } 1253 1254 } while ((pendingBrowserDispatch = getNextBrowserDispatch()) != null); 1255 } 1256 1257 if (pendingWakeups.get() > 0) { 1258 pendingWakeups.decrementAndGet(); 1259 } 1260 return pendingWakeups.get() > 0; 1261 } 1262 } 1263 1264 protected MessageReferenceFilter createMessageIdFilter(final String messageId) { 1265 return new MessageReferenceFilter() { 1266 public boolean evaluate(ConnectionContext context, MessageReference r) { 1267 return messageId.equals(r.getMessageId().toString()); 1268 } 1269 1270 public String toString() { 1271 return "MessageIdFilter: " + messageId; 1272 } 1273 }; 1274 } 1275 1276 protected MessageReferenceFilter createSelectorFilter(String selector) throws InvalidSelectorException { 1277 final BooleanExpression selectorExpression = SelectorParser.parse(selector); 1278 1279 return new MessageReferenceFilter() { 1280 public boolean evaluate(ConnectionContext context, MessageReference r) throws JMSException { 1281 MessageEvaluationContext messageEvaluationContext = context.getMessageEvaluationContext(); 1282 1283 messageEvaluationContext.setMessageReference(r); 1284 if (messageEvaluationContext.getDestination() == null) { 1285 messageEvaluationContext.setDestination(getActiveMQDestination()); 1286 } 1287 1288 return selectorExpression.matches(messageEvaluationContext); 1289 } 1290 }; 1291 } 1292 1293 protected void removeMessage(ConnectionContext c, QueueMessageReference r) throws IOException { 1294 removeMessage(c, null, r); 1295 synchronized (dispatchMutex) { 1296 synchronized (pagedInPendingDispatch) { 1297 pagedInPendingDispatch.remove(r); 1298 } 1299 } 1300 } 1301 1302 protected void removeMessage(ConnectionContext c, Subscription subs, QueueMessageReference r) throws IOException { 1303 MessageAck ack = new MessageAck(); 1304 ack.setAckType(MessageAck.STANDARD_ACK_TYPE); 1305 ack.setDestination(destination); 1306 ack.setMessageID(r.getMessageId()); 1307 removeMessage(c, subs, r, ack); 1308 } 1309 1310 protected void removeMessage(ConnectionContext context, Subscription sub, final QueueMessageReference reference, MessageAck ack) throws IOException { 1311 reference.setAcked(true); 1312 // This sends the ack the the journal.. 1313 if (!ack.isInTransaction()) { 1314 acknowledge(context, sub, ack, reference); 1315 getDestinationStatistics().getDequeues().increment(); 1316 dropMessage(reference); 1317 } else { 1318 try { 1319 acknowledge(context, sub, ack, reference); 1320 } finally { 1321 context.getTransaction().addSynchronization(new Synchronization() { 1322 1323 public void afterCommit() throws Exception { 1324 getDestinationStatistics().getDequeues().increment(); 1325 dropMessage(reference); 1326 wakeup(); 1327 } 1328 1329 public void afterRollback() throws Exception { 1330 reference.setAcked(false); 1331 } 1332 }); 1333 } 1334 } 1335 if (ack.isPoisonAck()) { 1336 // message gone to DLQ, is ok to allow redelivery 1337 synchronized (messages) { 1338 messages.rollback(reference.getMessageId()); 1339 } 1340 } 1341 1342 } 1343 1344 private void dropMessage(QueueMessageReference reference) { 1345 reference.drop(); 1346 destinationStatistics.getMessages().decrement(); 1347 synchronized (pagedInMessages) { 1348 pagedInMessages.remove(reference.getMessageId()); 1349 } 1350 } 1351 1352 public void messageExpired(ConnectionContext context, MessageReference reference) { 1353 messageExpired(context, null, reference); 1354 } 1355 1356 public void messageExpired(ConnectionContext context, Subscription subs, MessageReference reference) { 1357 if (LOG.isDebugEnabled()) { 1358 LOG.debug("message expired: " + reference); 1359 } 1360 broker.messageExpired(context, reference); 1361 destinationStatistics.getExpired().increment(); 1362 try { 1363 removeMessage(context, subs, (QueueMessageReference) reference); 1364 } catch (IOException e) { 1365 LOG.error("Failed to remove expired Message from the store ", e); 1366 } 1367 } 1368 1369 protected ConnectionContext createConnectionContext() { 1370 ConnectionContext answer = new ConnectionContext(new NonCachedMessageEvaluationContext()); 1371 answer.setBroker(this.broker); 1372 answer.getMessageEvaluationContext().setDestination(getActiveMQDestination()); 1373 answer.setSecurityContext(SecurityContext.BROKER_SECURITY_CONTEXT); 1374 return answer; 1375 } 1376 1377 final void sendMessage(final ConnectionContext context, Message msg) throws Exception { 1378 if (!msg.isPersistent() && messages.getSystemUsage() != null) { 1379 if (systemUsage.getTempUsage().isFull()) { 1380 final String logMessage = "Usage Manager Temp Store is Full. Stopping producer (" + msg.getProducerId() + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "." 1381 + " See http://activemq.apache.org/producer-flow-control.html for more info"; 1382 if (systemUsage.isSendFailIfNoSpace()) { 1383 throw new ResourceAllocationException(logMessage); 1384 } 1385 1386 waitForSpace(context, messages.getSystemUsage().getTempUsage(), logMessage); 1387 } 1388 1389 } 1390 synchronized (messages) { 1391 messages.addMessageLast(msg); 1392 } 1393 destinationStatistics.getEnqueues().increment(); 1394 destinationStatistics.getMessages().increment(); 1395 messageDelivered(context, msg); 1396 synchronized (consumers) { 1397 if (consumers.isEmpty()) { 1398 onMessageWithNoConsumers(context, msg); 1399 } 1400 } 1401 wakeup(); 1402 } 1403 1404 public void wakeup() { 1405 if (optimizedDispatch || isSlave()) { 1406 iterate(); 1407 pendingWakeups.incrementAndGet(); 1408 } else { 1409 asyncWakeup(); 1410 } 1411 } 1412 1413 private void asyncWakeup() { 1414 try { 1415 pendingWakeups.incrementAndGet(); 1416 this.taskRunner.wakeup(); 1417 } catch (InterruptedException e) { 1418 LOG.warn("Async task tunner failed to wakeup ", e); 1419 } 1420 } 1421 1422 private boolean isSlave() { 1423 return broker.getBrokerService().isSlave(); 1424 } 1425 1426 private List<QueueMessageReference> doPageIn(boolean force) throws Exception { 1427 List<QueueMessageReference> result = null; 1428 List<QueueMessageReference> resultList = null; 1429 synchronized (dispatchMutex) { 1430 int toPageIn = Math.min(getMaxPageSize(), messages.size()); 1431 if (LOG.isDebugEnabled()) { 1432 LOG.debug(destination.getPhysicalName() + " toPageIn: " + toPageIn + ", Inflight: " + destinationStatistics.getInflight().getCount() + ", pagedInMessages.size " 1433 + pagedInMessages.size() + ", enqueueSize: " + destinationStatistics.getEnqueues().getCount()); 1434 } 1435 1436 if (isLazyDispatch() && !force) { 1437 // Only page in the minimum number of messages which can be dispatched immediately. 1438 toPageIn = Math.min(getConsumerMessageCountBeforeFull(), toPageIn); 1439 } 1440 if (toPageIn > 0 && (force || (!consumers.isEmpty() && pagedInPendingDispatch.size() < getMaxPageSize()))) { 1441 int count = 0; 1442 result = new ArrayList<QueueMessageReference>(toPageIn); 1443 synchronized (messages) { 1444 try { 1445 messages.setMaxBatchSize(toPageIn); 1446 messages.reset(); 1447 while (messages.hasNext() && count < toPageIn) { 1448 MessageReference node = messages.next(); 1449 messages.remove(); 1450 QueueMessageReference ref = createMessageReference(node.getMessage()); 1451 if (ref.isExpired()) { 1452 if (broker.isExpired(ref)) { 1453 messageExpired(createConnectionContext(), ref); 1454 } else { 1455 ref.decrementReferenceCount(); 1456 } 1457 } else { 1458 result.add(ref); 1459 count++; 1460 } 1461 } 1462 } finally { 1463 messages.release(); 1464 } 1465 } 1466 // Only add new messages, not already pagedIn to avoid multiple dispatch attempts 1467 synchronized (pagedInMessages) { 1468 resultList = new ArrayList<QueueMessageReference>(result.size()); 1469 for (QueueMessageReference ref : result) { 1470 if (!pagedInMessages.containsKey(ref.getMessageId())) { 1471 pagedInMessages.put(ref.getMessageId(), ref); 1472 resultList.add(ref); 1473 } else { 1474 ref.decrementReferenceCount(); 1475 } 1476 } 1477 } 1478 } else { 1479 // Avoid return null list, if condition is not validated 1480 resultList = new ArrayList<QueueMessageReference>(); 1481 } 1482 } 1483 return resultList; 1484 } 1485 1486 private void doDispatch(List<QueueMessageReference> list) throws Exception { 1487 boolean doWakeUp = false; 1488 synchronized (dispatchMutex) { 1489 1490 synchronized (pagedInPendingDispatch) { 1491 if (!pagedInPendingDispatch.isEmpty()) { 1492 // Try to first dispatch anything that had not been 1493 // dispatched before. 1494 pagedInPendingDispatch = doActualDispatch(pagedInPendingDispatch); 1495 } 1496 // and now see if we can dispatch the new stuff.. and append to the pending 1497 // list anything that does not actually get dispatched. 1498 if (list != null && !list.isEmpty()) { 1499 if (pagedInPendingDispatch.isEmpty()) { 1500 pagedInPendingDispatch.addAll(doActualDispatch(list)); 1501 } else { 1502 for (QueueMessageReference qmr : list) { 1503 if (!pagedInPendingDispatch.contains(qmr)) { 1504 pagedInPendingDispatch.add(qmr); 1505 } 1506 } 1507 doWakeUp = true; 1508 } 1509 } 1510 } 1511 } 1512 if (doWakeUp) { 1513 // avoid lock order contention 1514 asyncWakeup(); 1515 } 1516 } 1517 1518 /** 1519 * @return list of messages that could get dispatched to consumers if they 1520 * were not full. 1521 */ 1522 private List<QueueMessageReference> doActualDispatch(List<QueueMessageReference> list) throws Exception { 1523 List<Subscription> consumers; 1524 1525 synchronized (this.consumers) { 1526 if (this.consumers.isEmpty() || isSlave()) { 1527 // slave dispatch happens in processDispatchNotification 1528 return list; 1529 } 1530 consumers = new ArrayList<Subscription>(this.consumers); 1531 } 1532 1533 List<QueueMessageReference> rc = new ArrayList<QueueMessageReference>(list.size()); 1534 Set<Subscription> fullConsumers = new HashSet<Subscription>(this.consumers.size()); 1535 1536 for (MessageReference node : list) { 1537 Subscription target = null; 1538 int interestCount = 0; 1539 for (Subscription s : consumers) { 1540 if (s instanceof QueueBrowserSubscription) { 1541 interestCount++; 1542 continue; 1543 } 1544 if (dispatchSelector.canSelect(s, node)) { 1545 if (!fullConsumers.contains(s)) { 1546 if (!s.isFull()) { 1547 // Dispatch it. 1548 s.add(node); 1549 target = s; 1550 break; 1551 } else { 1552 // no further dispatch of list to a full consumer to avoid out of order message receipt 1553 fullConsumers.add(s); 1554 } 1555 } 1556 interestCount++; 1557 } else { 1558 // makes sure it gets dispatched again 1559 if (!node.isDropped() && !((QueueMessageReference) node).isAcked() && (!node.isDropped() || s.getConsumerInfo().isBrowser())) { 1560 interestCount++; 1561 } 1562 } 1563 } 1564 1565 if ((target == null && interestCount > 0) || consumers.size() == 0) { 1566 // This means all subs were full or that there are no consumers... 1567 rc.add((QueueMessageReference) node); 1568 } 1569 1570 // If it got dispatched, rotate the consumer list to get round robin distribution. 1571 if (target != null && !strictOrderDispatch && consumers.size() > 1 && !dispatchSelector.isExclusiveConsumer(target)) { 1572 synchronized (this.consumers) { 1573 if (removeFromConsumerList(target)) { 1574 addToConsumerList(target); 1575 consumers = new ArrayList<Subscription>(this.consumers); 1576 } 1577 } 1578 } 1579 } 1580 1581 return rc; 1582 } 1583 1584 protected void pageInMessages(boolean force) throws Exception { 1585 doDispatch(doPageIn(force)); 1586 } 1587 1588 private void addToConsumerList(Subscription sub) { 1589 if (useConsumerPriority) { 1590 consumers.add(sub); 1591 Collections.sort(consumers, orderedCompare); 1592 } else { 1593 consumers.add(sub); 1594 } 1595 } 1596 1597 private boolean removeFromConsumerList(Subscription sub) { 1598 return consumers.remove(sub); 1599 } 1600 1601 private int getConsumerMessageCountBeforeFull() throws Exception { 1602 int total = 0; 1603 boolean zeroPrefetch = false; 1604 synchronized (consumers) { 1605 for (Subscription s : consumers) { 1606 zeroPrefetch |= s.getPrefetchSize() == 0; 1607 int countBeforeFull = s.countBeforeFull(); 1608 total += countBeforeFull; 1609 } 1610 } 1611 if (total == 0 && zeroPrefetch) { 1612 total = 1; 1613 } 1614 return total; 1615 } 1616 1617 /* 1618 * In slave mode, dispatch is ignored till we get this notification as the 1619 * dispatch process is non deterministic between master and slave. On a 1620 * notification, the actual dispatch to the subscription (as chosen by the 1621 * master) is completed. (non-Javadoc) 1622 * 1623 * @see 1624 * org.apache.activemq.broker.region.BaseDestination#processDispatchNotification 1625 * (org.apache.activemq.command.MessageDispatchNotification) 1626 */ 1627 public void processDispatchNotification(MessageDispatchNotification messageDispatchNotification) throws Exception { 1628 // do dispatch 1629 Subscription sub = getMatchingSubscription(messageDispatchNotification); 1630 if (sub != null) { 1631 MessageReference message = getMatchingMessage(messageDispatchNotification); 1632 sub.add(message); 1633 sub.processMessageDispatchNotification(messageDispatchNotification); 1634 } 1635 } 1636 1637 private QueueMessageReference getMatchingMessage(MessageDispatchNotification messageDispatchNotification) throws Exception { 1638 QueueMessageReference message = null; 1639 MessageId messageId = messageDispatchNotification.getMessageId(); 1640 1641 synchronized (dispatchMutex) { 1642 synchronized (pagedInPendingDispatch) { 1643 for (QueueMessageReference ref : pagedInPendingDispatch) { 1644 if (messageId.equals(ref.getMessageId())) { 1645 message = ref; 1646 pagedInPendingDispatch.remove(ref); 1647 break; 1648 } 1649 } 1650 } 1651 1652 if (message == null) { 1653 synchronized (pagedInMessages) { 1654 message = pagedInMessages.get(messageId); 1655 } 1656 } 1657 1658 if (message == null) { 1659 synchronized (messages) { 1660 try { 1661 messages.setMaxBatchSize(getMaxPageSize()); 1662 messages.reset(); 1663 while (messages.hasNext()) { 1664 MessageReference node = messages.next(); 1665 messages.remove(); 1666 if (messageId.equals(node.getMessageId())) { 1667 message = this.createMessageReference(node.getMessage()); 1668 break; 1669 } 1670 } 1671 } finally { 1672 messages.release(); 1673 } 1674 } 1675 } 1676 1677 if (message == null) { 1678 Message msg = loadMessage(messageId); 1679 if (msg != null) { 1680 message = this.createMessageReference(msg); 1681 } 1682 } 1683 1684 } 1685 if (message == null) { 1686 throw new JMSException("Slave broker out of sync with master - Message: " + messageDispatchNotification.getMessageId() + " on " + messageDispatchNotification.getDestination() 1687 + " does not exist among pending(" + pagedInPendingDispatch.size() + ") for subscription: " + messageDispatchNotification.getConsumerId()); 1688 } 1689 return message; 1690 } 1691 1692 /** 1693 * Find a consumer that matches the id in the message dispatch notification 1694 * 1695 * @param messageDispatchNotification 1696 * @return sub or null if the subscription has been removed before dispatch 1697 * @throws JMSException 1698 */ 1699 private Subscription getMatchingSubscription(MessageDispatchNotification messageDispatchNotification) throws JMSException { 1700 Subscription sub = null; 1701 synchronized (consumers) { 1702 for (Subscription s : consumers) { 1703 if (messageDispatchNotification.getConsumerId().equals(s.getConsumerInfo().getConsumerId())) { 1704 sub = s; 1705 break; 1706 } 1707 } 1708 } 1709 return sub; 1710 } 1711 1712 public void onUsageChanged(Usage usage, int oldPercentUsage, int newPercentUsage) { 1713 if (oldPercentUsage > newPercentUsage) { 1714 asyncWakeup(); 1715 } 1716 } 1717 1718 private final void waitForSpace(ConnectionContext context, Usage<?> usage, String warning) throws IOException, InterruptedException, ResourceAllocationException { 1719 if (systemUsage.getSendFailIfNoSpaceAfterTimeout() != 0) { 1720 if (!usage.waitForSpace(systemUsage.getSendFailIfNoSpaceAfterTimeout())) { 1721 throw new ResourceAllocationException(warning); 1722 } 1723 } else { 1724 long start = System.currentTimeMillis(); 1725 long nextWarn = start + blockedProducerWarningInterval; 1726 while (!usage.waitForSpace(1000)) { 1727 if (context.getStopping().get()) { 1728 throw new IOException("Connection closed, send aborted."); 1729 } 1730 1731 long now = System.currentTimeMillis(); 1732 if (now >= nextWarn) { 1733 LOG.info(warning + " (blocking for: " + (now - start) / 1000 + "s)"); 1734 nextWarn = now + blockedProducerWarningInterval; 1735 } 1736 } 1737 } 1738 } 1739 }