Validation and Replacement of Queue names in the Device profile

@@ -486,7 +486,6 @@ public class ActorSystemContext {
         return partitionService.resolve(serviceType, queueName, tenantId, entityId);
     }
 
-
     public String getServiceId() {
         return serviceInfoProvider.getServiceId();
     }

@@ -15,7 +15,7 @@
  */
 package org.thingsboard.server.controller;
 
-import org.springframework.beans.factory.annotation.Autowired;
+import lombok.RequiredArgsConstructor;
 import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
@@ -24,41 +24,26 @@ import org.springframework.web.bind.annotation.ResponseBody;
 import org.springframework.web.bind.annotation.RestController;
 import org.thingsboard.server.common.data.exception.ThingsboardException;
 import org.thingsboard.server.common.msg.queue.ServiceType;
-import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
-import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
+import org.thingsboard.server.queue.QueueService;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.stream.Collectors;
+import java.util.Set;
 
 @RestController
 @TbCoreComponent
 @RequestMapping("/api")
+@RequiredArgsConstructor
 public class QueueController extends BaseController {
 
-    @Autowired(required = false)
-    private TbQueueRuleEngineSettings ruleEngineSettings;
+    private final QueueService queueService;
 
     @PreAuthorize("hasAuthority('TENANT_ADMIN')")
     @RequestMapping(value = "/tenant/queues", params = {"serviceType"}, method = RequestMethod.GET)
     @ResponseBody
-    public List<String> getTenantQueuesByServiceType(@RequestParam String serviceType) throws ThingsboardException {
+    public Set<String> getTenantQueuesByServiceType(@RequestParam String serviceType) throws ThingsboardException {
         checkParameter("serviceType", serviceType);
         try {
-            ServiceType type = ServiceType.valueOf(serviceType);
-            switch (type) {
-                case TB_RULE_ENGINE:
-                    if (ruleEngineSettings == null) {
-                        return Arrays.asList("Main", "HighPriority", "SequentialByOriginator");
-                    }
-                    return ruleEngineSettings.getQueues().stream()
-                            .map(TbRuleEngineQueueConfiguration::getName)
-                            .collect(Collectors.toList());
-                default:
-                    return Collections.emptyList();
-            }
+            return queueService.getQueuesByServiceType(ServiceType.valueOf(serviceType));
         } catch (Exception e) {
             throw handleException(e);
         }

@@ -21,11 +21,13 @@ import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.test.util.ReflectionTestUtils;
 import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.queue.QueueService;
 import org.thingsboard.server.queue.discovery.HashPartitionService;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
@@ -57,6 +59,7 @@ public class HashPartitionServiceTest {
     private TenantRoutingInfoService routingInfoService;
     private ApplicationEventPublisher applicationEventPublisher;
     private TbQueueRuleEngineSettings ruleEngineSettings;
+    private QueueService queueService;
 
     private String hashFunctionName = "sha256";
 
@@ -67,10 +70,12 @@ public class HashPartitionServiceTest {
         applicationEventPublisher = mock(ApplicationEventPublisher.class);
         routingInfoService = mock(TenantRoutingInfoService.class);
         ruleEngineSettings = mock(TbQueueRuleEngineSettings.class);
+        queueService = mock(QueueService.class);
         clusterRoutingService = new HashPartitionService(discoveryService,
                 routingInfoService,
                 applicationEventPublisher,
-                ruleEngineSettings
+                ruleEngineSettings,
+                queueService
         );
         when(ruleEngineSettings.getQueues()).thenReturn(Collections.emptyList());
         ReflectionTestUtils.setField(clusterRoutingService, "coreTopic", "tb.core");
@@ -82,6 +87,7 @@ public class HashPartitionServiceTest {
                 .setTenantIdLSB(TenantId.NULL_UUID.getLeastSignificantBits())
                 .addAllServiceTypes(Collections.singletonList(ServiceType.TB_CORE.name()))
                 .build();
+//        when(queueService.resolve(Mockito.any(), Mockito.anyString())).thenAnswer(i -> i.getArguments()[1]);
 //        when(discoveryService.getServiceInfo()).thenReturn(currentServer);
         List<TransportProtos.ServiceInfo> otherServers = new ArrayList<>();
         for (int i = 1; i < SERVER_COUNT; i++) {

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * 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 org.thingsboard.server.queue;
+
+import org.thingsboard.server.common.msg.queue.ServiceType;
+
+import java.util.Set;
+
+public interface QueueService {
+
+    Set<String> getQueuesByServiceType(ServiceType serviceType);
+
+    String resolve(ServiceType serviceType, String queueName);
+
+}

@@ -42,6 +42,10 @@
         </dependency>
         <dependency>
             <groupId>org.thingsboard.common</groupId>
+            <artifactId>cluster-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.thingsboard.common</groupId>
             <artifactId>message</artifactId>
         </dependency>
         <dependency>

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * 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 org.thingsboard.server.queue;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+import org.thingsboard.server.common.msg.queue.ServiceQueue;
+import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
+import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
+
+import javax.annotation.PostConstruct;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+@Service
+@RequiredArgsConstructor
+public class DefaultQueueService implements QueueService {
+
+    private final TbQueueRuleEngineSettings ruleEngineSettings;
+    private Set<String> ruleEngineQueues;
+
+    @PostConstruct
+    public void init() {
+        ruleEngineQueues = ruleEngineSettings.getQueues().stream()
+                .map(TbRuleEngineQueueConfiguration::getName).collect(Collectors.toCollection(LinkedHashSet::new));
+    }
+
+    @Override
+    public Set<String> getQueuesByServiceType(ServiceType type) {
+        if (type == ServiceType.TB_RULE_ENGINE) {
+            return ruleEngineQueues;
+        } else {
+            return Collections.emptySet();
+        }
+    }
+
+    @Override
+    public String resolve(ServiceType serviceType, String queueName) {
+        if (StringUtils.isEmpty(queueName) || !getQueuesByServiceType(serviceType).contains(queueName)) {
+            return ServiceQueue.MAIN;
+        } else {
+            return queueName;
+        }
+    }
+}

@@ -22,6 +22,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.msg.queue.ServiceQueue;
@@ -30,6 +31,7 @@ import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.ServiceInfo;
+import org.thingsboard.server.queue.QueueService;
 import org.thingsboard.server.queue.discovery.event.ClusterTopologyChangeEvent;
 import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
 import org.thingsboard.server.queue.discovery.event.ServiceListChangedEvent;
@@ -64,6 +66,7 @@ public class HashPartitionService implements PartitionService {
     private final TbServiceInfoProvider serviceInfoProvider;
     private final TenantRoutingInfoService tenantRoutingInfoService;
     private final TbQueueRuleEngineSettings tbQueueRuleEngineSettings;
+    private final QueueService queueService;
     private final ConcurrentMap<ServiceQueue, String> partitionTopics = new ConcurrentHashMap<>();
     private final ConcurrentMap<ServiceQueue, Integer> partitionSizes = new ConcurrentHashMap<>();
     private final ConcurrentMap<TenantId, TenantRoutingInfo> tenantRoutingInfoMap = new ConcurrentHashMap<>();
@@ -81,11 +84,13 @@ public class HashPartitionService implements PartitionService {
     public HashPartitionService(TbServiceInfoProvider serviceInfoProvider,
                                 TenantRoutingInfoService tenantRoutingInfoService,
                                 ApplicationEventPublisher applicationEventPublisher,
-                                TbQueueRuleEngineSettings tbQueueRuleEngineSettings) {
+                                TbQueueRuleEngineSettings tbQueueRuleEngineSettings,
+                                QueueService queueService) {
         this.serviceInfoProvider = serviceInfoProvider;
         this.tenantRoutingInfoService = tenantRoutingInfoService;
         this.applicationEventPublisher = applicationEventPublisher;
         this.tbQueueRuleEngineSettings = tbQueueRuleEngineSettings;
+        this.queueService = queueService;
     }
 
     @PostConstruct
@@ -106,6 +111,7 @@ public class HashPartitionService implements PartitionService {
 
     @Override
     public TopicPartitionInfo resolve(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId) {
+        queueName = queueService.resolve(serviceType, queueName);
         return resolve(new ServiceQueue(serviceType, queueName), tenantId, entityId);
     }
 

@@ -36,8 +36,6 @@ public class TbQueueRuleEngineSettings {
     private String topic;
     private List<TbRuleEngineQueueConfiguration> queues;
 
-    //TODO 2.5 ybondarenko: make sure the queue names are valid to all queue providers.
-    // See how they are used in TbRuleEngineQueueFactory.createToRuleEngineMsgConsumer and all producers
     @PostConstruct
     public void validate() {
         queues.stream().filter(queue -> queue.getName().equals("Main")).findFirst().orElseThrow(() -> {

@@ -28,7 +28,7 @@ import com.squareup.wire.schema.internal.parser.ProtoFileElement;
 import com.squareup.wire.schema.internal.parser.ProtoParser;
 import com.squareup.wire.schema.internal.parser.TypeElement;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
+import org.thingsboard.server.common.data.StringUtils;
 import org.hibernate.exception.ConstraintViolationException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cache.Cache;
@@ -63,6 +63,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.common.data.rule.RuleChain;
+import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.dao.dashboard.DashboardService;
 import org.thingsboard.server.dao.entity.AbstractEntityService;
 import org.thingsboard.server.dao.exception.DataValidationException;
@@ -72,6 +73,7 @@ import org.thingsboard.server.dao.service.DataValidator;
 import org.thingsboard.server.dao.service.PaginatedRemover;
 import org.thingsboard.server.dao.service.Validator;
 import org.thingsboard.server.dao.tenant.TenantDao;
+import org.thingsboard.server.queue.QueueService;
 
 import java.util.Arrays;
 import java.util.Collections;
@@ -82,7 +84,6 @@ import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
 
-import static com.google.protobuf.FieldType.MESSAGE;
 import static org.thingsboard.server.common.data.CacheConstants.DEVICE_PROFILE_CACHE;
 import static org.thingsboard.server.dao.service.Validator.validateId;
 
@@ -104,6 +105,9 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D
         return "[Transport Configuration] invalid " + schemaName + " provided!";
     }
 
+    @Autowired(required = false)
+    private QueueService queueService;
+
     @Autowired
     private DeviceProfileDao deviceProfileDao;
 
@@ -373,6 +377,11 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D
                             throw new DataValidationException("Another default device profile is present in scope of current tenant!");
                         }
                     }
+                    if (!StringUtils.isEmpty(deviceProfile.getDefaultQueueName()) && queueService != null){
+                        if(!queueService.getQueuesByServiceType(ServiceType.TB_RULE_ENGINE).contains(deviceProfile.getDefaultQueueName())){
+                            throw new DataValidationException("Device profile is referencing to non-existent queue!");
+                        }
+                    }
                     if (deviceProfile.getProvisionType() == null) {
                         deviceProfile.setProvisionType(DeviceProfileProvisionType.DISABLED);
                     }

@@ -187,6 +187,7 @@ export class AddDeviceProfileDialogComponent extends
         name: this.deviceProfileDetailsFormGroup.get('name').value,
         type: this.deviceProfileDetailsFormGroup.get('type').value,
         image: this.deviceProfileDetailsFormGroup.get('image').value,
+        defaultQueueName: this.deviceProfileDetailsFormGroup.get('defaultQueueName').value,
         transportType: this.transportConfigFormGroup.get('transportType').value,
         provisionType: deviceProvisionConfiguration.type,
         provisionDeviceKey,

@@ -155,12 +155,7 @@ export class QueueTypeListComponent implements ControlValueAccessor, OnInit, Aft
           return searchText ? queue.toUpperCase().startsWith(searchText.toUpperCase()) : true;
         });
         if (result.length) {
-          if (searchText && searchText.length && result.indexOf(searchText) === -1) {
-            result.push(searchText);
-          }
           result.sort();
-        } else if (searchText && searchText.length) {
-          result.push(searchText);
         }
         return result;
       })