Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Test for https://issues.apache.org/jira/browse/AMQ-5290

This commit is contained in:
Christian Posta 2014-08-06 15:53:16 -07:00
parent 413e4840d6
commit c42b874972
3 changed files with 246 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/FuseMQQTTClientProvider.java → activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/FuseMQTTClientProvider.java
View File

@ -26,7 +26,7 @@ import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
import static org.fusesource.hawtbuf.UTF8Buffer.utf8;
class FuseMQQTTClientProvider implements MQTTClientProvider {
public class FuseMQTTClientProvider implements MQTTClientProvider {
private final MQTT mqtt = new MQTT();
private BlockingConnection connection;
@Override

2
activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTestSupport.java
View File

@ -339,7 +339,7 @@ public class MQTTTestSupport {
}
protected MQTTClientProvider getMQTTClientProvider() {
return new FuseMQQTTClientProvider();
return new FuseMQTTClientProvider();
}
protected MQTT createMQTTConnection() throws Exception {

244
activemq-unit-tests/src/test/java/org/apache/activemq/MQTTNetworkOfBrokersFailoverTest.java Normal file
View File

@ -0,0 +1,244 @@
/**
* 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;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnector;
import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.broker.jmx.DurableSubscriptionViewMBean;
import org.apache.activemq.network.DiscoveryNetworkConnector;
import org.apache.activemq.network.NetworkConnector;
import org.apache.activemq.network.NetworkTestSupport;
import org.apache.activemq.store.PersistenceAdapter;
import org.apache.activemq.util.Wait;
import org.apache.commons.lang.ArrayUtils;
import org.fusesource.hawtdispatch.Dispatch;
import org.fusesource.mqtt.client.*;
import org.fusesource.mqtt.client.Topic;
import org.fusesource.mqtt.codec.MQTTFrame;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.jms.*;
import javax.jms.Message;
import javax.management.ObjectName;
import java.net.URI;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
/**
* Created by ceposta
* <a href="http://christianposta.com/blog>http://christianposta.com/blog</a>.
*/
public class MQTTNetworkOfBrokersFailoverTest extends NetworkTestSupport {
private static final Logger LOG = LoggerFactory.getLogger(MQTTNetworkOfBrokersFailoverTest.class);
private final String subName = "Subscriber1";
private final String subName2 = "Subscriber2";
private final String topicName = "TEST.FOO";
private int localBrokerMQTTPort = -1;
private int remoteBrokerMQTTPort = -1;
protected void setUp() throws Exception {
useJmx=true;
super.setUp();
URI ncUri = new URI("static:(" + connector.getConnectUri().toString() + ")");
NetworkConnector nc = new DiscoveryNetworkConnector(ncUri);
nc.setDuplex(true);
remoteBroker.addNetworkConnector(nc);
nc.start();
// mqtt port should have been assigned by now
assertFalse(localBrokerMQTTPort == -1);
assertFalse(remoteBrokerMQTTPort == -1);
}
protected void tearDown() throws Exception {
if (remoteBroker.isStarted()) {
remoteBroker.stop();
remoteBroker.waitUntilStopped();
}
if (broker.isStarted()) {
broker.stop();
broker.waitUntilStopped();
}
super.tearDown();
}
@Test
public void testNoStaleSubscriptionAcrossNetwork() throws Exception {
// before we get started, we want an async way to be able to know when
// the durable consumer has been networked so we can assert that it indeed
// would have a durable subscriber. for example, when we subscribe on remote broker,
// a network-sub would be created on local broker and we want to listen for when that
// even happens. we do that with advisory messages and a latch:
CountDownLatch consumerNetworked = listenForConsumersOn(broker);
// create a subscription with Clean == 0 (durable sub for QoS==1 && QoS==2)
// on the remote broker. this sub should still be there after we disconnect
MQTT remoteMqtt = createMQTTTcpConnection("foo", false, remoteBrokerMQTTPort);
BlockingConnection remoteConn = remoteMqtt.blockingConnection();
remoteConn.connect();
remoteConn.subscribe(new Topic[]{new Topic("foo/bar", QoS.AT_LEAST_ONCE)});
remoteConn.disconnect();
consumerNetworked.await(1, TimeUnit.SECONDS);
assertOneDurableSubOn(remoteBroker, "foo");
assertOneDurableSubOn(broker, "NC_localhost_inbound_local");
// now we reconnect the same sub on the local broker, again with clean==0
MQTT localMqtt = createMQTTTcpConnection("foo", false, localBrokerMQTTPort);
BlockingConnection localConn = localMqtt.blockingConnection();
localConn.connect();
localConn.subscribe(new Topic[]{new Topic("foo/bar", QoS.AT_LEAST_ONCE)});
// now let's connect back up to remote broker and send a message
remoteConn = remoteMqtt.blockingConnection();
remoteConn.connect();
remoteConn.publish("foo/bar", "Hello, World!".getBytes(), QoS.AT_LEAST_ONCE, false);
// now we should see that message on the local broker because the subscription
// should have been properly networked... we'll give a sec of grace for the
// networking and forwarding to have happened properly
org.fusesource.mqtt.client.Message msg = localConn.receive(1, TimeUnit.SECONDS);
assertNotNull(msg);
msg.ack();
String response = new String(msg.getPayload());
assertEquals("Hello, World!", response);
// Now... we SHOULD NOT see a message on the remote broker because we already
// consumed it on the local broker... having the same message on the remote broker
// would effectively give us duplicates in a distributed topic scenario:
remoteConn.subscribe(new Topic[]{new Topic("foo/bar", QoS.AT_LEAST_ONCE)});
msg = remoteConn.receive(500, TimeUnit.MILLISECONDS);
assertNull("We have duplicate messages across the cluster for a distributed topic",
msg);
}
private CountDownLatch listenForConsumersOn(BrokerService broker) throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
URI brokerUri = broker.getVmConnectorURI();
System.out.println(brokerUri.toASCIIString());
final ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(brokerUri.toASCIIString());
final Connection connection = cf.createConnection();
final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination dest = session.createTopic("ActiveMQ.Advisory.Consumer.Topic.foo.bar.>");
MessageConsumer consumer = session.createConsumer(dest);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
System.out.println("got message! " + message);
latch.countDown();
// shutdown this connection
Dispatch.getGlobalQueue().execute(new Runnable() {
@Override
public void run() {
try {
session.close();
connection.close();
} catch (JMSException e) {
e.printStackTrace();
}
}
});
}
});
return latch;
}
private void assertOneDurableSubOn(BrokerService broker, String subName) throws Exception {
BrokerViewMBean brokerView = broker.getAdminView();
ObjectName[] activeDurableSubs = brokerView.getDurableTopicSubscribers();
ObjectName[] inactiveDurableSubs = brokerView.getInactiveDurableTopicSubscribers();
ObjectName[] allDurables = (ObjectName[]) ArrayUtils.addAll(activeDurableSubs, inactiveDurableSubs);
assertEquals(1, allDurables.length);
// at this point our assertions should prove that we have only on durable sub
DurableSubscriptionViewMBean durableSubView = (DurableSubscriptionViewMBean)
broker.getManagementContext().newProxyInstance(allDurables[0], DurableSubscriptionViewMBean.class, true);
assertEquals(subName, durableSubView.getClientId());
}
@Override
protected BrokerService createBroker() throws Exception {
BrokerService broker = super.createBroker();
broker.setPersistent(true);
broker.setBrokerName("local");
broker.setDataDirectory("target/activemq-data");
broker.setDeleteAllMessagesOnStartup(true);
TransportConnector tc = broker.addConnector("mqtt://localhost:0");
localBrokerMQTTPort = tc.getConnectUri().getPort();
return broker;
}
@Override
protected BrokerService createRemoteBroker(PersistenceAdapter persistenceAdapter) throws Exception {
BrokerService broker = super.createRemoteBroker(persistenceAdapter);
broker.setPersistent(true);
broker.setDeleteAllMessagesOnStartup(true);
broker.setDataDirectory("target/activemq-data");
TransportConnector tc = broker.addConnector("mqtt://localhost:0");
remoteBrokerMQTTPort = tc.getConnectUri().getPort();
return broker;
}
private MQTT createMQTTTcpConnection(String clientId, boolean clean, int port) throws Exception {
MQTT mqtt = new MQTT();
mqtt.setConnectAttemptsMax(1);
mqtt.setReconnectAttemptsMax(0);
mqtt.setTracer(createTracer());
if (clientId != null) {
mqtt.setClientId(clientId);
}
mqtt.setCleanSession(clean);
mqtt.setHost("localhost", port);
return mqtt;
}
protected Tracer createTracer() {
return new Tracer() {
@Override
public void onReceive(MQTTFrame frame) {
LOG.info("Client Received:\n" + frame);
}
@Override
public void onSend(MQTTFrame frame) {
LOG.info("Client Sent:\n" + frame);
}
@Override
public void debug(String message, Object... args) {
LOG.info(String.format(message, args));
}
};
}
}