Push entity created event to the device profile rule chain and queue if specified (#4131)

* Remove device from cache in case null value cached in the distributed redis

* Handle case when device was removed from db but message in the queue exists

* Code review chagnes

* Added usage statistics configuration to yml file

* Use msg queue instead of default

* Make private

* Make private

* Push entity created event to the device profile rule chain and queue if specified

application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java

@@ -278,7 +278,21 @@ class DefaultTbContext implements TbContext {
     }
 
     public TbMsg deviceCreatedMsg(Device device, RuleNodeId ruleNodeId) {
-        return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED);
+        RuleChainId ruleChainId = null;
+        String queueName = ServiceQueue.MAIN;
+        if (device.getDeviceProfileId() != null) {
+            DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().find(device.getDeviceProfileId());
+            if (deviceProfile == null) {
+                log.warn("[{}] Device profile is null!", device.getDeviceProfileId());
+                ruleChainId = null;
+                queueName = ServiceQueue.MAIN;
+            } else {
+                ruleChainId = deviceProfile.getDefaultRuleChainId();
+                String defaultQueueName = deviceProfile.getDefaultQueueName();
+                queueName = defaultQueueName != null ? defaultQueueName : ServiceQueue.MAIN;
+            }
+        }
+        return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, queueName, ruleChainId);
     }
 
     public TbMsg assetCreatedMsg(Asset asset, RuleNodeId ruleNodeId) {
@@ -290,8 +304,12 @@ class DefaultTbContext implements TbContext {
     }
 
     public <E, I extends EntityId> TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action) {
+        return entityActionMsg(entity, id, ruleNodeId, action, ServiceQueue.MAIN, null);
+    }
+
+    public <E, I extends EntityId> TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, String queueName, RuleChainId ruleChainId) {
         try {
-            return TbMsg.newMsg(action, id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)));
+            return TbMsg.newMsg(queueName, action, id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), ruleChainId, null);
         } catch (JsonProcessingException | IllegalArgumentException e) {
             throw new RuntimeException("Failed to process " + id.getEntityType().name().toLowerCase() + " " + action + " msg: " + e);
         }