https://issues.apache.org/jira/browse/AMQ-3382 - statically included destination should be always forwarded

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1141124 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java

@@ -954,7 +954,18 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
             }
         }
 
-        ActiveMQDestination[] dests = excludedDestinations;
+        ActiveMQDestination[] dests = staticallyIncludedDestinations;
+        if (dests != null && dests.length > 0) {
+            for (int i = 0; i < dests.length; i++) {
+                ActiveMQDestination match = dests[i];
+                DestinationFilter inclusionFilter = DestinationFilter.parseFilter(match);
+                if (match != null && inclusionFilter.matches(destination) && dests[i].getDestinationType() == destination.getDestinationType()) {
+                    return true;
+                }
+            }
+        }
+
+        dests = excludedDestinations;
         if (dests != null && dests.length > 0) {
             for (int i = 0; i < dests.length; i++) {
                 ActiveMQDestination match = dests[i];

activemq-core/src/test/java/org/apache/activemq/usecases/TestBrokerConnectionDuplexExcludedDestinations.java

@@ -101,6 +101,12 @@ public class TestBrokerConnectionDuplexExcludedDestinations extends TestCase {
 		hubProducer.setDisableMessageID(true);
 		hubProducer.setDisableMessageTimestamp(true);
 
+		//create spoke producer
+		MessageProducer spokeProducer = hubSession.createProducer(null);
+		spokeProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+		spokeProducer.setDisableMessageID(true);
+		spokeProducer.setDisableMessageTimestamp(true);
+
 		Queue excludedQueueHub = hubSession.createQueue("exclude.test.foo");
 		TextMessage excludedMsgHub = hubSession.createTextMessage();
 		excludedMsgHub.setText(excludedQueueHub.toString());
@@ -108,11 +114,17 @@ public class TestBrokerConnectionDuplexExcludedDestinations extends TestCase {
 		Queue includedQueueHub = hubSession.createQueue("include.test.foo");
 
 		TextMessage includedMsgHub = hubSession.createTextMessage();
-		includedMsgHub.setText(includedQueueHub.toString());		
+		includedMsgHub.setText(includedQueueHub.toString());
+
+		Queue alwaysIncludedQueueHub = hubSession.createQueue("always.include.test.foo");
+
+		TextMessage alwaysIncludedMsgHub = hubSession.createTextMessage();
+		alwaysIncludedMsgHub.setText(alwaysIncludedQueueHub.toString());
 
 		// Sending from Hub queue
 		hubProducer.send(excludedQueueHub, excludedMsgHub);
 		hubProducer.send(includedQueueHub, includedMsgHub);
+        hubProducer.send(alwaysIncludedQueueHub, alwaysIncludedMsgHub);
 
 
 		Queue excludedQueueSpoke = spokeSession.createQueue("exclude.test.foo");
@@ -121,7 +133,21 @@ public class TestBrokerConnectionDuplexExcludedDestinations extends TestCase {
 		Thread.sleep(100);
 
 	    Queue includedQueueSpoke = spokeSession.createQueue("include.test.foo");
-		MessageConsumer includedConsumerSpoke = spokeSession.createConsumer(includedQueueSpoke);		
+		MessageConsumer includedConsumerSpoke = spokeSession.createConsumer(includedQueueSpoke);
+
+		Thread.sleep(100);
+
+	    Queue alwaysIncludedQueueSpoke = spokeSession.createQueue("always.include.test.foo");
+		MessageConsumer alwaysIncludedConsumerSpoke = spokeSession.createConsumer(alwaysIncludedQueueHub);
+
+        Thread.sleep(100);
+        TextMessage alwaysIncludedMsgSpoke = spokeSession.createTextMessage();
+		alwaysIncludedMsgSpoke.setText(alwaysIncludedQueueSpoke.toString());
+        spokeProducer.send(alwaysIncludedQueueSpoke, alwaysIncludedMsgSpoke);
+
+
+		MessageConsumer alwaysIncludedConsumerHub = spokeSession.createConsumer(alwaysIncludedQueueHub);
+
 		
 		// Receiving from excluded Spoke queue
 		Message msg = excludedConsumerSpoke.receive(200);
@@ -130,6 +156,10 @@ public class TestBrokerConnectionDuplexExcludedDestinations extends TestCase {
 		// Receiving from included Spoke queue
 		msg = includedConsumerSpoke.receive(200);
 		assertEquals(includedMsgHub, msg);
+
+		// Receiving from included Spoke queue
+		msg = alwaysIncludedConsumerSpoke.receive(200);
+		assertEquals(alwaysIncludedMsgHub, msg);
 		
 		// we should be able to receive excluded queue message on Hub
 		MessageConsumer excludedConsumerHub = hubSession.createConsumer(excludedQueueHub);

activemq-core/src/test/resources/org/apache/activemq/usecases/sender-duplex.xml

@@ -40,8 +40,8 @@
       		<topic physicalName="exclude.test.bar"/>
       	</excludedDestinations>
         <staticallyIncludedDestinations>
-      		<queue physicalName="always.include.queue"/>
-      		<topic physicalName="always.include.topic"/>
+      		<queue physicalName="always.include.test.foo"/>
+      		<topic physicalName="always.include.test.bar"/>
       	</staticallyIncludedDestinations>
 
       </networkConnector>