You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ta...@apache.org on 2015/05/15 22:02:07 UTC
activemq git commit: https://issues.apache.org/jira/browse/AMQ-5441
Repository: activemq
Updated Branches:
refs/heads/master 3e599a595 -> fd2805fe1
https://issues.apache.org/jira/browse/AMQ-5441
Don't dip into the store to recover durable topic subscriptions or queue
subscriptions depending on the subscription strategy in use. The needed
information is always kept in either TopicRegion or QueueRegion even
after restart as all stored destinations are reloaded as the broker is
started.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/fd2805fe
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/fd2805fe
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/fd2805fe
Branch: refs/heads/master
Commit: fd2805fe1cc5950302d7230dc0fb23d33a9fedc9
Parents: 3e599a5
Author: Timothy Bish <ta...@gmail.com>
Authored: Fri May 15 16:01:54 2015 -0400
Committer: Timothy Bish <ta...@gmail.com>
Committed: Fri May 15 16:01:54 2015 -0400
----------------------------------------------------------------------
.../activemq/broker/region/TopicRegion.java | 13 ++
.../AbstractMQTTSubscriptionStrategy.java | 2 +-
.../MQTTDefaultSubscriptionStrategy.java | 42 +++-
.../MQTTVirtualTopicSubscriptionStrategy.java | 44 ++--
.../mqtt/MQTTCompositeQueueRetainedTest.java | 6 +-
.../mqtt/MQTTSubscriptionRecoveryTest.java | 208 +++++++++++++++++++
.../activemq/transport/mqtt/MQTTTest.java | 3 +-
.../transport/mqtt/MQTTTestSupport.java | 80 ++++---
8 files changed, 340 insertions(+), 58 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/fd2805fe/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
----------------------------------------------------------------------
diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
index 8fe6a6c..684ac88 100755
--- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
+++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/TopicRegion.java
@@ -16,6 +16,7 @@
*/
package org.apache.activemq.broker.region;
+import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@@ -395,6 +396,18 @@ public class TopicRegion extends AbstractRegion {
return null;
}
+ public List<DurableTopicSubscription> lookupSubscriptions(String clientId) {
+ List<DurableTopicSubscription> result = new ArrayList<DurableTopicSubscription>();
+
+ for (Map.Entry<SubscriptionKey, DurableTopicSubscription> subscriptionEntry : durableSubscriptions.entrySet()) {
+ if (subscriptionEntry.getKey().getClientId().equals(clientId)) {
+ result.add(subscriptionEntry.getValue());
+ }
+ }
+
+ return result;
+ }
+
public boolean isKeepDurableSubsActive() {
return keepDurableSubsActive;
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/fd2805fe/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/AbstractMQTTSubscriptionStrategy.java
----------------------------------------------------------------------
diff --git a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/AbstractMQTTSubscriptionStrategy.java b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/AbstractMQTTSubscriptionStrategy.java
index 883bbfb..bb84192 100644
--- a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/AbstractMQTTSubscriptionStrategy.java
+++ b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/AbstractMQTTSubscriptionStrategy.java
@@ -207,7 +207,7 @@ public abstract class AbstractMQTTSubscriptionStrategy implements MQTTSubscripti
LOG.warn("Error subscribing to {}", topicName, throwable);
// version 3.1 don't supports silent fail
// version 3.1.1 send "error" qos
- if (protocol.version == protocol.V3_1_1) {
+ if (protocol.version == MQTTProtocolConverter.V3_1_1) {
qos[0] = SUBSCRIBE_ERROR;
} else {
qos[0] = (byte) qoS.ordinal();
http://git-wip-us.apache.org/repos/asf/activemq/blob/fd2805fe/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/MQTTDefaultSubscriptionStrategy.java
----------------------------------------------------------------------
diff --git a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/MQTTDefaultSubscriptionStrategy.java b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/MQTTDefaultSubscriptionStrategy.java
index 61619d2..1b3ac5d 100644
--- a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/MQTTDefaultSubscriptionStrategy.java
+++ b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/MQTTDefaultSubscriptionStrategy.java
@@ -17,19 +17,22 @@
package org.apache.activemq.transport.mqtt.strategy;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import org.apache.activemq.ActiveMQPrefetchPolicy;
+import org.apache.activemq.broker.region.DurableTopicSubscription;
+import org.apache.activemq.broker.region.RegionBroker;
+import org.apache.activemq.broker.region.TopicRegion;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.RemoveSubscriptionInfo;
import org.apache.activemq.command.Response;
import org.apache.activemq.command.SubscriptionInfo;
-import org.apache.activemq.store.PersistenceAdapterSupport;
import org.apache.activemq.transport.mqtt.MQTTProtocolConverter;
import org.apache.activemq.transport.mqtt.MQTTProtocolException;
import org.apache.activemq.transport.mqtt.MQTTProtocolSupport;
@@ -52,12 +55,8 @@ public class MQTTDefaultSubscriptionStrategy extends AbstractMQTTSubscriptionStr
@Override
public void onConnect(CONNECT connect) throws MQTTProtocolException {
- List<SubscriptionInfo> subs;
- try {
- subs = PersistenceAdapterSupport.listSubscriptions(brokerService.getPersistenceAdapter(), protocol.getClientId());
- } catch (IOException e) {
- throw new MQTTProtocolException("Error loading store subscriptions", true, e);
- }
+ List<SubscriptionInfo> subs = lookupSubscription(protocol.getClientId());
+
if (connect.cleanSession()) {
deleteDurableSubs(subs);
} else {
@@ -159,4 +158,33 @@ public class MQTTDefaultSubscriptionStrategy extends AbstractMQTTSubscriptionStr
LOG.warn("Could not restore the MQTT durable subs.", e);
}
}
+
+ List<SubscriptionInfo> lookupSubscription(String clientId) throws MQTTProtocolException {
+ List<SubscriptionInfo> result = new ArrayList<SubscriptionInfo>();
+ RegionBroker regionBroker;
+
+ try {
+ regionBroker = (RegionBroker) brokerService.getBroker().getAdaptor(RegionBroker.class);
+ } catch (Exception e) {
+ throw new MQTTProtocolException("Error recovering durable subscriptions: " + e.getMessage(), false, e);
+ }
+
+ final TopicRegion topicRegion = (TopicRegion) regionBroker.getTopicRegion();
+ List<DurableTopicSubscription> subscriptions = topicRegion.lookupSubscriptions(clientId);
+ if (subscriptions != null) {
+ for (DurableTopicSubscription subscription : subscriptions) {
+ LOG.debug("Recovered durable sub:{} on connect", subscription);
+
+ SubscriptionInfo info = new SubscriptionInfo();
+
+ info.setDestination(subscription.getActiveMQDestination());
+ info.setSubcriptionName(subscription.getSubscriptionKey().getSubscriptionName());
+ info.setClientId(clientId);
+
+ result.add(info);
+ }
+ }
+
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/fd2805fe/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/MQTTVirtualTopicSubscriptionStrategy.java
----------------------------------------------------------------------
diff --git a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/MQTTVirtualTopicSubscriptionStrategy.java b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/MQTTVirtualTopicSubscriptionStrategy.java
index 99917c7..d0735e1 100644
--- a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/MQTTVirtualTopicSubscriptionStrategy.java
+++ b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/strategy/MQTTVirtualTopicSubscriptionStrategy.java
@@ -20,6 +20,7 @@ import static org.apache.activemq.transport.mqtt.MQTTProtocolSupport.convertActi
import static org.apache.activemq.transport.mqtt.MQTTProtocolSupport.convertMQTTToActiveMQ;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
@@ -27,13 +28,14 @@ import java.util.Set;
import java.util.StringTokenizer;
import org.apache.activemq.ActiveMQPrefetchPolicy;
+import org.apache.activemq.broker.region.QueueRegion;
+import org.apache.activemq.broker.region.RegionBroker;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.DestinationInfo;
import org.apache.activemq.command.Response;
-import org.apache.activemq.store.PersistenceAdapterSupport;
import org.apache.activemq.transport.mqtt.MQTTProtocolConverter;
import org.apache.activemq.transport.mqtt.MQTTProtocolException;
import org.apache.activemq.transport.mqtt.MQTTSubscription;
@@ -59,22 +61,7 @@ public class MQTTVirtualTopicSubscriptionStrategy extends AbstractMQTTSubscripti
@Override
public void onConnect(CONNECT connect) throws MQTTProtocolException {
- List<ActiveMQQueue> queues;
- try {
- queues = PersistenceAdapterSupport.listQueues(brokerService.getPersistenceAdapter(), new PersistenceAdapterSupport.DestinationMatcher() {
-
- @Override
- public boolean matches(ActiveMQDestination destination) {
- if (destination.getPhysicalName().startsWith("Consumer." + protocol.getClientId())) {
- LOG.debug("Recovered client sub: {}", destination.getPhysicalName());
- return true;
- }
- return false;
- }
- });
- } catch (IOException e) {
- throw new MQTTProtocolException("Error restoring durable subscriptions", true, e);
- }
+ List<ActiveMQQueue> queues = lookupQueues(protocol.getClientId());
if (connect.cleanSession()) {
deleteDurableQueues(queues);
@@ -232,4 +219,27 @@ public class MQTTVirtualTopicSubscriptionStrategy extends AbstractMQTTSubscripti
LOG.warn("Could not restore the MQTT durable subs.", e);
}
}
+
+ List<ActiveMQQueue> lookupQueues(String clientId) throws MQTTProtocolException {
+ List<ActiveMQQueue> result = new ArrayList<ActiveMQQueue>();
+ RegionBroker regionBroker;
+
+ try {
+ regionBroker = (RegionBroker) brokerService.getBroker().getAdaptor(RegionBroker.class);
+ } catch (Exception e) {
+ throw new MQTTProtocolException("Error recovering queues: " + e.getMessage(), false, e);
+ }
+
+ final QueueRegion queueRegion = (QueueRegion) regionBroker.getQueueRegion();
+ for (ActiveMQDestination destination : queueRegion.getDestinationMap().keySet()) {
+ if (destination.isQueue() && !destination.isTemporary()) {
+ if (destination.getPhysicalName().startsWith("Consumer." + clientId)) {
+ LOG.debug("Recovered client sub: {} on connect", destination.getPhysicalName());
+ result.add((ActiveMQQueue) destination);
+ }
+ }
+ }
+
+ return result;
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/fd2805fe/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTCompositeQueueRetainedTest.java
----------------------------------------------------------------------
diff --git a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTCompositeQueueRetainedTest.java b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTCompositeQueueRetainedTest.java
index d195ed0..8ca6318 100644
--- a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTCompositeQueueRetainedTest.java
+++ b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTCompositeQueueRetainedTest.java
@@ -55,8 +55,8 @@ public class MQTTCompositeQueueRetainedTest extends MQTTTestSupport {
private static final int NUM_MESSAGES = 25;
@Override
- protected void createBroker() throws Exception {
- brokerService = new BrokerService();
+ protected BrokerService createBroker(boolean deleteAllOnStart) throws Exception {
+ BrokerService brokerService = new BrokerService();
brokerService.setPersistent(isPersistent());
brokerService.setAdvisorySupport(false);
brokerService.setSchedulerSupport(isSchedulerSupportEnabled());
@@ -75,6 +75,8 @@ public class MQTTCompositeQueueRetainedTest extends MQTTTestSupport {
final VirtualDestinationInterceptor destinationInterceptor = new VirtualDestinationInterceptor();
destinationInterceptor.setVirtualDestinations(new VirtualDestination[] {compositeTopic} );
brokerService.setDestinationInterceptors(new DestinationInterceptor[] { destinationInterceptor });
+
+ return brokerService;
}
@Test(timeout = 60 * 1000)
http://git-wip-us.apache.org/repos/asf/activemq/blob/fd2805fe/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTSubscriptionRecoveryTest.java
----------------------------------------------------------------------
diff --git a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTSubscriptionRecoveryTest.java b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTSubscriptionRecoveryTest.java
new file mode 100644
index 0000000..61f74cc
--- /dev/null
+++ b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTSubscriptionRecoveryTest.java
@@ -0,0 +1,208 @@
+/**
+ * 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.transport.mqtt;
+
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.activemq.util.Wait;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test that all previous QoS 2 subscriptions are recovered on Broker restart.
+ */
+@RunWith(Parameterized.class)
+public class MQTTSubscriptionRecoveryTest extends MQTTTestSupport {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MQTTSubscriptionRecoveryTest.class);
+
+ protected boolean defaultStrategy = false;
+
+ @Parameters(name="{0}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {
+ { "mqtt-virtual-topic-subscriptions", false},
+ { "mqtt-default-subscriptions", true} });
+ }
+
+ public MQTTSubscriptionRecoveryTest(String subscriptionStrategy, boolean defaultStrategy) {
+ this.defaultStrategy = defaultStrategy;
+ }
+
+ @Override
+ public boolean isPersistent() {
+ return true;
+ }
+
+ @Override
+ public String getProtocolConfig() {
+ if (defaultStrategy) {
+ return "transport.subscriptionStrategy=mqtt-default-subscriptions";
+ } else {
+ return "transport.subscriptionStrategy=mqtt-virtual-topic-subscriptions";
+ }
+ }
+
+ @Test
+ public void testDurableSubscriptionsAreRecovered() throws Exception {
+
+ MqttClient connection = createClient(getName());
+
+ final String[] topics = { "TopicA/", "TopicB/", "TopicC/" };
+ for (int i = 0; i < topics.length; i++) {
+ LOG.debug("Subscribing to Topic:{}", topics[i]);
+ connection.subscribe(topics[i], EXACTLY_ONCE);
+ }
+
+ assertStatsForConnectedClient(topics.length);
+
+ disconnect(connection);
+
+ assertStatsForDisconnectedClient(topics.length);
+
+ restartBroker();
+
+ assertStatsForDisconnectedClient(topics.length);
+
+ connection = createClient(getName());
+
+ assertStatsForConnectedClient(topics.length);
+ }
+
+ private void assertStatsForConnectedClient(final int numDestinations) throws Exception {
+ if (defaultStrategy) {
+ assertTopicStatsForConnectedClient(numDestinations);
+ } else {
+ assertQueueStatsForConnectedClient(numDestinations);
+ }
+ }
+
+ private void assertStatsForDisconnectedClient(final int numDestinations) throws Exception {
+ if (defaultStrategy) {
+ assertTopicStatsForDisconnectedClient(numDestinations);
+ } else {
+ assertQueueStatsForDisconnectedClient(numDestinations);
+ }
+ }
+
+ //----- Assert implementations based on subscription strategy ------------//
+
+ private void assertQueueStatsForConnectedClient(final int numDestinations) throws Exception {
+ assertTrue(Wait.waitFor(new Wait.Condition() {
+
+ @Override
+ public boolean isSatisified() throws Exception {
+ return getProxyToBroker().getQueueSubscribers().length == numDestinations;
+ }
+ }));
+ }
+
+ private void assertQueueStatsForDisconnectedClient(final int numDestinations) throws Exception {
+ assertTrue(Wait.waitFor(new Wait.Condition() {
+
+ @Override
+ public boolean isSatisified() throws Exception {
+ return getProxyToBroker().getQueueSubscribers().length == 0;
+ }
+ }));
+ }
+
+ private void assertTopicStatsForConnectedClient(final int numDestinations) throws Exception {
+ assertTrue(Wait.waitFor(new Wait.Condition() {
+
+ @Override
+ public boolean isSatisified() throws Exception {
+ return getProxyToBroker().getDurableTopicSubscribers().length == numDestinations;
+ }
+ }));
+
+ assertTrue(Wait.waitFor(new Wait.Condition() {
+
+ @Override
+ public boolean isSatisified() throws Exception {
+ return getProxyToBroker().getInactiveDurableTopicSubscribers().length == 0;
+ }
+ }));
+ }
+
+ private void assertTopicStatsForDisconnectedClient(final int numDestinations) throws Exception {
+ assertTrue(Wait.waitFor(new Wait.Condition() {
+
+ @Override
+ public boolean isSatisified() throws Exception {
+ return getProxyToBroker().getDurableTopicSubscribers().length == 0;
+ }
+ }));
+
+ assertTrue(Wait.waitFor(new Wait.Condition() {
+
+ @Override
+ public boolean isSatisified() throws Exception {
+ return getProxyToBroker().getInactiveDurableTopicSubscribers().length == numDestinations;
+ }
+ }));
+ }
+
+ //----- Client Connect and Disconnect using Paho MQTT --------------------//
+
+ protected MqttClient createClient(String clientId) throws Exception {
+ return createClient(false, clientId, null);
+ }
+
+ protected MqttClient createClient(boolean cleanSession, String clientId, MqttCallback listener) throws Exception {
+ MqttConnectOptions options = new MqttConnectOptions();
+ options.setCleanSession(cleanSession);
+ options.setKeepAliveInterval(30);
+
+ final MqttClient client = new MqttClient("tcp://localhost:" + getPort(), clientId, new MemoryPersistence());
+ client.setCallback(listener);
+ client.connect(options);
+
+ Wait.waitFor(new Wait.Condition() {
+ @Override
+ public boolean isSatisified() throws Exception {
+ return client.isConnected();
+ }
+ }, TimeUnit.SECONDS.toMillis(15), TimeUnit.MILLISECONDS.toMillis(100));
+
+ return client;
+ }
+
+ protected void disconnect(final MqttClient client) throws Exception {
+ client.disconnect();
+ client.close();
+
+ Wait.waitFor(new Wait.Condition() {
+ @Override
+ public boolean isSatisified() throws Exception {
+ return !client.isConnected();
+ }
+ }, TimeUnit.SECONDS.toMillis(15), TimeUnit.MILLISECONDS.toMillis(100));
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq/blob/fd2805fe/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java
----------------------------------------------------------------------
diff --git a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java
index 584d111..360ed64 100644
--- a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java
+++ b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java
@@ -1465,8 +1465,7 @@ public class MQTTTest extends MQTTTestSupport {
connectionPub.disconnect();
- stopBroker();
- startBroker();
+ restartBroker();
MQTT mqttSub = createMQTTConnection("MQTT-Sub-Client", false);
BlockingConnection connectionSub = mqttSub.blockingConnection();
http://git-wip-us.apache.org/repos/asf/activemq/blob/fd2805fe/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTestSupport.java
----------------------------------------------------------------------
diff --git a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTestSupport.java b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTestSupport.java
index 5c31f10..18aee54 100644
--- a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTestSupport.java
+++ b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTestSupport.java
@@ -115,11 +115,47 @@ public class MQTTTestSupport {
}
public void startBroker() throws Exception {
+ brokerService = createBroker(true);
- createBroker();
+ configureBroker(brokerService);
- applyBrokerPolicies();
- applyMemoryLimitPolicy();
+ brokerService.start();
+ brokerService.waitUntilStarted();
+ port = brokerService.getTransportConnectorByName("mqtt").getConnectUri().getPort();
+ jmsUri = brokerService.getTransportConnectorByName("openwire").getPublishableConnectString();
+ cf = new ActiveMQConnectionFactory(jmsUri);
+ }
+
+ public void restartBroker() throws Exception {
+ stopBroker();
+
+ brokerService = createBroker(false);
+
+ configureBroker(brokerService);
+
+ brokerService.start();
+ brokerService.waitUntilStarted();
+ port = brokerService.getTransportConnectorByName("mqtt").getConnectUri().getPort();
+ jmsUri = brokerService.getTransportConnectorByName("openwire").getPublishableConnectString();
+ cf = new ActiveMQConnectionFactory(jmsUri);
+ }
+
+ protected BrokerService createBroker(boolean deleteAllMessages) throws Exception {
+ BrokerService brokerService = new BrokerService();
+ brokerService.setDeleteAllMessagesOnStartup(deleteAllMessages);
+ brokerService.setPersistent(isPersistent());
+ brokerService.setAdvisorySupport(false);
+ brokerService.setUseJmx(true);
+ brokerService.getManagementContext().setCreateConnector(false);
+ brokerService.setSchedulerSupport(isSchedulerSupportEnabled());
+ brokerService.setPopulateJMSXUserID(true);
+
+ return brokerService;
+ }
+
+ protected void configureBroker(BrokerService brokerService) throws Exception {
+ applyBrokerPolicies(brokerService);
+ applyMemoryLimitPolicy(brokerService);
// Setup SSL context...
File keyStore = new File(basedir(), "src/test/resources/server.keystore");
@@ -133,10 +169,8 @@ public class MQTTTestSupport {
sslContext.afterPropertiesSet();
brokerService.setSslContext(sslContext);
- addMQTTConnector();
- addOpenWireConnector();
-
- cf = new ActiveMQConnectionFactory(jmsUri);
+ addMQTTConnector(brokerService);
+ addOpenWireConnector(brokerService);
ArrayList<BrokerPlugin> plugins = new ArrayList<BrokerPlugin>();
createPlugins(plugins);
@@ -155,23 +189,6 @@ public class MQTTTestSupport {
BrokerPlugin[] array = new BrokerPlugin[plugins.size()];
brokerService.setPlugins(plugins.toArray(array));
}
-
- brokerService.start();
- brokerService.waitUntilStarted();
- port = brokerService.getTransportConnectorByName("mqtt").getConnectUri().getPort();
- }
-
- protected void applyMemoryLimitPolicy() throws Exception {
- }
-
- protected void createBroker() throws Exception {
- brokerService = new BrokerService();
- brokerService.setPersistent(isPersistent());
- brokerService.setAdvisorySupport(false);
- brokerService.setUseJmx(true);
- brokerService.getManagementContext().setCreateConnector(false);
- brokerService.setSchedulerSupport(isSchedulerSupportEnabled());
- brokerService.setPopulateJMSXUserID(true);
}
/**
@@ -197,16 +214,21 @@ public class MQTTTestSupport {
return null;
}
- protected void applyBrokerPolicies() throws Exception {
+ protected void applyBrokerPolicies(BrokerService brokerService) throws Exception {
// NOOP here
}
- protected void addOpenWireConnector() throws Exception {
- TransportConnector connector = brokerService.addConnector("tcp://0.0.0.0:0");
- jmsUri = connector.getPublishableConnectString();
+ protected void applyMemoryLimitPolicy(BrokerService brokerService) throws Exception {
+ }
+
+ protected void addOpenWireConnector(BrokerService brokerService) throws Exception {
+ TransportConnector connector = new TransportConnector();
+ connector.setUri(new URI("tcp://0.0.0.0:0"));
+ connector.setName("openwire");
+ brokerService.addConnector(connector);
}
- protected void addMQTTConnector() throws Exception {
+ protected void addMQTTConnector(BrokerService brokerService) throws Exception {
// Overrides of this method can add additional configuration options or add multiple
// MQTT transport connectors as needed, the port variable is always supposed to be
// assigned the primary MQTT connector's port.