You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by de...@apache.org on 2009/01/22 18:22:15 UTC
svn commit: r736720 - in
/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region:
Queue.java QueueDispatchSelector.java
Author: dejanb
Date: Thu Jan 22 09:22:12 2009
New Revision: 736720
URL: http://svn.apache.org/viewvc?rev=736720&view=rev
Log:
Recovery dispatch refactoring as the part of the solution for the https://issues.apache.org/activemq/browse/AMQ-2016
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/QueueDispatchSelector.java
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java?rev=736720&r1=736719&r2=736720&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java Thu Jan 22 09:22:12 2009
@@ -233,16 +233,18 @@
}
}
- // any newly paged in messages that are not dispatched are added to pagedInPending in iterate()
- doPageIn(false);
+ // do recovery dispatch only if it is a browser subscription
+ if(sub instanceof QueueBrowserSubscription ) {
+ // any newly paged in messages that are not dispatched are added to pagedInPending in iterate()
+ doPageIn(false);
+
+ synchronized (pagedInMessages) {
+ RecoveryDispatch rd = new RecoveryDispatch();
+ rd.messages = new ArrayList<QueueMessageReference>(pagedInMessages.values());
+ rd.subscription = sub;
+ recoveries.addLast(rd);
+ }
- synchronized (pagedInMessages) {
- RecoveryDispatch rd = new RecoveryDispatch();
- rd.messages = new ArrayList<QueueMessageReference>(pagedInMessages.values());
- rd.subscription = sub;
- recoveries.addLast(rd);
- }
- if( sub instanceof QueueBrowserSubscription ) {
((QueueBrowserSubscription)sub).incrementQueueRef();
}
if (!(this.optimizedDispatch || isSlave())) {
@@ -303,9 +305,14 @@
doDispatch(list);
}
}
-
- if (consumers.isEmpty()) {
- messages.gc();
+ //if it is a last consumer (and not a browser) dispatch all pagedIn messages
+ if (consumers.isEmpty() && !(sub instanceof QueueBrowserSubscription)) {
+ List<QueueMessageReference> list = new ArrayList<QueueMessageReference>();
+ for (QueueMessageReference ref : pagedInMessages.values()) {
+ list.add(ref);
+ }
+ pagedInPendingDispatch.clear();
+ doDispatch(list);
}
if (!(this.optimizedDispatch || isSlave())) {
wakeup();
@@ -615,6 +622,7 @@
int count = 0;
List<Message> l = new ArrayList<Message>();
try {
+ pageInMessages(false);
synchronized (this.pagedInPendingDispatch) {
for (Iterator<QueueMessageReference> i = this.pagedInPendingDispatch
.iterator(); i.hasNext()
@@ -657,7 +665,7 @@
}
}
}
- } catch (IOException e) {
+ } catch (Exception e) {
LOG.error("Problem retrieving message in browse() ", e);
}
return l.toArray(new Message[l.size()]);
@@ -899,7 +907,7 @@
int movedCounter = 0;
Set<MessageReference> set = new CopyOnWriteArraySet<MessageReference>();
do {
- pageInMessages();
+ doPageIn(true);
synchronized (pagedInMessages) {
set.addAll(pagedInMessages.values());
}
@@ -981,7 +989,7 @@
e.printStackTrace();
}
}
-
+
boolean pageInMoreMessages = false;
synchronized (messages) {
pageInMoreMessages = !messages.isEmpty();
@@ -1230,6 +1238,7 @@
pagedInPendingDispatch.add(qmr);
}
}
+ wakeup();
}
}
}
@@ -1268,11 +1277,16 @@
}
}
interestCount++;
+ } else {
+ // makes sure it gets dispatched again
+ if (!node.isDropped() && !((QueueMessageReference)node).isAcked() && (!node.isDropped() || s.getConsumerInfo().isBrowser())) {
+ interestCount++;
+ }
}
}
- if (target == null && interestCount>0) {
- // This means all subs were full...
+ if ((target == null && interestCount>0) || consumers.size() == 0) {
+ // This means all subs were full or that there are no consumers...
rc.add((QueueMessageReference)node);
}
@@ -1288,10 +1302,6 @@
}
}
- //LOG.info(getName()+" Pending messages:");
- //for (MessageReference n : rc) {
- // LOG.info(getName()+" - " + n.getMessageId());
- // }
return rc;
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/QueueDispatchSelector.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/QueueDispatchSelector.java?rev=736720&r1=736719&r2=736720&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/QueueDispatchSelector.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/QueueDispatchSelector.java Thu Jan 22 09:22:12 2009
@@ -70,7 +70,7 @@
if (result) {
result = exclusiveConsumer == null
|| exclusiveConsumer == subscription;
- if (result) {
+ if (result && !subscription.isFull()) {
QueueMessageReference node = (QueueMessageReference) m;
// Keep message groups together.
String groupId = node.getGroupID();