From 7dbc1c3cc64518fffe86e4a5369981a215d88c3a Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Thu, 10 Sep 2015 13:56:20 -0400 Subject: [PATCH 1/8] improving jdk detection and javadoc options --- pom.xml | 67 +++++++++++++++++++++++++++++++-------------------------- 1 file changed, 37 insertions(+), 30 deletions(-) diff --git a/pom.xml b/pom.xml index c06e961a463..f26dd51c540 100644 --- a/pom.xml +++ b/pom.xml @@ -33,6 +33,26 @@ artemis-protocols artemis-dto artemis-boot + artemis-web + artemis-website + artemis-cli + artemis-commons + artemis-selector + artemis-core-client + artemis-server + artemis-jms-client + artemis-jms-server + artemis-native + artemis-journal + artemis-ra + artemis-rest + artemis-service-extensions + artemis-maven-plugin + integration/activemq-spring-integration + integration/activemq-aerogear-integration + integration/activemq-vertx-integration + artemis-distribution + tests ActiveMQ Artemis Parent @@ -482,6 +502,23 @@ javac + + jdk18 + + 1.8 + + + + + org.apache.maven.plugins + maven-javadoc-plugin + + -Xdoclint:none + + + + + dev @@ -517,29 +554,6 @@ true - - artemis-dto - artemis-web - artemis-website - artemis-cli - artemis-commons - artemis-selector - artemis-core-client - artemis-server - artemis-jms-client - artemis-jms-server - artemis-native - artemis-journal - artemis-ra - artemis-rest - artemis-service-extensions - artemis-maven-plugin - integration/activemq-spring-integration - integration/activemq-aerogear-integration - integration/activemq-vertx-integration - artemis-distribution - tests - release @@ -594,13 +608,6 @@ - - org.apache.maven.plugins - maven-javadoc-plugin - - -Xdoclint:none - - From 9e7cabeede75a31349275fda6527344169f2033d Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Thu, 10 Sep 2015 13:57:18 -0400 Subject: [PATCH 2/8] using new qpid-jms client on AMQP --- artemis-protocols/artemis-proton-plug/pom.xml | 4 +- .../org/proton/plug/test/AbstractJMSTest.java | 34 +--- .../java/org/proton/plug/test/ProtonTest.java | 29 ++-- examples/protocols/amqp/pom.xml | 4 +- .../activemq/server0/artemis-roles.properties | 17 -- .../activemq/server0/artemis-users.properties | 17 -- .../resources/activemq/server0/broker.xml | 68 -------- .../amqp/{proton-j => queue}/pom.xml | 8 +- .../amqp/{proton-j => queue}/readme.html | 49 ------ .../artemis/jms/example/ProtonJExample.java | 39 +++-- tests/activemq5-unit-tests/pom.xml | 8 - tests/integration-tests/pom.xml | 4 +- .../tests/integration/proton/ProtonTest.java | 160 ++---------------- 13 files changed, 62 insertions(+), 379 deletions(-) delete mode 100644 examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/artemis-roles.properties delete mode 100644 examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/artemis-users.properties delete mode 100644 examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/broker.xml rename examples/protocols/amqp/{proton-j => queue}/pom.xml (94%) rename examples/protocols/amqp/{proton-j => queue}/readme.html (59%) rename examples/protocols/amqp/{proton-j => queue}/src/main/java/org/apache/activemq/artemis/jms/example/ProtonJExample.java (59%) diff --git a/artemis-protocols/artemis-proton-plug/pom.xml b/artemis-protocols/artemis-proton-plug/pom.xml index cedbc872e52..47e3a91137e 100644 --- a/artemis-protocols/artemis-proton-plug/pom.xml +++ b/artemis-protocols/artemis-proton-plug/pom.xml @@ -88,8 +88,8 @@ org.apache.qpid - qpid-amqp-1-0-client-jms - 0.24 + qpid-jms-client + 0.5.0 test diff --git a/artemis-protocols/artemis-proton-plug/src/test/java/org/proton/plug/test/AbstractJMSTest.java b/artemis-protocols/artemis-proton-plug/src/test/java/org/proton/plug/test/AbstractJMSTest.java index bccea5fe4d1..90fff621d9f 100644 --- a/artemis-protocols/artemis-proton-plug/src/test/java/org/proton/plug/test/AbstractJMSTest.java +++ b/artemis-protocols/artemis-proton-plug/src/test/java/org/proton/plug/test/AbstractJMSTest.java @@ -21,24 +21,22 @@ import javax.jms.ExceptionListener; import javax.jms.JMSException; import javax.jms.Queue; +import javax.jms.Session; import java.lang.ref.WeakReference; -import org.apache.qpid.amqp_1_0.jms.impl.ConnectionFactoryImpl; -import org.apache.qpid.amqp_1_0.jms.impl.QueueImpl; +import org.apache.qpid.jms.JmsConnectionFactory; import org.proton.plug.test.minimalserver.DumbServer; import org.proton.plug.test.minimalserver.MinimalServer; public class AbstractJMSTest { - protected final boolean useHawtJMS; protected final boolean useSASL; protected String address = "exampleQueue"; protected MinimalServer server = new MinimalServer(); - public AbstractJMSTest(boolean useHawtJMS, boolean useSASL) { - this.useHawtJMS = useHawtJMS; + public AbstractJMSTest(boolean useSASL) { this.useSASL = useSASL; } @@ -77,34 +75,16 @@ public void onException(JMSException exception) { protected ConnectionFactory createConnectionFactory() { if (useSASL) { - if (useHawtJMS) { - // return new JmsConnectionFactory("aaaaaaaa", "aaaaaaa", "amqp://localhost:" + Constants.PORT); - return null; - } - else { - return new ConnectionFactoryImpl("localhost", Constants.PORT, "aaaaaaaa", "aaaaaaa"); - } + return new JmsConnectionFactory("aaaaaaaa", "aaaaaaa", "amqp://localhost:5672"); } else { - if (useHawtJMS) { - // return new JmsConnectionFactory("amqp://localhost:" + Constants.PORT); - return null; - } - else { - return new ConnectionFactoryImpl("localhost", Constants.PORT, null, null); - } + return new JmsConnectionFactory( "amqp://localhost:5672"); } } - protected Queue createQueue() { - if (useHawtJMS) { - // return new JmsQueue(address); - return null; - } - else { - return new QueueImpl(address); - } + protected Queue createQueue(Session session) throws Exception { + return session.createQueue(address); } } diff --git a/artemis-protocols/artemis-proton-plug/src/test/java/org/proton/plug/test/ProtonTest.java b/artemis-protocols/artemis-proton-plug/src/test/java/org/proton/plug/test/ProtonTest.java index bf972b6fed7..e8be646089a 100644 --- a/artemis-protocols/artemis-proton-plug/src/test/java/org/proton/plug/test/ProtonTest.java +++ b/artemis-protocols/artemis-proton-plug/src/test/java/org/proton/plug/test/ProtonTest.java @@ -32,7 +32,6 @@ import java.util.HashMap; import java.util.List; -import org.apache.qpid.amqp_1_0.jms.impl.QueueImpl; import org.apache.qpid.proton.amqp.messaging.AmqpValue; import org.apache.qpid.proton.amqp.messaging.Properties; import org.apache.qpid.proton.message.Message; @@ -59,16 +58,15 @@ public class ProtonTest extends AbstractJMSTest { protected Connection connection; - @Parameterized.Parameters(name = "useHawt={0} sasl={1}") + @Parameterized.Parameters(name = "sasl={0}") public static Collection data() { - List list = Arrays.asList(new Object[][]{{Boolean.FALSE, Boolean.TRUE}, {Boolean.FALSE, Boolean.FALSE}}); - + List list = Arrays.asList(new Object[][]{{Boolean.TRUE}, {Boolean.FALSE}}); System.out.println("Size = " + list.size()); return list; } - public ProtonTest(boolean useHawtJMS, boolean useSASL) { - super(useHawtJMS, useSASL); + public ProtonTest(boolean useSASL) { + super(useSASL); } @Before @@ -92,7 +90,6 @@ public void tearDown() throws Exception { public void testMessagesReceivedInParallel() throws Throwable { final int numMessages = getNumberOfMessages(); long time = System.currentTimeMillis(); - final Queue queue = createQueue(); final ArrayList exceptions = new ArrayList<>(); @@ -105,6 +102,7 @@ public void run() { // connectionConsumer = connection; connectionConsumer.start(); Session sessionConsumer = connectionConsumer.createSession(false, Session.AUTO_ACKNOWLEDGE); + final Queue queue = createQueue(sessionConsumer); final MessageConsumer consumer = sessionConsumer.createConsumer(queue); int count = numMessages; @@ -143,6 +141,7 @@ public void run() { Session session = connection.createSession(false, Session.DUPS_OK_ACKNOWLEDGE); t.start(); + final Queue queue = createQueue(session); MessageProducer p = session.createProducer(queue); p.setDeliveryMode(DeliveryMode.PERSISTENT); @@ -156,14 +155,14 @@ public void run() { } long taken = (System.currentTimeMillis() - time); - System.out.println("taken on send = " + taken + " usehawt = " + useHawtJMS + " sasl = " + useSASL); + System.out.println("taken on send = " + taken + " sasl = " + useSASL); t.join(); for (Throwable e : exceptions) { throw e; } taken = (System.currentTimeMillis() - time); - System.out.println("taken = " + taken + " usehawt = " + useHawtJMS + " sasl = " + useSASL); + System.out.println("taken = " + taken + " sasl = " + useSASL); connection.close(); // assertEquals(0, q.getMessageCount()); @@ -171,9 +170,9 @@ public void run() { @Test public void testSimpleCreateSessionAndClose() throws Throwable { - final QueueImpl queue = new QueueImpl(address); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue queue = session.createQueue(address); Thread.sleep(1000); session.close(); connection.close(); @@ -183,10 +182,10 @@ public void testSimpleCreateSessionAndClose() throws Throwable { public void testSimpleBinary() throws Throwable { final int numMessages = 5; long time = System.currentTimeMillis(); - final QueueImpl queue = new QueueImpl(address); - Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue queue = createQueue(session); + byte[] bytes = new byte[0xf + 1]; for (int i = 0; i <= 0xf; i++) { bytes[i] = (byte) i; @@ -230,8 +229,8 @@ public void testSimpleBinary() throws Throwable { @Test public void testMapMessage() throws Exception { - Queue queue = createQueue(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue queue = createQueue(session); MessageProducer p = session.createProducer(queue); for (int i = 0; i < 10; i++) { MapMessage message = session.createMapMessage(); @@ -252,8 +251,8 @@ public void testMapMessage() throws Exception { @Test public void testProperties() throws Exception { - Queue queue = createQueue(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Queue queue = createQueue(session); MessageProducer p = session.createProducer(queue); TextMessage message = session.createTextMessage(); message.setText("msg:0"); @@ -310,7 +309,7 @@ public void testSendWithSimpleClient() throws Exception { Session clientSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); connection.start(); - MessageConsumer consumer = clientSession.createConsumer(createQueue()); + MessageConsumer consumer = clientSession.createConsumer(createQueue(clientSession)); for (int i = 0; i < 1; i++) { MapMessage msg = (MapMessage) consumer.receive(5000); System.out.println("Msg " + msg); diff --git a/examples/protocols/amqp/pom.xml b/examples/protocols/amqp/pom.xml index 5841ad496e5..0f0973265b9 100644 --- a/examples/protocols/amqp/pom.xml +++ b/examples/protocols/amqp/pom.xml @@ -48,7 +48,7 @@ under the License. release proton-cpp - proton-j + queue proton-ruby @@ -58,7 +58,7 @@ under the License. - proton-j + queue proton-ruby diff --git a/examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/artemis-roles.properties b/examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/artemis-roles.properties deleted file mode 100644 index 4e2d44cec4f..00000000000 --- a/examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/artemis-roles.properties +++ /dev/null @@ -1,17 +0,0 @@ -## --------------------------------------------------------------------------- -## 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. -## --------------------------------------------------------------------------- -guest=guest \ No newline at end of file diff --git a/examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/artemis-users.properties b/examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/artemis-users.properties deleted file mode 100644 index 4e2d44cec4f..00000000000 --- a/examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/artemis-users.properties +++ /dev/null @@ -1,17 +0,0 @@ -## --------------------------------------------------------------------------- -## 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. -## --------------------------------------------------------------------------- -guest=guest \ No newline at end of file diff --git a/examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/broker.xml b/examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/broker.xml deleted file mode 100644 index f1df1c6a9e0..00000000000 --- a/examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/broker.xml +++ /dev/null @@ -1,68 +0,0 @@ - - - - - - - - - - - - - ${data.dir}/server0/data/messaging/bindings - - ${data.dir}/server0/data/messaging/journal - - ${data.dir}/server0/data/messaging/largemessages - - ${data.dir}/server0/data/messaging/paging - - - - - tcp://localhost:5672 - tcp://localhost:61616 - - - - -
testQueue
- - - - - - - - - - - - - - - - - - - diff --git a/examples/protocols/amqp/proton-j/pom.xml b/examples/protocols/amqp/queue/pom.xml similarity index 94% rename from examples/protocols/amqp/proton-j/pom.xml rename to examples/protocols/amqp/queue/pom.xml index 4f57fdc3923..8a2b0286324 100644 --- a/examples/protocols/amqp/proton-j/pom.xml +++ b/examples/protocols/amqp/queue/pom.xml @@ -27,7 +27,7 @@ under the License. 1.1.1-SNAPSHOT - artemis-proton-j + queue jar ActiveMQ Artemis Proton-J Example @@ -38,8 +38,8 @@ under the License. org.apache.qpid - qpid-amqp-1-0-client - 0.22 + qpid-jms-client + 0.5.0 @@ -97,7 +97,7 @@ under the License. org.apache.activemq.examples.amqp - artemis-proton-j + queue ${project.version} diff --git a/examples/protocols/amqp/proton-j/readme.html b/examples/protocols/amqp/queue/readme.html similarity index 59% rename from examples/protocols/amqp/proton-j/readme.html rename to examples/protocols/amqp/queue/readme.html index b20054e4a50..4eb1f0c76b1 100644 --- a/examples/protocols/amqp/proton-j/readme.html +++ b/examples/protocols/amqp/queue/readme.html @@ -42,54 +42,5 @@

Proton qpid java example

<acceptor name="proton-acceptor">tcp://localhost:5672</acceptor> -

Example step-by-step

-
    -
  1. Create an amqp qpid 1.0 connection.
    2. - 
    -           connection= new Connection("localhost", 5672, null, null);
-
    - -
  2. Create a session
    3. - 
    -           Session session = connection.createSession();
-
    - -
  3. Create a sender
    4. - 
    -           Sender sender = session.createSender("testQueue");
-
    - -
  4. send a simple message
    5. - 
    -           sender.send(new Message("I am an amqp message"));
-
    - -
  5. create a moving receiver, this means the message will be removed from the queue
    6. - 
    -           Receiver rec = session.createMovingReceiver("testQueue");
-
    - -
  6. set some credit so we can receive
    7. - 
    -          rec.setCredit(UnsignedInteger.valueOf(1), false);
-
    - -
  7. receive the simple message
    8. - 
    -          Message m = rec.receive(5000);
-                System.out.println("message = " + m.getPayload());
-
    - -
  8. acknowledge the message
    9. - 
    -          rec.acknowledge(m);
-
    - -
  9. close the connection
    10. - 
    -          connection.close();
-
    -
- diff --git a/examples/protocols/amqp/proton-j/src/main/java/org/apache/activemq/artemis/jms/example/ProtonJExample.java b/examples/protocols/amqp/queue/src/main/java/org/apache/activemq/artemis/jms/example/ProtonJExample.java similarity index 59% rename from examples/protocols/amqp/proton-j/src/main/java/org/apache/activemq/artemis/jms/example/ProtonJExample.java rename to examples/protocols/amqp/queue/src/main/java/org/apache/activemq/artemis/jms/example/ProtonJExample.java index 85334b45687..fa50e1aadf7 100644 --- a/examples/protocols/amqp/proton-j/src/main/java/org/apache/activemq/artemis/jms/example/ProtonJExample.java +++ b/examples/protocols/amqp/queue/src/main/java/org/apache/activemq/artemis/jms/example/ProtonJExample.java @@ -16,43 +16,46 @@ */ package org.apache.activemq.artemis.jms.example; -import org.apache.qpid.amqp_1_0.client.Connection; -import org.apache.qpid.amqp_1_0.client.Message; -import org.apache.qpid.amqp_1_0.client.Receiver; -import org.apache.qpid.amqp_1_0.client.Sender; -import org.apache.qpid.amqp_1_0.client.Session; -import org.apache.qpid.amqp_1_0.type.UnsignedInteger; +import javax.jms.Connection; +import javax.jms.ConnectionFactory; +import javax.jms.MessageProducer; +import javax.jms.*; + + +import org.apache.qpid.jms.JmsConnectionFactory; public class ProtonJExample { public static void main(String[] args) throws Exception { Connection connection = null; + ConnectionFactory connectionFactory = new JmsConnectionFactory("amqp://localhost:5672"); try { + + // Step 1. Create an amqp qpid 1.0 connection - connection = new Connection("localhost", 5672, null, null); + connection = connectionFactory.createConnection(); // Step 2. Create a session - Session session = connection.createSession(); + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // Step 3. Create a sender - Sender sender = session.createSender("jms.queue.exampleQueue"); + Queue queue = session.createQueue("jms.queue.exampleQueue"); + MessageProducer sender = session.createProducer(queue); + + // Step 4. send a few simple message + sender.send(session.createTextMessage("Hello world ")); - // Step 4. send a simple message - sender.send(new Message("I am an amqp message")); + connection.start(); // Step 5. create a moving receiver, this means the message will be removed from the queue - Receiver rec = session.createMovingReceiver("jms.queue.exampleQueue"); + MessageConsumer consumer = session.createConsumer(queue); - // Step 6. set some credit so we can receive - rec.setCredit(UnsignedInteger.valueOf(1), false); // Step 7. receive the simple message - Message m = rec.receive(5000); - System.out.println("message = " + m.getPayload()); + TextMessage m = (TextMessage) consumer.receive(5000); + System.out.println("message = " + m.getText()); - // Step 8. acknowledge the message - rec.acknowledge(m); } finally { if (connection != null) { diff --git a/tests/activemq5-unit-tests/pom.xml b/tests/activemq5-unit-tests/pom.xml index 8ee1050b5ce..9ac5c2a5453 100644 --- a/tests/activemq5-unit-tests/pom.xml +++ b/tests/activemq5-unit-tests/pom.xml @@ -37,7 +37,6 @@ 10.11.1.1 2.4 3.3 - 0.30 3.18 1.3 1.7.10 @@ -195,13 +194,6 @@ ${commons-net-version} - - org.apache.qpid - qpid-amqp-1-0-client-jms - ${qpid-jms-version} - test - - org.apache.xbean xbean-spring diff --git a/tests/integration-tests/pom.xml b/tests/integration-tests/pom.xml index c71286c9bd3..de8ca770f92 100644 --- a/tests/integration-tests/pom.xml +++ b/tests/integration-tests/pom.xml @@ -187,8 +187,8 @@ org.apache.qpid - qpid-amqp-1-0-client-jms - 0.24 + qpid-jms-client + 0.5.0 org.apache.qpid diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proton/ProtonTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proton/ProtonTest.java index 036be19de31..e779c14985f 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proton/ProtonTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proton/ProtonTest.java @@ -40,22 +40,15 @@ import java.util.HashMap; import java.util.Random; -import org.apache.activemq.artemis.api.core.management.ResourceNames; -import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; -import org.apache.activemq.artemis.tests.util.RandomUtil; -import org.apache.qpid.amqp_1_0.client.Receiver; -import org.apache.qpid.amqp_1_0.client.Sender; -import org.apache.qpid.amqp_1_0.jms.impl.ConnectionFactoryImpl; -import org.apache.qpid.amqp_1_0.jms.impl.QueueImpl; -import org.apache.qpid.amqp_1_0.type.UnsignedInteger; import org.apache.activemq.artemis.api.core.SimpleString; import org.apache.activemq.artemis.api.core.TransportConfiguration; -import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient; import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants; import org.apache.activemq.artemis.core.server.ActiveMQServer; import org.apache.activemq.artemis.core.server.Queue; import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; import org.apache.activemq.artemis.utils.ByteUtil; +import org.apache.qpid.jms.JmsConnectionFactory; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -737,153 +730,20 @@ public void testProperties() throws Exception { connection.close(); } - @Test - public void testUsingPlainAMQP() throws Exception { - if (this.protocol != 0 && protocol != 3) { - return; - } - - org.apache.qpid.amqp_1_0.client.Connection connection = null; - - try { - // Step 1. Create an amqp qpid 1.0 connection - connection = new org.apache.qpid.amqp_1_0.client.Connection("localhost", 5672, null, null); - - // Step 2. Create a session - org.apache.qpid.amqp_1_0.client.Session session = connection.createSession(); - - // Step 3. Create a sender - Sender sender = session.createSender("jms.queue.exampleQueue"); - - // Step 4. send a simple message - sender.send(new org.apache.qpid.amqp_1_0.client.Message("I am an amqp message")); - - // Step 5. create a moving receiver, this means the message will be removed from the queue - Receiver rec = session.createMovingReceiver("jms.queue.exampleQueue"); - - // Step 6. set some credit so we can receive - rec.setCredit(UnsignedInteger.valueOf(1), false); - - // Step 7. receive the simple message - org.apache.qpid.amqp_1_0.client.Message m = rec.receive(5000); - System.out.println("message = " + m.getPayload()); - - // Step 8. acknowledge the message - rec.acknowledge(m); - } - finally { - if (connection != null) { - // Step 9. close the connection - connection.close(); - } - } - } - - @Test - public void testUsingPlainAMQPSenderWithNonExistentQueue() throws Exception { - if (this.protocol != 0 && protocol != 3) { - return; - } - - String queue = ResourceNames.JMS_QUEUE + RandomUtil.randomString(); - - org.apache.qpid.amqp_1_0.client.Connection connection = null; - - try { - // Step 1. Create an amqp qpid 1.0 connection - connection = new org.apache.qpid.amqp_1_0.client.Connection("localhost", 5672, null, null); - - // Step 2. Create a session - org.apache.qpid.amqp_1_0.client.Session session = connection.createSession(); - - // Step 3. Create a sender - Sender sender = session.createSender(queue); - - Assert.assertNotNull(server.locateQueue(new SimpleString(queue))); - - // Step 4. send a simple message - sender.send(new org.apache.qpid.amqp_1_0.client.Message("I am an amqp message")); - - // Step 5. create a moving receiver, this means the message will be removed from the queue - Receiver rec = session.createMovingReceiver(queue); - - // Step 6. set some credit so we can receive - rec.setCredit(UnsignedInteger.valueOf(1), false); - - // Step 7. receive the simple message - org.apache.qpid.amqp_1_0.client.Message m = rec.receive(5000); - System.out.println("message = " + m.getPayload()); - - // Step 8. acknowledge the message - rec.acknowledge(m); - } - finally { - if (connection != null) { - // Step 9. close the connection - connection.close(); - } - } - } - - @Test - public void testUsingPlainAMQPReceiverWithNonExistentQueue() throws Exception { - if (this.protocol != 0 && protocol != 3) { - return; - } - - String queue = ResourceNames.JMS_QUEUE + RandomUtil.randomString(); - - org.apache.qpid.amqp_1_0.client.Connection connection = null; - + private javax.jms.Queue createQueue(String address) throws Exception { + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); try { - // Step 1. Create an amqp qpid 1.0 connection - connection = new org.apache.qpid.amqp_1_0.client.Connection("localhost", 5672, null, null); - - // Step 2. Create a session - org.apache.qpid.amqp_1_0.client.Session session = connection.createSession(); - - // Step 3. create a moving receiver, this means the message will be removed from the queue - Receiver rec = session.createMovingReceiver(queue); - - Assert.assertNotNull(server.locateQueue(new SimpleString(queue))); - - // Step 4. Create a sender - Sender sender = session.createSender(queue); - - // Step 5. send a simple message - sender.send(new org.apache.qpid.amqp_1_0.client.Message("I am an amqp message")); - - // Step 6. set some credit so we can receive - rec.setCredit(UnsignedInteger.valueOf(1), false); - - // Step 7. receive the simple message - org.apache.qpid.amqp_1_0.client.Message m = rec.receive(5000); - System.out.println("message = " + m.getPayload()); - - // Step 8. acknowledge the message - rec.acknowledge(m); + return session.createQueue(address); } finally { - if (connection != null) { - // Step 9. close the connection - connection.close(); - } - } - } - - private javax.jms.Queue createQueue(String address) { - if (protocol == 0 || protocol == 3) { - return new QueueImpl(address); - } - else { - return ActiveMQJMSClient.createQueue(address); + session.close(); } } private javax.jms.Connection createConnection() throws JMSException { Connection connection; if (protocol == 3) { - factory = new ConnectionFactoryImpl("localhost", 5672, null, null); + factory = new JmsConnectionFactory("amqp://localhost:5672"); connection = factory.createConnection(); connection.setExceptionListener(new ExceptionListener() { @Override @@ -894,7 +754,7 @@ public void onException(JMSException exception) { connection.start(); } else if (protocol == 0) { - factory = new ConnectionFactoryImpl("localhost", 5672, "guest", "guest"); + factory = new JmsConnectionFactory("guest", "guest", "amqp://localhost:5672"); connection = factory.createConnection(); connection.setExceptionListener(new ExceptionListener() { @Override @@ -909,12 +769,12 @@ public void onException(JMSException exception) { if (protocol == 1) { transport = new TransportConfiguration(INVM_CONNECTOR_FACTORY); + factory = new ActiveMQConnectionFactory("vm:/0"); } else { - transport = new TransportConfiguration(NETTY_CONNECTOR_FACTORY); + factory = new ActiveMQConnectionFactory(); } - factory = new ActiveMQConnectionFactory(false, transport); connection = factory.createConnection("guest", "guest"); connection.setExceptionListener(new ExceptionListener() { @Override From e4e1cbbb357113d972abb4085013a065607cf7b6 Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Thu, 10 Sep 2015 15:52:56 -0400 Subject: [PATCH 3/8] Removing a non open wire test -- this test is only validating AMQP <-> MQTT converstion through ActiveMQ5. No value for OpenWire tests --- .../activemq/conversions/AmqpAndMqttTest.java | 119 ------------------ 1 file changed, 119 deletions(-) delete mode 100644 tests/activemq5-unit-tests/src/test/java/org/apache/activemq/conversions/AmqpAndMqttTest.java diff --git a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/conversions/AmqpAndMqttTest.java b/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/conversions/AmqpAndMqttTest.java deleted file mode 100644 index d9b7add1e86..00000000000 --- a/tests/activemq5-unit-tests/src/test/java/org/apache/activemq/conversions/AmqpAndMqttTest.java +++ /dev/null @@ -1,119 +0,0 @@ -/** - * 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.conversions; - -import org.apache.activemq.CombinationTestSupport; -import org.apache.activemq.broker.BrokerService; -import org.apache.activemq.broker.TransportConnector; -import org.apache.qpid.amqp_1_0.jms.impl.ConnectionFactoryImpl; -import org.apache.qpid.amqp_1_0.jms.impl.QueueImpl; -import org.apache.qpid.amqp_1_0.jms.impl.TopicImpl; -import org.fusesource.mqtt.client.BlockingConnection; -import org.fusesource.mqtt.client.MQTT; -import org.fusesource.mqtt.client.QoS; - -import javax.jms.*; -import java.io.UnsupportedEncodingException; -import java.util.Arrays; - -/** - */ -public class AmqpAndMqttTest extends CombinationTestSupport { - - protected BrokerService broker; - private TransportConnector amqpConnector; - private TransportConnector mqttConnector; - - @Override - protected void setUp() throws Exception { - super.setUp(); - broker = createBroker(); - broker.start(); - broker.waitUntilStarted(); - } - - @Override - protected void tearDown() throws Exception { - if (broker != null) { - broker.stop(); - broker.waitUntilStopped(); - broker = null; - } - super.tearDown(); - } - - protected BrokerService createBroker() throws Exception { - BrokerService broker = new BrokerService(); - broker.setPersistent(false); - amqpConnector = broker.addConnector("amqp://0.0.0.0:0"); - mqttConnector = broker.addConnector("mqtt://0.0.0.0:0"); - return broker; - } - - public void testFromMqttToAmqp() throws Exception { - Connection amqp = createAmqpConnection(); - Session session = amqp.createSession(false, Session.AUTO_ACKNOWLEDGE); - MessageConsumer consumer = session.createConsumer(session.createTopic("topic://FOO")); - - final BlockingConnection mqtt = createMQTTConnection().blockingConnection(); - mqtt.connect(); - byte[] payload = bytes("Hello World"); - mqtt.publish("FOO", payload, QoS.AT_LEAST_ONCE, false); - mqtt.disconnect(); - - Message msg = consumer.receive(1000 * 5); - assertNotNull(msg); - assertTrue(msg instanceof BytesMessage); - - BytesMessage bmsg = (BytesMessage) msg; - byte[] actual = new byte[(int) bmsg.getBodyLength()]; - bmsg.readBytes(actual); - assertTrue(Arrays.equals(actual, payload)); - amqp.close(); - } - - private byte[] bytes(String value) { - try { - return value.getBytes("UTF-8"); - } - catch (UnsupportedEncodingException e) { - throw new RuntimeException(e); - } - } - - protected MQTT createMQTTConnection() throws Exception { - MQTT mqtt = new MQTT(); - mqtt.setConnectAttemptsMax(1); - mqtt.setReconnectAttemptsMax(0); - mqtt.setHost("localhost", mqttConnector.getConnectUri().getPort()); - return mqtt; - } - - public Connection createAmqpConnection() throws Exception { - final ConnectionFactoryImpl factory = new ConnectionFactoryImpl("localhost", amqpConnector.getConnectUri().getPort(), "admin", "password"); - final Connection connection = factory.createConnection(); - connection.setExceptionListener(new ExceptionListener() { - @Override - public void onException(JMSException exception) { - exception.printStackTrace(); - } - }); - connection.start(); - return connection; - } - -} From dc53bfd7c70ddec992e7174f5f998d21508f7e4e Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Thu, 10 Sep 2015 16:22:38 -0400 Subject: [PATCH 4/8] fixing a deadlock that appeared on the testsuite (MultipleThreadsOpeningTest) --- .../core/client/impl/ServerLocatorImpl.java | 48 ++++++++++--------- .../cluster/MultipleThreadsOpeningTest.java | 33 +++++++++---- 2 files changed, 49 insertions(+), 32 deletions(-) diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorImpl.java index bed47b7a218..506e084975e 100644 --- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorImpl.java +++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ServerLocatorImpl.java @@ -561,32 +561,34 @@ private ServerLocatorImpl(ServerLocatorImpl locator) { clusterTransportConfiguration = locator.clusterTransportConfiguration; } - private synchronized TransportConfiguration selectConnector() { + private TransportConfiguration selectConnector() { Pair[] usedTopology; synchronized (topologyArrayGuard) { usedTopology = topologyArray; } - // if the topologyArray is null, we will use the initialConnectors - if (usedTopology != null) { - if (ActiveMQClientLogger.LOGGER.isTraceEnabled()) { - ActiveMQClientLogger.LOGGER.trace("Selecting connector from toplogy."); - } - int pos = loadBalancingPolicy.select(usedTopology.length); - Pair pair = usedTopology[pos]; + synchronized (this) { + // if the topologyArray is null, we will use the initialConnectors + if (usedTopology != null) { + if (ActiveMQClientLogger.LOGGER.isTraceEnabled()) { + ActiveMQClientLogger.LOGGER.trace("Selecting connector from toplogy."); + } + int pos = loadBalancingPolicy.select(usedTopology.length); + Pair pair = usedTopology[pos]; - return pair.getA(); - } - else { - // Get from initialconnectors - if (ActiveMQClientLogger.LOGGER.isTraceEnabled()) { - ActiveMQClientLogger.LOGGER.trace("Selecting connector from initial connectors."); + return pair.getA(); } + else { + // Get from initialconnectors + if (ActiveMQClientLogger.LOGGER.isTraceEnabled()) { + ActiveMQClientLogger.LOGGER.trace("Selecting connector from initial connectors."); + } - int pos = loadBalancingPolicy.select(initialConnectors.length); + int pos = loadBalancingPolicy.select(initialConnectors.length); - return initialConnectors[pos]; + return initialConnectors[pos]; + } } } @@ -1494,16 +1496,18 @@ public void run() { } public void factoryClosed(final ClientSessionFactory factory) { + boolean isEmpty; synchronized (factories) { factories.remove(factory); + isEmpty = factories.isEmpty(); + } - if (!clusterConnection && factories.isEmpty()) { - // Go back to using the broadcast or static list - synchronized (topologyArrayGuard) { - receivedTopology = false; + if (!clusterConnection && isEmpty) { + // Go back to using the broadcast or static list + synchronized (topologyArrayGuard) { + receivedTopology = false; - topologyArray = null; - } + topologyArray = null; } } } diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/cluster/MultipleThreadsOpeningTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/cluster/MultipleThreadsOpeningTest.java index 47669daf2d4..6475a1334f9 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/cluster/MultipleThreadsOpeningTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/cluster/MultipleThreadsOpeningTest.java @@ -23,8 +23,8 @@ import org.apache.activemq.artemis.api.core.TransportConfiguration; import org.apache.activemq.artemis.api.jms.ActiveMQJMSClient; import org.apache.activemq.artemis.api.jms.JMSFactoryType; -import org.apache.activemq.artemis.tests.util.JMSClusteredTestBase; import org.apache.activemq.artemis.core.remoting.impl.invm.InVMConnectorFactory; +import org.apache.activemq.artemis.tests.util.JMSClusteredTestBase; import org.junit.Test; public class MultipleThreadsOpeningTest extends JMSClusteredTestBase { @@ -33,7 +33,7 @@ public class MultipleThreadsOpeningTest extends JMSClusteredTestBase { public void testMultipleOpen() throws Exception { cf1 = ActiveMQJMSClient.createConnectionFactoryWithHA(JMSFactoryType.CF, new TransportConfiguration(InVMConnectorFactory.class.getName(), generateInVMParams(1))); - final int numberOfOpens = 2000; + final int numberOfOpens = 500; int numberOfThreads = 20; // I want all the threads aligned, just ready to start creating connections like in a car race final CountDownLatch flagAlignSemaphore = new CountDownLatch(numberOfThreads); @@ -41,6 +41,10 @@ public void testMultipleOpen() throws Exception { class ThreadOpen extends Thread { + ThreadOpen(int i) { + super("MultipleThreadsOpeningTest/ThreadOpen::" + i); + } + int errors = 0; public void run() { @@ -50,8 +54,8 @@ public void run() { flagStartRace.await(); for (int i = 0; i < numberOfOpens; i++) { - if (i % 1000 == 0) - System.out.println("tests " + i); + if (i % 100 == 0) + System.out.println("connections created on Thread " + Thread.currentThread() + " " + i); Connection conn = cf1.createConnection(); Session sess = conn.createSession(true, Session.AUTO_ACKNOWLEDGE); sess.close(); @@ -68,18 +72,27 @@ public void run() { ThreadOpen[] threads = new ThreadOpen[numberOfThreads]; for (int i = 0; i < numberOfThreads; i++) { - threads[i] = new ThreadOpen(); + threads[i] = new ThreadOpen(i); threads[i].start(); } flagAlignSemaphore.await(); flagStartRace.countDown(); - for (ThreadOpen t : threads) { - // 5 minutes seems long but this may take a bit of time in a slower box - t.join(300000); - assertFalse(t.isAlive()); - assertEquals("There are Errors on the test thread", 0, t.errors); + try { + for (ThreadOpen t : threads) { + t.join(60000); + assertFalse(t.isAlive()); + assertEquals("There are Errors on the test thread", 0, t.errors); + } + } + finally { + for (ThreadOpen t : threads) { + if (t.isAlive()) { + t.interrupt(); + } + t.join(1000); + } } } } From 85277698a90e88482f84370fbe799e9031247107 Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Thu, 10 Sep 2015 16:31:47 -0400 Subject: [PATCH 5/8] removing old doc (non relevant any longer) --- docs/user-manual/en/persistence.md | 28 +--------------------------- 1 file changed, 1 insertion(+), 27 deletions(-) diff --git a/docs/user-manual/en/persistence.md b/docs/user-manual/en/persistence.md index c5f0862904e..a9269b127d7 100644 --- a/docs/user-manual/en/persistence.md +++ b/docs/user-manual/en/persistence.md @@ -269,7 +269,7 @@ The message journal is configured using the following attributes in journal. The compacting algorithm won't start until you have at least `journal-compact-min-files` - Setting this to 0 will disable the feature to compact completely. + Setting this to 0 will disable the feature to compact completely. This could be dangerous though as the journal could grow indefinitely. Use it wisely! @@ -354,29 +354,3 @@ straightforward. Simply set the parameter `persistence-enabled` in Please note that if you set this parameter to false, then *zero* persistence will occur. That means no bindings data, message data, large message data, duplicate id caches or paging data will be persisted. - -## Import/Export the Journal Data - -You may want to inspect the existent records on each one of the journals -used by Apache ActiveMQ Artemis, and you can use the export/import tool for that -purpose. -you can export the journal as a text file by using this command: - -`java -cp activemq-tools-jar-with-dependencies.jar export-journal ` - -To import the file as binary data on the journal (Notice you also -require netty.jar): - -`java -cp activemq-tools-jar-with-dependencies.jar import-journal ` - -- JournalDirectory: Use the configured folder for your selected folder. Example: ./activemq/data/journal - -- JournalPrefix: Use the prefix for your selected journal, as discussed above - -- FileExtension: Use the extension for your selected journal, as discussed above - -- FileSize: Use the size for your selected journal, as discussed above - -- FileOutput or FileInput: text file that will contain the exported data - -See [Tools](tools.md) for more information. From d5fca7213bf5a3615ebca076f9f386d6fcda88ca Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Thu, 10 Sep 2015 16:51:43 -0400 Subject: [PATCH 6/8] Sending interrupts on Leaking Threads through the testsuite, trying to speed up tests --- .../apache/activemq/artemis/tests/util/ActiveMQTestBase.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java b/artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java index 436ec3c6c7d..05f31deb45d 100644 --- a/artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java +++ b/artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java @@ -1943,6 +1943,9 @@ private boolean checkThread(StringBuffer buffer) { for (StackTraceElement el : elements) { buffer.append(el + "\n"); } + + buffer.append("\n-- Interrupting being sent on " + aliveThread + "--\n"); + aliveThread.interrupt(); } } From 1e9db8e82f7305b0086e70da5d4a7fec0c546453 Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Thu, 10 Sep 2015 17:50:07 -0400 Subject: [PATCH 7/8] Fixing test that eventually hangs on Virtual Machines (runs fine if one real machines) --- .../artemis/tests/integration/proton/ProtonTest.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proton/ProtonTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proton/ProtonTest.java index e779c14985f..7663f69f19a 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proton/ProtonTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proton/ProtonTest.java @@ -131,6 +131,7 @@ public void setUp() throws Exception { @After public void tearDown() throws Exception { try { + Thread.sleep(250); if (connection != null) { connection.close(); } @@ -257,6 +258,10 @@ public void testConnection() throws Exception { cons.close(); + for (int i = 0; i < 100 && serverQueue.getConsumerCount() != 0; i++) { + Thread.sleep(500); + } + assertEquals(0, serverQueue.getConsumerCount()); session.close(); @@ -330,7 +335,7 @@ public void testCancelMessages() throws Exception { connection = createConnection(); session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageConsumer consumer = session.createConsumer(queue); - Thread.sleep(1000); + Thread.sleep(100); consumer.close(); connection.close(); Assert.assertEquals(numMessages, getMessageCount(q)); From 3bff87cad85c300546ab5e9262c36cfb31c0048f Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Thu, 10 Sep 2015 18:26:06 -0400 Subject: [PATCH 8/8] Revert "Sending interrupts on Leaking Threads through the testsuite, trying to speed up tests" This reverts commit d5fca7213bf5a3615ebca076f9f386d6fcda88ca. --- .../apache/activemq/artemis/tests/util/ActiveMQTestBase.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java b/artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java index 05f31deb45d..436ec3c6c7d 100644 --- a/artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java +++ b/artemis-server/src/test/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java @@ -1943,9 +1943,6 @@ private boolean checkThread(StringBuffer buffer) { for (StackTraceElement el : elements) { buffer.append(el + "\n"); } - - buffer.append("\n-- Interrupting being sent on " + aliveThread + "--\n"); - aliveThread.interrupt(); } }