You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2017/06/22 15:51:25 UTC
[1/3] activemq-artemis git commit: This closes #1357
Repository: activemq-artemis
Updated Branches:
refs/heads/master 902e0c49a -> c2ad9cab0
This closes #1357
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/c2ad9cab
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/c2ad9cab
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/c2ad9cab
Branch: refs/heads/master
Commit: c2ad9cab0d0a4d3187789c36224480366554feeb
Parents: 902e0c4 2b80426
Author: Clebert Suconic <cl...@apache.org>
Authored: Thu Jun 22 11:51:11 2017 -0400
Committer: Clebert Suconic <cl...@apache.org>
Committed: Thu Jun 22 11:51:11 2017 -0400
----------------------------------------------------------------------
.../amqp/AmqpSendReceiveInterceptorTest.java | 102 ++++++++++++++
.../imported/MQTTInterceptorPropertiesTest.java | 137 +++++++++++++++++++
.../stomp/StompTestPropertiesInterceptor.java | 98 +++++++++++++
3 files changed, 337 insertions(+)
----------------------------------------------------------------------
[3/3] activemq-artemis git commit: ARTEMIS-607 New tests AMQP, MQTT,
Stomp interceptor properties tests
Posted by cl...@apache.org.
ARTEMIS-607 New tests AMQP,MQTT,Stomp interceptor properties tests
Added tests for AMQP, MQTT and Stomp protocol for interceptors.
Tests are checking intercepted message properties.
Linked issue ARTEMIS-607
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/e801df79
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/e801df79
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/e801df79
Branch: refs/heads/master
Commit: e801df79747bfe8c79a8e6642712822603a76190
Parents: 902e0c4
Author: Michal Tóth <mt...@redhat.com>
Authored: Wed Jun 21 11:37:45 2017 +0200
Committer: Clebert Suconic <cl...@apache.org>
Committed: Thu Jun 22 11:51:11 2017 -0400
----------------------------------------------------------------------
.../amqp/AmqpSendReceiveInterceptorTest.java | 102 ++++++++++++++
.../imported/MQTTInterceptorPropertiesTest.java | 136 +++++++++++++++++++
.../stomp/StompTestPropertiesInterceptor.java | 98 +++++++++++++
3 files changed, 336 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e801df79/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpSendReceiveInterceptorTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpSendReceiveInterceptorTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpSendReceiveInterceptorTest.java
index beb6a88..20be85d 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpSendReceiveInterceptorTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpSendReceiveInterceptorTest.java
@@ -26,8 +26,12 @@ import org.apache.activemq.transport.amqp.client.AmqpMessage;
import org.apache.activemq.transport.amqp.client.AmqpReceiver;
import org.apache.activemq.transport.amqp.client.AmqpSender;
import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.apache.qpid.proton.amqp.messaging.Header;
+import org.apache.qpid.proton.amqp.messaging.Properties;
import org.junit.Test;
+import java.util.HashMap;
+import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -71,6 +75,104 @@ public class AmqpSendReceiveInterceptorTest extends AmqpClientTestSupport {
AmqpMessage amqpMessage = receiver.receive(5, TimeUnit.SECONDS);
assertNotNull(amqpMessage);
assertEquals(latch2.getCount(), 0);
+ sender.close();
+ receiver.close();
+ connection.close();
+ }
+
+ private static final String ADDRESS = "address";
+ private static final String MESSAGE_ID = "messageId";
+ private static final String CORRELATION_ID = "correlationId";
+ private static final String MESSAGE_TEXT = "messageText";
+ private static final String DURABLE = "durable";
+ private static final String PRIORITY = "priority";
+ private static final String REPLY_TO = "replyTo";
+ private static final String TIME_TO_LIVE = "timeToLive";
+
+
+ private boolean checkMessageProperties(AMQPMessage message, Map<String, Object> expectedProperties) {
+ assertNotNull(message);
+ assertNotNull(server.getNodeID());
+
+ assertNotNull(message.getConnectionID());
+ assertEquals(message.getAddress(), expectedProperties.get(ADDRESS));
+ assertEquals(message.isDurable(), expectedProperties.get(DURABLE));
+
+ Properties props = message.getProperties();
+ assertEquals(props.getCorrelationId(), expectedProperties.get(CORRELATION_ID));
+ assertEquals(props.getReplyTo(), expectedProperties.get(REPLY_TO));
+ assertEquals(props.getMessageId(), expectedProperties.get(MESSAGE_ID));
+
+ Header header = message.getHeader();
+ assertEquals(header.getDurable(), expectedProperties.get(DURABLE));
+ assertEquals(header.getTtl().toString(), expectedProperties.get(TIME_TO_LIVE).toString());
+ assertEquals(header.getPriority().toString(), expectedProperties.get(PRIORITY).toString());
+ return true;
+ }
+
+ @Test(timeout = 60000)
+ public void testCheckInterceptedMessageProperties() throws Exception {
+ final CountDownLatch latch = new CountDownLatch(1);
+
+ final String addressQueue = getTestName();
+ final String messageId = "lala200";
+ final String correlationId = "lala-corrId";
+ final String msgText = "Test intercepted message";
+ final boolean durableMsg = false;
+ final short priority = 8;
+ final long timeToLive = 10000;
+ final String replyTo = "reply-to-myQueue";
+
+ Map<String, Object> expectedProperties = new HashMap<>();
+ expectedProperties.put(ADDRESS, addressQueue);
+ expectedProperties.put(MESSAGE_ID, messageId);
+ expectedProperties.put(CORRELATION_ID, correlationId);
+ expectedProperties.put(MESSAGE_TEXT, msgText);
+ expectedProperties.put(DURABLE, durableMsg);
+ expectedProperties.put(PRIORITY, priority);
+ expectedProperties.put(REPLY_TO, replyTo);
+ expectedProperties.put(TIME_TO_LIVE, timeToLive);
+
+ server.getRemotingService().addIncomingInterceptor(new AmqpInterceptor() {
+ @Override
+ public boolean intercept(AMQPMessage message, RemotingConnection connection) throws ActiveMQException {
+ latch.countDown();
+ return checkMessageProperties(message, expectedProperties);
+ }
+ });
+
+ AmqpClient client = createAmqpClient();
+ AmqpConnection connection = addConnection(client.connect());
+ AmqpSession session = connection.createSession();
+
+ AmqpSender sender = session.createSender(getTestName());
+ AmqpMessage message = new AmqpMessage();
+
+ message.setMessageId(messageId);
+ message.setCorrelationId(correlationId);
+ message.setText(msgText);
+ message.setDurable(durableMsg);
+ message.setPriority(priority);
+ message.setReplyToAddress(replyTo);
+ message.setTimeToLive(timeToLive);
+
+ sender.send(message);
+
+ assertTrue(latch.await(2, TimeUnit.SECONDS));
+ final CountDownLatch latch2 = new CountDownLatch(1);
+ server.getRemotingService().addOutgoingInterceptor(new AmqpInterceptor() {
+ @Override
+ public boolean intercept(AMQPMessage packet, RemotingConnection connection) throws ActiveMQException {
+ latch2.countDown();
+ return checkMessageProperties(packet, expectedProperties);
+ }
+ });
+ AmqpReceiver receiver = session.createReceiver(getTestName());
+ receiver.flow(2);
+ AmqpMessage amqpMessage = receiver.receive(5, TimeUnit.SECONDS);
+ assertNotNull(amqpMessage);
+ assertEquals(latch2.getCount(), 0);
+ sender.close();
receiver.close();
connection.close();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e801df79/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTInterceptorPropertiesTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTInterceptorPropertiesTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTInterceptorPropertiesTest.java
new file mode 100644
index 0000000..375e2f2
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTInterceptorPropertiesTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.artemis.tests.integration.mqtt.imported;
+
+import io.netty.handler.codec.mqtt.MqttFixedHeader;
+import io.netty.handler.codec.mqtt.MqttMessage;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.core.protocol.mqtt.MQTTConnectionManager;
+import org.apache.activemq.artemis.core.protocol.mqtt.MQTTInterceptor;
+import org.apache.activemq.artemis.core.protocol.mqtt.MQTTSession;
+import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
+import org.apache.activemq.artemis.utils.collections.ConcurrentHashSet;
+import org.apache.felix.resolver.util.ArrayMap;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ErrorCollector;
+
+import java.lang.reflect.Field;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class MQTTInterceptorPropertiesTest extends MQTTTestSupport {
+
+ @Override
+ @Before
+ public void setUp() throws Exception {
+ Field sessions = MQTTSession.class.getDeclaredField("SESSIONS");
+ sessions.setAccessible(true);
+ sessions.set(null, new ConcurrentHashMap<>());
+
+ Field connectedClients = MQTTConnectionManager.class.getDeclaredField("CONNECTED_CLIENTS");
+ connectedClients.setAccessible(true);
+ connectedClients.set(null, new ConcurrentHashSet<>());
+ super.setUp();
+ }
+
+ private static final String ADDRESS = "address";
+ private static final String MESSAGE_TEXT = "messageText";
+ private static final String RETAINED = "retained";
+
+
+ private boolean checkMessageProperties(MqttMessage message, Map<String, Object> expectedProperties) {
+ System.out.println("Checking properties in interceptor");
+ try {
+ assertNotNull(message);
+ assertNotNull(server.getNodeID());
+ MqttFixedHeader header = message.fixedHeader();
+ assertNotNull(header.messageType());
+ assertEquals(header.qosLevel().value(), AT_MOST_ONCE);
+ assertEquals(header.isRetain(), expectedProperties.get(RETAINED));
+ } catch (Throwable t) {
+ collector.addError(t);
+ }
+ return true;
+ }
+
+ @Rule
+ public ErrorCollector collector = new ErrorCollector();
+
+ @Test(timeout = 60000)
+ public void testCheckInterceptedMQTTMessageProperties() throws Exception {
+ final String addressQueue = name.getMethodName();
+ final String msgText = "Test intercepted message";
+ final boolean retained = true;
+
+ Map<String, Object> expectedProperties = new ArrayMap<>();
+ expectedProperties.put(ADDRESS, addressQueue);
+ expectedProperties.put(MESSAGE_TEXT, msgText);
+ expectedProperties.put(RETAINED, retained);
+
+
+ final MQTTClientProvider subscribeProvider = getMQTTClientProvider();
+ initializeConnection(subscribeProvider);
+
+ subscribeProvider.subscribe(addressQueue, AT_MOST_ONCE);
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ MQTTInterceptor incomingInterceptor = new MQTTInterceptor() {
+ @Override
+ public boolean intercept(MqttMessage packet, RemotingConnection connection) throws ActiveMQException {
+ System.out.println("incoming");
+ return checkMessageProperties(packet, expectedProperties);
+ }
+ };
+
+ MQTTInterceptor outgoingInterceptor = new MQTTInterceptor() {
+ @Override
+ public boolean intercept(MqttMessage packet, RemotingConnection connection) throws ActiveMQException {
+ System.out.println("outgoing");
+ return checkMessageProperties(packet, expectedProperties);
+ }
+ };
+ server.getRemotingService().addIncomingInterceptor(incomingInterceptor);
+ server.getRemotingService().addOutgoingInterceptor(outgoingInterceptor);
+
+
+ Thread thread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ byte[] payload = subscribeProvider.receive(10000);
+ assertNotNull("Should get a message", payload);
+ latch.countDown();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ });
+ thread.start();
+
+ final MQTTClientProvider publishProvider = getMQTTClientProvider();
+ initializeConnection(publishProvider);
+ publishProvider.publish(addressQueue, msgText.getBytes(), AT_MOST_ONCE, retained);
+
+ latch.await(10, TimeUnit.SECONDS);
+ subscribeProvider.disconnect();
+ publishProvider.disconnect();
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/e801df79/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTestPropertiesInterceptor.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTestPropertiesInterceptor.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTestPropertiesInterceptor.java
new file mode 100644
index 0000000..d380911
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTestPropertiesInterceptor.java
@@ -0,0 +1,98 @@
+/*
+ * 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.artemis.tests.integration.stomp;
+
+import org.apache.activemq.artemis.core.protocol.stomp.StompFrame;
+import org.apache.activemq.artemis.core.protocol.stomp.StompFrameInterceptor;
+import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
+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.felix.resolver.util.ArrayMap;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class StompTestPropertiesInterceptor extends StompTestBase {
+
+ @Override
+ public List<String> getIncomingInterceptors() {
+ List<String> stompIncomingInterceptor = new ArrayList<>();
+ stompIncomingInterceptor.add("org.apache.activemq.artemis.tests.integration.stomp.StompTestPropertiesInterceptor$StompFramePropertiesInterceptor");
+ return stompIncomingInterceptor;
+ }
+
+ @Override
+ public List<String> getOutgoingInterceptors() {
+ List<String> stompOutgoingInterceptor = new ArrayList<>();
+ stompOutgoingInterceptor.add("org.apache.activemq.artemis.tests.integration.stomp.StompTestPropertiesInterceptor$StompFramePropertiesInterceptor");
+
+ return stompOutgoingInterceptor;
+ }
+
+ public static class StompFramePropertiesInterceptor implements StompFrameInterceptor {
+
+ @Override
+ public boolean intercept(StompFrame stompFrame, RemotingConnection connection) {
+ if (stompFrame.getCommand().equals("CONNECT") || stompFrame.getCommand().equals("CONNECTED")) {
+ return true;
+ }
+ System.out.println("Checking properties in interceptor");
+ assertNotNull(stompFrame);
+ assertEquals(stompFrame.getHeader(MY_HEADER), expectedProperties.get(MY_HEADER));
+ assertEquals(stompFrame.getBody(), expectedProperties.get(MESSAGE_TEXT));
+ return true;
+ }
+ }
+
+
+ private static final String MESSAGE_TEXT = "messageText";
+ private static final String MY_HEADER = "my-header";
+ private static Map<String, Object> expectedProperties = new ArrayMap<>();
+
+ @Test(timeout = 60000)
+ public void testCheckInterceptedStompMessageProperties() throws Exception {
+ final String msgText = "Test intercepted message";
+ final String myHeader = "TestInterceptedHeader";
+ expectedProperties.put(MESSAGE_TEXT, msgText);
+ expectedProperties.put(MY_HEADER, myHeader);
+
+ StompClientConnection conn = StompClientConnectionFactory.createClientConnection("1.0", hostname, port);
+ conn.connect(defUser, defPass);
+
+ ClientStompFrame subFrame = conn.createFrame("SUBSCRIBE");
+
+ subFrame.addHeader("subscription-type", "ANYCAST");
+ subFrame.addHeader("destination", name.getMethodName());
+ subFrame.addHeader("ack", "auto");
+ subFrame.addHeader(MY_HEADER, myHeader);
+ subFrame.setBody(msgText);
+
+ conn.sendFrame(subFrame);
+
+ ClientStompFrame frame = conn.createFrame("SEND");
+ frame.addHeader("destination", name.getMethodName());
+ frame.addHeader("ack", "auto");
+ frame.addHeader(MY_HEADER, myHeader);
+ conn.sendFrame(frame);
+
+ conn.disconnect();
+
+ }
+}
[2/3] activemq-artemis git commit: ARTEMIS-1244 Retain flag code todo
comment
Posted by cl...@apache.org.
ARTEMIS-1244 Retain flag code todo comment
Identified possible bug in MQTT receive of missing retained flag.
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/2b804261
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/2b804261
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/2b804261
Branch: refs/heads/master
Commit: 2b804261532ae10db073cdccbfdf04ace999dc95
Parents: e801df7
Author: Michal Tóth <mt...@redhat.com>
Authored: Wed Jun 21 13:17:54 2017 +0200
Committer: Clebert Suconic <cl...@apache.org>
Committed: Thu Jun 22 11:51:11 2017 -0400
----------------------------------------------------------------------
.../integration/mqtt/imported/MQTTInterceptorPropertiesTest.java | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/2b804261/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTInterceptorPropertiesTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTInterceptorPropertiesTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTInterceptorPropertiesTest.java
index 375e2f2..0b62194 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTInterceptorPropertiesTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/mqtt/imported/MQTTInterceptorPropertiesTest.java
@@ -64,6 +64,7 @@ public class MQTTInterceptorPropertiesTest extends MQTTTestSupport {
MqttFixedHeader header = message.fixedHeader();
assertNotNull(header.messageType());
assertEquals(header.qosLevel().value(), AT_MOST_ONCE);
+ // TODO resolve the following line based on result of ARTEMIS-1244, currently fails (2.1.0)
assertEquals(header.isRetain(), expectedProperties.get(RETAINED));
} catch (Throwable t) {
collector.addError(t);