activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tab...@apache.org
Subject [06/14] activemq git commit: [AMQ-6778] fix and test - no longer gate firing advisory on broker start flag b/c consumers may be present once transport connectors have started
Date Thu, 07 Sep 2017 16:17:42 GMT
[AMQ-6778] fix and test - no longer gate firing advisory on broker start flag b/c consumers
may be present once transport connectors have started

(cherry picked from commit ddfa96028c4c803467b0c074b8c1b4d3e5f9687d)


Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/1bba73dc
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/1bba73dc
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/1bba73dc

Branch: refs/heads/activemq-5.15.x
Commit: 1bba73dcb65c0320dc21b606bccc28f21ad0752f
Parents: a023c7a
Author: gtully <gary.tully@gmail.com>
Authored: Tue Jul 25 13:02:01 2017 +0100
Committer: Timothy Bish <tabish121@gmail.com>
Committed: Thu Sep 7 12:10:55 2017 -0400

----------------------------------------------------------------------
 .../activemq/advisory/AdvisoryBroker.java       |  72 +++++++------
 .../advisory/AdvisoryDuringStartTest.java       | 105 +++++++++++++++++++
 2 files changed, 140 insertions(+), 37 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/activemq/blob/1bba73dc/activemq-broker/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
b/activemq-broker/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
index 4118dbe..1acd524 100644
--- a/activemq-broker/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
@@ -832,43 +832,41 @@ public class AdvisoryBroker extends BrokerFilter {
     }
 
     public void fireAdvisory(ConnectionContext context, ActiveMQTopic topic, Command command,
ConsumerId targetConsumerId, ActiveMQMessage advisoryMessage) throws Exception {
-        if (getBrokerService().isStarted()) {
-            //set properties
-            advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_NAME,
getBrokerName());
-            String id = getBrokerId() != null ? getBrokerId().getValue() : "NOT_SET";
-            advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID,
id);
-
-            String url = getBrokerService().getVmConnectorURI().toString();
-            //try and find the URL on the transport connector and use if it exists else
-            //try and find a default URL
-            if (context.getConnector() instanceof TransportConnector
-                    && ((TransportConnector) context.getConnector()).getPublishableConnectString()
!= null) {
-                url = ((TransportConnector) context.getConnector()).getPublishableConnectString();
-            } else if (getBrokerService().getDefaultSocketURIString() != null) {
-                url = getBrokerService().getDefaultSocketURIString();
-            }
-            advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_URL,
url);
-
-            //set the data structure
-            advisoryMessage.setDataStructure(command);
-            advisoryMessage.setPersistent(false);
-            advisoryMessage.setType(AdvisorySupport.ADIVSORY_MESSAGE_TYPE);
-            advisoryMessage.setMessageId(new MessageId(advisoryProducerId, messageIdGenerator.getNextSequenceId()));
-            advisoryMessage.setTargetConsumerId(targetConsumerId);
-            advisoryMessage.setDestination(topic);
-            advisoryMessage.setResponseRequired(false);
-            advisoryMessage.setProducerId(advisoryProducerId);
-            boolean originalFlowControl = context.isProducerFlowControl();
-            final ProducerBrokerExchange producerExchange = new ProducerBrokerExchange();
-            producerExchange.setConnectionContext(context);
-            producerExchange.setMutable(true);
-            producerExchange.setProducerState(new ProducerState(new ProducerInfo()));
-            try {
-                context.setProducerFlowControl(false);
-                next.send(producerExchange, advisoryMessage);
-            } finally {
-                context.setProducerFlowControl(originalFlowControl);
-            }
+        //set properties
+        advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_NAME,
getBrokerName());
+        String id = getBrokerId() != null ? getBrokerId().getValue() : "NOT_SET";
+        advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_ID,
id);
+
+        String url = getBrokerService().getVmConnectorURI().toString();
+        //try and find the URL on the transport connector and use if it exists else
+        //try and find a default URL
+        if (context.getConnector() instanceof TransportConnector
+                && ((TransportConnector) context.getConnector()).getPublishableConnectString()
!= null) {
+            url = ((TransportConnector) context.getConnector()).getPublishableConnectString();
+        } else if (getBrokerService().getDefaultSocketURIString() != null) {
+            url = getBrokerService().getDefaultSocketURIString();
+        }
+        advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_ORIGIN_BROKER_URL,
url);
+
+        //set the data structure
+        advisoryMessage.setDataStructure(command);
+        advisoryMessage.setPersistent(false);
+        advisoryMessage.setType(AdvisorySupport.ADIVSORY_MESSAGE_TYPE);
+        advisoryMessage.setMessageId(new MessageId(advisoryProducerId, messageIdGenerator.getNextSequenceId()));
+        advisoryMessage.setTargetConsumerId(targetConsumerId);
+        advisoryMessage.setDestination(topic);
+        advisoryMessage.setResponseRequired(false);
+        advisoryMessage.setProducerId(advisoryProducerId);
+        boolean originalFlowControl = context.isProducerFlowControl();
+        final ProducerBrokerExchange producerExchange = new ProducerBrokerExchange();
+        producerExchange.setConnectionContext(context);
+        producerExchange.setMutable(true);
+        producerExchange.setProducerState(new ProducerState(new ProducerInfo()));
+        try {
+            context.setProducerFlowControl(false);
+            next.send(producerExchange, advisoryMessage);
+        } finally {
+            context.setProducerFlowControl(originalFlowControl);
         }
     }
 

http://git-wip-us.apache.org/repos/asf/activemq/blob/1bba73dc/activemq-unit-tests/src/test/java/org/apache/activemq/broker/advisory/AdvisoryDuringStartTest.java
----------------------------------------------------------------------
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/advisory/AdvisoryDuringStartTest.java
b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/advisory/AdvisoryDuringStartTest.java
new file mode 100644
index 0000000..503a9b8
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/advisory/AdvisoryDuringStartTest.java
@@ -0,0 +1,105 @@
+/**
+ * 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.
+ */
+package org.apache.activemq.broker.advisory;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.command.ConsumerInfo;
+import org.apache.activemq.network.DiscoveryNetworkConnector;
+import org.apache.activemq.util.ServiceStopper;
+import org.junit.After;
+import org.junit.Test;
+
+import javax.jms.Connection;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.*;
+
+public class AdvisoryDuringStartTest {
+
+    BrokerService brokerService;
+
+    @After
+    public void stopBroker() throws Exception {
+        if (brokerService != null) {
+            brokerService.stop();
+        }
+    }
+
+    @Test
+    public void testConsumerAdvisoryDuringSlowStart() throws Exception {
+
+        brokerService = new BrokerService();
+        brokerService.setPersistent(false);
+        brokerService.setUseJmx(false);
+        brokerService.addConnector("tcp://localhost:0");
+
+        final CountDownLatch resumeStart = new CountDownLatch(1);
+        brokerService.addNetworkConnector(new DiscoveryNetworkConnector() {
+            @Override
+            protected void handleStart() throws Exception {
+                // delay broker started flag
+                resumeStart.await(5, TimeUnit.SECONDS);
+            }
+
+            @Override
+            protected void handleStop(ServiceStopper s) throws Exception {}
+        });
+        Executors.newCachedThreadPool().submit(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    brokerService.start();
+                } catch (Exception e) {
+                    e.printStackTrace();
+                    fail("error on start: " + e.toString());
+                }
+            }
+        });
+        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:("
+ brokerService.getTransportConnectorByScheme("tcp").getPublishableConnectString() + ")");
+        Connection advisoryConnection = connectionFactory.createConnection();
+        advisoryConnection.start();
+        Session advisorySession = advisoryConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageConsumer advisoryConsumer = advisorySession.createConsumer(advisorySession.createTopic("ActiveMQ.Advisory.Consumer.>"));
+
+        Connection consumerConnection = connectionFactory.createConnection();
+        Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        consumerConnection.start();
+        ActiveMQTopic dest = new ActiveMQTopic("SomeTopic");
+
+        // real consumer
+        consumerSession.createConsumer(dest);
+
+        resumeStart.countDown();
+        
+
+        ActiveMQMessage advisory = (ActiveMQMessage)advisoryConsumer.receive(4000);
+        assertNotNull(advisory);
+        assertTrue(advisory.getDataStructure() instanceof ConsumerInfo);
+        assertTrue(((ConsumerInfo)advisory.getDataStructure()).getDestination().equals(dest));
+        advisoryConnection.close();
+
+        consumerConnection.close();
+    }
+
+}


Mime
View raw message