Avoid stack overflow in SubscriptionFlowTree.findMatching

SgtSilvio · SgtSilvio · commit a3054bbe1008 · 2019-11-13T21:23:53.000+01:00
diff --git a/src/main/java/com/hivemq/client/internal/mqtt/datatypes/MqttTopicLevel.java b/src/main/java/com/hivemq/client/internal/mqtt/datatypes/MqttTopicLevel.java
@@ -35,15 +35,12 @@ public class MqttTopicLevel extends ByteArray.Range {
             new ByteArray(new byte[]{MqttTopicFilter.SINGLE_LEVEL_WILDCARD});
 
     public static @NotNull MqttTopicLevel root(final @NotNull MqttTopicImpl topic) {
-        final byte[] binary = topic.toBinary();
-        final int end = nextEnd(binary, 0);
-        return new MqttTopicLevel(binary, 0, end);
+        return new MqttTopicLevel(topic.toBinary(), -1, -1);
     }
 
     public static @NotNull MqttTopicLevel root(final @NotNull MqttTopicFilterImpl topicFilter) {
-        final byte[] binary = topicFilter.toBinary();
         final int start = topicFilter.getFilterByteStart() - 1;
-        return new MqttTopicLevel(binary, start, start);
+        return new MqttTopicLevel(topicFilter.toBinary(), start, start);
     }
 
     private static int nextEnd(final @NotNull byte[] array, final int start) {
diff --git a/src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttIncomingPublishFlows.java b/src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttIncomingPublishFlows.java
@@ -106,17 +106,17 @@ void cancel(final @NotNull MqttSubscribedPublishFlow flow) {
     }
 
     @NotNull HandleList<MqttIncomingPublishFlow> findMatching(final @NotNull MqttStatefulPublish publish) {
-        final HandleList<MqttIncomingPublishFlow> matchingFlows = new HandleList<>();
+        final MqttMatchingPublishFlows matchingFlows = new MqttMatchingPublishFlows();
         findMatching(publish, matchingFlows);
         return matchingFlows;
     }
 
     void findMatching(
-            final @NotNull MqttStatefulPublish publish,
-            final @NotNull HandleList<MqttIncomingPublishFlow> matchingFlows) {
+            final @NotNull MqttStatefulPublish publish, final @NotNull MqttMatchingPublishFlows matchingFlows) {
 
         final MqttTopicImpl topic = publish.stateless().getTopic();
-        if (subscriptionFlows.findMatching(topic, matchingFlows) || !matchingFlows.isEmpty()) {
+        subscriptionFlows.findMatching(topic, matchingFlows);
+        if (matchingFlows.subscriptionFound) {
             add(matchingFlows, globalFlows[MqttGlobalPublishFilter.SUBSCRIBED.ordinal()]);
         } else {
             add(matchingFlows, globalFlows[MqttGlobalPublishFilter.UNSOLICITED.ordinal()]);
diff --git a/src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttIncomingPublishFlowsWithId.java b/src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttIncomingPublishFlowsWithId.java
@@ -23,7 +23,6 @@
 import com.hivemq.client.internal.mqtt.message.publish.MqttStatefulPublish;
 import com.hivemq.client.internal.mqtt.message.subscribe.MqttStatefulSubscribe;
 import com.hivemq.client.internal.mqtt.message.subscribe.suback.MqttSubAck;
-import com.hivemq.client.internal.util.collections.HandleList;
 import com.hivemq.client.internal.util.collections.ImmutableIntList;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -127,8 +126,7 @@ void cancel(final @NotNull MqttSubscribedPublishFlow flow) {
 
     @Override
     void findMatching(
-            final @NotNull MqttStatefulPublish publish,
-            final @NotNull HandleList<MqttIncomingPublishFlow> matchingFlows) {
+            final @NotNull MqttStatefulPublish publish, final @NotNull MqttMatchingPublishFlows matchingFlows) {
 
         final ImmutableIntList subscriptionIdentifiers = publish.getSubscriptionIdentifiers();
         if (!subscriptionIdentifiers.isEmpty()) {
@@ -139,6 +137,9 @@ void findMatching(
                     matchingFlows.add(flow);
                 }
             }
+            if (!matchingFlows.isEmpty()) {
+                matchingFlows.subscriptionFound = true;
+            }
         }
         super.findMatching(publish, matchingFlows);
     }
diff --git a/src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttMatchingPublishFlows.java b/src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttMatchingPublishFlows.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright 2018 dc-square and the HiveMQ MQTT Client Project
+ *
+ * Licensed 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 com.hivemq.client.internal.mqtt.handler.publish.incoming;
+
+import com.hivemq.client.internal.util.collections.HandleList;
+
+/**
+ * @author Silvio Giebl
+ */
+class MqttMatchingPublishFlows extends HandleList<MqttIncomingPublishFlow> {
+
+    boolean subscriptionFound;
+}
diff --git a/src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttSubscriptionFlowList.java b/src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttSubscriptionFlowList.java
@@ -121,8 +121,8 @@ public void cancel(final @NotNull MqttSubscribedPublishFlow flow) {
     }
 
     @Override
-    public boolean findMatching(
-            final @NotNull MqttTopicImpl topic, final @NotNull HandleList<MqttIncomingPublishFlow> matchingFlows) {
+    public void findMatching(
+            final @NotNull MqttTopicImpl topic, final @NotNull MqttMatchingPublishFlows matchingFlows) {
 
         for (final MqttSubscribedPublishFlow flow : flows) {
             for (final MqttTopicFilterImpl topicFilter : flow.getTopicFilters()) {
@@ -133,14 +133,15 @@ public boolean findMatching(
             }
         }
         if (!matchingFlows.isEmpty()) {
-            return true;
+            matchingFlows.subscriptionFound = true;
+            return;
         }
         for (final MqttTopicFilterImpl subscribedTopicFilter : subscribedTopicFilters.keySet()) {
             if (subscribedTopicFilter.matches(topic)) {
-                return true;
+                matchingFlows.subscriptionFound = true;
+                return;
             }
         }
-        return false;
     }
 
     @Override
diff --git a/src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttSubscriptionFlowTree.java b/src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttSubscriptionFlowTree.java
@@ -108,10 +108,14 @@ public void cancel(final @NotNull MqttSubscribedPublishFlow flow) {
     }
 
     @Override
-    public boolean findMatching(
-            final @NotNull MqttTopicImpl topic, final @NotNull HandleList<MqttIncomingPublishFlow> matchingFlows) {
+    public void findMatching(
+            final @NotNull MqttTopicImpl topic, final @NotNull MqttMatchingPublishFlows matchingFlows) {
 
-        return (rootNode != null) && rootNode.findMatching(MqttTopicLevel.root(topic), matchingFlows);
+        final MqttTopicLevel level = MqttTopicLevel.root(topic);
+        TopicTreeNode node = rootNode;
+        while (node != null) {
+            node = node.findMatching(level.next(), matchingFlows);
+        }
     }
 
     @Override
@@ -300,27 +304,35 @@ private static boolean cancel(
             return false;
         }
 
-        boolean findMatching(
-                final @Nullable MqttTopicLevel level,
-                final @NotNull HandleList<MqttIncomingPublishFlow> matchingFlows) {
+        @Nullable TopicTreeNode findMatching(
+                final @Nullable MqttTopicLevel level, final @NotNull MqttMatchingPublishFlows matchingFlows) {
 
             if (level == null) {
                 add(matchingFlows, entries);
                 add(matchingFlows, multiLevelEntries);
-                return (subscriptions != 0) || (multiLevelSubscriptions != 0);
+                if ((subscriptions != 0) || (multiLevelSubscriptions != 0)) {
+                    matchingFlows.subscriptionFound = true;
+                }
+                return null;
             }
             add(matchingFlows, multiLevelEntries);
-            boolean subscriptionFound = (multiLevelSubscriptions != 0);
+            if (multiLevelSubscriptions != 0) {
+                matchingFlows.subscriptionFound = true;
+            }
             if (next != null) {
-                if (singleLevel != null) {
-                    subscriptionFound |= singleLevel.findMatching(level.fork().next(), matchingFlows);
-                }
-                final TopicTreeNode node = next.get(level);
-                if (node != null) {
-                    subscriptionFound |= node.findMatching(level.next(), matchingFlows);
+                final TopicTreeNode nextNode = next.get(level);
+                if (nextNode != null) {
+                    if (singleLevel != null) {
+                        TopicTreeNode node = singleLevel;
+                        final MqttTopicLevel levelFork = level.fork();
+                        while (node != null) {
+                            node = node.findMatching(levelFork.next(), matchingFlows);
+                        }
+                    }
+                    return nextNode;
                 }
             }
-            return subscriptionFound;
+            return singleLevel;
         }
 
         private static void add(
@@ -369,12 +381,12 @@ private static void add(
             return null;
         }
 
-        private void removeNext(final @NotNull TopicTreeNode node) {
-            assert next != null;
+        void removeNext(final @NotNull TopicTreeNode node) {
             assert node.parentLevel != null;
             if (node.parentLevel == MqttTopicLevel.SINGLE_LEVEL_WILDCARD) {
                 singleLevel = null;
             } else {
+                assert next != null;
                 next.remove(node.parentLevel);
                 if (next.isEmpty()) {
                     next = null;
diff --git a/src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttSubscriptionFlows.java b/src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttSubscriptionFlows.java
@@ -20,7 +20,6 @@
 import com.hivemq.client.internal.annotations.NotThreadSafe;
 import com.hivemq.client.internal.mqtt.datatypes.MqttTopicFilterImpl;
 import com.hivemq.client.internal.mqtt.datatypes.MqttTopicImpl;
-import com.hivemq.client.internal.util.collections.HandleList;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -42,7 +41,7 @@ void unsubscribe(
 
     void cancel(@NotNull MqttSubscribedPublishFlow flow);
 
-    boolean findMatching(@NotNull MqttTopicImpl topic, @NotNull HandleList<MqttIncomingPublishFlow> matchingFlows);
+    void findMatching(@NotNull MqttTopicImpl topic, @NotNull MqttMatchingPublishFlows matchingFlows);
 
     void clear(@NotNull Throwable cause);
 }
diff --git a/src/test/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttSubscriptionFlowsTest.java b/src/test/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttSubscriptionFlowsTest.java

Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,6 @@`
`23`	`23`	`import com.hivemq.client.internal.mqtt.message.publish.MqttStatefulPublish;`
`24`	`24`	`import com.hivemq.client.internal.mqtt.message.subscribe.MqttStatefulSubscribe;`
`25`	`25`	`import com.hivemq.client.internal.mqtt.message.subscribe.suback.MqttSubAck;`
`26`		`-import com.hivemq.client.internal.util.collections.HandleList;`
`27`	`26`	`import com.hivemq.client.internal.util.collections.ImmutableIntList;`
`28`	`27`	`import org.jetbrains.annotations.NotNull;`
`29`	`28`	`import org.jetbrains.annotations.Nullable;`
`@@ -127,8 +126,7 @@ void cancel(final @NotNull MqttSubscribedPublishFlow flow) {`
`127`	`126`
`128`	`127`	`@Override`
`129`	`128`	`void findMatching(`
`130`		`- final @NotNull MqttStatefulPublish publish,`
`131`		`- final @NotNull HandleList<MqttIncomingPublishFlow> matchingFlows) {`
	`129`	`+ final @NotNull MqttStatefulPublish publish, final @NotNull MqttMatchingPublishFlows matchingFlows) {`
`132`	`130`
`133`	`131`	`final ImmutableIntList subscriptionIdentifiers = publish.getSubscriptionIdentifiers();`
`134`	`132`	`if (!subscriptionIdentifiers.isEmpty()) {`
`@@ -139,6 +137,9 @@ void findMatching(`
`139`	`137`	`matchingFlows.add(flow);`
`140`	`138`	`}`
`141`	`139`	`}`
	`140`	`+ if (!matchingFlows.isEmpty()) {`
	`141`	`+ matchingFlows.subscriptionFound = true;`
	`142`	`+ }`
`142`	`143`	`}`
`143`	`144`	`super.findMatching(publish, matchingFlows);`
`144`	`145`	`}`
Original file line number	Diff line number	Diff line change
`@@ -121,8 +121,8 @@ public void cancel(final @NotNull MqttSubscribedPublishFlow flow) {`
`121`	`121`	`}`
`122`	`122`
`123`	`123`	`@Override`
`124`		`- public boolean findMatching(`
`125`		`- final @NotNull MqttTopicImpl topic, final @NotNull HandleList<MqttIncomingPublishFlow> matchingFlows) {`
	`124`	`+ public void findMatching(`
	`125`	`+ final @NotNull MqttTopicImpl topic, final @NotNull MqttMatchingPublishFlows matchingFlows) {`
`126`	`126`
`127`	`127`	`for (final MqttSubscribedPublishFlow flow : flows) {`
`128`	`128`	`for (final MqttTopicFilterImpl topicFilter : flow.getTopicFilters()) {`
`@@ -133,14 +133,15 @@ public boolean findMatching(`
`133`	`133`	`}`
`134`	`134`	`}`
`135`	`135`	`if (!matchingFlows.isEmpty()) {`
`136`		`- return true;`
	`136`	`+ matchingFlows.subscriptionFound = true;`
	`137`	`+ return;`
`137`	`138`	`}`
`138`	`139`	`for (final MqttTopicFilterImpl subscribedTopicFilter : subscribedTopicFilters.keySet()) {`
`139`	`140`	`if (subscribedTopicFilter.matches(topic)) {`
`140`		`- return true;`
	`141`	`+ matchingFlows.subscriptionFound = true;`
	`142`	`+ return;`
`141`	`143`	`}`
`142`	`144`	`}`
`143`		`- return false;`
`144`	`145`	`}`
`145`	`146`
`146`	`147`	`@Override`