Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-3596 - FilePendingMessageCursor memory list does not respect priority for non persistent messages. Fix with test. Reuse pendinglist from vm cursor for the file pending message cursor in memory list 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1202153 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Tully 2011-11-15 12:21:20 +00:00
parent 1462fd234c
commit 3557361e86
2 changed files with 20 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
View File

@ -53,7 +53,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
protected Broker broker; protected Broker broker;
private final PListStore store; private final PListStore store;
private final String name; private final String name;
private LinkedList<MessageReference> memoryList = new LinkedList<MessageReference>(); private PendingList memoryList;
private PList diskList; private PList diskList;
private Iterator<MessageReference> iter; private Iterator<MessageReference> iter;
private Destination regionDestination; private Destination regionDestination;
@ -68,6 +68,11 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
*/ */
public FilePendingMessageCursor(Broker broker, String name, boolean prioritizedMessages) { public FilePendingMessageCursor(Broker broker, String name, boolean prioritizedMessages) {
super(prioritizedMessages); super(prioritizedMessages);
if (this.prioritizedMessages) {
this.memoryList = new PrioritizedPendingList();
} else {
this.memoryList = new OrderedPendingList();
}
this.broker = broker; this.broker = broker;
// the store can be null if the BrokerService has persistence // the store can be null if the BrokerService has persistence
// turned off // turned off
@ -204,7 +209,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
regionDestination = node.getMessage().getRegionDestination(); regionDestination = node.getMessage().getRegionDestination();
if (isDiskListEmpty()) { if (isDiskListEmpty()) {
if (hasSpace() || this.store == null) { if (hasSpace() || this.store == null) {
memoryList.add(node); memoryList.addMessageLast(node);
node.incrementReferenceCount(); node.incrementReferenceCount();
setCacheEnabled(true); setCacheEnabled(true);
return true; return true;
@ -214,7 +219,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
if (isDiskListEmpty()) { if (isDiskListEmpty()) {
expireOldMessages(); expireOldMessages();
if (hasSpace()) { if (hasSpace()) {
memoryList.add(node); memoryList.addMessageLast(node);
node.incrementReferenceCount(); node.incrementReferenceCount();
return true; return true;
} else { } else {
@ -252,7 +257,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
regionDestination = node.getMessage().getRegionDestination(); regionDestination = node.getMessage().getRegionDestination();
if (isDiskListEmpty()) { if (isDiskListEmpty()) {
if (hasSpace()) { if (hasSpace()) {
memoryList.addFirst(node); memoryList.addMessageFirst(node);
node.incrementReferenceCount(); node.incrementReferenceCount();
setCacheEnabled(true); setCacheEnabled(true);
return; return;
@ -262,7 +267,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
if (isDiskListEmpty()) { if (isDiskListEmpty()) {
expireOldMessages(); expireOldMessages();
if (hasSpace()) { if (hasSpace()) {
memoryList.addFirst(node); memoryList.addMessageFirst(node);
node.incrementReferenceCount(); node.incrementReferenceCount();
return; return;
} else { } else {
@ -325,7 +330,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
*/ */
@Override @Override
public synchronized void remove(MessageReference node) { public synchronized void remove(MessageReference node) {
if (memoryList.remove(node)) { if (memoryList.remove(node) != null) {
node.decrementReferenceCount(); node.decrementReferenceCount();
} }
if (!isDiskListEmpty()) { if (!isDiskListEmpty()) {
@ -406,19 +411,15 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
protected synchronized void expireOldMessages() { protected synchronized void expireOldMessages() {
if (!memoryList.isEmpty()) { if (!memoryList.isEmpty()) {
LinkedList<MessageReference> tmpList = new LinkedList<MessageReference>(this.memoryList); for (Iterator<MessageReference> iterator = memoryList.iterator(); iterator.hasNext();) {
this.memoryList = new LinkedList<MessageReference>(); MessageReference node = iterator.next();
while (!tmpList.isEmpty()) {
MessageReference node = tmpList.removeFirst();
if (node.isExpired()) { if (node.isExpired()) {
node.decrementReferenceCount(); node.decrementReferenceCount();
discardExpiredMessage(node); discardExpiredMessage(node);
} else { iterator.remove();
memoryList.add(node);
} }
} }
} }
} }
protected synchronized void flushToDisk() { protected synchronized void flushToDisk() {
@ -428,8 +429,8 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
start = System.currentTimeMillis(); start = System.currentTimeMillis();
LOG.trace("" + name + ", flushToDisk() mem list size: " +memoryList.size() + " " + (systemUsage != null ? systemUsage.getMemoryUsage() : "") ); LOG.trace("" + name + ", flushToDisk() mem list size: " +memoryList.size() + " " + (systemUsage != null ? systemUsage.getMemoryUsage() : "") );
} }
while (!memoryList.isEmpty()) { for (Iterator<MessageReference> iterator = memoryList.iterator(); iterator.hasNext();) {
MessageReference node = memoryList.removeFirst(); MessageReference node = iterator.next();
node.decrementReferenceCount(); node.decrementReferenceCount();
ByteSequence bs; ByteSequence bs;
try { try {

4
activemq-core/src/test/java/org/apache/activemq/store/MessagePriorityTest.java
View File

@ -18,6 +18,7 @@
package org.apache.activemq.store; package org.apache.activemq.store;
import javax.jms.Connection; import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Message; import javax.jms.Message;
import javax.jms.MessageConsumer; import javax.jms.MessageConsumer;
import javax.jms.MessageProducer; import javax.jms.MessageProducer;
@ -53,6 +54,7 @@ abstract public class MessagePriorityTest extends CombinationTestSupport {
protected Session sess; protected Session sess;
public boolean useCache = true; public boolean useCache = true;
public int deliveryMode = Message.DEFAULT_DELIVERY_MODE;
public boolean dispatchAsync = true; public boolean dispatchAsync = true;
public boolean prioritizeMessages = true; public boolean prioritizeMessages = true;
public boolean immediatePriorityDispatch = true; public boolean immediatePriorityDispatch = true;
@ -150,6 +152,7 @@ abstract public class MessagePriorityTest extends CombinationTestSupport {
try { try {
MessageProducer producer = sess.createProducer(dest); MessageProducer producer = sess.createProducer(dest);
producer.setPriority(priority); producer.setPriority(priority);
producer.setDeliveryMode(deliveryMode);
for (int i = 0; i < messageCount; i++) { for (int i = 0; i < messageCount; i++) {
producer.send(sess.createTextMessage("message priority: " + priority)); producer.send(sess.createTextMessage("message priority: " + priority));
} }
@ -170,6 +173,7 @@ abstract public class MessagePriorityTest extends CombinationTestSupport {
public void initCombosForTestQueues() { public void initCombosForTestQueues() {
addCombinationValues("useCache", new Object[] {new Boolean(true), new Boolean(false)}); addCombinationValues("useCache", new Object[] {new Boolean(true), new Boolean(false)});
addCombinationValues("deliveryMode", new Object[] {new Integer(DeliveryMode.NON_PERSISTENT), new Integer(DeliveryMode.PERSISTENT)});
} }
public void testQueues() throws Exception { public void testQueues() throws Exception {