activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dej...@apache.org
Subject svn commit: r1214700 - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/command/NetworkBridgeFilter.java test/java/org/apache/activemq/network/SimpleNetworkTest.java
Date Thu, 15 Dec 2011 11:10:07 GMT
Author: dejanb
Date: Thu Dec 15 11:10:07 2011
New Revision: 1214700

URL: http://svn.apache.org/viewvc?rev=1214700&view=rev
Log:
fix for DuplexNetworkTest failure

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/command/NetworkBridgeFilter.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/command/NetworkBridgeFilter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/command/NetworkBridgeFilter.java?rev=1214700&r1=1214699&r2=1214700&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/command/NetworkBridgeFilter.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/command/NetworkBridgeFilter.java
Thu Dec 15 11:10:07 2011
@@ -16,17 +16,16 @@
  */
 package org.apache.activemq.command;
 
-import java.io.IOException;
-import java.util.Arrays;
-
-import javax.jms.JMSException;
-
 import org.apache.activemq.filter.BooleanExpression;
 import org.apache.activemq.filter.MessageEvaluationContext;
 import org.apache.activemq.util.JMSExceptionSupport;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Arrays;
+
 /**
  * @openwire:marshaller code="91"
  * 
@@ -106,6 +105,7 @@ public class NetworkBridgeFilter impleme
                 LOG.trace("ConsumerInfo advisory all ready routed once through target broker
("
                         + networkBrokerId + "), path: "
                         + Arrays.toString(info.getBrokerPath()) + " - ignoring: " + message);
+                return false;
             }
         }
         return true;

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java?rev=1214700&r1=1214699&r2=1214700&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
(original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
Thu Dec 15 11:10:07 2011
@@ -16,22 +16,9 @@
  */
 package org.apache.activemq.network;
 
-import java.net.URI;
-import java.util.concurrent.ConcurrentHashMap;
-import javax.jms.Connection;
-import javax.jms.DeliveryMode;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageListener;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.TextMessage;
-import javax.jms.TopicRequestor;
-import javax.jms.TopicSession;
 import org.apache.activemq.ActiveMQConnectionFactory;
 import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQTopic;
 import org.apache.activemq.command.ConsumerId;
 import org.apache.activemq.util.Wait;
@@ -42,6 +29,11 @@ import org.springframework.context.suppo
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.Resource;
 
+import javax.jms.*;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.concurrent.ConcurrentHashMap;
+
 public class SimpleNetworkTest extends org.apache.activemq.TestSupport {
 
     protected static final int MESSAGE_COUNT = 10;
@@ -89,7 +81,6 @@ public class SimpleNetworkTest extends o
     }
 
     public void testFiltering() throws Exception {
-
         MessageConsumer includedConsumer = remoteSession.createConsumer(included);
         MessageConsumer excludedConsumer = remoteSession.createConsumer(excluded);
         MessageProducer includedProducer = localSession.createProducer(included);
@@ -109,7 +100,7 @@ public class SimpleNetworkTest extends o
         MessageProducer producer = localSession.createProducer(included);
         producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
 
-        waitForConsumerRegistration(localBroker, 2);
+        waitForConsumerRegistration(localBroker, 2, included);
 
         for (int i = 0; i < MESSAGE_COUNT; i++) {
             Message test = localSession.createTextMessage("test-" + i);
@@ -122,20 +113,23 @@ public class SimpleNetworkTest extends o
         assertNull(consumer2.receive(1000));
     }
 
-    private void waitForConsumerRegistration(final BrokerService brokerService, final int
min) throws Exception {
+    private void waitForConsumerRegistration(final BrokerService brokerService, final int
min, final ActiveMQDestination destination) throws Exception {
         assertTrue("Internal bridge consumers registered in time", Wait.waitFor(new Wait.Condition()
{
             @Override
             public boolean isSatisified() throws Exception {
                 Object[] bridges = brokerService.getNetworkConnectors().get(0).bridges.values().toArray();
                 if (bridges.length > 0) {
-                    LOG.info(brokerService + " bridges "  + bridges);
+                    LOG.info(brokerService + " bridges "  + Arrays.toString(bridges));
                     DemandForwardingBridgeSupport demandForwardingBridgeSupport = (DemandForwardingBridgeSupport)
bridges[0];
                     ConcurrentHashMap<ConsumerId, DemandSubscription> forwardingBridges
= demandForwardingBridgeSupport.getLocalSubscriptionMap();
                     LOG.info(brokerService + " bridge "  + demandForwardingBridgeSupport
+ ", localSubs: " + forwardingBridges);
                     if (!forwardingBridges.isEmpty()) {
-                        DemandSubscription demandSubscription = (DemandSubscription) forwardingBridges.values().toArray()[0];
-                        LOG.info(brokerService + " DemandSubscription "  + demandSubscription
+ ", size: " + demandSubscription.size());
-                        return demandSubscription.size() >= min;
+                        for (DemandSubscription demandSubscription : forwardingBridges.values())
{
+                            if (demandSubscription.getLocalInfo().getDestination().equals(destination))
{
+                                LOG.info(brokerService + " DemandSubscription "  + demandSubscription
+ ", size: " + demandSubscription.size());
+                                return demandSubscription.size() >= min;
+                            }
+                        }
                     }
                 }
                 return false;
@@ -220,8 +214,10 @@ public class SimpleNetworkTest extends o
     protected void doSetUp() throws Exception {
         remoteBroker = createRemoteBroker();
         remoteBroker.start();
+        remoteBroker.waitUntilStarted();
         localBroker = createLocalBroker();
         localBroker.start();
+        localBroker.waitUntilStarted();
         URI localURI = localBroker.getVmConnectorURI();
         ActiveMQConnectionFactory fac = new ActiveMQConnectionFactory(localURI);
         fac.setAlwaysSyncSend(true);



Mime
View raw message