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

Fix for http://issues.apache.org/activemq/browse/AMQ-2530 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@902009 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2010-01-22 06:48:29 +00:00
parent dad4f49213
commit 8af25fd239
2 changed files with 12 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
activemq-core/src/main/java/org/apache/activemq/network/ConduitBridge.java
View File

@ -21,7 +21,6 @@ import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.Iterator; import java.util.Iterator;
import java.util.List; import java.util.List;
import org.apache.activemq.command.BrokerId; import org.apache.activemq.command.BrokerId;
import org.apache.activemq.command.ConsumerId; import org.apache.activemq.command.ConsumerId;
import org.apache.activemq.command.ConsumerInfo; import org.apache.activemq.command.ConsumerInfo;
@ -48,6 +47,7 @@ public class ConduitBridge extends DemandForwardingBridge {
super(configuration, localBroker, remoteBroker); super(configuration, localBroker, remoteBroker);
} }
@Override
protected DemandSubscription createDemandSubscription(ConsumerInfo info) throws IOException { protected DemandSubscription createDemandSubscription(ConsumerInfo info) throws IOException {
if (addToAlreadyInterestedConsumers(info)) { if (addToAlreadyInterestedConsumers(info)) {
return null; // don't want this subscription added return null; // don't want this subscription added
@ -94,7 +94,9 @@ public class ConduitBridge extends DemandForwardingBridge {
return matched; return matched;
} }
@Override
protected void removeDemandSubscription(ConsumerId id) throws IOException { protected void removeDemandSubscription(ConsumerId id) throws IOException {
super.removeDemandSubscription(id);
List<DemandSubscription> tmpList = new ArrayList<DemandSubscription>(); List<DemandSubscription> tmpList = new ArrayList<DemandSubscription>();
for (Iterator i = subscriptionMapByLocalId.values().iterator(); i.hasNext();) { for (Iterator i = subscriptionMapByLocalId.values().iterator(); i.hasNext();) {

12
activemq-core/src/test/java/org/apache/activemq/usecases/ThreeBrokerQueueNetworkTest.java
View File

@ -23,10 +23,8 @@ import java.util.Iterator;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.Destination; import javax.jms.Destination;
import javax.jms.MessageConsumer; import javax.jms.MessageConsumer;
import org.apache.activemq.JmsMultipleBrokersTestSupport; import org.apache.activemq.JmsMultipleBrokersTestSupport;
import org.apache.activemq.broker.Broker; import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerFilter; import org.apache.activemq.broker.BrokerFilter;
@ -367,7 +365,14 @@ public class ThreeBrokerQueueNetworkTest extends JmsMultipleBrokersTestSupport {
assertEquals(messageCount, msgsA.getMessageCount()); assertEquals(messageCount, msgsA.getMessageCount());
} }
public void testMigrateConsumerStuckMessages() throws Exception { /*
* This test is disabled - as it fails with a fix for
* http://issues.apache.org/activemq/browse/AMQ-2530 - which highlights that
* For a Conduit bridge - local subscription Ids weren't removed in a ConduitBridge
* The test fails because on closing clientA - clientB correctly receives all the
* messages - ie. half dont get stuck on BrokerA -
*/
public void XtestMigrateConsumerStuckMessages() throws Exception {
boolean suppressQueueDuplicateSubscriptions = false; boolean suppressQueueDuplicateSubscriptions = false;
bridgeAllBrokers("default", 3, suppressQueueDuplicateSubscriptions); bridgeAllBrokers("default", 3, suppressQueueDuplicateSubscriptions);
startAllBrokers(); startAllBrokers();
@ -626,6 +631,7 @@ public class ThreeBrokerQueueNetworkTest extends JmsMultipleBrokersTestSupport {
} }
} }
@Override
public void setUp() throws Exception { public void setUp() throws Exception {
super.setAutoFail(true); super.setAutoFail(true);
super.setUp(); super.setUp();