From 1f5694ed60041d3b75fcb77070bdf1dae4996bd6 Mon Sep 17 00:00:00 2001 From: "Timothy A. Bish" Date: Wed, 5 Jun 2013 17:34:50 +0000 Subject: [PATCH] fix for: https://issues.apache.org/jira/browse/AMQ-4487 and https://issues.apache.org/jira/browse/AMQ-4372 git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1489978 13f79535-47bb-0310-9956-ffa450edef68 --- .../region/QueueBrowserSubscription.java | 36 +++++++++---------- .../broker/region/policy/PolicyEntry.java | 8 ++++- .../org/apache/activemq/bugs/AMQ4487Test.java | 12 +++---- 3 files changed, 29 insertions(+), 27 deletions(-) diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/QueueBrowserSubscription.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/QueueBrowserSubscription.java index d8d5c763d5..74a673d960 100755 --- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/QueueBrowserSubscription.java +++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/QueueBrowserSubscription.java @@ -19,12 +19,11 @@ package org.apache.activemq.broker.region; import java.io.IOException; import java.util.ArrayList; import java.util.List; -import javax.jms.InvalidSelectorException; + import javax.jms.JMSException; import org.apache.activemq.broker.Broker; import org.apache.activemq.broker.ConnectionContext; -import org.apache.activemq.broker.region.cursors.VMPendingMessageCursor; import org.apache.activemq.command.ConsumerInfo; import org.apache.activemq.command.MessageAck; import org.apache.activemq.filter.MessageEvaluationContext; @@ -36,19 +35,20 @@ public class QueueBrowserSubscription extends QueueSubscription { boolean browseDone; boolean destinationsAdded; - public QueueBrowserSubscription(Broker broker,SystemUsage usageManager, ConnectionContext context, ConsumerInfo info) - throws JMSException { - super(broker,usageManager, context, info); + public QueueBrowserSubscription(Broker broker, SystemUsage usageManager, ConnectionContext context, ConsumerInfo info) throws JMSException { + super(broker, usageManager, context, info); } + @Override protected boolean canDispatch(MessageReference node) { - return !((QueueMessageReference)node).isAcked(); + return !((QueueMessageReference) node).isAcked(); } + @Override public synchronized String toString() { - return "QueueBrowserSubscription:" + " consumer=" + info.getConsumerId() + ", destinations=" - + destinations.size() + ", dispatched=" + dispatched.size() + ", delivered=" - + this.prefetchExtension + ", pending=" + getPendingQueueSize(); + return "QueueBrowserSubscription:" + " consumer=" + info.getConsumerId() + + ", destinations=" + destinations.size() + ", dispatched=" + dispatched.size() + + ", delivered=" + this.prefetchExtension + ", pending=" + getPendingQueueSize(); } synchronized public void destinationsAdded() throws Exception { @@ -57,12 +57,13 @@ public class QueueBrowserSubscription extends QueueSubscription { } private void checkDone() throws Exception { - if( !browseDone && queueRefs == 0 && destinationsAdded) { - browseDone=true; + if (!browseDone && queueRefs == 0 && destinationsAdded) { + browseDone = true; add(QueueMessageReference.NULL_MESSAGE); } } + @Override public boolean matches(MessageReference node, MessageEvaluationContext context) throws IOException { return !browseDone && super.matches(node, context); } @@ -70,15 +71,15 @@ public class QueueBrowserSubscription extends QueueSubscription { /** * Since we are a browser we don't really remove the message from the queue. */ - protected void acknowledge(ConnectionContext context, final MessageAck ack, final MessageReference n) - throws IOException { - if (info.isNetworkSubscription()) { - super.acknowledge(context, ack, n); - } + @Override + protected void acknowledge(ConnectionContext context, final MessageAck ack, final MessageReference n) throws IOException { + if (info.isNetworkSubscription()) { + super.acknowledge(context, ack, n); + } } synchronized public void incrementQueueRef() { - queueRefs++; + queueRefs++; } synchronized public void decrementQueueRef() throws Exception { @@ -88,7 +89,6 @@ public class QueueBrowserSubscription extends QueueSubscription { checkDone(); } - @Override public List remove(ConnectionContext context, Destination destination) throws Exception { super.remove(context, destination); diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java index b5134a3937..26bb387f57 100644 --- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java +++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/PolicyEntry.java @@ -238,7 +238,13 @@ public class PolicyEntry extends DestinationMapEntry { configurePrefetch(sub); sub.setCursorMemoryHighWaterMark(getCursorMemoryHighWaterMark()); sub.setUsePrefetchExtension(isUsePrefetchExtension()); - sub.setMaxProducersToAudit(getMaxProducersToAudit()); + + // TODO + // We currently need an infinite audit because of the way that browser dispatch + // is done. We should refactor the browsers to better handle message dispatch so + // we can remove this and perform a more efficient dispatch. + sub.setMaxProducersToAudit(Integer.MAX_VALUE); + sub.setMaxAuditDepth(Integer.MAX_VALUE); } public void configure(Broker broker, SystemUsage memoryManager, QueueSubscription sub) { diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4487Test.java b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4487Test.java index d0324c9471..346650e7c1 100644 --- a/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4487Test.java +++ b/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ4487Test.java @@ -19,7 +19,6 @@ package org.apache.activemq.bugs; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; -import java.net.URI; import java.util.Enumeration; import javax.jms.Connection; @@ -31,7 +30,6 @@ import javax.jms.Session; import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.broker.BrokerService; -import org.apache.activemq.broker.TransportConnector; import org.apache.activemq.broker.region.policy.PolicyEntry; import org.apache.activemq.broker.region.policy.PolicyMap; import org.junit.After; @@ -46,14 +44,14 @@ public class AMQ4487Test { private final String destinationName = "TEST.QUEUE"; private BrokerService broker; - private URI connectUri; private ActiveMQConnectionFactory factory; @Before public void startBroker() throws Exception { broker = new BrokerService(); - TransportConnector connector = broker.addConnector("tcp://0.0.0.0:0"); broker.deleteAllMessages(); + broker.setUseJmx(false); + broker.setAdvisorySupport(false); PolicyEntry policy = new PolicyEntry(); policy.setQueue(">"); @@ -64,8 +62,7 @@ public class AMQ4487Test { broker.start(); broker.waitUntilStarted(); - connectUri = connector.getConnectUri(); - factory = new ActiveMQConnectionFactory(connectUri); + factory = new ActiveMQConnectionFactory("vm://localhost"); } @After @@ -101,7 +98,7 @@ public class AMQ4487Test { @Test public void testBrowsingWithMoreThanMaxAuditDepth() throws Exception { - doTestBrowsing(76); + doTestBrowsing(300); } @SuppressWarnings("rawtypes") @@ -124,7 +121,6 @@ public class AMQ4487Test { if (LOG.isDebugEnabled()) { LOG.debug("Browsed Message: {}", m.getJMSMessageID()); } - LOG.info("Browsed Message: {}", m.getJMSMessageID()); received++; if (received > messagesToSend) {