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 c5df8d9ddf..2e2d4f0b77 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.BrokerFilter; 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.IdGenerator; 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 @@ public class StatisticsBroker extends BrokerFilter { 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 @@ public class StatisticsBroker extends BrokerFilter { 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 @@ public class StatisticsBroker extends BrokerFilter { 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 @@ 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 { super.start(); LOG.info("Starting StatisticsBroker"); @@ -174,6 +195,34 @@ public class StatisticsBroker extends BrokerFilter { 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 053403839a..b9baeb38b7 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 @@ public class NIOSSLTransport extends NIOTransport { 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 @@ public class NIOSSLTransport extends NIOTransport { } } - processCommand(plain); + if (status == SSLEngineResult.Status.OK && handshakeStatus != SSLEngineResult.HandshakeStatus.NEED_UNWRAP) { + processCommand(plain); + } } } catch (IOException e) { @@ -192,7 +198,7 @@ public class NIOSSLTransport extends NIOTransport { 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 282443eb96..d276f67388 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 Stomp protocol. @@ -201,7 +207,7 @@ public class StompWireFormat implements WireFormat { 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 ee2b3fe9cb..533ae0c939 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 class AMQ2571Test extends EmbeddedBrokerTestSupport { 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 @@ public class AMQ2571Test extends EmbeddedBrokerTestSupport { 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 @@ public class AMQ2571Test extends EmbeddedBrokerTestSupport { // 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 @@ public class AMQ2571Test extends EmbeddedBrokerTestSupport { @Override protected void setUp() throws Exception { bindAddress = "vm://localhost"; + setAutoFail(true); super.setUp(); } @@ -103,7 +105,6 @@ public class AMQ2571Test extends EmbeddedBrokerTestSupport { 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 79f07cc4cb..c6b415bd41 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.MessageConsumer; import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.Session; -import junit.framework.TestCase; +import java.net.URI; /** * 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 { 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 28db485b9e..6f0565891e 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.Socket; 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 @@ public class SoWriteTimeoutTest extends JmsTestSupport { 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 1ee49d3150..ef76f8a0e3 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 @@ public class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 @@ public class Stomp11Test extends CombinationTestSupport { 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 @@ public class Stomp11Test extends CombinationTestSupport { 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 @@ public class Stomp11Test extends CombinationTestSupport { 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 class Stomp11Test extends CombinationTestSupport { 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 @@ public class Stomp11Test extends CombinationTestSupport { "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 0000000000..7bf22fbaeb --- /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 headers = new HashMap(); + 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 0000000000..7ac0975715 --- /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 0000000000..d7555274cb --- /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 0000000000..f4f059b01c --- /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 bc03c9a488..ff7c71bf5a 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 class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { } 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { //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 class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { } 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { 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 class StompTest extends CombinationTestSupport { } 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 class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 @@ public class StompTest extends CombinationTestSupport { 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 9a0e117c44..c8596a2d25 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 @@ public class DurableSubscriptionOfflineTest extends org.apache.activemq.TestSupp } 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 @@ public class DurableSubscriptionOfflineTest extends org.apache.activemq.TestSupp } } 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 89a76f3e62..cd8d651d6f 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 @@ public class ThreeBrokerStompTemporaryQueueTest extends JmsMultipleBrokersTestSu 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 @@ public class ThreeBrokerStompTemporaryQueueTest extends JmsMultipleBrokersTestSu 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 d8331ab622..6c7e7f1ea4 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 @@ dos - + src/release / bin/*.bat - bin/activemq - bin/activemq-admin + bin/activemq + bin/activemq-admin bin/win32/*.bat bin/win32/*.conf + bin/win64/*.bat + bin/win64/*.conf dos - + src/release @@ -55,13 +57,15 @@ bin/win32/*.exe bin/win32/*.dll + bin/win64/*.exe + bin/win64/*.dll - + src/main/descriptors/common-bin.xml - + diff --git a/assembly/src/release/bin/win64/InstallService.bat b/assembly/src/release/bin/win64/InstallService.bat new file mode 100644 index 0000000000..6b27259488 --- /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 0000000000..72b2034d14 --- /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 0000000000..c49922f257 --- /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 0000000000..c4b612536b --- /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 0000000000..f07fc9e582 Binary files /dev/null and b/assembly/src/release/bin/win64/wrapper.dll differ diff --git a/assembly/src/release/bin/win64/wrapper.exe b/assembly/src/release/bin/win64/wrapper.exe new file mode 100644 index 0000000000..db2dddae8c Binary files /dev/null and b/assembly/src/release/bin/win64/wrapper.exe differ