process subscriptions improvements - when collecting persistent subscriptions and delivering msg for non-persistent

Author: dmytro-landiak

application/src/main/java/org/thingsboard/mqtt/broker/service/processing/MsgDispatcherServiceImpl.java

@@ -150,38 +150,26 @@ private void processPersistentSubscriptions(PublishMsgWithId publishMsgWithId, P
 
     PersistentMsgSubscriptions processBasicAndCollectPersistentSubscriptions(MsgSubscriptions msgSubscriptions,
                                                                              PublishMsgProto publishMsgProto) {
-        List<Subscription> applicationSubscriptions = null;
-        List<Subscription> deviceSubscriptions = null;
+        final PersistentMsgSubscriptions persistentSubscriptions = new PersistentMsgSubscriptions(
+                null, null, msgSubscriptions.getAllApplicationSharedSubscriptions()
+        );
         long startTime = System.nanoTime();
 
         if (!CollectionUtils.isEmpty(msgSubscriptions.getCommonSubscriptions())) {
-            int commonSubsSize = msgSubscriptions.getCommonSubscriptions().size();
-            applicationSubscriptions = initArrayList(commonSubsSize);
-            deviceSubscriptions = initArrayList(commonSubsSize);
-            processSubscriptions(msgSubscriptions.getCommonSubscriptions(), publishMsgProto,
-                    applicationSubscriptions, deviceSubscriptions);
+            processSubscriptions(msgSubscriptions.getCommonSubscriptions(), publishMsgProto, persistentSubscriptions);
         }
-
         if (!CollectionUtils.isEmpty(msgSubscriptions.getTargetDeviceSharedSubscriptions())) {
-            int targetDeviceSharedSubsSize = msgSubscriptions.getTargetDeviceSharedSubscriptions().size();
-            applicationSubscriptions = initSubscriptionListIfNull(applicationSubscriptions, targetDeviceSharedSubsSize);
-            deviceSubscriptions = initSubscriptionListIfNull(deviceSubscriptions, targetDeviceSharedSubsSize);
-            processSubscriptions(msgSubscriptions.getTargetDeviceSharedSubscriptions(), publishMsgProto,
-                    applicationSubscriptions, deviceSubscriptions);
+            processSubscriptions(msgSubscriptions.getTargetDeviceSharedSubscriptions(), publishMsgProto, persistentSubscriptions);
         }
 
         if (publishMsgProcessingTimerStats != null) {
             publishMsgProcessingTimerStats.logNotPersistentMessagesProcessing(startTime, TimeUnit.NANOSECONDS);
         }
-        return new PersistentMsgSubscriptions(
-                deviceSubscriptions,
-                applicationSubscriptions,
-                msgSubscriptions.getAllApplicationSharedSubscriptions()
-        );
+        return persistentSubscriptions;
     }
 
     private void processSubscriptions(List<Subscription> subscriptions, PublishMsgProto publishMsgProto,
-                                      List<Subscription> applicationSubscriptions, List<Subscription> deviceSubscriptions) {
+                                      final PersistentMsgSubscriptions persistentMsgSubscriptions) {
         boolean nonPersistentByPubQos = publishMsgProto.getQos() == MqttQoS.AT_MOST_ONCE.value();
         if (nonPersistentByPubQos) {
             if (processSubscriptionsInParallel) {
@@ -194,13 +182,25 @@ private void processSubscriptions(List<Subscription> subscriptions, PublishMsgPr
                 }
             }
         } else {
+            persistentMsgSubscriptions.setDeviceSubscriptions(initSubscriptionListIfNull(persistentMsgSubscriptions.getDeviceSubscriptions(), subscriptions.size()));
+            persistentMsgSubscriptions.setApplicationSubscriptions(initSubscriptionListIfNull(persistentMsgSubscriptions.getApplicationSubscriptions(), subscriptions.size()));
             if (processSubscriptionsInParallel) {
                 subscriptions
                         .parallelStream()
-                        .forEach(subscription -> processSubscription(subscription, publishMsgProto, applicationSubscriptions, deviceSubscriptions));
+                        .forEach(subscription -> processSubscription(
+                                subscription,
+                                publishMsgProto,
+                                persistentMsgSubscriptions.getApplicationSubscriptions(),
+                                persistentMsgSubscriptions.getDeviceSubscriptions())
+                        );
             } else {
                 for (Subscription subscription : subscriptions) {
-                    processSubscription(subscription, publishMsgProto, applicationSubscriptions, deviceSubscriptions);
+                    processSubscription(
+                            subscription,
+                            publishMsgProto,
+                            persistentMsgSubscriptions.getApplicationSubscriptions(),
+                            persistentMsgSubscriptions.getDeviceSubscriptions()
+                    );
                 }
             }
         }

application/src/main/java/org/thingsboard/mqtt/broker/service/processing/data/PersistentMsgSubscriptions.java

@@ -15,21 +15,21 @@
  */
 package org.thingsboard.mqtt.broker.service.processing.data;
 
+import lombok.AllArgsConstructor;
 import lombok.Data;
-import lombok.RequiredArgsConstructor;
 import org.springframework.util.CollectionUtils;
 import org.thingsboard.mqtt.broker.service.subscription.Subscription;
 
 import java.util.List;
 import java.util.Set;
 
 @Data
-@RequiredArgsConstructor
+@AllArgsConstructor
 public class PersistentMsgSubscriptions {
 
-    private final List<Subscription> deviceSubscriptions;
-    private final List<Subscription> applicationSubscriptions;
-    private final Set<Subscription> allApplicationSharedSubscriptions;
+    private List<Subscription> deviceSubscriptions;
+    private List<Subscription> applicationSubscriptions;
+    private Set<Subscription> allApplicationSharedSubscriptions;
 
     public boolean isNotEmpty() {
         return !CollectionUtils.isEmpty(deviceSubscriptions) ||

application/src/test/java/org/thingsboard/mqtt/broker/service/processing/MsgDispatcherServiceImplTest.java

@@ -311,29 +311,39 @@ public void testProcessBasicAndCollectPersistentSubscriptions() {
         ClientSessionInfo clientSessionInfo3 = mock(ClientSessionInfo.class);
         ClientSessionInfo clientSessionInfo4 = mock(ClientSessionInfo.class);
         ClientSessionInfo clientSessionInfo5 = mock(ClientSessionInfo.class);
+        ClientSessionInfo clientSessionInfo6 = mock(ClientSessionInfo.class);
+        ClientSessionInfo clientSessionInfo7 = mock(ClientSessionInfo.class);
 
         when(clientSessionInfo1.isPersistent()).thenReturn(true);
         when(clientSessionInfo2.isPersistent()).thenReturn(true);
         when(clientSessionInfo5.isPersistent()).thenReturn(false);
+        when(clientSessionInfo6.isPersistent()).thenReturn(true);
+        when(clientSessionInfo7.isPersistent()).thenReturn(true);
 
         when(clientSessionInfo1.getType()).thenReturn(ClientType.APPLICATION);
         when(clientSessionInfo2.getType()).thenReturn(ClientType.APPLICATION);
         when(clientSessionInfo5.getType()).thenReturn(ClientType.DEVICE);
+        when(clientSessionInfo6.getType()).thenReturn(ClientType.DEVICE);
+        when(clientSessionInfo7.getType()).thenReturn(ClientType.DEVICE);
 
         mockClientSessionGetClientId(clientSessionInfo1, "clientId1");
         mockClientSessionGetClientId(clientSessionInfo2, "clientId2");
+        mockClientSessionGetClientId(clientSessionInfo6, "clientId6");
+        mockClientSessionGetClientId(clientSessionInfo7, "clientId7");
 
         MsgSubscriptions msgSubscriptions = new MsgSubscriptions(
                 List.of(
                         new Subscription("test/topic/1", 1, clientSessionInfo1),
-                        new Subscription("test/+/1", 2, clientSessionInfo2)
+                        new Subscription("test/+/1", 2, clientSessionInfo2),
+                        new Subscription("test/+/1", 1, clientSessionInfo7)
                 ),
                 Set.of(
                         new Subscription("#", 2, clientSessionInfo3),
                         new Subscription("test/#", 0, clientSessionInfo4)
                 ),
                 List.of(
-                        new Subscription("test/topic/#", 1, clientSessionInfo5)
+                        new Subscription("test/topic/#", 1, clientSessionInfo5),
+                        new Subscription("+/topic/1", 2, clientSessionInfo6)
                 )
         );
         QueueProtos.PublishMsgProto publishMsgProto = QueueProtos.PublishMsgProto
@@ -346,7 +356,11 @@ public void testProcessBasicAndCollectPersistentSubscriptions() {
                 msgDispatcherService.processBasicAndCollectPersistentSubscriptions(msgSubscriptions, publishMsgProto);
 
         assertEquals(persistentMsgSubscriptions.getAllApplicationSharedSubscriptions(), msgSubscriptions.getAllApplicationSharedSubscriptions());
-        assertTrue(persistentMsgSubscriptions.getDeviceSubscriptions().isEmpty());
+
+        assertEquals(2, persistentMsgSubscriptions.getDeviceSubscriptions().size());
+        List<String> devClientIds = getClientIds(persistentMsgSubscriptions.getDeviceSubscriptions().stream());
+        assertTrue(devClientIds.containsAll(List.of("clientId6", "clientId7")));
+
         assertEquals(2, persistentMsgSubscriptions.getApplicationSubscriptions().size());
         List<String> appClientIds = getClientIds(persistentMsgSubscriptions.getApplicationSubscriptions().stream());
         assertTrue(appClientIds.containsAll(List.of("clientId1", "clientId2")));