From 5de0c8e2fbdc992e0873d92fceac3ffcd5a5e468 Mon Sep 17 00:00:00 2001 From: Timothy Bish Date: Thu, 31 Jul 2014 17:23:35 -0400 Subject: [PATCH] https://issues.apache.org/jira/browse/AMQ-5298 Fix the NPE case --- .../activemq/transport/mqtt/MQTTProtocolConverter.java | 8 ++++++-- .../java/org/apache/activemq/transport/mqtt/MQTTTest.java | 7 +++---- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTProtocolConverter.java b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTProtocolConverter.java index eb5bb2b7cf..4ca23c370f 100644 --- a/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTProtocolConverter.java +++ b/activemq-mqtt/src/main/java/org/apache/activemq/transport/mqtt/MQTTProtocolConverter.java @@ -40,6 +40,7 @@ import org.apache.activemq.broker.region.RegionBroker; import org.apache.activemq.broker.region.Subscription; import org.apache.activemq.broker.region.TopicRegion; import org.apache.activemq.broker.region.policy.RetainedMessageSubscriptionRecoveryPolicy; +import org.apache.activemq.broker.region.virtual.VirtualTopicInterceptor; import org.apache.activemq.command.ActiveMQBytesMessage; import org.apache.activemq.command.ActiveMQDestination; import org.apache.activemq.command.ActiveMQMapMessage; @@ -500,7 +501,11 @@ public class MQTTProtocolConverter { for (Subscription subscription : dest.getConsumers()) { if (subscription.getConsumerInfo().getConsumerId().equals(consumerId)) { try { - ((org.apache.activemq.broker.region.Topic)dest).recoverRetroactiveMessages(connectionContext, subscription); + if (dest instanceof org.apache.activemq.broker.region.Topic) { + ((org.apache.activemq.broker.region.Topic)dest).recoverRetroactiveMessages(connectionContext, subscription); + } else if (dest instanceof VirtualTopicInterceptor) { + ((VirtualTopicInterceptor)dest).getTopic().recoverRetroactiveMessages(connectionContext, subscription); + } if (subscription instanceof PrefetchSubscription) { // request dispatch for prefetch subs PrefetchSubscription prefetchSubscription = (PrefetchSubscription) subscription; @@ -917,7 +922,6 @@ public class MQTTProtocolConverter { case '>': chars[i] = '#'; break; - case '+': chars[i] = '*'; break; diff --git a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java index e332c48e23..c4571dc500 100644 --- a/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java +++ b/activemq-mqtt/src/test/java/org/apache/activemq/transport/mqtt/MQTTTest.java @@ -518,14 +518,14 @@ public class MQTTTest extends MQTTTestSupport { assertEquals(RETAIN, new String(msg.getPayload())); msg.ack(); assertNull(connection.receive(500, TimeUnit.MILLISECONDS)); - connection.unsubscribe(new String[]{"TopicA"}); + connection.unsubscribe(new String[]{TOPICA}); // clear retained message and check that we don't receive it connection.publish(TOPICA, "".getBytes(), QoS.AT_MOST_ONCE, true); connection.subscribe(new Topic[]{new Topic(TOPICA, QoS.AT_LEAST_ONCE)}); msg = connection.receive(500, TimeUnit.MILLISECONDS); assertNull("Retained message not cleared for " + clientId, msg); - connection.unsubscribe(new String[]{"TopicA"}); + connection.unsubscribe(new String[]{TOPICA}); // set retained message again and check connection.publish(TOPICA, RETAIN.getBytes(), QoS.EXACTLY_ONCE, true); @@ -547,8 +547,7 @@ public class MQTTTest extends MQTTTestSupport { msg.ack(); assertNull(connection.receive(500, TimeUnit.MILLISECONDS)); - connection.unsubscribe(new String[]{"TopicA"}); - + connection.unsubscribe(new String[]{TOPICA}); connection.disconnect(); } }