[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
diff --git a/activemq-core/src/main/java/org/apache/activemq/plugin/StatisticsBroker.java b/activemq-core/src/main/java/org/apache/activemq/plugin/StatisticsBroker.java
index c5df8d9..2e2d4f0 100644
--- a/activemq-core/src/main/java/org/apache/activemq/plugin/StatisticsBroker.java
+++ b/activemq-core/src/main/java/org/apache/activemq/plugin/StatisticsBroker.java
@@ -22,6 +22,8 @@
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.ConnectionContext;
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.DestinationStatistics;
import org.apache.activemq.broker.region.RegionBroker;
@@ -37,6 +39,9 @@
import org.apache.activemq.util.LongSequenceGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+
+import javax.jms.JMSException;
+import javax.management.ObjectName;
import java.io.File;
import java.net.URI;
import java.util.Set;
@@ -50,9 +55,11 @@
private static Logger LOG = LoggerFactory.getLogger(StatisticsBroker.class);
static final String STATS_DESTINATION_PREFIX = "ActiveMQ.Statistics.Destination";
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 final LongSequenceGenerator messageIdGenerator = new LongSequenceGenerator();
protected final ProducerId advisoryProducerId = new ProducerId();
+ protected BrokerViewMBean brokerView;
/**
*
@@ -80,6 +87,10 @@
STATS_DESTINATION_PREFIX.length());
boolean brokerStats = physicalName.regionMatches(true, 0, STATS_BROKER_PREFIX, 0, STATS_BROKER_PREFIX
.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) {
String queueryName = physicalName.substring(STATS_DESTINATION_PREFIX.length(), physicalName.length());
ActiveMQDestination queryDest = ActiveMQDestination.createDestination(queueryName,msgDest.getDestinationType());
@@ -108,10 +119,11 @@
sendStats(producerExchange.getConnectionContext(), statsMessage, replyTo);
}
}
+ } else if (subStats) {
+ sendSubStats(producerExchange.getConnectionContext(), getBrokerView().getQueueSubscribers(), replyTo);
+ sendSubStats(producerExchange.getConnectionContext(), getBrokerView().getTopicSubscribers(), replyTo);
} else if (brokerStats) {
ActiveMQMapMessage statsMessage = new ActiveMQMapMessage();
- BrokerService brokerService = getBrokerService();
- RegionBroker regionBroker = (RegionBroker) brokerService.getRegionBroker();
SystemUsage systemUsage = brokerService.getSystemUsage();
DestinationStatistics stats = regionBroker.getDestinationStatistics();
statsMessage.setString("brokerName", regionBroker.getBrokerName());
@@ -165,6 +177,15 @@
}
}
+ 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 {
super.start();
LOG.info("Starting StatisticsBroker");
@@ -174,6 +195,34 @@
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)
throws Exception {
msg.setPersistent(false);
diff --git a/activemq-core/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java b/activemq-core/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
index 0534038..b9baeb3 100644
--- a/activemq-core/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
+++ b/activemq-core/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
@@ -135,7 +135,11 @@
while(true) {
if (!plain.hasRemaining()) {
- plain.clear();
+ if (status == SSLEngineResult.Status.OK && handshakeStatus != SSLEngineResult.HandshakeStatus.NEED_UNWRAP) {
+ plain.clear();
+ } else {
+ plain.compact();
+ }
int readCount = secureRead(plain);
@@ -150,7 +154,9 @@
}
}
- processCommand(plain);
+ if (status == SSLEngineResult.Status.OK && handshakeStatus != SSLEngineResult.HandshakeStatus.NEED_UNWRAP) {
+ processCommand(plain);
+ }
}
} catch (IOException e) {
@@ -192,7 +198,7 @@
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);
if (bytesRead == -1) {
diff --git a/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompWireFormat.java b/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompWireFormat.java
index 282443e..d276f67 100644
--- a/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompWireFormat.java
+++ b/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompWireFormat.java
@@ -16,15 +16,21 @@
*/
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.ByteArrayOutputStream;
import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.wireformat.WireFormat;
-import java.io.*;
-import java.util.HashMap;
-import java.util.Map;
-
/**
* Implements marshalling and unmarsalling the <a
* href="http://stomp.codehaus.org/">Stomp</a> protocol.
@@ -201,7 +207,7 @@
ByteSequence nameSeq = stream.toByteSequence();
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);
} catch (Exception e) {
throw new ProtocolException("Unable to parser header line [" + line + "]", true);
diff --git a/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2571Test.java b/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2571Test.java
index ee2b3fe..533ae0c 100644
--- a/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2571Test.java
+++ b/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2571Test.java
@@ -32,6 +32,7 @@
public void testTempQueueClosing() {
try {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(this.bindAddress);
+ connectionFactory.setAlwaysSyncSend(true);
// First create session that will own the TempQueue
Connection connectionA = connectionFactory.createConnection();
@@ -56,13 +57,13 @@
Thread sendingThread = new Thread(new Runnable() {
public void run() {
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);
}
} catch (JMSException e) {
- // We don't get this exception every time.
- // Not getting it means that we don't know if the
- // creator of the TempQueue has disconnected.
+ e.printStackTrace();
}
}
});
@@ -72,7 +73,7 @@
// Now close connection A. This will remove the TempQueue.
connectionA.close();
// 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
// TempQueue is gone.
@@ -95,6 +96,7 @@
@Override
protected void setUp() throws Exception {
bindAddress = "vm://localhost";
+ setAutoFail(true);
super.setUp();
}
@@ -103,7 +105,6 @@
BrokerService answer = new BrokerService();
answer.setPersistent(false);
answer.setUseJmx(false);
- answer.addConnector(bindAddress);
return answer;
}
}
\ No newline at end of file
diff --git a/activemq-core/src/test/java/org/apache/activemq/plugin/BrokerStatisticsPluginTest.java b/activemq-core/src/test/java/org/apache/activemq/plugin/BrokerStatisticsPluginTest.java
index 79f07cc..c6b415b 100644
--- a/activemq-core/src/test/java/org/apache/activemq/plugin/BrokerStatisticsPluginTest.java
+++ b/activemq-core/src/test/java/org/apache/activemq/plugin/BrokerStatisticsPluginTest.java
@@ -16,14 +16,14 @@
*/
package org.apache.activemq.plugin;
+import junit.framework.TestCase;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerFactory;
import org.apache.activemq.broker.BrokerPlugin;
import org.apache.activemq.broker.BrokerService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.net.URI;
-import java.util.Enumeration;
+
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MapMessage;
@@ -32,7 +32,7 @@
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
-import junit.framework.TestCase;
+import java.net.URI;
/**
* A BrokerStatisticsPluginTest
@@ -92,6 +92,32 @@
}
+
+ 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 {
broker = createBroker();
diff --git a/activemq-core/src/test/java/org/apache/activemq/transport/SoWriteTimeoutTest.java b/activemq-core/src/test/java/org/apache/activemq/transport/SoWriteTimeoutTest.java
index 28db485..6f05658 100644
--- a/activemq-core/src/test/java/org/apache/activemq/transport/SoWriteTimeoutTest.java
+++ b/activemq-core/src/test/java/org/apache/activemq/transport/SoWriteTimeoutTest.java
@@ -20,12 +20,15 @@
import java.net.SocketException;
import java.net.URI;
import java.util.concurrent.TimeUnit;
+
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
+
import junit.framework.Test;
+
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.JmsTestSupport;
import org.apache.activemq.broker.BrokerService;
@@ -110,7 +113,7 @@
stompConnection.open(new Socket("localhost", proxy.getUrl().getPort()));
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);
frame = stompConnection.receiveFrame();
assertTrue(frame.startsWith("CONNECTED"));
diff --git a/activemq-core/src/test/java/org/apache/activemq/transport/stomp/Stomp11Test.java b/activemq-core/src/test/java/org/apache/activemq/transport/stomp/Stomp11Test.java
index 1ee49d3..ef76f8a 100644
--- a/activemq-core/src/test/java/org/apache/activemq/transport/stomp/Stomp11Test.java
+++ b/activemq-core/src/test/java/org/apache/activemq/transport/stomp/Stomp11Test.java
@@ -68,7 +68,6 @@
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
queue = new ActiveMQQueue(getQueueName());
connection.start();
-
}
private void stompConnect() throws IOException, URISyntaxException, UnknownHostException {
@@ -106,11 +105,11 @@
public void testConnect() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"host:localhost\n" +
- "request-id: 1\n" +
+ "request-id:1\n" +
"\n" + Stomp.NULL;
stompConnection.sendFrame(connectFrame);
@@ -129,8 +128,8 @@
public void testConnectWithVersionOptions() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.0,1.1\n" +
"host:localhost\n" +
"\n" + Stomp.NULL;
@@ -150,8 +149,8 @@
public void testConnectWithValidFallback() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.0,10.1\n" +
"host:localhost\n" +
"\n" + Stomp.NULL;
@@ -171,8 +170,8 @@
public void testConnectWithInvalidFallback() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:9.0,10.1\n" +
"host:localhost\n" +
"\n" + Stomp.NULL;
@@ -189,8 +188,8 @@
public void testHeartbeats() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"heart-beat:0,1000\n" +
"host:localhost\n" +
@@ -231,8 +230,8 @@
public void testHeartbeatsDropsIdleConnection() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"heart-beat:1000,0\n" +
"host:localhost\n" +
@@ -261,8 +260,8 @@
public void testSendAfterMissingHeartbeat() throws Exception {
- String connectFrame = "STOMP\n" + "login: system\n" +
- "passcode: manager\n" +
+ String connectFrame = "STOMP\n" + "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"heart-beat:1000,0\n" +
"host:localhost\n" +
@@ -290,8 +289,8 @@
public void testRejectInvalidHeartbeats1() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"heart-beat:0\n" +
"host:localhost\n" +
@@ -309,8 +308,8 @@
public void testRejectInvalidHeartbeats2() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"heart-beat:T,0\n" +
"host:localhost\n" +
@@ -328,8 +327,8 @@
public void testRejectInvalidHeartbeats3() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"heart-beat:100,10,50\n" +
"host:localhost\n" +
@@ -347,8 +346,8 @@
public void testSubscribeAndUnsubscribe() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"host:localhost\n" +
"\n" + Stomp.NULL;
@@ -392,8 +391,8 @@
public void testSubscribeWithNoId() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"host:localhost\n" +
"\n" + Stomp.NULL;
@@ -418,8 +417,8 @@
public void testUnsubscribeWithNoId() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"host:localhost\n" +
"\n" + Stomp.NULL;
@@ -449,8 +448,8 @@
public void testAckMessageWithId() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"host:localhost\n" +
"\n" + Stomp.NULL;
@@ -487,8 +486,8 @@
public void testAckMessageWithNoId() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"host:localhost\n" +
"\n" + Stomp.NULL;
@@ -530,8 +529,8 @@
final int MSG_COUNT = 10;
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"host:localhost\n" +
"\n" + Stomp.NULL;
@@ -591,12 +590,11 @@
stompConnection.sendFrame(frame);
}
-
public void testSendMessageWithStandardHeadersEncoded() throws Exception {
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;
stompConnection.sendFrame(frame);
@@ -622,10 +620,9 @@
assertEquals("GroupID", "abc", amqMessage.getGroupID());
}
-
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);
frame = stompConnection.receiveFrame();
@@ -654,8 +651,8 @@
public void testNackMessage() throws Exception {
String connectFrame = "STOMP\n" +
- "login: system\n" +
- "passcode: manager\n" +
+ "login:system\n" +
+ "passcode:manager\n" +
"accept-version:1.1\n" +
"host:localhost\n" +
"\n" + Stomp.NULL;
@@ -702,9 +699,49 @@
"id:12345\n\n" + Stomp.NULL;
stompConnection.sendFrame(frame);
-
frame = "DISCONNECT\n" + "\n\n" + Stomp.NULL;
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);
+ }
}
diff --git a/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompLoadTest.java b/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompLoadTest.java
new file mode 100644
index 0000000..7bf22fb
--- /dev/null
+++ b/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompLoadTest.java
@@ -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;
+ }
+ }
+}
diff --git a/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompNIOLoadTest.java b/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompNIOLoadTest.java
new file mode 100644
index 0000000..7ac0975
--- /dev/null
+++ b/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompNIOLoadTest.java
@@ -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();
+ }
+}
diff --git a/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompNIOSSLLoadTest.java b/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompNIOSSLLoadTest.java
new file mode 100644
index 0000000..d755527
--- /dev/null
+++ b/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompNIOSSLLoadTest.java
@@ -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());
+ }
+}
diff --git a/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompSSLLoadTest.java b/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompSSLLoadTest.java
new file mode 100644
index 0000000..f4f059b
--- /dev/null
+++ b/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompSSLLoadTest.java
@@ -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());
+ }
+}
diff --git a/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java b/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java
index bc03c9a..ff7c71b 100644
--- a/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java
+++ b/activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java
@@ -188,7 +188,7 @@
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);
String f = stompConnection.receiveFrame();
@@ -201,7 +201,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -226,13 +226,13 @@
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);
frame = stompConnection.receiveFrame();
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);
@@ -245,7 +245,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -266,7 +266,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -286,7 +286,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -315,7 +315,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -337,7 +337,7 @@
StompConnection receiver = new StompConnection();
URI connectUri = new URI(bindAddress);
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);
frame = receiver.receiveFrame();
@@ -346,7 +346,7 @@
frame = "SUBSCRIBE\n" + "destination:/queue/" + getQueueName() + "\n" + "ack:auto\n\n" + Stomp.NULL;
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);
frame = stompConnection.receiveFrame();
@@ -395,7 +395,7 @@
StompConnection sender = new StompConnection();
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);
frame = sender.receiveFrame();
@@ -411,7 +411,7 @@
StompConnection receiver = new StompConnection();
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);
frame = receiver.receiveFrame();
@@ -439,7 +439,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -459,7 +459,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -488,7 +488,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -516,7 +516,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -546,7 +546,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -579,7 +579,7 @@
int ctr = 10;
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);
frame = stompConnection.receiveFrame();
@@ -617,7 +617,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -639,7 +639,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -666,7 +666,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -693,7 +693,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -720,7 +720,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -744,7 +744,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -788,7 +788,7 @@
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);
frame = stompConnection.receiveFrame();
assertTrue(frame.startsWith("CONNECTED"));
@@ -824,7 +824,7 @@
public void testTransactionCommit() throws Exception {
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);
String f = stompConnection.receiveFrame();
@@ -846,7 +846,7 @@
public void testTransactionRollback() throws Exception {
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);
String f = stompConnection.receiveFrame();
@@ -879,7 +879,7 @@
public void testDisconnectedClientsAreRemovedFromTheBroker() throws Exception {
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);
@@ -892,7 +892,7 @@
}
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);
String f = stompConnection.receiveFrame();
@@ -903,7 +903,7 @@
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);
String f = stompConnection.receiveFrame();
@@ -914,7 +914,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -929,7 +929,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -946,7 +946,7 @@
public void testTransformationUnknownTranslator() throws Exception {
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);
frame = stompConnection.receiveFrame();
@@ -964,7 +964,7 @@
public void testTransformationFailed() throws Exception {
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);
frame = stompConnection.receiveFrame();
@@ -983,7 +983,7 @@
public void testTransformationSendXMLObject() throws Exception {
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);
frame = stompConnection.receiveFrame();
@@ -1002,7 +1002,7 @@
public void testTransformationSendJSONObject() throws Exception {
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);
frame = stompConnection.receiveFrame();
@@ -1024,7 +1024,7 @@
ObjectMessage message = session.createObjectMessage(new SamplePojo("Dejan", "Belgrade"));
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);
frame = stompConnection.receiveFrame();
@@ -1046,7 +1046,7 @@
ObjectMessage message = session.createObjectMessage(new SamplePojo("Dejan", "Belgrade"));
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);
frame = stompConnection.receiveFrame();
@@ -1069,7 +1069,7 @@
ObjectMessage message = session.createObjectMessage(new SamplePojo("Dejan", "Belgrade"));
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);
frame = stompConnection.receiveFrame();
@@ -1092,7 +1092,7 @@
ObjectMessage message = session.createObjectMessage(new SamplePojo("Dejan", "Belgrade"));
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);
frame = stompConnection.receiveFrame();
@@ -1119,7 +1119,7 @@
mapMessage.setString("city", "Belgrade");
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);
frame = stompConnection.receiveFrame();
@@ -1150,7 +1150,7 @@
mapMessage.setString("city", "Belgrade");
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);
frame = stompConnection.receiveFrame();
@@ -1173,7 +1173,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -1195,7 +1195,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -1222,7 +1222,7 @@
message.writeBytes(new byte[]{1, 2, 3, 4, 5});
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);
frame = stompConnection.receiveFrame();
@@ -1251,7 +1251,7 @@
message.setStringProperty("transformation", Stomp.Transformations.JMS_OBJECT_XML.toString());
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);
frame = stompConnection.receiveFrame();
@@ -1274,7 +1274,7 @@
message.setStringProperty("transformation", Stomp.Transformations.JMS_OBJECT_XML.toString());
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);
frame = stompConnection.receiveFrame();
@@ -1294,7 +1294,7 @@
public void testTransformationSendXMLMap() throws Exception {
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);
frame = stompConnection.receiveFrame();
@@ -1312,7 +1312,7 @@
public void testTransformationSendJSONMap() throws Exception {
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);
frame = stompConnection.receiveFrame();
@@ -1335,7 +1335,7 @@
message.setString("city", "Belgrade");
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);
frame = stompConnection.receiveFrame();
@@ -1360,7 +1360,7 @@
message.setString("city", "Belgrade");
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);
frame = stompConnection.receiveFrame();
@@ -1386,7 +1386,7 @@
BrokerViewMBean view = (BrokerViewMBean)broker.getManagementContext().newProxyInstance(brokerName, BrokerViewMBean.class, true);
// 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);
frame = stompConnection.receiveFrame();
@@ -1412,7 +1412,7 @@
//reconnect
stompConnect();
// 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);
frame = stompConnection.receiveFrame();
assertTrue(frame.startsWith("CONNECTED"));
@@ -1530,7 +1530,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -1550,7 +1550,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -1567,7 +1567,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -1672,7 +1672,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -1707,7 +1707,7 @@
}
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);
frame = stompConnection.receiveFrame();
@@ -1748,7 +1748,7 @@
public void testReplytoModification() throws Exception {
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);
frame = stompConnection.receiveFrame();
@@ -1769,7 +1769,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -1780,7 +1780,7 @@
}
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);
frame = stompConnection.receiveFrame();
@@ -1841,7 +1841,7 @@
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);
frame = stompConnection.receiveFrame();
@@ -1856,7 +1856,7 @@
StompConnection responder = new StompConnection();
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);
frame = responder.receiveFrame();
@@ -1950,7 +1950,7 @@
private void doTestConnectionLeak() throws Exception {
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);
frame = stompConnection.receiveFrame();
diff --git a/activemq-core/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineTest.java b/activemq-core/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineTest.java
index 9a0e117..c8596a2 100644
--- a/activemq-core/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineTest.java
+++ b/activemq-core/src/test/java/org/apache/activemq/usecases/DurableSubscriptionOfflineTest.java
@@ -1610,7 +1610,7 @@
}
executorService.shutdown();
executorService.awaitTermination(10, TimeUnit.MINUTES);
- assertTrue("No exceptions", exceptions.isEmpty());
+ assertTrue("No exceptions " + exceptions.elements(), exceptions.isEmpty());
}
public static class Listener implements MessageListener {
@@ -1649,6 +1649,7 @@
}
}
catch (JMSException e) {
+ e.printStackTrace();
exceptions.add(e);
}
}
diff --git a/activemq-core/src/test/java/org/apache/activemq/usecases/ThreeBrokerStompTemporaryQueueTest.java b/activemq-core/src/test/java/org/apache/activemq/usecases/ThreeBrokerStompTemporaryQueueTest.java
index 89a76f3..cd8d651 100644
--- a/activemq-core/src/test/java/org/apache/activemq/usecases/ThreeBrokerStompTemporaryQueueTest.java
+++ b/activemq-core/src/test/java/org/apache/activemq/usecases/ThreeBrokerStompTemporaryQueueTest.java
@@ -70,7 +70,7 @@
stompConnection = new StompConnection();
stompConnection.open("localhost", 61614);
// 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();
assertTrue(frame.toString().startsWith("CONNECTED"));
@@ -98,29 +98,29 @@
assertEquals("Advisory topic should be present", 1, advisoryTopicsForTempQueues);
stompConnection.disconnect();
-
+
Thread.sleep(1000);
-
+
advisoryTopicsForTempQueues = countTopicsByName("BrokerA", "ActiveMQ.Advisory.Consumer.Queue.ID");
assertEquals("Advisory topic should have been deleted", 0, advisoryTopicsForTempQueues);
advisoryTopicsForTempQueues = countTopicsByName("BrokerB", "ActiveMQ.Advisory.Consumer.Queue.ID");
assertEquals("Advisory topic should have been deleted", 0, advisoryTopicsForTempQueues);
advisoryTopicsForTempQueues = countTopicsByName("BrokerC", "ActiveMQ.Advisory.Consumer.Queue.ID");
- assertEquals("Advisory topic should have been deleted", 0, advisoryTopicsForTempQueues);
-
+ assertEquals("Advisory topic should have been deleted", 0, advisoryTopicsForTempQueues);
+
LOG.info("Restarting brokerA");
BrokerItem brokerItem = brokers.remove("BrokerA");
if (brokerItem != null) {
brokerItem.destroy();
}
-
+
BrokerService restartedBroker = createAndConfigureBroker(new URI("broker:(tcp://localhost:61616,stomp://localhost:61613)/BrokerA"));
bridgeAndConfigureBrokers("BrokerA", "BrokerB");
bridgeAndConfigureBrokers("BrokerA", "BrokerC");
restartedBroker.start();
waitForBridgeFormation();
-
+
Thread.sleep(3000);
assertEquals("Destination", 0, brokers.get("BrokerA").broker.getAdminView().getTemporaryQueues().length);
diff --git a/assembly/src/main/descriptors/windows-bin.xml b/assembly/src/main/descriptors/windows-bin.xml
index d8331ab..6c7e7f1 100644
--- a/assembly/src/main/descriptors/windows-bin.xml
+++ b/assembly/src/main/descriptors/windows-bin.xml
@@ -6,9 +6,9 @@
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.
@@ -34,20 +34,22 @@
</excludes>
<lineEnding>dos</lineEnding>
</fileSet>
-
+
<fileSet>
<directory>src/release</directory>
<outputDirectory>/</outputDirectory>
<includes>
<include>bin/*.bat</include>
- <include>bin/activemq</include>
- <include>bin/activemq-admin</include>
+ <include>bin/activemq</include>
+ <include>bin/activemq-admin</include>
<include>bin/win32/*.bat</include>
<include>bin/win32/*.conf</include>
+ <include>bin/win64/*.bat</include>
+ <include>bin/win64/*.conf</include>
</includes>
<lineEnding>dos</lineEnding>
</fileSet>
-
+
<!-- Copy over files that should not get the line endings converted -->
<fileSet>
<directory>src/release</directory>
@@ -55,13 +57,15 @@
<includes>
<include>bin/win32/*.exe</include>
<include>bin/win32/*.dll</include>
+ <include>bin/win64/*.exe</include>
+ <include>bin/win64/*.dll</include>
</includes>
</fileSet>
</fileSets>
-
+
<componentDescriptors>
<componentDescriptor>src/main/descriptors/common-bin.xml</componentDescriptor>
</componentDescriptors>
-
+
</assembly>
diff --git a/assembly/src/release/bin/win64/InstallService.bat b/assembly/src/release/bin/win64/InstallService.bat
new file mode 100644
index 0000000..6b27259
--- /dev/null
+++ b/assembly/src/release/bin/win64/InstallService.bat
@@ -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
+
diff --git a/assembly/src/release/bin/win64/UninstallService.bat b/assembly/src/release/bin/win64/UninstallService.bat
new file mode 100644
index 0000000..72b2034
--- /dev/null
+++ b/assembly/src/release/bin/win64/UninstallService.bat
@@ -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
+
diff --git a/assembly/src/release/bin/win64/activemq.bat b/assembly/src/release/bin/win64/activemq.bat
new file mode 100644
index 0000000..c49922f
--- /dev/null
+++ b/assembly/src/release/bin/win64/activemq.bat
@@ -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=
diff --git a/assembly/src/release/bin/win64/wrapper.conf b/assembly/src/release/bin/win64/wrapper.conf
new file mode 100644
index 0000000..c4b6125
--- /dev/null
+++ b/assembly/src/release/bin/win64/wrapper.conf
@@ -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
diff --git a/assembly/src/release/bin/win64/wrapper.dll b/assembly/src/release/bin/win64/wrapper.dll
new file mode 100644
index 0000000..f07fc9e
--- /dev/null
+++ b/assembly/src/release/bin/win64/wrapper.dll
Binary files differ
diff --git a/assembly/src/release/bin/win64/wrapper.exe b/assembly/src/release/bin/win64/wrapper.exe
new file mode 100644
index 0000000..db2ddda
--- /dev/null
+++ b/assembly/src/release/bin/win64/wrapper.exe
Binary files differ
