https://issues.apache.org/activemq/browse/AMQ-2993 - virtual topics and advisory messages

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

activemq-core/src/main/java/org/apache/activemq/broker/region/virtual/VirtualTopicInterceptor.java

@@ -41,8 +41,10 @@ public class VirtualTopicInterceptor extends DestinationFilter {
     }
 
     public void send(ProducerBrokerExchange context, Message message) throws Exception {
+        if (!message.isAdvisory()) {
             ActiveMQDestination queueConsumers = getQueueConsumersWildcard(message.getDestination());
             send(context, message, queueConsumers);
+        }
         super.send(context, message);
     }
 

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

@@ -28,21 +28,26 @@ import org.apache.activemq.broker.region.QueueSubscription;
 import org.apache.activemq.util.MessageIdList;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.core.io.ClassPathResource;
 
 public class BrowseOverNetworkTest extends JmsMultipleBrokersTestSupport {
     private static final Log LOG = LogFactory.getLog(QueueSubscription.class);
     protected static final int MESSAGE_COUNT = 10;
 
     public void testBrowse() throws Exception {
+        createBroker(new URI("broker:(tcp://localhost:61617)/BrokerB?persistent=false&useJmx=false"));
+        createBroker(new URI("broker:(tcp://localhost:61616)/BrokerA?persistent=false&useJmx=false"));
+
         bridgeBrokers("BrokerA", "BrokerB");
 
+
         startAllBrokers();
 
         Destination dest = createDestination("TEST.FOO", false);
 
         sendMessages("BrokerA", dest, MESSAGE_COUNT);
 
-        browseMessages(dest);
+        browseMessages("BrokerB", dest);
 
         Thread.sleep(2000);
 
@@ -61,23 +66,40 @@ public class BrowseOverNetworkTest extends JmsMultipleBrokersTestSupport {
                 + msgsB.getMessageCount());
     }
 
-    protected void browseMessages(Destination dest) throws Exception {
+    public void testconsumerInfo() throws Exception {
-        QueueBrowser browser = createBrowser("BrokerB", dest);
+        createBroker(new ClassPathResource("org/apache/activemq/usecases/browse-broker1.xml"));
+        createBroker(new ClassPathResource("org/apache/activemq/usecases/browse-broker2.xml"));
+
+        startAllBrokers();
+
+        brokers.get("broker1").broker.waitUntilStarted();
+
+        
+        Destination dest = createDestination("QUEUE.A,QUEUE.B", false);
+
+
+        int broker1 = browseMessages("broker1", dest);
+        assertEquals("Browsed a message on an empty queue", 0, broker1);
+        Thread.sleep(1000);
+        int broker2 = browseMessages("broker2", dest);
+        assertEquals("Browsed a message on an empty queue", 0, broker2);
+
+    }
+
+    protected int browseMessages(String broker, Destination dest) throws Exception {
+        QueueBrowser browser = createBrowser(broker, dest);
         Enumeration msgs = browser.getEnumeration();
         int browsedMessage = 0;
         while (msgs.hasMoreElements()) {
             browsedMessage++;
             msgs.nextElement();
         }
+        return browsedMessage;
     }
 
     public void setUp() throws Exception {
         super.setAutoFail(true);
         super.setUp();
-        createBroker(new URI(
-                "broker:(tcp://localhost:61616)/BrokerA?persistent=false&useJmx=false"));
-        createBroker(new URI(
-                "broker:(tcp://localhost:61617)/BrokerB?persistent=false&useJmx=false"));
     }
 
 }

activemq-core/src/test/resources/org/apache/activemq/usecases/browse-broker1.xml

@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+    Licensed to the Apache Software Foundation (ASF) under one or more
+    contributor license agreements.  See the NOTICE file distributed with
+    this work for additional information regarding copyright ownership.
+    The ASF licenses this file to You under the Apache License, Version 2.0
+    (the "License"); you may not use this file except in compliance with
+    the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+-->        
+<beans 
+  xmlns="http://www.springframework.org/schema/beans" 
+  xmlns:amq="http://activemq.apache.org/schema/core"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
+           http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
+           http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
+  
+  <!-- Broker1 ?useQueueForAccept=false -->
+  <amq:broker brokerName="broker1" id="broker1" useJmx="true" 
+  			  persistent="true" start="false" advisorySupport="true">
+ 
+ 
+   <amq:destinationInterceptors>
+            <amq:virtualDestinationInterceptor>
+              <amq:virtualDestinations>
+                <amq:virtualTopic name=">" prefix="VTopic.*." />
+              </amq:virtualDestinations>
+           </amq:virtualDestinationInterceptor>
+    </amq:destinationInterceptors>
+
+    <amq:networkConnectors>
+      
+      <amq:networkConnector uri="static:(tcp://localhost:61617)" name="broker1_broker2">
+        
+        <amq:excludedDestinations>
+          <amq:queue physicalName="QUEUE.A"/>
+        </amq:excludedDestinations>
+        <amq:staticallyIncludedDestinations>
+          <amq:queue physicalName="QUEUE.B"/>
+        </amq:staticallyIncludedDestinations>
+      </amq:networkConnector>
+ 
+    </amq:networkConnectors>
+
+    <amq:transportConnectors>
+      <amq:transportConnector uri="tcp://localhost:61616" />
+    </amq:transportConnectors>
+
+  </amq:broker>
+  
+
+</beans>

activemq-core/src/test/resources/org/apache/activemq/usecases/browse-broker2.xml

@@ -0,0 +1,52 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+    Licensed to the Apache Software Foundation (ASF) under one or more
+    contributor license agreements.  See the NOTICE file distributed with
+    this work for additional information regarding copyright ownership.
+    The ASF licenses this file to You under the Apache License, Version 2.0
+    (the "License"); you may not use this file except in compliance with
+    the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+-->        
+<beans 
+  xmlns="http://www.springframework.org/schema/beans" 
+  xmlns:amq="http://activemq.apache.org/schema/core"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
+           http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
+           http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
+
+    
+  <!-- Broker2 (lonb) -->
+  <amq:broker brokerName="broker2" id="broker2" useJmx="true" 
+  				persistent="true" start="false" advisorySupport="true">
+
+    <!--  Network connectors -->
+    <amq:networkConnectors>
+      
+      <amq:networkConnector uri="static:(tcp://localhost:61616)" name="broker2_broker1">
+       
+        <amq:excludedDestinations>
+          <amq:queue physicalName="QUEUE.B"/>
+        </amq:excludedDestinations>
+        <amq:staticallyIncludedDestinations>
+          <amq:queue physicalName="QUEUE.A"/>
+        </amq:staticallyIncludedDestinations>
+      </amq:networkConnector>
+ 
+    </amq:networkConnectors>
+
+    <amq:transportConnectors>
+      <amq:transportConnector uri="tcp://localhost:61617" />
+    </amq:transportConnectors>
+        
+  </amq:broker>
+  
+  </beans>