+ * 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.artemis.tests.integration.crossprotocol; + +import java.net.URI; +import java.util.concurrent.TimeUnit; + +import org.apache.activemq.artemis.api.core.RoutingType; +import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.core.config.Configuration; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.settings.impl.AddressSettings; +import org.apache.activemq.artemis.tests.integration.stomp.StompTestBase; +import org.apache.activemq.artemis.tests.integration.stomp.util.ClientStompFrame; +import org.apache.activemq.artemis.tests.integration.stomp.util.StompClientConnection; +import org.apache.activemq.artemis.tests.integration.stomp.util.StompClientConnectionFactory; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; +import org.apache.activemq.transport.amqp.client.AmqpClient; +import org.apache.activemq.transport.amqp.client.AmqpConnection; +import org.apache.activemq.transport.amqp.client.AmqpMessage; +import org.apache.activemq.transport.amqp.client.AmqpSender; +import org.apache.activemq.transport.amqp.client.AmqpSession; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class AMQPToStompTest extends ActiveMQTestBase { + + private ActiveMQServer server; + protected String queueName = "amqpToStompTestQueue1"; + private SimpleString coreQueue; + + @Override + @Before + public void setUp() throws Exception { + super.setUp(); + server = createServer(true, true); + server.start(); + server.waitForActivation(10, TimeUnit.SECONDS); + + Configuration serverConfig = server.getConfiguration(); + serverConfig.getAddressesSettings().put("#", new AddressSettings().setAutoCreateQueues(false) + .setAutoCreateAddresses(false) + .setDeadLetterAddress(new SimpleString("ActiveMQ.DLQ"))); + serverConfig.setSecurityEnabled(false); + coreQueue = new SimpleString(queueName); + server.createQueue(coreQueue, RoutingType.ANYCAST, coreQueue, null, false, false); + } + + @Override + @After + public void tearDown() throws Exception { + server.stop(); + super.tearDown(); + } + + @Test + public void testSendAmqpReceiveStomp() throws Exception { + AmqpClient client = new AmqpClient(new URI("tcp://127.0.0.1:61616"), null, null); + AmqpConnection amqpconnection = client.connect(); + try { + AmqpSession session = amqpconnection.createSession(); + AmqpSender sender = session.createSender(queueName); + AmqpMessage message = new AmqpMessage(); + message.setText("mine"); + sender.send(message); + } finally { + amqpconnection.close(); + } + + StompClientConnection conn = StompClientConnectionFactory.createClientConnection(new URI("tcp://127.0.0.1:61616")); + conn.connect(null, null); + try { + StompTestBase.subscribeQueue(conn, null, queueName); + ClientStompFrame frame = conn.receiveFrame(); + assertNotNull(frame); + assertNotNull(frame.getBody()); + assertTrue(frame.getBody().contains("mine")); + } finally { + conn.closeTransport(); + } + } +} diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTestBase.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTestBase.java index 08f6be3389..ec166dd388 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTestBase.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTestBase.java @@ -219,19 +219,19 @@ public abstract class StompTestBase extends ActiveMQTestBase { return new ActiveMQJMSConnectionFactory(false, new TransportConfiguration(InVMConnectorFactory.class.getName())); } - protected String getQueueName() { + protected static String getQueueName() { return "testQueue"; } - protected String getQueuePrefix() { + protected static String getQueuePrefix() { return ""; } - protected String getTopicName() { + protected static String getTopicName() { return "testtopic"; } - protected String getTopicPrefix() { + protected static String getTopicPrefix() { return ""; } @@ -264,25 +264,25 @@ public abstract class StompTestBase extends ActiveMQTestBase { producer.send(message); } - public void abortTransaction(StompClientConnection conn, String txID) throws IOException, InterruptedException { + public static void abortTransaction(StompClientConnection conn, String txID) throws IOException, InterruptedException { ClientStompFrame abortFrame = conn.createFrame(Stomp.Commands.ABORT) .addHeader(Stomp.Headers.TRANSACTION, txID); conn.sendFrame(abortFrame); } - public void beginTransaction(StompClientConnection conn, String txID) throws IOException, InterruptedException { + public static void beginTransaction(StompClientConnection conn, String txID) throws IOException, InterruptedException { ClientStompFrame beginFrame = conn.createFrame(Stomp.Commands.BEGIN) .addHeader(Stomp.Headers.TRANSACTION, txID); conn.sendFrame(beginFrame); } - public void commitTransaction(StompClientConnection conn, String txID) throws IOException, InterruptedException { + public static void commitTransaction(StompClientConnection conn, String txID) throws IOException, InterruptedException { commitTransaction(conn, txID, false); } - public void commitTransaction(StompClientConnection conn, + public static void commitTransaction(StompClientConnection conn, String txID, boolean receipt) throws IOException, InterruptedException { ClientStompFrame beginFrame = conn.createFrame(Stomp.Commands.COMMIT) @@ -297,7 +297,7 @@ public abstract class StompTestBase extends ActiveMQTestBase { } } - public void ack(StompClientConnection conn, + public static void ack(StompClientConnection conn, String subscriptionId, ClientStompFrame messageIdFrame) throws IOException, InterruptedException { String messageID = messageIdFrame.getHeader(Stomp.Headers.Message.MESSAGE_ID); @@ -315,7 +315,7 @@ public abstract class StompTestBase extends ActiveMQTestBase { } } - public void ack(StompClientConnection conn, + public static void ack(StompClientConnection conn, String subscriptionId, String mid, String txID) throws IOException, InterruptedException { @@ -329,7 +329,7 @@ public abstract class StompTestBase extends ActiveMQTestBase { conn.sendFrame(frame); } - public void nack(StompClientConnection conn, String subscriptionId, String messageId) throws IOException, InterruptedException { + public static void nack(StompClientConnection conn, String subscriptionId, String messageId) throws IOException, InterruptedException { ClientStompFrame frame = conn.createFrame(Stomp.Commands.NACK) .addHeader(Stomp.Headers.Ack.SUBSCRIPTION, subscriptionId) .addHeader(Stomp.Headers.Message.MESSAGE_ID, messageId); @@ -337,25 +337,25 @@ public abstract class StompTestBase extends ActiveMQTestBase { conn.sendFrame(frame); } - public ClientStompFrame subscribe(StompClientConnection conn, + public static ClientStompFrame subscribe(StompClientConnection conn, String subscriptionId) throws IOException, InterruptedException { return subscribe(conn, subscriptionId, Stomp.Headers.Subscribe.AckModeValues.AUTO); } - public ClientStompFrame subscribe(StompClientConnection conn, + public static ClientStompFrame subscribe(StompClientConnection conn, String subscriptionId, String ack) throws IOException, InterruptedException { return subscribe(conn, subscriptionId, ack, null); } - public ClientStompFrame subscribe(StompClientConnection conn, + public static ClientStompFrame subscribe(StompClientConnection conn, String subscriptionId, String ack, String durableId) throws IOException, InterruptedException { return subscribe(conn, subscriptionId, ack, durableId, true); } - public ClientStompFrame subscribe(StompClientConnection conn, + public static ClientStompFrame subscribe(StompClientConnection conn, String subscriptionId, String ack, String durableId, @@ -363,7 +363,7 @@ public abstract class StompTestBase extends ActiveMQTestBase { return subscribe(conn, subscriptionId, ack, durableId, null, receipt); } - public ClientStompFrame subscribe(StompClientConnection conn, + public static ClientStompFrame subscribe(StompClientConnection conn, String subscriptionId, String ack, String durableId, @@ -371,7 +371,7 @@ public abstract class StompTestBase extends ActiveMQTestBase { return subscribe(conn, subscriptionId, ack, durableId, selector, true); } - public ClientStompFrame subscribe(StompClientConnection conn, + public static ClientStompFrame subscribe(StompClientConnection conn, String subscriptionId, String ack, String durableId, @@ -380,11 +380,11 @@ public abstract class StompTestBase extends ActiveMQTestBase { return subscribe(conn, subscriptionId, ack, durableId, selector, getQueuePrefix() + getQueueName(), receipt); } - public ClientStompFrame subscribeQueue(StompClientConnection conn, String subId, String destination) throws IOException, InterruptedException { + public static ClientStompFrame subscribeQueue(StompClientConnection conn, String subId, String destination) throws IOException, InterruptedException { return subscribe(conn, subId, Stomp.Headers.Subscribe.AckModeValues.AUTO, null, null, destination, true); } - public ClientStompFrame subscribe(StompClientConnection conn, + public static ClientStompFrame subscribe(StompClientConnection conn, String subscriptionId, String ack, String durableId, @@ -426,14 +426,14 @@ public abstract class StompTestBase extends ActiveMQTestBase { return frame; } - public ClientStompFrame subscribeTopic(StompClientConnection conn, + public static ClientStompFrame subscribeTopic(StompClientConnection conn, String subscriptionId, String ack, String durableId) throws IOException, InterruptedException { return subscribeTopic(conn, subscriptionId, ack, durableId, true); } - public ClientStompFrame subscribeTopic(StompClientConnection conn, + public static ClientStompFrame subscribeTopic(StompClientConnection conn, String subscriptionId, String ack, String durableId, @@ -441,7 +441,7 @@ public abstract class StompTestBase extends ActiveMQTestBase { return subscribeTopic(conn, subscriptionId, ack, durableId, receipt, false); } - public ClientStompFrame subscribeTopic(StompClientConnection conn, + public static ClientStompFrame subscribeTopic(StompClientConnection conn, String subscriptionId, String ack, String durableId, @@ -481,17 +481,17 @@ public abstract class StompTestBase extends ActiveMQTestBase { return frame; } - public ClientStompFrame unsubscribe(StompClientConnection conn, String subscriptionId) throws IOException, InterruptedException { + public static ClientStompFrame unsubscribe(StompClientConnection conn, String subscriptionId) throws IOException, InterruptedException { return unsubscribe(conn, subscriptionId, null, false, false); } - public ClientStompFrame unsubscribe(StompClientConnection conn, + public static ClientStompFrame unsubscribe(StompClientConnection conn, String subscriptionId, boolean receipt) throws IOException, InterruptedException { return unsubscribe(conn, subscriptionId, null, receipt, false); } - public ClientStompFrame unsubscribe(StompClientConnection conn, + public static ClientStompFrame unsubscribe(StompClientConnection conn, String subscriptionId, String destination, boolean receipt, @@ -523,19 +523,19 @@ public abstract class StompTestBase extends ActiveMQTestBase { return frame; } - public ClientStompFrame send(StompClientConnection conn, String destination, String contentType, String body) throws IOException, InterruptedException { + public static ClientStompFrame send(StompClientConnection conn, String destination, String contentType, String body) throws IOException, InterruptedException { return send(conn, destination, contentType, body, false); } - public ClientStompFrame send(StompClientConnection conn, String destination, String contentType, String body, boolean receipt) throws IOException, InterruptedException { + public static ClientStompFrame send(StompClientConnection conn, String destination, String contentType, String body, boolean receipt) throws IOException, InterruptedException { return send(conn, destination, contentType, body, receipt, null); } - public ClientStompFrame send(StompClientConnection conn, String destination, String contentType, String body, boolean receipt, RoutingType destinationType) throws IOException, InterruptedException { + public static ClientStompFrame send(StompClientConnection conn, String destination, String contentType, String body, boolean receipt, RoutingType destinationType) throws IOException, InterruptedException { return send(conn, destination, contentType, body, receipt, destinationType, null); } - public ClientStompFrame send(StompClientConnection conn, String destination, String contentType, String body, boolean receipt, RoutingType destinationType, String txId) throws IOException, InterruptedException { + public static ClientStompFrame send(StompClientConnection conn, String destination, String contentType, String body, boolean receipt, RoutingType destinationType, String txId) throws IOException, InterruptedException { ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND) .addHeader(Stomp.Headers.Send.DESTINATION, destination) .setBody(body); @@ -573,7 +573,7 @@ public abstract class StompTestBase extends ActiveMQTestBase { return frame; } - public URI createStompClientUri(String scheme, String hostname, int port) throws URISyntaxException { + public static URI createStompClientUri(String scheme, String hostname, int port) throws URISyntaxException { return new URI(scheme + "://" + hostname + ":" + port); } } diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/v11/StompV11Test.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/v11/StompV11Test.java index 15f7146374..99ad1fba3a 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/v11/StompV11Test.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/v11/StompV11Test.java @@ -1153,7 +1153,7 @@ public class StompV11Test extends StompTestBase { } //ack the last - this.ack(conn, "sub1", frame); + ack(conn, "sub1", frame); unsubscribe(conn, "sub1"); @@ -1186,7 +1186,7 @@ public class StompV11Test extends StompTestBase { //ack the 49th if (i == num - 2) { - this.ack(conn, "sub1", frame); + ack(conn, "sub1", frame); } } @@ -1252,7 +1252,7 @@ public class StompV11Test extends StompTestBase { IntegrationTestLogger.LOGGER.info(i + " == received: " + frame); //ack on even numbers if (i % 2 == 0) { - this.ack(conn, "sub1", frame); + ack(conn, "sub1", frame); } } @@ -1279,12 +1279,12 @@ public class StompV11Test extends StompTestBase { public void testTwoSubscribers() throws Exception { conn.connect(defUser, defPass, CLIENT_ID); - this.subscribeTopic(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, null); + subscribeTopic(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, null); StompClientConnection newConn = StompClientConnectionFactory.createClientConnection(uri); newConn.connect(defUser, defPass, "myclientid2"); - this.subscribeTopic(newConn, "sub2", Stomp.Headers.Subscribe.AckModeValues.AUTO, null); + subscribeTopic(newConn, "sub2", Stomp.Headers.Subscribe.AckModeValues.AUTO, null); send(newConn, getTopicPrefix() + getTopicName(), null, "Hello World"); @@ -1302,8 +1302,8 @@ public class StompV11Test extends StompTestBase { assertEquals("sub2", frame.getHeader(Stomp.Headers.Message.SUBSCRIPTION)); // remove suscription - this.unsubscribe(conn, "sub1", true); - this.unsubscribe(newConn, "sub2", true); + unsubscribe(conn, "sub1", true); + unsubscribe(newConn, "sub2", true); conn.disconnect(); newConn.disconnect(); @@ -1318,7 +1318,7 @@ public class StompV11Test extends StompTestBase { StompClientConnection connV11_2 = StompClientConnectionFactory.createClientConnection(uri); connV11_2.connect(defUser, defPass); - this.subscribe(connV11_2, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(connV11_2, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); ClientStompFrame frame = connV11_2.receiveFrame(2000); @@ -1347,7 +1347,7 @@ public class StompV11Test extends StompTestBase { public void testBodyWithUTF8() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, getName(), Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, getName(), Stomp.Headers.Subscribe.AckModeValues.AUTO); String text = "A" + "\u00ea" + "\u00f1" + "\u00fc" + "C"; IntegrationTestLogger.LOGGER.info(text); @@ -1366,7 +1366,7 @@ public class StompV11Test extends StompTestBase { public void testClientAckNotPartOfTransaction() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, getName(), Stomp.Headers.Subscribe.AckModeValues.CLIENT); + subscribe(conn, getName(), Stomp.Headers.Subscribe.AckModeValues.CLIENT); sendJmsMessage(getName()); @@ -1381,7 +1381,7 @@ public class StompV11Test extends StompTestBase { beginTransaction(conn, "tx1"); - this.ack(conn, getName(), messageID, "tx1"); + ack(conn, getName(), messageID, "tx1"); abortTransaction(conn, "tx1"); @@ -1389,7 +1389,7 @@ public class StompV11Test extends StompTestBase { assertNull(frame); - this.unsubscribe(conn, getName()); + unsubscribe(conn, getName()); conn.disconnect(); } @@ -1398,7 +1398,7 @@ public class StompV11Test extends StompTestBase { public void testDisconnectAndError() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, getName(), Stomp.Headers.Subscribe.AckModeValues.CLIENT); + subscribe(conn, getName(), Stomp.Headers.Subscribe.AckModeValues.CLIENT); String uuid = UUID.randomUUID().toString(); @@ -1453,9 +1453,9 @@ public class StompV11Test extends StompTestBase { public void testDurableSubscriber() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.CLIENT, getName()); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.CLIENT, getName()); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.CLIENT, getName(), false); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.CLIENT, getName(), false); ClientStompFrame frame = conn.receiveFrame(); Assert.assertTrue(frame.getCommand().equals(Stomp.Responses.ERROR)); @@ -1467,7 +1467,7 @@ public class StompV11Test extends StompTestBase { public void testDurableSubscriberWithReconnection() throws Exception { conn.connect(defUser, defPass, CLIENT_ID); - this.subscribeTopic(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, getName()); + subscribeTopic(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, getName()); String uuid = UUID.randomUUID().toString(); @@ -1487,7 +1487,7 @@ public class StompV11Test extends StompTestBase { conn = StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass, CLIENT_ID); - this.subscribeTopic(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, getName()); + subscribeTopic(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, getName()); // we must have received the message frame = conn.receiveFrame(); @@ -1496,7 +1496,7 @@ public class StompV11Test extends StompTestBase { Assert.assertNotNull(frame.getHeader(Stomp.Headers.Message.DESTINATION)); Assert.assertEquals(getName(), frame.getBody()); - this.unsubscribe(conn, "sub1"); + unsubscribe(conn, "sub1"); conn.disconnect(); } @@ -1505,14 +1505,14 @@ public class StompV11Test extends StompTestBase { public void testDurableUnSubscribe() throws Exception { conn.connect(defUser, defPass, CLIENT_ID); - this.subscribeTopic(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO, getName()); + subscribeTopic(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO, getName()); conn.disconnect(); conn.destroy(); conn = StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass, CLIENT_ID); - this.unsubscribe(conn, getName(), null, false, true); + unsubscribe(conn, getName(), null, false, true); long start = System.currentTimeMillis(); SimpleString queueName = SimpleString.toSimpleString(CLIENT_ID + "." + getName()); @@ -1552,7 +1552,7 @@ public class StompV11Test extends StompTestBase { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); for (int i = 0; i < ctr; ++i) { data[i] = getName() + i; @@ -1583,7 +1583,7 @@ public class StompV11Test extends StompTestBase { public void testSubscribeWithAutoAckAndSelector() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, null, "foo = 'zzz'"); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, null, "foo = 'zzz'"); sendJmsMessage("Ignored message", "foo", "1234"); sendJmsMessage("Real message", "foo", "zzz"); @@ -1599,7 +1599,7 @@ public class StompV11Test extends StompTestBase { public void testRedeliveryWithClientAck() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "subscriptionId", Stomp.Headers.Subscribe.AckModeValues.CLIENT); + subscribe(conn, "subscriptionId", Stomp.Headers.Subscribe.AckModeValues.CLIENT); sendJmsMessage(getName()); @@ -1834,7 +1834,7 @@ public class StompV11Test extends StompTestBase { public void testSubscribeToTopic() throws Exception { conn.connect(defUser, defPass); - this.subscribeTopic(conn, "sub1", null, null, true); + subscribeTopic(conn, "sub1", null, null, true); sendJmsMessage(getName(), topic); @@ -1844,7 +1844,7 @@ public class StompV11Test extends StompTestBase { Assert.assertTrue(frame.getHeader(Stomp.Headers.Message.DESTINATION).equals(getTopicPrefix() + getTopicName())); Assert.assertTrue(frame.getBody().equals(getName())); - this.unsubscribe(conn, "sub1", true); + unsubscribe(conn, "sub1", true); sendJmsMessage(getName(), topic); @@ -1858,7 +1858,7 @@ public class StompV11Test extends StompTestBase { public void testSubscribeToTopicWithNoLocal() throws Exception { conn.connect(defUser, defPass); - this.subscribeTopic(conn, "sub1", null, null, true, true); + subscribeTopic(conn, "sub1", null, null, true, true); send(conn, getTopicPrefix() + getTopicName(), null, "Hello World"); @@ -1875,7 +1875,7 @@ public class StompV11Test extends StompTestBase { Assert.assertTrue(frame.getHeader(Stomp.Headers.Message.DESTINATION).equals(getTopicPrefix() + getTopicName())); Assert.assertTrue(frame.getBody().equals(getName())); - this.unsubscribe(conn, "sub1"); + unsubscribe(conn, "sub1"); conn.disconnect(); } @@ -1884,7 +1884,7 @@ public class StompV11Test extends StompTestBase { public void testSubscribeWithAutoAck() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); sendJmsMessage(getName()); @@ -1906,7 +1906,7 @@ public class StompV11Test extends StompTestBase { public void testSubscribeWithAutoAckAndBytesMessage() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); byte[] payload = new byte[]{1, 2, 3, 4, 5}; sendJmsMessage(payload, queue); @@ -1930,7 +1930,7 @@ public class StompV11Test extends StompTestBase { public void testSubscribeWithClientAck() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.CLIENT); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.CLIENT); sendJmsMessage(getName()); @@ -1938,7 +1938,7 @@ public class StompV11Test extends StompTestBase { assertEquals(getName().length(), Integer.parseInt(frame.getHeader(Stomp.Headers.CONTENT_LENGTH))); - this.ack(conn, "sub1", frame); + ack(conn, "sub1", frame); conn.disconnect(); @@ -1962,7 +1962,7 @@ public class StompV11Test extends StompTestBase { public void testSubscribeWithID() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "mysubid", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "mysubid", Stomp.Headers.Subscribe.AckModeValues.AUTO); sendJmsMessage(getName()); @@ -1977,7 +1977,7 @@ public class StompV11Test extends StompTestBase { public void testSubscribeWithMessageSentWithProperties() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); MessageProducer producer = session.createProducer(queue); BytesMessage message = session.createBytesMessage(); @@ -2015,7 +2015,7 @@ public class StompV11Test extends StompTestBase { conn.connect(defUser, defPass); // first tx - this.beginTransaction(conn, "tx1"); + beginTransaction(conn, "tx1"); ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND) .addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()) @@ -2024,13 +2024,13 @@ public class StompV11Test extends StompTestBase { conn.sendFrame(frame); - this.commitTransaction(conn, "tx1"); + commitTransaction(conn, "tx1"); Message message = consumer.receive(1000); Assert.assertNotNull("Should have received a message", message); // 2nd tx with same tx ID - this.beginTransaction(conn, "tx1"); + beginTransaction(conn, "tx1"); frame = conn.createFrame(Stomp.Commands.SEND) .addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()) @@ -2039,7 +2039,7 @@ public class StompV11Test extends StompTestBase { conn.sendFrame(frame); - this.commitTransaction(conn, "tx1"); + commitTransaction(conn, "tx1"); message = consumer.receive(1000); Assert.assertNotNull("Should have received a message", message); @@ -2053,7 +2053,7 @@ public class StompV11Test extends StompTestBase { conn.connect(defUser, defPass); - this.beginTransaction(conn, "tx1"); + beginTransaction(conn, "tx1"); ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND) .addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()) @@ -2068,7 +2068,7 @@ public class StompV11Test extends StompTestBase { // check the message is not committed assertNull(consumer.receive(100)); - this.commitTransaction(conn, "tx1", true); + commitTransaction(conn, "tx1", true); Message message = consumer.receive(1000); Assert.assertNotNull("Should have received a message", message); @@ -2082,7 +2082,7 @@ public class StompV11Test extends StompTestBase { conn.connect(defUser, defPass); - this.beginTransaction(conn, "tx1"); + beginTransaction(conn, "tx1"); ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND) .addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()) @@ -2092,9 +2092,9 @@ public class StompV11Test extends StompTestBase { conn.sendFrame(frame); // rollback first message - this.abortTransaction(conn, "tx1"); + abortTransaction(conn, "tx1"); - this.beginTransaction(conn, "tx1"); + beginTransaction(conn, "tx1"); frame = conn.createFrame(Stomp.Commands.SEND) .addHeader(Stomp.Headers.Send.DESTINATION, getQueuePrefix() + getQueueName()) @@ -2103,7 +2103,7 @@ public class StompV11Test extends StompTestBase { conn.sendFrame(frame); - this.commitTransaction(conn, "tx1", true); + commitTransaction(conn, "tx1", true); // only second msg should be received since first msg was rolled back TextMessage message = (TextMessage) consumer.receive(1000); @@ -2117,7 +2117,7 @@ public class StompV11Test extends StompTestBase { public void testUnsubscribe() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); // send a message to our queue sendJmsMessage("first message"); @@ -2128,7 +2128,7 @@ public class StompV11Test extends StompTestBase { Assert.assertTrue(frame.getCommand().equals(Stomp.Responses.MESSAGE)); // remove suscription - this.unsubscribe(conn, "sub1", true); + unsubscribe(conn, "sub1", true); // send a message to our queue sendJmsMessage("second message"); @@ -2185,7 +2185,7 @@ public class StompV11Test extends StompTestBase { protected void assertSubscribeWithClientAckThenConsumeWithAutoAck(boolean sendDisconnect) throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.CLIENT); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.CLIENT); sendJmsMessage(getName()); @@ -2206,7 +2206,7 @@ public class StompV11Test extends StompTestBase { // message should be received since message was not acknowledged conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", null); + subscribe(conn, "sub1", null); frame = conn.receiveFrame(); Assert.assertTrue(frame.getCommand().equals(Stomp.Responses.MESSAGE)); @@ -2218,7 +2218,7 @@ public class StompV11Test extends StompTestBase { conn = StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", null, null, true); + subscribe(conn, "sub1", null, null, true); sendJmsMessage("shouldBeNextMessage"); @@ -2273,7 +2273,7 @@ public class StompV11Test extends StompTestBase { public void testSendContentType() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); MessageProducer producer = session.createProducer(queue); BytesMessage message = session.createBytesMessage(); diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/v12/StompV12Test.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/v12/StompV12Test.java index f80052be01..d94d7c1421 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/v12/StompV12Test.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/v12/StompV12Test.java @@ -1268,12 +1268,12 @@ public class StompV12Test extends StompTestBase { public void testTwoSubscribers() throws Exception { conn.connect(defUser, defPass, CLIENT_ID); - this.subscribeTopic(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, null); + subscribeTopic(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, null); StompClientConnection newConn = StompClientConnectionFactory.createClientConnection(v11Uri); newConn.connect(defUser, defPass, "myclientid2"); - this.subscribeTopic(newConn, "sub2", Stomp.Headers.Subscribe.AckModeValues.AUTO, null); + subscribeTopic(newConn, "sub2", Stomp.Headers.Subscribe.AckModeValues.AUTO, null); send(conn, getTopicPrefix() + getTopicName(), null, "Hello World"); @@ -1291,8 +1291,8 @@ public class StompV12Test extends StompTestBase { Assert.assertEquals("sub2", frame.getHeader(Stomp.Headers.Message.SUBSCRIPTION)); // remove suscription - this.unsubscribe(conn, "sub1", true); - this.unsubscribe(newConn, "sub2", true); + unsubscribe(conn, "sub1", true); + unsubscribe(newConn, "sub2", true); conn.disconnect(); newConn.disconnect(); @@ -1307,7 +1307,7 @@ public class StompV12Test extends StompTestBase { StompClientConnection connV12_2 = StompClientConnectionFactory.createClientConnection(v11Uri); connV12_2.connect(defUser, defPass); - this.subscribe(connV12_2, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(connV12_2, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); ClientStompFrame frame = connV12_2.receiveFrame(2000); @@ -1336,7 +1336,7 @@ public class StompV12Test extends StompTestBase { public void testBodyWithUTF8() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, getName(), Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, getName(), Stomp.Headers.Subscribe.AckModeValues.AUTO); String text = "A" + "\u00ea" + "\u00f1" + "\u00fc" + "C"; System.out.println(text); @@ -1355,7 +1355,7 @@ public class StompV12Test extends StompTestBase { public void testClientAckNotPartOfTransaction() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, getName(), "client"); + subscribe(conn, getName(), "client"); sendJmsMessage(getName()); @@ -1379,7 +1379,7 @@ public class StompV12Test extends StompTestBase { Assert.assertNull(frame); - this.unsubscribe(conn, getName()); + unsubscribe(conn, getName()); conn.disconnect(); } @@ -1388,7 +1388,7 @@ public class StompV12Test extends StompTestBase { public void testDisconnectAndError() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, getName(), "client"); + subscribe(conn, getName(), "client"); ClientStompFrame frame = conn.createFrame("DISCONNECT"); frame.addHeader(Stomp.Headers.RECEIPT_REQUESTED, "1"); @@ -1441,9 +1441,9 @@ public class StompV12Test extends StompTestBase { public void testDurableSubscriber() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", "client", getName()); + subscribe(conn, "sub1", "client", getName()); - ClientStompFrame frame = this.subscribe(conn, "sub1", "client", getName()); + ClientStompFrame frame = subscribe(conn, "sub1", "client", getName()); Assert.assertTrue(frame.getCommand().equals(Stomp.Responses.ERROR)); @@ -1455,7 +1455,7 @@ public class StompV12Test extends StompTestBase { public void testDurableSubscriberWithReconnection() throws Exception { conn.connect(defUser, defPass, CLIENT_ID); - this.subscribeTopic(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, getName()); + subscribeTopic(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, getName()); ClientStompFrame frame = conn.createFrame("DISCONNECT"); frame.addHeader(Stomp.Headers.RECEIPT_REQUESTED, "1"); @@ -1473,7 +1473,7 @@ public class StompV12Test extends StompTestBase { conn = (StompClientConnectionV12) StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass, CLIENT_ID); - this.subscribeTopic(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, getName()); + subscribeTopic(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, getName()); // we must have received the message frame = conn.receiveFrame(); @@ -1482,7 +1482,7 @@ public class StompV12Test extends StompTestBase { Assert.assertNotNull(frame.getHeader(Stomp.Headers.Subscribe.DESTINATION)); Assert.assertEquals(getName(), frame.getBody()); - this.unsubscribe(conn, "sub1"); + unsubscribe(conn, "sub1"); conn.disconnect(); } @@ -1491,14 +1491,14 @@ public class StompV12Test extends StompTestBase { public void testDurableUnSubscribe() throws Exception { conn.connect(defUser, defPass, CLIENT_ID); - this.subscribeTopic(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO, getName()); + subscribeTopic(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO, getName()); conn.disconnect(); conn.destroy(); conn = (StompClientConnectionV12) StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass, CLIENT_ID); - this.unsubscribe(conn, getName(), null, false, true); + unsubscribe(conn, getName(), null, false, true); long start = System.currentTimeMillis(); SimpleString queueName = SimpleString.toSimpleString(CLIENT_ID + "." + getName()); @@ -1538,7 +1538,7 @@ public class StompV12Test extends StompTestBase { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); for (int i = 0; i < ctr; ++i) { data[i] = getName() + i; @@ -1569,7 +1569,7 @@ public class StompV12Test extends StompTestBase { public void testSubscribeWithAutoAckAndSelector() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, null, "foo = 'zzz'"); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO, null, "foo = 'zzz'"); sendJmsMessage("Ignored message", "foo", "1234"); sendJmsMessage("Real message", "foo", "zzz"); @@ -1585,7 +1585,7 @@ public class StompV12Test extends StompTestBase { public void testRedeliveryWithClientAck() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "subId", "client"); + subscribe(conn, "subId", "client"); sendJmsMessage(getName()); @@ -1825,7 +1825,7 @@ public class StompV12Test extends StompTestBase { public void testSubscribeToTopic() throws Exception { conn.connect(defUser, defPass); - this.subscribeTopic(conn, "sub1", null, null, true); + subscribeTopic(conn, "sub1", null, null, true); sendJmsMessage(getName(), topic); @@ -1835,7 +1835,7 @@ public class StompV12Test extends StompTestBase { Assert.assertTrue(frame.getHeader(Stomp.Headers.Subscribe.DESTINATION).equals(getTopicPrefix() + getTopicName())); Assert.assertTrue(frame.getBody().equals(getName())); - this.unsubscribe(conn, "sub1", true); + unsubscribe(conn, "sub1", true); sendJmsMessage(getName(), topic); @@ -1849,7 +1849,7 @@ public class StompV12Test extends StompTestBase { public void testSubscribeToTopicWithNoLocal() throws Exception { conn.connect(defUser, defPass); - this.subscribeTopic(conn, "sub1", null, null, true, true); + subscribeTopic(conn, "sub1", null, null, true, true); // send a message on the same connection => it should not be received send(conn, getTopicPrefix() + getTopicName(), null, "Hello World"); @@ -1867,7 +1867,7 @@ public class StompV12Test extends StompTestBase { Assert.assertTrue(frame.getHeader(Stomp.Headers.Subscribe.DESTINATION).equals(getTopicPrefix() + getTopicName())); Assert.assertTrue(frame.getBody().equals(getName())); - this.unsubscribe(conn, "sub1"); + unsubscribe(conn, "sub1"); conn.disconnect(); } @@ -1876,7 +1876,7 @@ public class StompV12Test extends StompTestBase { public void testSubscribeWithAutoAck() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); sendJmsMessage(getName()); @@ -1898,7 +1898,7 @@ public class StompV12Test extends StompTestBase { public void testSubscribeWithAutoAckAndBytesMessage() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); byte[] payload = new byte[]{1, 2, 3, 4, 5}; sendJmsMessage(payload, queue); @@ -1922,7 +1922,7 @@ public class StompV12Test extends StompTestBase { public void testSubscribeWithClientAck() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", "client"); + subscribe(conn, "sub1", "client"); sendJmsMessage(getName()); @@ -1952,7 +1952,7 @@ public class StompV12Test extends StompTestBase { public void testSubscribeWithID() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "mysubid", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "mysubid", Stomp.Headers.Subscribe.AckModeValues.AUTO); sendJmsMessage(getName()); @@ -1967,7 +1967,7 @@ public class StompV12Test extends StompTestBase { public void testSubscribeWithMessageSentWithProperties() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); MessageProducer producer = session.createProducer(queue); BytesMessage message = session.createBytesMessage(); @@ -2005,7 +2005,7 @@ public class StompV12Test extends StompTestBase { conn.connect(defUser, defPass); // first tx - this.beginTransaction(conn, "tx1"); + beginTransaction(conn, "tx1"); ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND) .addHeader(Stomp.Headers.Subscribe.DESTINATION, getQueuePrefix() + getQueueName()) @@ -2014,13 +2014,13 @@ public class StompV12Test extends StompTestBase { conn.sendFrame(frame); - this.commitTransaction(conn, "tx1"); + commitTransaction(conn, "tx1"); Message message = consumer.receive(1000); Assert.assertNotNull("Should have received a message", message); // 2nd tx with same tx ID - this.beginTransaction(conn, "tx1"); + beginTransaction(conn, "tx1"); frame = conn.createFrame(Stomp.Commands.SEND) .addHeader(Stomp.Headers.Subscribe.DESTINATION, getQueuePrefix() + getQueueName()) @@ -2029,7 +2029,7 @@ public class StompV12Test extends StompTestBase { conn.sendFrame(frame); - this.commitTransaction(conn, "tx1"); + commitTransaction(conn, "tx1"); message = consumer.receive(1000); Assert.assertNotNull("Should have received a message", message); @@ -2043,7 +2043,7 @@ public class StompV12Test extends StompTestBase { conn.connect(defUser, defPass); - this.beginTransaction(conn, "tx1"); + beginTransaction(conn, "tx1"); ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND) .addHeader(Stomp.Headers.Subscribe.DESTINATION, getQueuePrefix() + getQueueName()) @@ -2058,7 +2058,7 @@ public class StompV12Test extends StompTestBase { // check the message is not committed Assert.assertNull(consumer.receive(100)); - this.commitTransaction(conn, "tx1", true); + commitTransaction(conn, "tx1", true); Message message = consumer.receive(1000); Assert.assertNotNull("Should have received a message", message); @@ -2072,7 +2072,7 @@ public class StompV12Test extends StompTestBase { conn.connect(defUser, defPass); - this.beginTransaction(conn, "tx1"); + beginTransaction(conn, "tx1"); ClientStompFrame frame = conn.createFrame(Stomp.Commands.SEND) .addHeader(Stomp.Headers.Subscribe.DESTINATION, getQueuePrefix() + getQueueName()) @@ -2082,9 +2082,9 @@ public class StompV12Test extends StompTestBase { conn.sendFrame(frame); // rollback first message - this.abortTransaction(conn, "tx1"); + abortTransaction(conn, "tx1"); - this.beginTransaction(conn, "tx1"); + beginTransaction(conn, "tx1"); frame = conn.createFrame(Stomp.Commands.SEND) .addHeader(Stomp.Headers.Subscribe.DESTINATION, getQueuePrefix() + getQueueName()) @@ -2093,7 +2093,7 @@ public class StompV12Test extends StompTestBase { conn.sendFrame(frame); - this.commitTransaction(conn, "tx1", true); + commitTransaction(conn, "tx1", true); // only second msg should be received since first msg was rolled back TextMessage message = (TextMessage) consumer.receive(1000); @@ -2107,7 +2107,7 @@ public class StompV12Test extends StompTestBase { public void testUnsubscribe() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); // send a message to our queue sendJmsMessage("first message"); @@ -2118,7 +2118,7 @@ public class StompV12Test extends StompTestBase { Assert.assertTrue(frame.getCommand().equals(Stomp.Responses.MESSAGE)); // remove suscription - this.unsubscribe(conn, "sub1", true); + unsubscribe(conn, "sub1", true); // send a message to our queue sendJmsMessage("second message"); @@ -2133,7 +2133,7 @@ public class StompV12Test extends StompTestBase { public void testDisconnectWithoutUnsubscribe() throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); // send a message to our queue sendJmsMessage("first message"); @@ -2157,7 +2157,7 @@ public class StompV12Test extends StompTestBase { Assert.assertNull("not expected: " + frame, frame); //subscribe again. - this.subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); + subscribe(conn, "sub1", Stomp.Headers.Subscribe.AckModeValues.AUTO); // receive message from socket frame = conn.receiveFrame(); @@ -2168,7 +2168,7 @@ public class StompV12Test extends StompTestBase { frame = conn.receiveFrame(1000); Assert.assertNull("not expected: " + frame, frame); - this.unsubscribe(conn, "sub1", true); + unsubscribe(conn, "sub1", true); frame = conn.receiveFrame(1000); Assert.assertNull(frame); @@ -2179,7 +2179,7 @@ public class StompV12Test extends StompTestBase { protected void assertSubscribeWithClientAckThenConsumeWithAutoAck(boolean sendDisconnect) throws Exception { conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", "client"); + subscribe(conn, "sub1", "client"); sendJmsMessage(getName()); @@ -2200,7 +2200,7 @@ public class StompV12Test extends StompTestBase { // message should be received since message was not acknowledged conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", null); + subscribe(conn, "sub1", null); frame = conn.receiveFrame(); Assert.assertTrue(frame.getCommand().equals(Stomp.Responses.MESSAGE)); @@ -2212,7 +2212,7 @@ public class StompV12Test extends StompTestBase { conn = (StompClientConnectionV12) StompClientConnectionFactory.createClientConnection(uri); conn.connect(defUser, defPass); - this.subscribe(conn, "sub1", null, null, true); + subscribe(conn, "sub1", null, null, true); sendJmsMessage("shouldBeNextMessage");