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

[maven-release-plugin] copy for tag activemq-5.6.0 

git-svn-id: https://svn.apache.org/repos/asf/activemq/tags/activemq-5.6.0@1332982 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Bosanac Dejan 2012-05-02 10:35:51 +00:00
parent b24e695941 b7e264f917
commit 30697b4e50
21 changed files with 956 additions and 141 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
activemq-core/src/main/java/org/apache/activemq/plugin/StatisticsBroker.java
View File

@ -22,6 +22,8 @@ import org.apache.activemq.broker.BrokerFilter;
import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.ConnectionContext; import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.ProducerBrokerExchange; import org.apache.activemq.broker.ProducerBrokerExchange;
import org.apache.activemq.broker.jmx.BrokerViewMBean;
import org.apache.activemq.broker.jmx.SubscriptionViewMBean;
import org.apache.activemq.broker.region.Destination; import org.apache.activemq.broker.region.Destination;
import org.apache.activemq.broker.region.DestinationStatistics; import org.apache.activemq.broker.region.DestinationStatistics;
import org.apache.activemq.broker.region.RegionBroker; import org.apache.activemq.broker.region.RegionBroker;
@ -37,6 +39,9 @@ import org.apache.activemq.util.IdGenerator;
import org.apache.activemq.util.LongSequenceGenerator; import org.apache.activemq.util.LongSequenceGenerator;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import javax.jms.JMSException;
import javax.management.ObjectName;
import java.io.File; import java.io.File;
import java.net.URI; import java.net.URI;
import java.util.Set; import java.util.Set;
@ -50,9 +55,11 @@ public class StatisticsBroker extends BrokerFilter {
private static Logger LOG = LoggerFactory.getLogger(StatisticsBroker.class); private static Logger LOG = LoggerFactory.getLogger(StatisticsBroker.class);
static final String STATS_DESTINATION_PREFIX = "ActiveMQ.Statistics.Destination"; static final String STATS_DESTINATION_PREFIX = "ActiveMQ.Statistics.Destination";
static final String STATS_BROKER_PREFIX = "ActiveMQ.Statistics.Broker"; static final String STATS_BROKER_PREFIX = "ActiveMQ.Statistics.Broker";
static final String STATS_SUBSCRIPTION_PREFIX = "ActiveMQ.Statistics.Subscription";
private static final IdGenerator ID_GENERATOR = new IdGenerator(); private static final IdGenerator ID_GENERATOR = new IdGenerator();
private final LongSequenceGenerator messageIdGenerator = new LongSequenceGenerator(); private final LongSequenceGenerator messageIdGenerator = new LongSequenceGenerator();
protected final ProducerId advisoryProducerId = new ProducerId(); protected final ProducerId advisoryProducerId = new ProducerId();
protected BrokerViewMBean brokerView;
/** /**
* *
@ -80,6 +87,10 @@ public class StatisticsBroker extends BrokerFilter {
STATS_DESTINATION_PREFIX.length()); STATS_DESTINATION_PREFIX.length());
boolean brokerStats = physicalName.regionMatches(true, 0, STATS_BROKER_PREFIX, 0, STATS_BROKER_PREFIX boolean brokerStats = physicalName.regionMatches(true, 0, STATS_BROKER_PREFIX, 0, STATS_BROKER_PREFIX
.length()); .length());
boolean subStats = physicalName.regionMatches(true, 0, STATS_SUBSCRIPTION_PREFIX, 0, STATS_SUBSCRIPTION_PREFIX
.length());
BrokerService brokerService = getBrokerService();
RegionBroker regionBroker = (RegionBroker) brokerService.getRegionBroker();
if (destStats) { if (destStats) {
String queueryName = physicalName.substring(STATS_DESTINATION_PREFIX.length(), physicalName.length()); String queueryName = physicalName.substring(STATS_DESTINATION_PREFIX.length(), physicalName.length());
ActiveMQDestination queryDest = ActiveMQDestination.createDestination(queueryName,msgDest.getDestinationType()); ActiveMQDestination queryDest = ActiveMQDestination.createDestination(queueryName,msgDest.getDestinationType());
@ -108,10 +119,11 @@ public class StatisticsBroker extends BrokerFilter {
sendStats(producerExchange.getConnectionContext(), statsMessage, replyTo); sendStats(producerExchange.getConnectionContext(), statsMessage, replyTo);
} }
} }
} else if (subStats) {
sendSubStats(producerExchange.getConnectionContext(), getBrokerView().getQueueSubscribers(), replyTo);
sendSubStats(producerExchange.getConnectionContext(), getBrokerView().getTopicSubscribers(), replyTo);
} else if (brokerStats) { } else if (brokerStats) {
ActiveMQMapMessage statsMessage = new ActiveMQMapMessage(); ActiveMQMapMessage statsMessage = new ActiveMQMapMessage();
BrokerService brokerService = getBrokerService();
RegionBroker regionBroker = (RegionBroker) brokerService.getRegionBroker();
SystemUsage systemUsage = brokerService.getSystemUsage(); SystemUsage systemUsage = brokerService.getSystemUsage();
DestinationStatistics stats = regionBroker.getDestinationStatistics(); DestinationStatistics stats = regionBroker.getDestinationStatistics();
statsMessage.setString("brokerName", regionBroker.getBrokerName()); statsMessage.setString("brokerName", regionBroker.getBrokerName());
@ -165,6 +177,15 @@ public class StatisticsBroker extends BrokerFilter {
} }
} }
BrokerViewMBean getBrokerView() throws Exception {
if (this.brokerView == null) {
ObjectName brokerName = getBrokerService().getBrokerObjectName();
this.brokerView = (BrokerViewMBean) getBrokerService().getManagementContext().newProxyInstance(brokerName,
BrokerViewMBean.class, true);
}
return this.brokerView;
}
public void start() throws Exception { public void start() throws Exception {
super.start(); super.start();
LOG.info("Starting StatisticsBroker"); LOG.info("Starting StatisticsBroker");
@ -174,6 +195,34 @@ public class StatisticsBroker extends BrokerFilter {
super.stop(); super.stop();
} }
protected void sendSubStats(ConnectionContext context, ObjectName[] subscribers, ActiveMQDestination replyTo) throws Exception {
for (int i = 0; i < subscribers.length; i++) {
ObjectName name = subscribers[i];
SubscriptionViewMBean subscriber = (SubscriptionViewMBean)getBrokerService().getManagementContext().newProxyInstance(name, SubscriptionViewMBean.class, true);
ActiveMQMapMessage statsMessage = prepareSubscriptionMessage(subscriber);
sendStats(context, statsMessage, replyTo);
}
}
protected ActiveMQMapMessage prepareSubscriptionMessage(SubscriptionViewMBean subscriber) throws JMSException {
ActiveMQMapMessage statsMessage = new ActiveMQMapMessage();
statsMessage.setString("destinationName", subscriber.getDestinationName());
statsMessage.setString("clientId", subscriber.getClientId());
statsMessage.setString("connectionId", subscriber.getConnectionId());
statsMessage.setLong("sessionId", subscriber.getSessionId());
statsMessage.setString("selector", subscriber.getSelector());
statsMessage.setLong("enqueueCounter", subscriber.getEnqueueCounter());
statsMessage.setLong("dequeueCounter", subscriber.getDequeueCounter());
statsMessage.setLong("dispatchedCounter", subscriber.getDispatchedCounter());
statsMessage.setLong("dispatchedQueueSize", subscriber.getDispatchedQueueSize());
statsMessage.setInt("prefetchSize", subscriber.getPrefetchSize());
statsMessage.setInt("maximumPendingMessageLimit", subscriber.getMaximumPendingMessageLimit());
statsMessage.setBoolean("exclusive", subscriber.isExclusive());
statsMessage.setBoolean("retroactive", subscriber.isRetroactive());
statsMessage.setBoolean("slowConsumer", subscriber.isSlowConsumer());
return statsMessage;
}
protected void sendStats(ConnectionContext context, ActiveMQMapMessage msg, ActiveMQDestination replyTo) protected void sendStats(ConnectionContext context, ActiveMQMapMessage msg, ActiveMQDestination replyTo)
throws Exception { throws Exception {
msg.setPersistent(false); msg.setPersistent(false);

8
activemq-core/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
View File

@ -135,7 +135,11 @@ public class NIOSSLTransport extends NIOTransport {
while(true) { while(true) {
if (!plain.hasRemaining()) { if (!plain.hasRemaining()) {
if (status == SSLEngineResult.Status.OK && handshakeStatus != SSLEngineResult.HandshakeStatus.NEED_UNWRAP) {
plain.clear(); plain.clear();
} else {
plain.compact();
}
int readCount = secureRead(plain); int readCount = secureRead(plain);
@ -150,7 +154,9 @@ public class NIOSSLTransport extends NIOTransport {
} }
} }
if (status == SSLEngineResult.Status.OK && handshakeStatus != SSLEngineResult.HandshakeStatus.NEED_UNWRAP) {
processCommand(plain); processCommand(plain);
}
} }
} catch (IOException e) { } catch (IOException e) {
@ -192,7 +198,7 @@ public class NIOSSLTransport extends NIOTransport {
protected int secureRead(ByteBuffer plain) throws Exception { protected int secureRead(ByteBuffer plain) throws Exception {
if (!(inputBuffer.position() != 0 && inputBuffer.hasRemaining())) { if (!(inputBuffer.position() != 0 && inputBuffer.hasRemaining()) || status == SSLEngineResult.Status.BUFFER_UNDERFLOW) {
int bytesRead = channel.read(inputBuffer); int bytesRead = channel.read(inputBuffer);
if (bytesRead == -1) { if (bytesRead == -1) {

16
activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompWireFormat.java
View File

@ -16,15 +16,21 @@
*/ */
package org.apache.activemq.transport.stomp; package org.apache.activemq.transport.stomp;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.PushbackInputStream;
import java.util.HashMap;
import java.util.Map;
import org.apache.activemq.util.ByteArrayInputStream; import org.apache.activemq.util.ByteArrayInputStream;
import org.apache.activemq.util.ByteArrayOutputStream; import org.apache.activemq.util.ByteArrayOutputStream;
import org.apache.activemq.util.ByteSequence; import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.wireformat.WireFormat; import org.apache.activemq.wireformat.WireFormat;
import java.io.*;
import java.util.HashMap;
import java.util.Map;
/** /**
* Implements marshalling and unmarsalling the <a * Implements marshalling and unmarsalling the <a
* href="http://stomp.codehaus.org/">Stomp</a> protocol. * href="http://stomp.codehaus.org/">Stomp</a> protocol.
@ -201,7 +207,7 @@ public class StompWireFormat implements WireFormat {
ByteSequence nameSeq = stream.toByteSequence(); ByteSequence nameSeq = stream.toByteSequence();
String name = new String(nameSeq.getData(), nameSeq.getOffset(), nameSeq.getLength(), "UTF-8").trim(); String name = new String(nameSeq.getData(), nameSeq.getOffset(), nameSeq.getLength(), "UTF-8").trim();
String value = decodeHeader(headerLine).trim(); String value = decodeHeader(headerLine);
headers.put(name, value); headers.put(name, value);
} catch (Exception e) { } catch (Exception e) {
throw new ProtocolException("Unable to parser header line [" + line + "]", true); throw new ProtocolException("Unable to parser header line [" + line + "]", true);

13
activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2571Test.java
View File

@ -32,6 +32,7 @@ public class AMQ2571Test extends EmbeddedBrokerTestSupport {
public void testTempQueueClosing() { public void testTempQueueClosing() {
try { try {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(this.bindAddress); ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(this.bindAddress);
connectionFactory.setAlwaysSyncSend(true);
// First create session that will own the TempQueue // First create session that will own the TempQueue
Connection connectionA = connectionFactory.createConnection(); Connection connectionA = connectionFactory.createConnection();
@ -56,13 +57,13 @@ public class AMQ2571Test extends EmbeddedBrokerTestSupport {
Thread sendingThread = new Thread(new Runnable() { Thread sendingThread = new Thread(new Runnable() {
public void run() { public void run() {
try { try {
for (int i = 0; i < 100000; i++) { long end = System.currentTimeMillis() + 5*60*1000;
// wait for exception on send
while (System.currentTimeMillis() < end) {
producerB.send(message); producerB.send(message);
} }
} catch (JMSException e) { } catch (JMSException e) {
// We don't get this exception every time. e.printStackTrace();
// Not getting it means that we don't know if the
// creator of the TempQueue has disconnected.
} }
} }
}); });
@ -72,7 +73,7 @@ public class AMQ2571Test extends EmbeddedBrokerTestSupport {
// Now close connection A. This will remove the TempQueue. // Now close connection A. This will remove the TempQueue.
connectionA.close(); connectionA.close();
// Wait for the thread to finish. // Wait for the thread to finish.
sendingThread.join(); sendingThread.join(5*60*1000);
// Sleep for a while to make sure that we should know that the // Sleep for a while to make sure that we should know that the
// TempQueue is gone. // TempQueue is gone.
@ -95,6 +96,7 @@ public class AMQ2571Test extends EmbeddedBrokerTestSupport {
@Override @Override
protected void setUp() throws Exception { protected void setUp() throws Exception {
bindAddress = "vm://localhost"; bindAddress = "vm://localhost";
setAutoFail(true);
super.setUp(); super.setUp();
} }
@ -103,7 +105,6 @@ public class AMQ2571Test extends EmbeddedBrokerTestSupport {
BrokerService answer = new BrokerService(); BrokerService answer = new BrokerService();
answer.setPersistent(false); answer.setPersistent(false);
answer.setUseJmx(false); answer.setUseJmx(false);
answer.addConnector(bindAddress);
return answer; return answer;
} }
} }

32
activemq-core/src/test/java/org/apache/activemq/plugin/BrokerStatisticsPluginTest.java
View File

@ -16,14 +16,14 @@
*/ */
package org.apache.activemq.plugin; package org.apache.activemq.plugin;
import junit.framework.TestCase;
import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerFactory; import org.apache.activemq.broker.BrokerFactory;
import org.apache.activemq.broker.BrokerPlugin; import org.apache.activemq.broker.BrokerPlugin;
import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.BrokerService;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.net.URI;
import java.util.Enumeration;
import javax.jms.Connection; import javax.jms.Connection;
import javax.jms.ConnectionFactory; import javax.jms.ConnectionFactory;
import javax.jms.MapMessage; import javax.jms.MapMessage;
@ -32,7 +32,7 @@ import javax.jms.MessageConsumer;
import javax.jms.MessageProducer; import javax.jms.MessageProducer;
import javax.jms.Queue; import javax.jms.Queue;
import javax.jms.Session; import javax.jms.Session;
import junit.framework.TestCase; import java.net.URI;
/** /**
* A BrokerStatisticsPluginTest * A BrokerStatisticsPluginTest
@ -91,6 +91,32 @@ public class BrokerStatisticsPluginTest extends TestCase{
*/ */
}
public void testSubscriptionStats() throws Exception{
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue replyTo = session.createTemporaryQueue();
MessageConsumer consumer = session.createConsumer(replyTo);
Queue testQueue = session.createQueue("Test.Queue");
MessageConsumer testConsumer = session.createConsumer(testQueue);
MessageProducer producer = session.createProducer(null);
Queue query = session.createQueue(StatisticsBroker.STATS_SUBSCRIPTION_PREFIX);
Message msg = session.createMessage();
producer.send(testQueue,msg);
msg.setJMSReplyTo(replyTo);
producer.send(query,msg);
MapMessage reply = (MapMessage) consumer.receive();
assertNotNull(reply);
assertTrue(reply.getMapNames().hasMoreElements());
/*for (Enumeration e = reply.getMapNames();e.hasMoreElements();) {
String name = e.nextElement().toString();
System.err.println(name+"="+reply.getObject(name));
}*/
} }
protected void setUp() throws Exception { protected void setUp() throws Exception {

5
activemq-core/src/test/java/org/apache/activemq/transport/SoWriteTimeoutTest.java
View File

@ -20,12 +20,15 @@ import java.net.Socket;
import java.net.SocketException; import java.net.SocketException;
import java.net.URI; import java.net.URI;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import javax.jms.Connection; import javax.jms.Connection;
import javax.jms.Destination; import javax.jms.Destination;
import javax.jms.JMSException; import javax.jms.JMSException;
import javax.jms.MessageConsumer; import javax.jms.MessageConsumer;
import javax.jms.Session; import javax.jms.Session;
import junit.framework.Test; import junit.framework.Test;
import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.JmsTestSupport; import org.apache.activemq.JmsTestSupport;
import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.BrokerService;
@ -110,7 +113,7 @@ public class SoWriteTimeoutTest extends JmsTestSupport {
stompConnection.open(new Socket("localhost", proxy.getUrl().getPort())); stompConnection.open(new Socket("localhost", proxy.getUrl().getPort()));
stompConnection.getStompSocket().setTcpNoDelay(true); stompConnection.getStompSocket().setTcpNoDelay(true);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
assertTrue(frame.startsWith("CONNECTED")); assertTrue(frame.startsWith("CONNECTED"));

119
activemq-core/src/test/java/org/apache/activemq/transport/stomp/Stomp11Test.java
View File

@ -68,7 +68,6 @@ public class Stomp11Test extends CombinationTestSupport {
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
queue = new ActiveMQQueue(getQueueName()); queue = new ActiveMQQueue(getQueueName());
connection.start(); connection.start();
} }
private void stompConnect() throws IOException, URISyntaxException, UnknownHostException { private void stompConnect() throws IOException, URISyntaxException, UnknownHostException {
@ -106,11 +105,11 @@ public class Stomp11Test extends CombinationTestSupport {
public void testConnect() throws Exception { public void testConnect() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"host:localhost\n" + "host:localhost\n" +
"request-id: 1\n" + "request-id:1\n" +
"\n" + Stomp.NULL; "\n" + Stomp.NULL;
stompConnection.sendFrame(connectFrame); stompConnection.sendFrame(connectFrame);
@ -129,8 +128,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testConnectWithVersionOptions() throws Exception { public void testConnectWithVersionOptions() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.0,1.1\n" + "accept-version:1.0,1.1\n" +
"host:localhost\n" + "host:localhost\n" +
"\n" + Stomp.NULL; "\n" + Stomp.NULL;
@ -150,8 +149,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testConnectWithValidFallback() throws Exception { public void testConnectWithValidFallback() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.0,10.1\n" + "accept-version:1.0,10.1\n" +
"host:localhost\n" + "host:localhost\n" +
"\n" + Stomp.NULL; "\n" + Stomp.NULL;
@ -171,8 +170,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testConnectWithInvalidFallback() throws Exception { public void testConnectWithInvalidFallback() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:9.0,10.1\n" + "accept-version:9.0,10.1\n" +
"host:localhost\n" + "host:localhost\n" +
"\n" + Stomp.NULL; "\n" + Stomp.NULL;
@ -189,8 +188,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testHeartbeats() throws Exception { public void testHeartbeats() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"heart-beat:0,1000\n" + "heart-beat:0,1000\n" +
"host:localhost\n" + "host:localhost\n" +
@ -231,8 +230,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testHeartbeatsDropsIdleConnection() throws Exception { public void testHeartbeatsDropsIdleConnection() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"heart-beat:1000,0\n" + "heart-beat:1000,0\n" +
"host:localhost\n" + "host:localhost\n" +
@ -261,8 +260,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testSendAfterMissingHeartbeat() throws Exception { public void testSendAfterMissingHeartbeat() throws Exception {
String connectFrame = "STOMP\n" + "login: system\n" + String connectFrame = "STOMP\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"heart-beat:1000,0\n" + "heart-beat:1000,0\n" +
"host:localhost\n" + "host:localhost\n" +
@ -290,8 +289,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testRejectInvalidHeartbeats1() throws Exception { public void testRejectInvalidHeartbeats1() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"heart-beat:0\n" + "heart-beat:0\n" +
"host:localhost\n" + "host:localhost\n" +
@ -309,8 +308,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testRejectInvalidHeartbeats2() throws Exception { public void testRejectInvalidHeartbeats2() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"heart-beat:T,0\n" + "heart-beat:T,0\n" +
"host:localhost\n" + "host:localhost\n" +
@ -328,8 +327,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testRejectInvalidHeartbeats3() throws Exception { public void testRejectInvalidHeartbeats3() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"heart-beat:100,10,50\n" + "heart-beat:100,10,50\n" +
"host:localhost\n" + "host:localhost\n" +
@ -347,8 +346,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testSubscribeAndUnsubscribe() throws Exception { public void testSubscribeAndUnsubscribe() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"host:localhost\n" + "host:localhost\n" +
"\n" + Stomp.NULL; "\n" + Stomp.NULL;
@ -392,8 +391,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testSubscribeWithNoId() throws Exception { public void testSubscribeWithNoId() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"host:localhost\n" + "host:localhost\n" +
"\n" + Stomp.NULL; "\n" + Stomp.NULL;
@ -418,8 +417,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testUnsubscribeWithNoId() throws Exception { public void testUnsubscribeWithNoId() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"host:localhost\n" + "host:localhost\n" +
"\n" + Stomp.NULL; "\n" + Stomp.NULL;
@ -449,8 +448,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testAckMessageWithId() throws Exception { public void testAckMessageWithId() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"host:localhost\n" + "host:localhost\n" +
"\n" + Stomp.NULL; "\n" + Stomp.NULL;
@ -487,8 +486,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testAckMessageWithNoId() throws Exception { public void testAckMessageWithNoId() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"host:localhost\n" + "host:localhost\n" +
"\n" + Stomp.NULL; "\n" + Stomp.NULL;
@ -530,8 +529,8 @@ public class Stomp11Test extends CombinationTestSupport {
final int MSG_COUNT = 10; final int MSG_COUNT = 10;
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"host:localhost\n" + "host:localhost\n" +
"\n" + Stomp.NULL; "\n" + Stomp.NULL;
@ -591,12 +590,11 @@ public class Stomp11Test extends CombinationTestSupport {
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
} }
public void testSendMessageWithStandardHeadersEncoded() throws Exception { public void testSendMessageWithStandardHeadersEncoded() throws Exception {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n" + String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n" +
"accept-version:1.1" + "\n\n" + Stomp.NULL; "accept-version:1.1" + "\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
@ -622,10 +620,9 @@ public class Stomp11Test extends CombinationTestSupport {
assertEquals("GroupID", "abc", amqMessage.getGroupID()); assertEquals("GroupID", "abc", amqMessage.getGroupID());
} }
public void testSubscribeWithMessageSentWithEncodedProperties() throws Exception { public void testSubscribeWithMessageSentWithEncodedProperties() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n" + "accept-version:1.1" + "\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n" + "accept-version:1.1" + "\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -654,8 +651,8 @@ public class Stomp11Test extends CombinationTestSupport {
public void testNackMessage() throws Exception { public void testNackMessage() throws Exception {
String connectFrame = "STOMP\n" + String connectFrame = "STOMP\n" +
"login: system\n" + "login:system\n" +
"passcode: manager\n" + "passcode:manager\n" +
"accept-version:1.1\n" + "accept-version:1.1\n" +
"host:localhost\n" + "host:localhost\n" +
"\n" + Stomp.NULL; "\n" + Stomp.NULL;
@ -702,9 +699,49 @@ public class Stomp11Test extends CombinationTestSupport {
"id:12345\n\n" + Stomp.NULL; "id:12345\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = "DISCONNECT\n" + "\n\n" + Stomp.NULL; frame = "DISCONNECT\n" + "\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
} }
public void testHeaderValuesAreNotWSTrimmed() throws Exception {
String connectFrame = "STOMP\n" +
"login:system\n" +
"passcode:manager\n" +
"accept-version:1.1\n" +
"host:localhost\n" +
"\n" + Stomp.NULL;
stompConnection.sendFrame(connectFrame);
String f = stompConnection.receiveFrame();
LOG.debug("Broker sent: " + f);
assertTrue(f.startsWith("CONNECTED"));
String message = "SEND\n" + "destination:/queue/" + getQueueName() +
"\ntest1: value" +
"\ntest2:value " +
"\ntest3: value " +
"\n\n" + "Hello World" + Stomp.NULL;
stompConnection.sendFrame(message);
String frame = "SUBSCRIBE\n" + "destination:/queue/" + getQueueName() + "\n" +
"id:12345\n" + "ack:auto\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame);
StompFrame received = stompConnection.receive();
assertTrue(received.getAction().equals("MESSAGE"));
assertEquals(" value", received.getHeaders().get("test1"));
assertEquals("value ", received.getHeaders().get("test2"));
assertEquals(" value ", received.getHeaders().get("test3"));
frame = "UNSUBSCRIBE\n" + "destination:/queue/" + getQueueName() + "\n" +
"id:12345\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame);
frame = "DISCONNECT\n" + "\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame);
}
} }

237
activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompLoadTest.java Normal file
View File

@ -0,0 +1,237 @@
/**
* 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.stomp;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.net.Socket;
import java.net.URI;
import java.util.HashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.Connection;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.util.Wait;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class StompLoadTest {
private static final Logger LOG = LoggerFactory.getLogger(StompLoadTest.class);
protected String bindAddress = "stomp://localhost:61613";
protected String confUri = "xbean:org/apache/activemq/transport/stomp/stomp-auth-broker.xml";
protected String jmsUri = "vm://localhost";
private static final int TASK_COUNT = 100;
private static final int MSG_COUNT = 250; // AMQ-3819: Above 250 or so and the CPU goes bonkers with NOI+SSL.
private BrokerService broker;
protected StompConnection stompConnection = new StompConnection();
protected Connection connection;
protected Session session;
protected ActiveMQQueue queue;
private ExecutorService executor;
private CountDownLatch started;
private CountDownLatch ready;
private AtomicInteger receiveCount;
@Before
public void setUp() throws Exception {
broker = BrokerFactory.createBroker(new URI(confUri));
broker.setDeleteAllMessagesOnStartup(true);
broker.start();
broker.waitUntilStarted();
stompConnect();
ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory(jmsUri);
connection = cf.createConnection("system", "manager");
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
queue = new ActiveMQQueue(getDestinationName());
connection.start();
executor = Executors.newFixedThreadPool(TASK_COUNT, new ThreadFactory() {
private long i = 0;
public Thread newThread(Runnable runnable) {
this.i++;
final Thread t = new Thread(runnable, "Test Worker " + this.i);
return t;
}
});
started = new CountDownLatch(TASK_COUNT);
ready = new CountDownLatch(1);
receiveCount = new AtomicInteger(0);
}
@After
public void tearDown() throws Exception {
try {
executor.shutdownNow();
connection.close();
stompDisconnect();
} catch(Exception e) {
} finally {
broker.stop();
broker.waitUntilStopped();
}
}
@Test(timeout=20*60*1000*1000)
public void testStompUnloadLoad() throws Exception {
for (int i = 0; i < TASK_COUNT; ++i) {
executor.execute(new Runnable() {
@Override
public void run() {
LOG.debug("Receive Thread Connecting to Broker.");
int numReceived = 0;
StompConnection connection = new StompConnection();
try {
stompConnect(connection);
} catch (Exception e) {
LOG.error("Caught Exception while connecting: " + e.getMessage());
}
try {
for (int i = 0; i < 10; i++) {
connection.subscribe("/queue/test-" + i, "auto");
connection.subscribe("/topic/test-" + i, "auto");
}
HashMap<String, String> headers = new HashMap<String, String>();
headers.put("activemq.prefetchSize", "1");
connection.subscribe("/topic/" + getDestinationName(), "auto", headers);
ready.await();
// Now that the main test thread is ready we wait a bit to let the tasks
// all subscribe and the CPU to settle a bit.
TimeUnit.SECONDS.sleep(3);
started.countDown();
while (true) {
// Read Timeout ends this task, we override the default here since there
// are so many threads running and we don't know how slow the test box is.
StompFrame frame = connection.receive(TimeUnit.SECONDS.toMillis(60));
assertNotNull(frame);
numReceived++;
if (LOG.isDebugEnabled() && (numReceived % 50) == 0 || numReceived == MSG_COUNT) {
LOG.debug("Receiver thread got message: " + frame.getHeaders().get("message-id"));
}
receiveCount.incrementAndGet();
}
} catch (Exception e) {
if (numReceived != MSG_COUNT) {
LOG.warn("Receive task caught exception after receipt of ["+numReceived+
"] messages: " + e.getMessage());
}
}
}
});
}
ready.countDown();
assertTrue("Timed out waiting for receivers to start.", started.await(5, TimeUnit.MINUTES));
String frame;
// Lets still wait a bit to make sure all subscribers get a fair shake at
// getting online before we send. Account for slow Hudson machines
TimeUnit.SECONDS.sleep(5);
for( int ix = 0; ix < MSG_COUNT; ix++) {
frame = "SEND\n destination:/topic/" + getDestinationName() +
"\nid:" + ix +
"\ncontent-length:5" + " \n\n" +
"\u0001\u0002\u0000\u0004\u0005" + Stomp.NULL;
stompConnection.sendFrame(frame);
}
LOG.info("All " + MSG_COUNT + " message have been sent, awaiting receipt.");
assertTrue("Should get [" + TASK_COUNT * MSG_COUNT + "] message but was: " + receiveCount.get(), Wait.waitFor(new Wait.Condition() {
@Override
public boolean isSatisified() throws Exception {
return receiveCount.get() == TASK_COUNT * MSG_COUNT;
}
}, TimeUnit.MINUTES.toMillis(10)));
LOG.info("Test Completed and all messages received, shutting down.");
executor.shutdown();
executor.awaitTermination(2, TimeUnit.MINUTES);
frame = "DISCONNECT\n" + "\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame);
LOG.info("Test Finished.");
}
protected void stompConnect() throws Exception {
URI connectUri = new URI(bindAddress);
LOG.debug("Attempting connection to: " + bindAddress);
stompConnection.open(createSocket(connectUri));
stompConnection.connect("system", "manager");
}
private void stompConnect(StompConnection connection) throws Exception {
URI connectUri = new URI(bindAddress);
LOG.debug("Attempting connection to: " + bindAddress);
connection.open(createSocket(connectUri));
connection.connect("system", "manager");
}
protected Socket createSocket(URI connectUri) throws IOException {
return new Socket("127.0.0.1", connectUri.getPort());
}
protected String getDestinationName() {
return getClass().getName() + ".Tester";
}
protected void stompDisconnect() throws IOException {
if (stompConnection != null) {
stompConnection.close();
stompConnection = null;
}
}
}

38
activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompNIOLoadTest.java Normal file
View File

@ -0,0 +1,38 @@
/**
* 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.stomp;
import org.junit.After;
import org.junit.Before;
public class StompNIOLoadTest extends StompLoadTest {
@Before
@Override
public void setUp() throws Exception {
bindAddress = "stomp+nio://localhost:61612";
confUri = "xbean:org/apache/activemq/transport/stomp/niostomp-auth-broker.xml";
super.setUp();
}
@After
@Override
public void tearDown() throws Exception {
super.tearDown();
}
}

57
activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompNIOSSLLoadTest.java Normal file
View File

@ -0,0 +1,57 @@
/**
* 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.stomp;
import java.io.IOException;
import java.net.Socket;
import java.net.URI;
import javax.net.SocketFactory;
import javax.net.ssl.SSLSocketFactory;
import org.junit.After;
import org.junit.Before;
public class StompNIOSSLLoadTest extends StompLoadTest {
@Before
@Override
public void setUp() throws Exception {
bindAddress = "stomp+nio+ssl://localhost:61613";
confUri = "xbean:org/apache/activemq/transport/stomp/sslstomp-auth-broker.xml";
System.setProperty("javax.net.ssl.trustStore", "src/test/resources/client.keystore");
System.setProperty("javax.net.ssl.trustStorePassword", "password");
System.setProperty("javax.net.ssl.trustStoreType", "jks");
System.setProperty("javax.net.ssl.keyStore", "src/test/resources/server.keystore");
System.setProperty("javax.net.ssl.keyStorePassword", "password");
System.setProperty("javax.net.ssl.keyStoreType", "jks");
super.setUp();
}
@After
@Override
public void tearDown() throws Exception {
super.tearDown();
}
@Override
protected Socket createSocket(URI connectUri) throws IOException {
SocketFactory factory = SSLSocketFactory.getDefault();
return factory.createSocket("127.0.0.1", connectUri.getPort());
}
}

57
activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompSSLLoadTest.java Normal file
View File

@ -0,0 +1,57 @@
/**
* 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.stomp;
import java.io.IOException;
import java.net.Socket;
import java.net.URI;
import javax.net.SocketFactory;
import javax.net.ssl.SSLSocketFactory;
import org.junit.After;
import org.junit.Before;
public class StompSSLLoadTest extends StompLoadTest {
@Before
@Override
public void setUp() throws Exception {
bindAddress = "stomp+ssl://localhost:61612";
confUri = "xbean:org/apache/activemq/transport/stomp/sslstomp-auth-broker.xml";
System.setProperty("javax.net.ssl.trustStore", "src/test/resources/client.keystore");
System.setProperty("javax.net.ssl.trustStorePassword", "password");
System.setProperty("javax.net.ssl.trustStoreType", "jks");
System.setProperty("javax.net.ssl.keyStore", "src/test/resources/server.keystore");
System.setProperty("javax.net.ssl.keyStorePassword", "password");
System.setProperty("javax.net.ssl.keyStoreType", "jks");
super.setUp();
}
@After
@Override
public void tearDown() throws Exception {
super.tearDown();
}
@Override
protected Socket createSocket(URI connectUri) throws IOException {
SocketFactory factory = SSLSocketFactory.getDefault();
return factory.createSocket("127.0.0.1", connectUri.getPort());
}
}

128
activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java
View File

@ -188,7 +188,7 @@ public class StompTest extends CombinationTestSupport {
public void testConnect() throws Exception { public void testConnect() throws Exception {
String connectFrame = "CONNECT\n" + "login: system\n" + "passcode: manager\n" + "request-id: 1\n" + "\n" + Stomp.NULL; String connectFrame = "CONNECT\n" + "login:system\n" + "passcode:manager\n" + "request-id:1\n" + "\n" + Stomp.NULL;
stompConnection.sendFrame(connectFrame); stompConnection.sendFrame(connectFrame);
String f = stompConnection.receiveFrame(); String f = stompConnection.receiveFrame();
@ -201,7 +201,7 @@ public class StompTest extends CombinationTestSupport {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -226,13 +226,13 @@ public class StompTest extends CombinationTestSupport {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
assertTrue(frame.startsWith("CONNECTED")); assertTrue(frame.startsWith("CONNECTED"));
frame = "SEND\n" + "destination:/queue/" + getQueueName() + "\n" + "JMSXGroupID: TEST\n\n" + "Hello World" + Stomp.NULL; frame = "SEND\n" + "destination:/queue/" + getQueueName() + "\n" + "JMSXGroupID:TEST\n\n" + "Hello World" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
@ -245,7 +245,7 @@ public class StompTest extends CombinationTestSupport {
MessageConsumer consumer = session.createConsumer(queue, "foo = 'abc'"); MessageConsumer consumer = session.createConsumer(queue, "foo = 'abc'");
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -266,7 +266,7 @@ public class StompTest extends CombinationTestSupport {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -286,7 +286,7 @@ public class StompTest extends CombinationTestSupport {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -315,7 +315,7 @@ public class StompTest extends CombinationTestSupport {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -337,7 +337,7 @@ public class StompTest extends CombinationTestSupport {
StompConnection receiver = new StompConnection(); StompConnection receiver = new StompConnection();
URI connectUri = new URI(bindAddress); URI connectUri = new URI(bindAddress);
receiver.open(createSocket(connectUri)); receiver.open(createSocket(connectUri));
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
receiver.sendFrame(frame); receiver.sendFrame(frame);
frame = receiver.receiveFrame(); frame = receiver.receiveFrame();
@ -346,7 +346,7 @@ public class StompTest extends CombinationTestSupport {
frame = "SUBSCRIBE\n" + "destination:/queue/" + getQueueName() + "\n" + "ack:auto\n\n" + Stomp.NULL; frame = "SUBSCRIBE\n" + "destination:/queue/" + getQueueName() + "\n" + "ack:auto\n\n" + Stomp.NULL;
receiver.sendFrame(frame); receiver.sendFrame(frame);
frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -395,7 +395,7 @@ public class StompTest extends CombinationTestSupport {
StompConnection sender = new StompConnection(); StompConnection sender = new StompConnection();
sender.open(createSocket(connectUri)); sender.open(createSocket(connectUri));
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
sender.sendFrame(frame); sender.sendFrame(frame);
frame = sender.receiveFrame(); frame = sender.receiveFrame();
@ -411,7 +411,7 @@ public class StompTest extends CombinationTestSupport {
StompConnection receiver = new StompConnection(); StompConnection receiver = new StompConnection();
receiver.open(createSocket(connectUri)); receiver.open(createSocket(connectUri));
frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
receiver.sendFrame(frame); receiver.sendFrame(frame);
frame = receiver.receiveFrame(); frame = receiver.receiveFrame();
@ -439,7 +439,7 @@ public class StompTest extends CombinationTestSupport {
public void testSubscribeWithAutoAck() throws Exception { public void testSubscribeWithAutoAck() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -459,7 +459,7 @@ public class StompTest extends CombinationTestSupport {
public void testSubscribeWithAutoAckAndBytesMessage() throws Exception { public void testSubscribeWithAutoAckAndBytesMessage() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -488,7 +488,7 @@ public class StompTest extends CombinationTestSupport {
public void testBytesMessageWithNulls() throws Exception { public void testBytesMessageWithNulls() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -516,7 +516,7 @@ public class StompTest extends CombinationTestSupport {
final int MSG_COUNT = 50; final int MSG_COUNT = 50;
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -546,7 +546,7 @@ public class StompTest extends CombinationTestSupport {
public void testSubscribeWithMessageSentWithProperties() throws Exception { public void testSubscribeWithMessageSentWithProperties() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -579,7 +579,7 @@ public class StompTest extends CombinationTestSupport {
int ctr = 10; int ctr = 10;
String[] data = new String[ctr]; String[] data = new String[ctr];
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -617,7 +617,7 @@ public class StompTest extends CombinationTestSupport {
public void testSubscribeWithAutoAckAndSelector() throws Exception { public void testSubscribeWithAutoAckAndSelector() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -639,7 +639,7 @@ public class StompTest extends CombinationTestSupport {
public void testSubscribeWithAutoAckAndNumericSelector() throws Exception { public void testSubscribeWithAutoAckAndNumericSelector() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -666,7 +666,7 @@ public class StompTest extends CombinationTestSupport {
public void testSubscribeWithAutoAckAndBooleanSelector() throws Exception { public void testSubscribeWithAutoAckAndBooleanSelector() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -693,7 +693,7 @@ public class StompTest extends CombinationTestSupport {
public void testSubscribeWithAutoAckAnFloatSelector() throws Exception { public void testSubscribeWithAutoAckAnFloatSelector() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -720,7 +720,7 @@ public class StompTest extends CombinationTestSupport {
public void testSubscribeWithClientAck() throws Exception { public void testSubscribeWithClientAck() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -744,7 +744,7 @@ public class StompTest extends CombinationTestSupport {
public void testSubscribeWithClientAckedAndContentLength() throws Exception { public void testSubscribeWithClientAckedAndContentLength() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -788,7 +788,7 @@ public class StompTest extends CombinationTestSupport {
public void testUnsubscribe() throws Exception { public void testUnsubscribe() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
assertTrue(frame.startsWith("CONNECTED")); assertTrue(frame.startsWith("CONNECTED"));
@ -824,7 +824,7 @@ public class StompTest extends CombinationTestSupport {
public void testTransactionCommit() throws Exception { public void testTransactionCommit() throws Exception {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
String f = stompConnection.receiveFrame(); String f = stompConnection.receiveFrame();
@ -846,7 +846,7 @@ public class StompTest extends CombinationTestSupport {
public void testTransactionRollback() throws Exception { public void testTransactionRollback() throws Exception {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
String f = stompConnection.receiveFrame(); String f = stompConnection.receiveFrame();
@ -879,7 +879,7 @@ public class StompTest extends CombinationTestSupport {
public void testDisconnectedClientsAreRemovedFromTheBroker() throws Exception { public void testDisconnectedClientsAreRemovedFromTheBroker() throws Exception {
assertClients(1); assertClients(1);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
@ -892,7 +892,7 @@ public class StompTest extends CombinationTestSupport {
} }
public void testConnectNotAuthenticatedWrongUser() throws Exception { public void testConnectNotAuthenticatedWrongUser() throws Exception {
String frame = "CONNECT\n" + "login: dejanb\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login: dejanb\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
String f = stompConnection.receiveFrame(); String f = stompConnection.receiveFrame();
@ -903,7 +903,7 @@ public class StompTest extends CombinationTestSupport {
public void testConnectNotAuthenticatedWrongPassword() throws Exception { public void testConnectNotAuthenticatedWrongPassword() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: dejanb\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode: dejanb\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
String f = stompConnection.receiveFrame(); String f = stompConnection.receiveFrame();
@ -914,7 +914,7 @@ public class StompTest extends CombinationTestSupport {
public void testSendNotAuthorized() throws Exception { public void testSendNotAuthorized() throws Exception {
String frame = "CONNECT\n" + "login: guest\n" + "passcode: password\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:guest\n" + "passcode:password\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -929,7 +929,7 @@ public class StompTest extends CombinationTestSupport {
public void testSubscribeNotAuthorized() throws Exception { public void testSubscribeNotAuthorized() throws Exception {
String frame = "CONNECT\n" + "login: guest\n" + "passcode: password\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:guest\n" + "passcode:password\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -946,7 +946,7 @@ public class StompTest extends CombinationTestSupport {
public void testTransformationUnknownTranslator() throws Exception { public void testTransformationUnknownTranslator() throws Exception {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -964,7 +964,7 @@ public class StompTest extends CombinationTestSupport {
public void testTransformationFailed() throws Exception { public void testTransformationFailed() throws Exception {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -983,7 +983,7 @@ public class StompTest extends CombinationTestSupport {
public void testTransformationSendXMLObject() throws Exception { public void testTransformationSendXMLObject() throws Exception {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1002,7 +1002,7 @@ public class StompTest extends CombinationTestSupport {
public void testTransformationSendJSONObject() throws Exception { public void testTransformationSendJSONObject() throws Exception {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1024,7 +1024,7 @@ public class StompTest extends CombinationTestSupport {
ObjectMessage message = session.createObjectMessage(new SamplePojo("Dejan", "Belgrade")); ObjectMessage message = session.createObjectMessage(new SamplePojo("Dejan", "Belgrade"));
producer.send(message); producer.send(message);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1046,7 +1046,7 @@ public class StompTest extends CombinationTestSupport {
ObjectMessage message = session.createObjectMessage(new SamplePojo("Dejan", "Belgrade")); ObjectMessage message = session.createObjectMessage(new SamplePojo("Dejan", "Belgrade"));
producer.send(message); producer.send(message);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1069,7 +1069,7 @@ public class StompTest extends CombinationTestSupport {
ObjectMessage message = session.createObjectMessage(new SamplePojo("Dejan", "Belgrade")); ObjectMessage message = session.createObjectMessage(new SamplePojo("Dejan", "Belgrade"));
producer.send(message); producer.send(message);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1092,7 +1092,7 @@ public class StompTest extends CombinationTestSupport {
ObjectMessage message = session.createObjectMessage(new SamplePojo("Dejan", "Belgrade")); ObjectMessage message = session.createObjectMessage(new SamplePojo("Dejan", "Belgrade"));
producer.send(message); producer.send(message);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1119,7 +1119,7 @@ public class StompTest extends CombinationTestSupport {
mapMessage.setString("city", "Belgrade"); mapMessage.setString("city", "Belgrade");
producer.send(mapMessage); producer.send(mapMessage);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1150,7 +1150,7 @@ public class StompTest extends CombinationTestSupport {
mapMessage.setString("city", "Belgrade"); mapMessage.setString("city", "Belgrade");
producer.send(mapMessage); producer.send(mapMessage);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1173,7 +1173,7 @@ public class StompTest extends CombinationTestSupport {
public void testTransformationSendAndReceiveXmlMap() throws Exception { public void testTransformationSendAndReceiveXmlMap() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1195,7 +1195,7 @@ public class StompTest extends CombinationTestSupport {
public void testTransformationSendAndReceiveJsonMap() throws Exception { public void testTransformationSendAndReceiveJsonMap() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1222,7 +1222,7 @@ public class StompTest extends CombinationTestSupport {
message.writeBytes(new byte[]{1, 2, 3, 4, 5}); message.writeBytes(new byte[]{1, 2, 3, 4, 5});
producer.send(message); producer.send(message);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1251,7 +1251,7 @@ public class StompTest extends CombinationTestSupport {
message.setStringProperty("transformation", Stomp.Transformations.JMS_OBJECT_XML.toString()); message.setStringProperty("transformation", Stomp.Transformations.JMS_OBJECT_XML.toString());
producer.send(message); producer.send(message);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1274,7 +1274,7 @@ public class StompTest extends CombinationTestSupport {
message.setStringProperty("transformation", Stomp.Transformations.JMS_OBJECT_XML.toString()); message.setStringProperty("transformation", Stomp.Transformations.JMS_OBJECT_XML.toString());
producer.send(message); producer.send(message);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1294,7 +1294,7 @@ public class StompTest extends CombinationTestSupport {
public void testTransformationSendXMLMap() throws Exception { public void testTransformationSendXMLMap() throws Exception {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1312,7 +1312,7 @@ public class StompTest extends CombinationTestSupport {
public void testTransformationSendJSONMap() throws Exception { public void testTransformationSendJSONMap() throws Exception {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1335,7 +1335,7 @@ public class StompTest extends CombinationTestSupport {
message.setString("city", "Belgrade"); message.setString("city", "Belgrade");
producer.send(message); producer.send(message);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1360,7 +1360,7 @@ public class StompTest extends CombinationTestSupport {
message.setString("city", "Belgrade"); message.setString("city", "Belgrade");
producer.send(message); producer.send(message);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1386,7 +1386,7 @@ public class StompTest extends CombinationTestSupport {
BrokerViewMBean view = (BrokerViewMBean)broker.getManagementContext().newProxyInstance(brokerName, BrokerViewMBean.class, true); BrokerViewMBean view = (BrokerViewMBean)broker.getManagementContext().newProxyInstance(brokerName, BrokerViewMBean.class, true);
// connect // connect
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\nclient-id:test\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\nclient-id:test\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1412,7 +1412,7 @@ public class StompTest extends CombinationTestSupport {
//reconnect //reconnect
stompConnect(); stompConnect();
// connect // connect
frame = "CONNECT\n" + "login: system\n" + "passcode: manager\nclient-id:test\n\n" + Stomp.NULL; frame = "CONNECT\n" + "login:system\n" + "passcode:manager\nclient-id:test\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
assertTrue(frame.startsWith("CONNECTED")); assertTrue(frame.startsWith("CONNECTED"));
@ -1530,7 +1530,7 @@ public class StompTest extends CombinationTestSupport {
public void testTempDestination() throws Exception { public void testTempDestination() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1550,7 +1550,7 @@ public class StompTest extends CombinationTestSupport {
MessageConsumer consumer = session.createConsumer(queue); MessageConsumer consumer = session.createConsumer(queue);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1567,7 +1567,7 @@ public class StompTest extends CombinationTestSupport {
public void testJMSXUserIDIsSetInStompMessage() throws Exception { public void testJMSXUserIDIsSetInStompMessage() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1672,7 +1672,7 @@ public class StompTest extends CombinationTestSupport {
public void testReceiptNewQueue() throws Exception { public void testReceiptNewQueue() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1707,7 +1707,7 @@ public class StompTest extends CombinationTestSupport {
} }
public void testTransactedClientAckBrokerStats() throws Exception { public void testTransactedClientAckBrokerStats() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1748,7 +1748,7 @@ public class StompTest extends CombinationTestSupport {
public void testReplytoModification() throws Exception { public void testReplytoModification() throws Exception {
String replyto = "some destination"; String replyto = "some destination";
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1769,7 +1769,7 @@ public class StompTest extends CombinationTestSupport {
public void testReplyToDestinationNaming() throws Exception { public void testReplyToDestinationNaming() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1780,7 +1780,7 @@ public class StompTest extends CombinationTestSupport {
} }
public void testSendNullBodyTextMessage() throws Exception { public void testSendNullBodyTextMessage() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1841,7 +1841,7 @@ public class StompTest extends CombinationTestSupport {
public void testReplyToAcrossConnections() throws Exception { public void testReplyToAcrossConnections() throws Exception {
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();
@ -1856,7 +1856,7 @@ public class StompTest extends CombinationTestSupport {
StompConnection responder = new StompConnection(); StompConnection responder = new StompConnection();
stompConnect(responder); stompConnect(responder);
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
responder.sendFrame(frame); responder.sendFrame(frame);
frame = responder.receiveFrame(); frame = responder.receiveFrame();
@ -1950,7 +1950,7 @@ public class StompTest extends CombinationTestSupport {
private void doTestConnectionLeak() throws Exception { private void doTestConnectionLeak() throws Exception {
stompConnect(); stompConnect();
String frame = "CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL; String frame = "CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame); stompConnection.sendFrame(frame);
frame = stompConnection.receiveFrame(); frame = stompConnection.receiveFrame();

3
activemq-core/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineTest.java
View File

@ -1610,7 +1610,7 @@ public class DurableSubscriptionOfflineTest extends org.apache.activemq.TestSupp
} }
executorService.shutdown(); executorService.shutdown();
executorService.awaitTermination(10, TimeUnit.MINUTES); executorService.awaitTermination(10, TimeUnit.MINUTES);
assertTrue("No exceptions", exceptions.isEmpty()); assertTrue("No exceptions " + exceptions.elements(), exceptions.isEmpty());
} }
public static class Listener implements MessageListener { public static class Listener implements MessageListener {
@ -1649,6 +1649,7 @@ public class DurableSubscriptionOfflineTest extends org.apache.activemq.TestSupp
} }
} }
catch (JMSException e) { catch (JMSException e) {
e.printStackTrace();
exceptions.add(e); exceptions.add(e);
} }
} }

2
activemq-core/src/test/java/org/apache/activemq/usecases/ThreeBrokerStompTemporaryQueueTest.java
View File

@ -70,7 +70,7 @@ public class ThreeBrokerStompTemporaryQueueTest extends JmsMultipleBrokersTestSu
stompConnection = new StompConnection(); stompConnection = new StompConnection();
stompConnection.open("localhost", 61614); stompConnection.open("localhost", 61614);
// Creating a temp queue // Creating a temp queue
stompConnection.sendFrame("CONNECT\n" + "login: system\n" + "passcode: manager\n\n" + Stomp.NULL); stompConnection.sendFrame("CONNECT\n" + "login:system\n" + "passcode:manager\n\n" + Stomp.NULL);
StompFrame frame = stompConnection.receive(); StompFrame frame = stompConnection.receive();
assertTrue(frame.toString().startsWith("CONNECTED")); assertTrue(frame.toString().startsWith("CONNECTED"));

4
assembly/src/main/descriptors/windows-bin.xml
View File

@ -44,6 +44,8 @@
<include>bin/activemq-admin</include> <include>bin/activemq-admin</include>
<include>bin/win32/*.bat</include> <include>bin/win32/*.bat</include>
<include>bin/win32/*.conf</include> <include>bin/win32/*.conf</include>
<include>bin/win64/*.bat</include>
<include>bin/win64/*.conf</include>
</includes> </includes>
<lineEnding>dos</lineEnding> <lineEnding>dos</lineEnding>
</fileSet> </fileSet>
@ -55,6 +57,8 @@
<includes> <includes>
<include>bin/win32/*.exe</include> <include>bin/win32/*.exe</include>
<include>bin/win32/*.dll</include> <include>bin/win32/*.dll</include>
<include>bin/win64/*.exe</include>
<include>bin/win64/*.dll</include>
</includes> </includes>
</fileSet> </fileSet>

52
assembly/src/release/bin/win64/InstallService.bat Normal file
View File

@ -0,0 +1,52 @@
@echo off
REM ------------------------------------------------------------------------
REM Licensed to the Apache Software Foundation (ASF) under one or more
REM contributor license agreements. See the NOTICE file distributed with
REM this work for additional information regarding copyright ownership.
REM The ASF licenses this file to You under the Apache License, Version 2.0
REM (the "License"); you may not use this file except in compliance with
REM the License. You may obtain a copy of the License at
REM
REM http://www.apache.org/licenses/LICENSE-2.0
REM
REM Unless required by applicable law or agreed to in writing, software
REM distributed under the License is distributed on an "AS IS" BASIS,
REM WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
REM See the License for the specific language governing permissions and
REM limitations under the License.
REM ------------------------------------------------------------------------
setlocal
rem Java Service Wrapper general NT service install script
if "%OS%"=="Windows_NT" goto nt
echo This script only works with NT-based versions of Windows.
goto :eof
:nt
rem
rem Find the application home.
rem
rem %~dp0 is location of current script under NT
set _REALPATH=%~dp0
set ACTIVEMQ_HOME=%~dp0..\..
set ACTIVEMQ_BASE=%~dp0..\..
:conf
set _WRAPPER_CONF="%ACTIVEMQ_HOME%\bin\win64\wrapper.conf"
set _ACTIVEMQ_HOME="set.ACTIVEMQ_HOME=%ACTIVEMQ_HOME%"
set _ACTIVEMQ_BASE="set.ACTIVEMQ_BASE=%ACTIVEMQ_BASE%"
rem
rem Install the Wrapper as an NT service.
rem
:startup
"wrapper.exe" -i %_WRAPPER_CONF% %_ACTIVEMQ_HOME% %_ACTIVEMQ_BASE%
if not errorlevel 1 goto :eof
pause

48
assembly/src/release/bin/win64/UninstallService.bat Normal file
View File

@ -0,0 +1,48 @@
@echo off
REM ------------------------------------------------------------------------
REM Licensed to the Apache Software Foundation (ASF) under one or more
REM contributor license agreements. See the NOTICE file distributed with
REM this work for additional information regarding copyright ownership.
REM The ASF licenses this file to You under the Apache License, Version 2.0
REM (the "License"); you may not use this file except in compliance with
REM the License. You may obtain a copy of the License at
REM
REM http://www.apache.org/licenses/LICENSE-2.0
REM
REM Unless required by applicable law or agreed to in writing, software
REM distributed under the License is distributed on an "AS IS" BASIS,
REM WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
REM See the License for the specific language governing permissions and
REM limitations under the License.
REM ------------------------------------------------------------------------
setlocal
rem Java Service Wrapper general NT service uninstall script
if "%OS%"=="Windows_NT" goto nt
echo This script only works with NT-based versions of Windows.
goto :eof
:nt
rem
rem Find the application home.
rem
rem %~dp0 is location of current script under NT
set _REALPATH=%~dp0
set ACTIVEMQ_HOME=%~dp0\..\..
:conf
set _WRAPPER_CONF="%ACTIVEMQ_HOME%\bin\win64\wrapper.conf"
rem
rem Uninstall the Wrapper as an NT service.
rem
:startup
"%_APP_HOME%wrapper.exe" -r %_WRAPPER_CONF%
if not errorlevel 1 goto :eof
pause

50
assembly/src/release/bin/win64/activemq.bat Normal file
View File

@ -0,0 +1,50 @@
@echo off
REM ------------------------------------------------------------------------
REM Licensed to the Apache Software Foundation (ASF) under one or more
REM contributor license agreements. See the NOTICE file distributed with
REM this work for additional information regarding copyright ownership.
REM The ASF licenses this file to You under the Apache License, Version 2.0
REM (the "License"); you may not use this file except in compliance with
REM the License. You may obtain a copy of the License at
REM
REM http://www.apache.org/licenses/LICENSE-2.0
REM
REM Unless required by applicable law or agreed to in writing, software
REM distributed under the License is distributed on an "AS IS" BASIS,
REM WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
REM See the License for the specific language governing permissions and
REM limitations under the License.
REM ------------------------------------------------------------------------
rem
rem Find the application home.
rem
if "%OS%"=="Windows_NT" goto nt
echo This is not NT, so please edit this script and set _APP_HOME manually
set _APP_HOME=..
goto conf
:nt
rem %~dp0 is name of current script under NT
set _APP_HOME=%~dp0
rem
rem Find the wrapper.conf
rem
:conf
set _WRAPPER_CONF=wrapper.conf
rem
rem Run the application.
rem At runtime, the current directory will be that of Wrapper.exe
rem
"%_APP_HOME%wrapper.exe" -c %_WRAPPER_CONF%
if not errorlevel 1 goto end
pause
:end
set _APP_HOME=
set _WRAPPER_CONF=

143
assembly/src/release/bin/win64/wrapper.conf Normal file
View File

@ -0,0 +1,143 @@
# ------------------------------------------------------------------------
# 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.
# ------------------------------------------------------------------------
#********************************************************************
# Wrapper Properties
#********************************************************************
# wrapper.debug=TRUE
set.default.ACTIVEMQ_HOME=../..
set.default.ACTIVEMQ_BASE=../..
set.default.ACTIVEMQ_CONF=%ACTIVEMQ_BASE%/conf
set.default.ACTIVEMQ_DATA=%ACTIVEMQ_BASE%/data
wrapper.working.dir=.
# Java Application
wrapper.java.command=java
# Java Main class. This class must implement the WrapperListener interface
# or guarantee that the WrapperManager class is initialized. Helper
# classes are provided to do this for you. See the Integration section
# of the documentation for details.
wrapper.java.mainclass=org.tanukisoftware.wrapper.WrapperSimpleApp
# Java Classpath (include wrapper.jar) Add class path elements as
# needed starting from 1
wrapper.java.classpath.1=%ACTIVEMQ_HOME%/bin/wrapper.jar
wrapper.java.classpath.2=%ACTIVEMQ_HOME%/bin/run.jar
# Java Library Path (location of Wrapper.DLL or libwrapper.so)
wrapper.java.library.path.1=%ACTIVEMQ_HOME%/bin/win64
# Java Additional Parameters
# note that n is the parameter number starting from 1.
wrapper.java.additional.1=-Dactivemq.home="%ACTIVEMQ_HOME%"
wrapper.java.additional.2=-Dactivemq.base="%ACTIVEMQ_BASE%"
wrapper.java.additional.3=-Djavax.net.ssl.keyStorePassword=password
wrapper.java.additional.4=-Djavax.net.ssl.trustStorePassword=password
wrapper.java.additional.5=-Djavax.net.ssl.keyStore=%ACTIVEMQ_CONF%/broker.ks
wrapper.java.additional.6=-Djavax.net.ssl.trustStore=%ACTIVEMQ_CONF%/broker.ts
wrapper.java.additional.7=-Dcom.sun.management.jmxremote
wrapper.java.additional.8=-Dorg.apache.activemq.UseDedicatedTaskRunner=true
wrapper.java.additional.9=-Djava.util.logging.config.file=logging.properties
wrapper.java.additional.10=-Dactivemq.conf="%ACTIVEMQ_CONF%"
wrapper.java.additional.11=-Dactivemq.data="%ACTIVEMQ_DATA%"
# Uncomment to enable remote jmx
#wrapper.java.additional.n=-Dcom.sun.management.jmxremote.port=1616
#wrapper.java.additional.n=-Dcom.sun.management.jmxremote.authenticate=false
#wrapper.java.additional.n=-Dcom.sun.management.jmxremote.ssl=false
# Uncomment to enable YourKit profiling
#wrapper.java.additional.n=-Xrunyjpagent
# Uncomment to enable remote debugging
#wrapper.java.additional.n=-Xdebug -Xnoagent -Djava.compiler=NONE
#wrapper.java.additional.n=-Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=5005
# Initial Java Heap Size (in MB)
#wrapper.java.initmemory=3
# Maximum Java Heap Size (in MB)
wrapper.java.maxmemory=1024
# Application parameters. Add parameters as needed starting from 1
wrapper.app.parameter.1=org.apache.activemq.console.Main
wrapper.app.parameter.2=start
#********************************************************************
# Wrapper Logging Properties
#********************************************************************
# Format of output for the console. (See docs for formats)
wrapper.console.format=PM
# Log Level for console output. (See docs for log levels)
wrapper.console.loglevel=INFO
# Log file to use for wrapper output logging.
wrapper.logfile=%ACTIVEMQ_DATA%/wrapper.log
# Format of output for the log file. (See docs for formats)
wrapper.logfile.format=LPTM
# Log Level for log file output. (See docs for log levels)
wrapper.logfile.loglevel=INFO
# Maximum size that the log file will be allowed to grow to before
# the log is rolled. Size is specified in bytes. The default value
# of 0, disables log rolling. May abbreviate with the 'k' (kb) or
# 'm' (mb) suffix. For example: 10m = 10 megabytes.
wrapper.logfile.maxsize=0
# Maximum number of rolled log files which will be allowed before old
# files are deleted. The default value of 0 implies no limit.
wrapper.logfile.maxfiles=0
# Log Level for sys/event log output. (See docs for log levels)
wrapper.syslog.loglevel=NONE
#********************************************************************
# Wrapper Windows Properties
#********************************************************************
# Title to use when running as a console
wrapper.console.title=ActiveMQ
#********************************************************************
# Wrapper Windows NT/2000/XP Service Properties
#********************************************************************
# WARNING - Do not modify any of these properties when an application
# using this configuration file has been installed as a service.
# Please uninstall the service before modifying this section. The
# service can then be reinstalled.
# Name of the service
wrapper.ntservice.name=ActiveMQ
# Display name of the service
wrapper.ntservice.displayname=ActiveMQ
# Description of the service
wrapper.ntservice.description=ActiveMQ Broker
# Service dependencies. Add dependencies as needed starting from 1
wrapper.ntservice.dependency.1=
# Mode in which the service is installed. AUTO_START or DEMAND_START
wrapper.ntservice.starttype=AUTO_START
# Allow the service to interact with the desktop.
wrapper.ntservice.interactive=false

BIN
assembly/src/release/bin/win64/wrapper.dll Normal file
View File

Binary file not shown.

BIN
assembly/src/release/bin/win64/wrapper.exe Normal file
View File

Binary file not shown.