Fixed MqttIncomingPublishFlowsWithId.findMatching

Author: SgtSilvio

src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttIncomingPublishFlows.java

@@ -19,7 +19,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.mqtt.ioc.ClientScope;
 import com.hivemq.client.internal.mqtt.message.publish.MqttStatefulPublish;
 import com.hivemq.client.internal.mqtt.message.subscribe.MqttStatefulSubscribe;
@@ -75,8 +74,9 @@ public void subAck(
 
         final ImmutableList<MqttSubscription> subscriptions = subscribe.stateless().getSubscriptions();
         final ImmutableList<Mqtt5SubAckReasonCode> reasonCodes = subAck.getReasonCodes();
+        final boolean countNotMatching = subscriptions.size() > reasonCodes.size();
         for (int i = 0; i < subscriptions.size(); i++) {
-            if (reasonCodes.get(i).isError()) {
+            if (countNotMatching || reasonCodes.get(i).isError()) {
                 remove(subscriptions.get(i).getTopicFilter(), flow);
             }
         }
@@ -89,9 +89,9 @@ void remove(final @NotNull MqttTopicFilterImpl topicFilter, final @Nullable Mqtt
     public void unsubscribe(final @NotNull MqttStatefulUnsubscribe unsubscribe, final @NotNull MqttUnsubAck unsubAck) {
         final ImmutableList<MqttTopicFilterImpl> topicFilters = unsubscribe.stateless().getTopicFilters();
         final ImmutableList<Mqtt5UnsubAckReasonCode> reasonCodes = unsubAck.getReasonCodes();
-        final boolean areAllSuccess = reasonCodes == Mqtt3UnsubAckView.REASON_CODES_ALL_SUCCESS;
+        final boolean allSuccess = reasonCodes == Mqtt3UnsubAckView.REASON_CODES_ALL_SUCCESS;
         for (int i = 0; i < topicFilters.size(); i++) {
-            if (areAllSuccess || !reasonCodes.get(i).isError()) {
+            if (allSuccess || !reasonCodes.get(i).isError()) {
                 unsubscribe(topicFilters.get(i));
             }
         }
@@ -114,8 +114,7 @@ void cancel(final @NotNull MqttSubscribedPublishFlow flow) {
     void findMatching(
             final @NotNull MqttStatefulPublish publish, final @NotNull MqttMatchingPublishFlows matchingFlows) {
 
-        final MqttTopicImpl topic = publish.stateless().getTopic();
-        subscriptionFlows.findMatching(topic, matchingFlows);
+        subscriptionFlows.findMatching(publish.stateless().getTopic(), matchingFlows);
         if (matchingFlows.subscriptionFound) {
             add(matchingFlows, globalFlows[MqttGlobalPublishFilter.SUBSCRIBED.ordinal()]);
         } else {

src/main/java/com/hivemq/client/internal/mqtt/handler/publish/incoming/MqttIncomingPublishFlowsWithId.java

@@ -29,13 +29,10 @@
 
 import javax.inject.Inject;
 import java.util.HashMap;
-import java.util.function.Consumer;
 
 import static com.hivemq.client.internal.mqtt.message.subscribe.MqttStatefulSubscribe.DEFAULT_NO_SUBSCRIPTION_IDENTIFIER;
 
 /**
- * single threaded, in channel eventloop
- *
  * @author Silvio Giebl
  */
 @ClientScope
@@ -44,7 +41,6 @@ public class MqttIncomingPublishFlowsWithId extends MqttIncomingPublishFlows {
 
     private final @NotNull HashMap<Integer, MqttSubscribedPublishFlow> flowsWithIdsMap = new HashMap<>();
     private final @NotNull MqttSubscriptionFlows flowsWithIds;
-    private final @NotNull Consumer<MqttSubscribedPublishFlow> flowWithIdUnsubscribedCallback = this::unsubscribed;
 
     @Inject
     MqttIncomingPublishFlowsWithId(
@@ -61,8 +57,8 @@ public void subscribe(
         if (flow != null) {
             final int subscriptionIdentifier = subscribe.getSubscriptionIdentifier();
             if (subscriptionIdentifier != DEFAULT_NO_SUBSCRIPTION_IDENTIFIER) {
-                flowsWithIdsMap.put(subscriptionIdentifier, flow);
                 flow.setSubscriptionIdentifier(subscriptionIdentifier);
+                flowsWithIdsMap.put(subscriptionIdentifier, flow);
             }
         }
         super.subscribe(subscribe, flow);
@@ -102,15 +98,12 @@ void remove(final @NotNull MqttTopicFilterImpl topicFilter, final @Nullable Mqtt
 
     @Override
     void unsubscribe(final @NotNull MqttTopicFilterImpl topicFilter) {
-        flowsWithIds.unsubscribe(topicFilter, flowWithIdUnsubscribedCallback);
+        flowsWithIds.unsubscribe(topicFilter, this::unsubscribed);
         super.unsubscribe(topicFilter);
     }
 
     private void unsubscribed(final @NotNull MqttSubscribedPublishFlow flow) {
-        final int subscriptionIdentifier = flow.getSubscriptionIdentifier();
-        if (subscriptionIdentifier != DEFAULT_NO_SUBSCRIPTION_IDENTIFIER) {
-            flowsWithIdsMap.remove(subscriptionIdentifier);
-        }
+        flowsWithIdsMap.remove(flow.getSubscriptionIdentifier());
     }
 
     @Override
@@ -131,13 +124,14 @@ void findMatching(
         final ImmutableIntList subscriptionIdentifiers = publish.getSubscriptionIdentifiers();
         if (!subscriptionIdentifiers.isEmpty()) {
             for (int i = 0; i < subscriptionIdentifiers.size(); i++) {
-                final int subscriptionIdentifier = subscriptionIdentifiers.get(i);
-                final MqttSubscribedPublishFlow flow = flowsWithIdsMap.get(subscriptionIdentifier);
+                final MqttSubscribedPublishFlow flow = flowsWithIdsMap.get(subscriptionIdentifiers.get(i));
                 if (flow != null) {
                     matchingFlows.add(flow);
                 }
             }
-            if (!matchingFlows.isEmpty()) {
+            if (matchingFlows.isEmpty()) {
+                flowsWithIds.findMatching(publish.stateless().getTopic(), matchingFlows);
+            } else {
                 matchingFlows.subscriptionFound = true;
             }
         }