From 45321c65bd5e95bf92b1c29c072334f11ed4ca00 Mon Sep 17 00:00:00 2001
From: Ingo Weiss <ingo@redhat.com>
Date: Tue, 6 Jun 2017 16:46:58 +0100
Subject: [PATCH 1/2] [ARTEMIS-1209] JMS OpenWire client cannot read
 notifications from activemq.notifications topic

Issue: https://issues.apache.org/jira/browse/ARTEMIS-1209
---
 .../openwire/OpenWireMessageConverter.java    |  5 ++++-
 .../openwire/SimpleOpenWireTest.java          | 22 +++++++++++++++++++
 2 files changed, 26 insertions(+), 1 deletion(-)

diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java
index dd7879c9b0..508bac95cd 100644
--- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java
+++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java
@@ -94,6 +94,8 @@ public class OpenWireMessageConverter implements MessageConverter<OpenwireMessag
    private static final String AMQ_MSG_DROPPABLE = AMQ_PREFIX + "DROPPABLE";
    private static final String AMQ_MSG_COMPRESSED = AMQ_PREFIX + "COMPRESSED";
 
+   private static final String AMQ_NOTIFICATIONS_DESTINATION = "activemq.notifications";
+
    private final WireFormat marshaller;
 
    public OpenWireMessageConverter(WireFormat marshaller) {
@@ -774,7 +776,8 @@ public class OpenWireMessageConverter implements MessageConverter<OpenwireMessag
       if (props != null) {
          for (SimpleString s : props) {
             String keyStr = s.toString();
-            if (keyStr.startsWith("_AMQ") || keyStr.startsWith("__HDR_")) {
+            if ((keyStr.startsWith("_AMQ") || keyStr.startsWith("__HDR_")) &&
+                    !(actualDestination.toString().contains(AMQ_NOTIFICATIONS_DESTINATION))) {
                continue;
             }
             Object prop = coreMessage.getObjectProperty(s);
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/SimpleOpenWireTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/SimpleOpenWireTest.java
index 6eb45a8245..cb4bd11d58 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/SimpleOpenWireTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/SimpleOpenWireTest.java
@@ -38,6 +38,7 @@ import javax.jms.TemporaryQueue;
 import javax.jms.TemporaryTopic;
 import javax.jms.TextMessage;
 import javax.jms.Topic;
+import javax.jms.TopicConnection;
 import javax.jms.TopicPublisher;
 import javax.jms.TopicSession;
 import javax.jms.TopicSubscriber;
@@ -1467,6 +1468,27 @@ public class SimpleOpenWireTest extends BasicOpenWireTest {
       }
    }
 
+   @Test
+   public void testNotificationProperties() throws Exception {
+      try (TopicConnection topicConnection = factory.createTopicConnection()) {
+         TopicSession topicSession = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
+         Topic notificationsTopic = topicSession.createTopic("activemq.notifications");
+         TopicSubscriber subscriber = topicSession.createSubscriber(notificationsTopic);
+         List<Message> receivedMessages = new ArrayList<>();
+         subscriber.setMessageListener(receivedMessages::add);
+         topicConnection.start();
+
+         while (receivedMessages.size() == 0) {
+            Thread.sleep(1000);
+         }
+
+         for (Message message : receivedMessages) {
+            assertNotNull(message);
+            assertNotNull(message.getStringProperty("_AMQ_NotifType"));
+         }
+      }
+   }
+
    private void checkQueueEmpty(String qName) {
       PostOffice po = server.getPostOffice();
       LocalQueueBinding binding = (LocalQueueBinding) po.getBinding(SimpleString.toSimpleString(qName));

From 4e173cfa8e61406650de08caba6c16f7ec4339ca Mon Sep 17 00:00:00 2001
From: Clebert Suconic <clebertsuconic@apache.org>
Date: Wed, 7 Jun 2017 16:26:25 -0400
Subject: [PATCH 2/2] [ARTEMIS-1209] Just enhancing the OpenWire test with a
 Wait.waitFor()

---
 .../tests/integration/openwire/SimpleOpenWireTest.java     | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/SimpleOpenWireTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/SimpleOpenWireTest.java
index cb4bd11d58..a41acc63f8 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/SimpleOpenWireTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/SimpleOpenWireTest.java
@@ -63,6 +63,7 @@ import org.apache.activemq.artemis.core.postoffice.PostOffice;
 import org.apache.activemq.artemis.core.postoffice.impl.LocalQueueBinding;
 import org.apache.activemq.artemis.api.core.RoutingType;
 import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.util.Wait;
 import org.apache.activemq.command.ActiveMQQueue;
 import org.apache.activemq.command.ActiveMQTopic;
 import org.junit.Assert;
@@ -1478,9 +1479,9 @@ public class SimpleOpenWireTest extends BasicOpenWireTest {
          subscriber.setMessageListener(receivedMessages::add);
          topicConnection.start();
 
-         while (receivedMessages.size() == 0) {
-            Thread.sleep(1000);
-         }
+         Wait.waitFor(() -> receivedMessages.size() > 0);
+
+         Assert.assertTrue(receivedMessages.size() > 0);
 
          for (Message message : receivedMessages) {
             assertNotNull(message);