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

using new qpid-jms client on AMQP

This commit is contained in:
Clebert Suconic 2015-09-10 13:57:18 -04:00
parent 649edd3a37
commit 7b2c50415a
13 changed files with 62 additions and 379 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
artemis-protocols/artemis-proton-plug/pom.xml
View File

@ -88,8 +88,8 @@
</dependency>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-amqp-1-0-client-jms</artifactId>
<version>0.24</version>
<artifactId>qpid-jms-client</artifactId>
<version>0.5.0</version>
<scope>test</scope>
</dependency>

34
artemis-protocols/artemis-proton-plug/src/test/java/org/proton/plug/test/AbstractJMSTest.java
View File

@ -21,24 +21,22 @@ import javax.jms.ConnectionFactory;
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 class AbstractJMSTest {
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);
}
}

29
artemis-protocols/artemis-proton-plug/src/test/java/org/proton/plug/test/ProtonTest.java
View File

@ -32,7 +32,6 @@ import java.util.Collection;
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<Object[]> data() {
List<Object[]> list = Arrays.asList(new Object[][]{{Boolean.FALSE, Boolean.TRUE}, {Boolean.FALSE, Boolean.FALSE}});
List<Object[]> 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 class ProtonTest extends AbstractJMSTest {
public void testMessagesReceivedInParallel() throws Throwable {
final int numMessages = getNumberOfMessages();
long time = System.currentTimeMillis();
final Queue queue = createQueue();
final ArrayList<Throwable> exceptions = new ArrayList<>();
@ -105,6 +102,7 @@ public class ProtonTest extends AbstractJMSTest {
// 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 class ProtonTest extends AbstractJMSTest {
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 class ProtonTest extends AbstractJMSTest {
}
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 class ProtonTest extends AbstractJMSTest {
@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 class ProtonTest extends AbstractJMSTest {
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 class ProtonTest extends AbstractJMSTest {
@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 class ProtonTest extends AbstractJMSTest {
@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 class ProtonTest extends AbstractJMSTest {
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);

4
examples/protocols/amqp/pom.xml
View File

@ -48,7 +48,7 @@ under the License.
<id>release</id>
<modules>
<module>proton-cpp</module>
<module>proton-j</module>
<module>queue</module>
<module>proton-ruby</module>
</modules>
</profile>
@ -58,7 +58,7 @@ under the License.
<!-- this one to be run manually
<module>proton-cpp</module> -->
<module>proton-j</module>
<module>queue</module>
<module>proton-ruby</module>
</modules>
</profile>

17
examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/artemis-roles.properties
View File

@ -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

17
examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/artemis-users.properties
View File

@ -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

68
examples/protocols/amqp/proton-j/src/main/resources/activemq/server0/broker.xml
View File

@ -1,68 +0,0 @@
<?xml version='1.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.
-->
<configuration xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="urn:activemq"
xsi:schemaLocation="urn:activemq /schema/artemis-server.xsd">
<jms xmlns="urn:activemq:jms">
<!--the queue used by the example-->
<queue name="exampleQueue"/>
</jms>
<core xmlns="urn:activemq:core">
<bindings-directory>${data.dir}/server0/data/messaging/bindings</bindings-directory>
<journal-directory>${data.dir}/server0/data/messaging/journal</journal-directory>
<large-messages-directory>${data.dir}/server0/data/messaging/largemessages</large-messages-directory>
<paging-directory>${data.dir}/server0/data/messaging/paging</paging-directory>
<!-- Acceptors -->
<acceptors>
<acceptor name="proton-acceptor">tcp://localhost:5672</acceptor>
<acceptor name="netty-acceptor">tcp://localhost:61616</acceptor>
</acceptors>
<queues>
<queue name="testQueue">
<address>testQueue</address>
</queue>
</queues>
<!-- Other config -->
<security-settings>
<!--security for example queue-->
<security-setting match="jms.queue.exampleQueue">
<permission type="createDurableQueue" roles="guest"/>
<permission type="deleteDurableQueue" roles="guest"/>
<permission type="createNonDurableQueue" roles="guest"/>
<permission type="deleteNonDurableQueue" roles="guest"/>
<permission type="consume" roles="guest"/>
<permission type="send" roles="guest"/>
</security-setting>
</security-settings>
</core>
</configuration>

8
examples/protocols/amqp/proton-j/pom.xml → examples/protocols/amqp/queue/pom.xml
View File

@ -27,7 +27,7 @@ under the License.
<version>1.1.1-SNAPSHOT</version>
</parent>
<artifactId>artemis-proton-j</artifactId>
<artifactId>queue</artifactId>
<packaging>jar</packaging>
<name>ActiveMQ Artemis Proton-J Example</name>
@ -38,8 +38,8 @@ under the License.
<dependencies>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-amqp-1-0-client</artifactId>
<version>0.22</version>
<artifactId>qpid-jms-client</artifactId>
<version>0.5.0</version>
</dependency>
</dependencies>
@ -97,7 +97,7 @@ under the License.
<dependencies>
<dependency>
<groupId>org.apache.activemq.examples.amqp</groupId>
<artifactId>artemis-proton-j</artifactId>
<artifactId>queue</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>

49
examples/protocols/amqp/proton-j/readme.html → examples/protocols/amqp/queue/readme.html
View File

@ -42,54 +42,5 @@ under the License.
&lt;acceptor name="proton-acceptor"&gt;tcp://localhost:5672&lt;/acceptor&gt;
</code>
</pre>
<h2>Example step-by-step</h2>
<ol>
<li> Create an amqp qpid 1.0 connection.</li>
<pre class="prettyprint">
<code>connection= new Connection("localhost", 5672, null, null);</code>
</pre>
<li>Create a session</li>
<pre class="prettyprint">
<code>Session session = connection.createSession();</code>
</pre>
<li>Create a sender</li>
<pre class="prettyprint">
<code>Sender sender = session.createSender("testQueue");</code>
</pre>
<li>send a simple message</li>
<pre class="prettyprint">
<code>sender.send(new Message("I am an amqp message"));</code>
</pre>
<li>create a moving receiver, this means the message will be removed from the queue</li>
<pre class="prettyprint">
<code>Receiver rec = session.createMovingReceiver("testQueue");</code>
</pre>
<li>set some credit so we can receive</li>
<pre class="prettyprint">
<code>rec.setCredit(UnsignedInteger.valueOf(1), false);</code>
</pre>
<li>receive the simple message</li>
<pre class="prettyprint">
<code>Message m = rec.receive(5000);
System.out.println("message = " + m.getPayload());</code>
</pre>
<li>acknowledge the message</li>
<pre class="prettyprint">
<code>rec.acknowledge(m);</code>
</pre>
<li>close the connection</li>
<pre class="prettyprint">
<code>connection.close();</code>
</pre>
</ol>
</body>
</html>

39
examples/protocols/amqp/proton-j/src/main/java/org/apache/activemq/artemis/jms/example/ProtonJExample.java → examples/protocols/amqp/queue/src/main/java/org/apache/activemq/artemis/jms/example/ProtonJExample.java
View File

@ -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 simple message
sender.send(new Message("I am an amqp message"));
// Step 4. send a few simple message
sender.send(session.createTextMessage("Hello world "));
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) {

8
tests/activemq5-unit-tests/pom.xml
View File

@ -37,7 +37,6 @@
<org-apache-derby-version>10.11.1.1</org-apache-derby-version>
<commons-io-version>2.4</commons-io-version>
<commons-net-version>3.3</commons-net-version>
<qpid-jms-version>0.30</qpid-jms-version>
<xbean-version>3.18</xbean-version>
<hamcrest-version>1.3</hamcrest-version>
<slf4j-version>1.7.10</slf4j-version>
@ -195,13 +194,6 @@
<version>${commons-net-version}</version>
</dependency>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-amqp-1-0-client-jms</artifactId>
<version>${qpid-jms-version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.xbean</groupId>
<artifactId>xbean-spring</artifactId>

4
tests/integration-tests/pom.xml
View File

@ -187,8 +187,8 @@
</dependency>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-amqp-1-0-client-jms</artifactId>
<version>0.24</version>
<artifactId>qpid-jms-client</artifactId>
<version>0.5.0</version>
</dependency>
<dependency>
<groupId>org.apache.qpid</groupId>

160
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/proton/ProtonTest.java
View File

@ -40,22 +40,15 @@ import java.util.Enumeration;
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 class ProtonTest extends ActiveMQTestBase {
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;
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 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);
return session.createQueue(address);
}
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;
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);
}
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 class ProtonTest extends ActiveMQTestBase {
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 class ProtonTest extends ActiveMQTestBase {
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