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

Apply patch for: https://issues.apache.org/jira/browse/AMQ-4338 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1450366 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Timothy A. Bish 2013-02-26 19:16:06 +00:00
parent 122656a20d
commit f60da9866f
2 changed files with 25 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/AbstractMQTTTest.java
View File

@ -16,6 +16,10 @@
*/ */
package org.apache.activemq.transport.mqtt; package org.apache.activemq.transport.mqtt;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import java.io.File; import java.io.File;
import java.io.IOException; import java.io.IOException;
import java.security.ProtectionDomain; import java.security.ProtectionDomain;
@ -27,6 +31,7 @@ import javax.jms.MessageConsumer;
import javax.jms.MessageProducer; import javax.jms.MessageProducer;
import javax.jms.Session; import javax.jms.Session;
import javax.jms.TextMessage; import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.BrokerService;
@ -36,8 +41,6 @@ import org.apache.activemq.util.ByteSequence;
import org.junit.After; import org.junit.After;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import static org.junit.Assert.*;
public abstract class AbstractMQTTTest { public abstract class AbstractMQTTTest {
protected TransportConnector mqttConnector; protected TransportConnector mqttConnector;
@ -72,7 +75,7 @@ public abstract class AbstractMQTTTest {
} }
} }
@Test(timeout=30000) @Test(timeout=300000)
public void testSendAndReceiveMQTT() throws Exception { public void testSendAndReceiveMQTT() throws Exception {
addMQTTConnector(); addMQTTConnector();
brokerService.start(); brokerService.start();
@ -85,6 +88,7 @@ public abstract class AbstractMQTTTest {
final CountDownLatch latch = new CountDownLatch(numberOfMessages); final CountDownLatch latch = new CountDownLatch(numberOfMessages);
Thread thread = new Thread(new Runnable() { Thread thread = new Thread(new Runnable() {
@Override
public void run() { public void run() {
for (int i = 0; i < numberOfMessages; i++){ for (int i = 0; i < numberOfMessages; i++){
try { try {
@ -115,7 +119,7 @@ public abstract class AbstractMQTTTest {
publishProvider.disconnect(); publishProvider.disconnect();
} }
@Test(timeout=30000) @Test(timeout=300000)
public void testSendAtMostOnceReceiveExactlyOnce() throws Exception { public void testSendAtMostOnceReceiveExactlyOnce() throws Exception {
/** /**
* Although subscribing with EXACTLY ONCE, the message gets published * Although subscribing with EXACTLY ONCE, the message gets published
@ -138,7 +142,7 @@ public abstract class AbstractMQTTTest {
provider.disconnect(); provider.disconnect();
} }
@Test(timeout=30000) @Test(timeout=300000)
public void testSendAtLeastOnceReceiveExactlyOnce() throws Exception { public void testSendAtLeastOnceReceiveExactlyOnce() throws Exception {
addMQTTConnector(); addMQTTConnector();
brokerService.start(); brokerService.start();
@ -156,7 +160,7 @@ public abstract class AbstractMQTTTest {
provider.disconnect(); provider.disconnect();
} }
@Test(timeout=30000) @Test(timeout=300000)
public void testSendAtLeastOnceReceiveAtMostOnce() throws Exception { public void testSendAtLeastOnceReceiveAtMostOnce() throws Exception {
addMQTTConnector(); addMQTTConnector();
brokerService.start(); brokerService.start();
@ -174,8 +178,7 @@ public abstract class AbstractMQTTTest {
provider.disconnect(); provider.disconnect();
} }
@Test(timeout=300000)
@Test(timeout=30000)
public void testSendAndReceiveAtMostOnce() throws Exception { public void testSendAndReceiveAtMostOnce() throws Exception {
addMQTTConnector(); addMQTTConnector();
brokerService.start(); brokerService.start();
@ -193,7 +196,7 @@ public abstract class AbstractMQTTTest {
provider.disconnect(); provider.disconnect();
} }
@Test(timeout=30000) @Test(timeout=300000)
public void testSendAndReceiveAtLeastOnce() throws Exception { public void testSendAndReceiveAtLeastOnce() throws Exception {
addMQTTConnector(); addMQTTConnector();
brokerService.start(); brokerService.start();
@ -211,7 +214,7 @@ public abstract class AbstractMQTTTest {
provider.disconnect(); provider.disconnect();
} }
@Test(timeout=30000) @Test(timeout=300000)
public void testSendAndReceiveExactlyOnce() throws Exception { public void testSendAndReceiveExactlyOnce() throws Exception {
addMQTTConnector(); addMQTTConnector();
brokerService.start(); brokerService.start();
@ -221,7 +224,6 @@ public abstract class AbstractMQTTTest {
final MQTTClientProvider subscriber = getMQTTClientProvider(); final MQTTClientProvider subscriber = getMQTTClientProvider();
initializeConnection(subscriber); initializeConnection(subscriber);
subscriber.subscribe("foo",EXACTLY_ONCE); subscriber.subscribe("foo",EXACTLY_ONCE);
for (int i = 0; i < numberOfMessages; i++) { for (int i = 0; i < numberOfMessages; i++) {
String payload = "Test Message: " + i; String payload = "Test Message: " + i;
@ -234,7 +236,7 @@ public abstract class AbstractMQTTTest {
publisher.disconnect(); publisher.disconnect();
} }
@Test(timeout=30000) @Test(timeout=300000)
public void testSendAndReceiveLargeMessages() throws Exception { public void testSendAndReceiveLargeMessages() throws Exception {
byte[] payload = new byte[1024 * 32]; byte[] payload = new byte[1024 * 32];
for (int i = 0; i < payload.length; i++){ for (int i = 0; i < payload.length; i++){
@ -261,8 +263,7 @@ public abstract class AbstractMQTTTest {
publisher.disconnect(); publisher.disconnect();
} }
@Test(timeout=300000)
@Test(timeout=30000)
public void testSendMQTTReceiveJMS() throws Exception { public void testSendMQTTReceiveJMS() throws Exception {
addMQTTConnector(); addMQTTConnector();
TransportConnector openwireTransport = brokerService.addConnector("tcp://localhost:0"); TransportConnector openwireTransport = brokerService.addConnector("tcp://localhost:0");
@ -272,7 +273,6 @@ public abstract class AbstractMQTTTest {
initializeConnection(provider); initializeConnection(provider);
final String DESTINATION_NAME = "foo.*"; final String DESTINATION_NAME = "foo.*";
ActiveMQConnection activeMQConnection = (ActiveMQConnection) new ActiveMQConnectionFactory(openwireTransport.getConnectUri()).createConnection(); ActiveMQConnection activeMQConnection = (ActiveMQConnection) new ActiveMQConnectionFactory(openwireTransport.getConnectUri()).createConnection();
activeMQConnection.start(); activeMQConnection.start();
Session s = activeMQConnection.createSession(false, Session.AUTO_ACKNOWLEDGE); Session s = activeMQConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
@ -288,12 +288,11 @@ public abstract class AbstractMQTTTest {
assertEquals(payload, new String(bs.data, bs.offset, bs.length)); assertEquals(payload, new String(bs.data, bs.offset, bs.length));
} }
activeMQConnection.close(); activeMQConnection.close();
provider.disconnect(); provider.disconnect();
} }
@Test(timeout=30000) @Test(timeout=300000)
public void testSendJMSReceiveMQTT() throws Exception { public void testSendJMSReceiveMQTT() throws Exception {
addMQTTConnector(); addMQTTConnector();
TransportConnector openwireTransport = brokerService.addConnector("tcp://localhost:0"); TransportConnector openwireTransport = brokerService.addConnector("tcp://localhost:0");

15
activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java
View File

@ -16,16 +16,16 @@
*/ */
package org.apache.activemq.transport.mqtt; package org.apache.activemq.transport.mqtt;
import static org.junit.Assert.assertTrue;
import org.apache.activemq.util.Wait; import org.apache.activemq.util.Wait;
import org.fusesource.mqtt.client.BlockingConnection; import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.MQTT; import org.fusesource.mqtt.client.MQTT;
import org.junit.Test; import org.junit.Test;
import static org.junit.Assert.assertTrue;
public class MQTTTest extends AbstractMQTTTest { public class MQTTTest extends AbstractMQTTTest {
@Test(timeout=30000) @Test(timeout=300000)
public void testPingKeepsInactivityMonitorAlive() throws Exception { public void testPingKeepsInactivityMonitorAlive() throws Exception {
addMQTTConnector(); addMQTTConnector();
brokerService.start(); brokerService.start();
@ -45,7 +45,7 @@ public class MQTTTest extends AbstractMQTTTest {
connection.disconnect(); connection.disconnect();
} }
@Test(timeout=30000) @Test(timeout=300000)
public void testTurnOffInactivityMonitor()throws Exception{ public void testTurnOffInactivityMonitor()throws Exception{
addMQTTConnector("?transport.useInactivityMonitor=false"); addMQTTConnector("?transport.useInactivityMonitor=false");
brokerService.start(); brokerService.start();
@ -65,7 +65,6 @@ public class MQTTTest extends AbstractMQTTTest {
connection.disconnect(); connection.disconnect();
} }
@Test(timeout=30000) @Test(timeout=30000)
public void testDefaultKeepAliveWhenClientSpecifiesZero() throws Exception { public void testDefaultKeepAliveWhenClientSpecifiesZero() throws Exception {
// default keep alive in milliseconds // default keep alive in milliseconds
@ -85,17 +84,19 @@ public class MQTTTest extends AbstractMQTTTest {
})); }));
} }
@Override
protected String getProtocolScheme() { protected String getProtocolScheme() {
return "mqtt"; return "mqtt";
} }
@Override
protected void addMQTTConnector() throws Exception { protected void addMQTTConnector() throws Exception {
addMQTTConnector(""); addMQTTConnector("");
} }
@Override
protected void addMQTTConnector(String config) throws Exception { protected void addMQTTConnector(String config) throws Exception {
mqttConnector= brokerService.addConnector(getProtocolScheme()+"://localhost:0" + config); mqttConnector = brokerService.addConnector(getProtocolScheme()+"://localhost:0" + config);
} }
@Override @Override