You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ta...@apache.org on 2014/03/27 16:34:09 UTC
[1/3] git commit: https://issues.apache.org/jira/browse/AMQ-4563
Repository: activemq
Updated Branches:
refs/heads/trunk c07514f57 -> da07a1176
https://issues.apache.org/jira/browse/AMQ-4563
Fix for client's that use a message Id similar to ActiveMQ's version
which can throw off the ack later on when a stored message is
dispatched.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/da07a117
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/da07a117
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/da07a117
Branch: refs/heads/trunk
Commit: da07a11760e18edb52035129495edc97193e3fe3
Parents: 6d8449f
Author: Timothy Bish <ta...@gmai.com>
Authored: Thu Mar 27 11:32:57 2014 -0400
Committer: Timothy Bish <ta...@gmai.com>
Committed: Thu Mar 27 11:33:50 2014 -0400
----------------------------------------------------------------------
.../transport/amqp/AmqpProtocolConverter.java | 26 ++++++++++++--------
1 file changed, 16 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/da07a117/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java
index 71a26e0..a0560b4 100644
--- a/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java
+++ b/activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java
@@ -561,14 +561,20 @@ class AmqpProtocolConverter implements IAmqpProtocolConverter {
}
message.setProducerId(producerId);
- MessageId messageId = message.getMessageId();
- if (messageId == null) {
- messageId = new MessageId();
- message.setMessageId(messageId);
+ // Always override the AMQP client's MessageId with our own. Preserve the
+ // original in the TextView property for later Ack.
+ MessageId messageId = new MessageId(producerId, messageIdGenerator.getNextSequenceId());
+
+ MessageId amqpMessageId = message.getMessageId();
+ if (amqpMessageId != null) {
+ if (amqpMessageId.getTextView() != null) {
+ messageId.setTextView(amqpMessageId.getTextView());
+ } else {
+ messageId.setTextView(amqpMessageId.toString());
+ }
}
- messageId.setProducerId(producerId);
- messageId.setProducerSequenceId(messageIdGenerator.getNextSequenceId());
+ message.setMessageId(messageId);
LOG.trace("Inbound Message:{} from Producer:{}", message.getMessageId(), producerId + ":" + messageId.getProducerSequenceId());
@@ -580,12 +586,12 @@ class AmqpProtocolConverter implements IAmqpProtocolConverter {
}
// Lets handle the case where the expiration was set, but the timestamp
- // was not set by the client. Lets assign the timestamp now, and adjust the
+ // was not set by the client. Lets assign the timestamp now, and adjust the
// expiration.
- if( message.getExpiration()!= 0 ) {
- if( message.getTimestamp()==0 ) {
+ if (message.getExpiration() != 0) {
+ if (message.getTimestamp() == 0) {
message.setTimestamp(System.currentTimeMillis());
- message.setExpiration(message.getTimestamp()+message.getExpiration());
+ message.setExpiration(message.getTimestamp() + message.getExpiration());
}
}
[3/3] git commit: Reduce the use of QPid JMS specific types in this
test.
Posted by ta...@apache.org.
Reduce the use of QPid JMS specific types in this test.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/359ec25e
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/359ec25e
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/359ec25e
Branch: refs/heads/trunk
Commit: 359ec25e215c49d2e53fcd8ae36530329c0acad6
Parents: c07514f
Author: Timothy Bish <ta...@gmai.com>
Authored: Thu Mar 27 10:32:03 2014 -0400
Committer: Timothy Bish <ta...@gmai.com>
Committed: Thu Mar 27 11:33:50 2014 -0400
----------------------------------------------------------------------
.../activemq/transport/amqp/AMQ4563Test.java | 23 ++++++++++----------
1 file changed, 12 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/359ec25e/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AMQ4563Test.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AMQ4563Test.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AMQ4563Test.java
index d0215be..ad2c31c 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AMQ4563Test.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AMQ4563Test.java
@@ -30,6 +30,7 @@ import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
+import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
@@ -52,13 +53,12 @@ public class AMQ4563Test extends AmqpTestSupport {
public void testMessagesAreAckedAMQProducer() throws Exception {
int messagesSent = 3;
ActiveMQAdmin.enableJMSFrameTracing();
- QueueImpl queue = new QueueImpl("queue://txqueue");
assertTrue(brokerService.isPersistent());
Connection connection = createAMQConnection();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Destination destination = session.createQueue("txqueue");
- MessageProducer p = session.createProducer(destination);
+ Queue queue = session.createQueue("txqueue");
+ MessageProducer p = session.createProducer(queue);
TextMessage message = null;
for (int i=0; i < messagesSent; i++) {
message = session.createTextMessage();
@@ -69,26 +69,26 @@ public class AMQ4563Test extends AmqpTestSupport {
}
// After the first restart we should get all messages sent above
+ QueueImpl qpidQueue = new QueueImpl("queue://txqueue");
restartBroker(connection, session);
- int messagesReceived = readAllMessages(queue);
+ int messagesReceived = readAllMessages(qpidQueue);
assertEquals(messagesSent, messagesReceived);
// This time there should be no messages on this queue
restartBroker(connection, session);
- messagesReceived = readAllMessages(queue);
+ messagesReceived = readAllMessages(qpidQueue);
assertEquals(0, messagesReceived);
}
@Test(timeout = 60000)
public void testSelectingOnAMQPMessageID() throws Exception {
ActiveMQAdmin.enableJMSFrameTracing();
- QueueImpl queue = new QueueImpl("queue://txqueue");
assertTrue(brokerService.isPersistent());
Connection connection = createAMQPConnection();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Destination destination = session.createQueue("txqueue");
- MessageProducer p = session.createProducer(destination);
+ Queue queue = session.createQueue("txqueue");
+ MessageProducer p = session.createProducer(queue);
TextMessage message = session.createTextMessage();
String messageText = "Hello sent at " + new java.util.Date().toString();
message.setText(messageText);
@@ -129,11 +129,11 @@ public class AMQ4563Test extends AmqpTestSupport {
public void testMessagesAreAckedAMQPProducer() throws Exception {
int messagesSent = 3;
ActiveMQAdmin.enableJMSFrameTracing();
- QueueImpl queue = new QueueImpl("queue://txqueue");
assertTrue(brokerService.isPersistent());
Connection connection = createAMQPConnection();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Queue queue = session.createQueue("txqueue");
MessageProducer p = session.createProducer(queue);
TextMessage message = null;
for (int i=0; i < messagesSent; i++) {
@@ -155,11 +155,11 @@ public class AMQ4563Test extends AmqpTestSupport {
assertEquals(0, messagesReceived);
}
- private int readAllMessages(QueueImpl queue) throws JMSException {
+ private int readAllMessages(Queue queue) throws JMSException {
return readAllMessages(queue, null);
}
- private int readAllMessages(QueueImpl queue, String selector) throws JMSException {
+ private int readAllMessages(Queue queue, String selector) throws JMSException {
Connection connection = createAMQPConnection();
try {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
@@ -231,6 +231,7 @@ public class AMQ4563Test extends AmqpTestSupport {
/**
* Copied from AmqpTestSupport, modified to use persistence
*/
+ @Override
public void createBroker(boolean deleteAllMessages) throws Exception {
KahaDBStore kaha = new KahaDBStore();
kaha.setDirectory(new File(KAHADB_DIRECTORY));
[2/3] git commit: Speed up these tests by only creating the connector
needed for the test so we spend less time starting and stopping the broker
for each test run.
Posted by ta...@apache.org.
Speed up these tests by only creating the connector needed for the test
so we spend less time starting and stopping the broker for each test
run.
Project: http://git-wip-us.apache.org/repos/asf/activemq/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/6d8449f0
Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/6d8449f0
Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/6d8449f0
Branch: refs/heads/trunk
Commit: 6d8449f0530cc53e0f7648c7c7182fdf392dd699
Parents: 359ec25
Author: Timothy Bish <ta...@gmai.com>
Authored: Thu Mar 27 11:29:06 2014 -0400
Committer: Timothy Bish <ta...@gmai.com>
Committed: Thu Mar 27 11:33:50 2014 -0400
----------------------------------------------------------------------
.../activemq/transport/amqp/AmqpNioTest.java | 2 +-
.../activemq/transport/amqp/AmqpSslTest.java | 10 ----
.../transport/amqp/AmqpTestSupport.java | 50 +++++++++++++++-----
.../transport/amqp/JMSClientNioPlusSslTest.java | 15 ++++++
.../transport/amqp/JMSClientNioTest.java | 15 ++++++
.../transport/amqp/JMSClientSslTest.java | 24 ++++++++--
.../activemq/transport/amqp/JMSClientTest.java | 6 ++-
.../transport/amqp/bugs/AMQ4753Test.java | 10 ++++
8 files changed, 104 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq/blob/6d8449f0/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpNioTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpNioTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpNioTest.java
index ca18674..432ec90 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpNioTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpNioTest.java
@@ -19,8 +19,8 @@ package org.apache.activemq.transport.amqp;
import org.apache.activemq.broker.BrokerService;
public class AmqpNioTest extends AmqpTestSupport {
+
protected void addAMQPConnector(BrokerService brokerService) throws Exception {
brokerService.addConnector("amqp+nio://localhost:1883?maxInactivityDuration=-1");
}
-
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/6d8449f0/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpSslTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpSslTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpSslTest.java
index 505fb53..9372854 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpSslTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpSslTest.java
@@ -41,15 +41,6 @@ public class AmqpSslTest extends AmqpTestSupport {
brokerService.addConnector("amqp+ssl://localhost:8883");
}
-// protected AMQP createAMQPConnection() throws Exception {
-// AMQP amqp = new AMQP();
-// amqp.setHost("ssl://localhost:8883");
-// SSLContext ctx = SSLContext.getInstance("TLS");
-// ctx.init(new KeyManager[0], new TrustManager[]{new DefaultTrustManager()}, new SecureRandom());
-// amqp.setSslContext(ctx);
-// return amqp;
-// }
-
static class DefaultTrustManager implements X509TrustManager {
@Override
@@ -65,5 +56,4 @@ public class AmqpSslTest extends AmqpTestSupport {
return new X509Certificate[0];
}
}
-
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/6d8449f0/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java
index bf3a4cb..2524300 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/AmqpTestSupport.java
@@ -117,18 +117,44 @@ public class AmqpTestSupport {
}
protected void addAMQPConnector() throws Exception {
- TransportConnector connector = brokerService.addConnector("amqp+ssl://0.0.0.0:" + sslPort);
- sslPort = connector.getConnectUri().getPort();
- LOG.debug("Using amqp+ssl port " + sslPort);
- connector = brokerService.addConnector("amqp://0.0.0.0:" + port);
- port = connector.getConnectUri().getPort();
- LOG.debug("Using amqp port " + port);
- connector = brokerService.addConnector("amqp+nio://0.0.0.0:" + nioPort);
- nioPort = connector.getConnectUri().getPort();
- LOG.debug("Using amqp+nio port " + nioPort);
- connector = brokerService.addConnector("amqp+nio+ssl://0.0.0.0:" + nioPlusSslPort);
- nioPlusSslPort = connector.getConnectUri().getPort();
- LOG.debug("Using amqp+nio+ssl port " + nioPlusSslPort);
+ TransportConnector connector = null;
+
+ if (isUseTcpConnector()) {
+ connector = brokerService.addConnector("amqp://0.0.0.0:" + port);
+ port = connector.getConnectUri().getPort();
+ LOG.debug("Using amqp port " + port);
+ }
+ if (isUseSslConnector()) {
+ connector = brokerService.addConnector("amqp+ssl://0.0.0.0:" + sslPort);
+ sslPort = connector.getConnectUri().getPort();
+ LOG.debug("Using amqp+ssl port " + sslPort);
+ }
+ if (isUseNioConnector()) {
+ connector = brokerService.addConnector("amqp+nio://0.0.0.0:" + nioPort);
+ nioPort = connector.getConnectUri().getPort();
+ LOG.debug("Using amqp+nio port " + nioPort);
+ }
+ if (isUseNioPlusSslConnector()) {
+ connector = brokerService.addConnector("amqp+nio+ssl://0.0.0.0:" + nioPlusSslPort);
+ nioPlusSslPort = connector.getConnectUri().getPort();
+ LOG.debug("Using amqp+nio+ssl port " + nioPlusSslPort);
+ }
+ }
+
+ protected boolean isUseTcpConnector() {
+ return true;
+ }
+
+ protected boolean isUseSslConnector() {
+ return false;
+ }
+
+ protected boolean isUseNioConnector() {
+ return false;
+ }
+
+ protected boolean isUseNioPlusSslConnector() {
+ return false;
}
public void startBroker() throws Exception {
http://git-wip-us.apache.org/repos/asf/activemq/blob/6d8449f0/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientNioPlusSslTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientNioPlusSslTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientNioPlusSslTest.java
index d39b53c..758607d 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientNioPlusSslTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientNioPlusSslTest.java
@@ -30,4 +30,19 @@ public class JMSClientNioPlusSslTest extends JMSClientSslTest {
LOG.debug("JMSClientNioPlusSslTest.getBrokerPort returning nioPlusSslPort {}", nioPlusSslPort);
return nioPlusSslPort;
}
+
+ @Override
+ protected boolean isUseTcpConnector() {
+ return false;
+ }
+
+ @Override
+ protected boolean isUseNioPlusSslConnector() {
+ return true;
+ }
+
+ @Override
+ protected String getTargetConnectorName() {
+ return "amqp+nio+ssl";
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/6d8449f0/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientNioTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientNioTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientNioTest.java
index 7ac67a6..fa7c6c0 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientNioTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientNioTest.java
@@ -30,4 +30,19 @@ public class JMSClientNioTest extends JMSClientTest {
LOG.debug("JMSClientNioTest.getBrokerPort returning nioPort {}", nioPort);
return nioPort;
}
+
+ @Override
+ protected boolean isUseTcpConnector() {
+ return false;
+ }
+
+ @Override
+ protected boolean isUseNioConnector() {
+ return true;
+ }
+
+ @Override
+ protected String getTargetConnectorName() {
+ return "amqp+nio";
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/6d8449f0/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientSslTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientSslTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientSslTest.java
index 2a7445b..9901217 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientSslTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientSslTest.java
@@ -16,16 +16,17 @@
*/
package org.apache.activemq.transport.amqp;
-import org.junit.BeforeClass;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.security.SecureRandom;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.net.ssl.KeyManager;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
-import java.security.SecureRandom;
+
+import org.junit.BeforeClass;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Test the JMS client when connected to the SSL transport.
@@ -51,4 +52,19 @@ public class JMSClientSslTest extends JMSClientTest {
LOG.debug("JMSClientSslTest.getBrokerPort returning sslPort {}", sslPort);
return sslPort;
}
+
+ @Override
+ protected boolean isUseTcpConnector() {
+ return false;
+ }
+
+ @Override
+ protected boolean isUseSslConnector() {
+ return true;
+ }
+
+ @Override
+ protected String getTargetConnectorName() {
+ return "amqp+ssl";
+ }
}
http://git-wip-us.apache.org/repos/asf/activemq/blob/6d8449f0/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java
index 313ac97..bd69444 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java
@@ -690,7 +690,7 @@ public class JMSClientTest extends AmqpTestSupport {
public void testConnectionsAreClosed() throws Exception {
ActiveMQAdmin.enableJMSFrameTracing();
- final ConnectorViewMBean connector = getProxyToConnectionView("amqp");
+ final ConnectorViewMBean connector = getProxyToConnectionView(getTargetConnectorName());
LOG.info("Current number of Connections is: {}", connector.connectionCount());
ArrayList<Connection> connections = new ArrayList<Connection>();
@@ -715,6 +715,10 @@ public class JMSClientTest extends AmqpTestSupport {
}));
}
+ protected String getTargetConnectorName() {
+ return "amqp";
+ }
+
@Test(timeout=30000)
public void testExecptionListenerCalledOnBrokerStop() throws Exception {
ActiveMQAdmin.enableJMSFrameTracing();
http://git-wip-us.apache.org/repos/asf/activemq/blob/6d8449f0/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/bugs/AMQ4753Test.java
----------------------------------------------------------------------
diff --git a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/bugs/AMQ4753Test.java b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/bugs/AMQ4753Test.java
index 3be3482..efa98b1 100644
--- a/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/bugs/AMQ4753Test.java
+++ b/activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/bugs/AMQ4753Test.java
@@ -36,6 +36,16 @@ import org.junit.Test;
public class AMQ4753Test extends AmqpTestSupport {
+ @Override
+ protected boolean isUseTcpConnector() {
+ return false;
+ }
+
+ @Override
+ protected boolean isUseNioPlusSslConnector() {
+ return true;
+ }
+
@Test(timeout = 120 * 1000)
public void testAmqpNioPlusSslSendReceive() throws JMSException{
Connection connection = createAMQPConnection(nioPlusSslPort, true);