Browse Source

used queueId instead of queueName in DeviceProfile and TbMsg

pull/6134/head
YevhenBondarenko 4 years ago
parent
commit
4e4ac0012c
  1. 11
      application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
  2. 71
      application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java
  3. 2
      application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
  4. 33
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java
  5. 10
      application/src/main/java/org/thingsboard/server/service/queue/DefaultQueueRoutingInfoService.java
  6. 15
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java
  7. 11
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java
  8. 7
      application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingResult.java
  9. 7
      application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java
  10. 15
      application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
  11. 32
      common/cluster-api/src/main/proto/queue.proto
  12. 5
      common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java
  13. 2
      common/data/src/main/java/org/thingsboard/server/common/data/id/QueueId.java
  14. 63
      common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
  15. 36
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java
  16. 3
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java
  17. 11
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/QueueRoutingInfo.java
  18. 5
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/QueueRoutingInfoService.java
  19. 4
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java
  20. 37
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
  21. 15
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/TransportQueueRoutingInfoService.java
  22. 12
      dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java
  23. 2
      dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java
  24. 13
      dao/src/main/java/org/thingsboard/server/dao/model/sql/DeviceProfileEntity.java
  25. 31
      dao/src/main/resources/sql/schema-entities.sql
  26. 6
      dao/src/test/resources/sql/system-data.sql
  27. 14
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java
  28. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java
  29. 4
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java
  30. 6
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java
  31. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
  32. 12
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNode.java
  33. 3
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNodeConfiguration.java
  34. 8
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java
  35. 4
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java
  36. 30
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java
  37. 15
      ui-ngx/src/app/core/http/entity.service.ts
  38. 4
      ui-ngx/src/app/core/http/queue.service.ts
  39. 6
      ui-ngx/src/app/modules/home/components/profile/add-device-profile-dialog.component.html
  40. 8
      ui-ngx/src/app/modules/home/components/profile/add-device-profile-dialog.component.ts
  41. 6
      ui-ngx/src/app/modules/home/components/profile/device-profile.component.html
  42. 8
      ui-ngx/src/app/modules/home/components/profile/device-profile.component.ts
  43. 6
      ui-ngx/src/app/modules/home/components/wizard/device-wizard-dialog.component.html
  44. 2
      ui-ngx/src/app/modules/home/components/wizard/device-wizard-dialog.component.scss
  45. 11
      ui-ngx/src/app/modules/home/components/wizard/device-wizard-dialog.component.ts
  46. 54
      ui-ngx/src/app/shared/components/queue/queue-autocomplete.component.html
  47. 215
      ui-ngx/src/app/shared/components/queue/queue-autocomplete.component.ts
  48. 3
      ui-ngx/src/app/shared/models/device.models.ts
  49. 3
      ui-ngx/src/app/shared/shared.module.ts
  50. 2
      ui-ngx/src/assets/locale/locale.constant-en_US.json

11
application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@ -40,6 +40,7 @@ import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.Event;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.msg.TbActorMsg;
@ -63,6 +64,7 @@ import org.thingsboard.server.dao.event.EventService;
import org.thingsboard.server.dao.nosql.CassandraBufferedRateReadExecutor;
import org.thingsboard.server.dao.nosql.CassandraBufferedRateWriteExecutor;
import org.thingsboard.server.dao.ota.OtaPackageService;
import org.thingsboard.server.dao.queue.QueueService;
import org.thingsboard.server.dao.relation.RelationService;
import org.thingsboard.server.dao.resource.ResourceService;
import org.thingsboard.server.dao.rule.RuleChainService;
@ -330,6 +332,11 @@ public class ActorSystemContext {
@Getter
private TbRpcService tbRpcService;
@Lazy
@Autowired(required = false)
@Getter
private QueueService queueService;
@Value("${actors.session.max_concurrent_sessions_per_device:1}")
@Getter
private long maxConcurrentSessionsPerDevice;
@ -495,8 +502,8 @@ public class ActorSystemContext {
return partitionService.resolve(serviceType, tenantId, entityId);
}
public TopicPartitionInfo resolve(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId) {
return partitionService.resolve(serviceType, queueName, tenantId, entityId);
public TopicPartitionInfo resolve(ServiceType serviceType, QueueId queueId, TenantId tenantId, EntityId entityId) {
return partitionService.resolve(serviceType, queueId, tenantId, entityId);
}
public String getServiceId() {

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

@ -46,6 +46,7 @@ import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.RuleNodeId;
import org.thingsboard.server.common.data.id.TenantId;
@ -57,7 +58,6 @@ import org.thingsboard.server.common.msg.TbActorMsg;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.common.msg.TbMsgProcessingStackItem;
import org.thingsboard.server.common.msg.queue.ServiceQueue;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.dao.asset.AssetService;
@ -72,6 +72,7 @@ import org.thingsboard.server.dao.entityview.EntityViewService;
import org.thingsboard.server.dao.nosql.CassandraStatementTask;
import org.thingsboard.server.dao.nosql.TbResultSetFuture;
import org.thingsboard.server.dao.ota.OtaPackageService;
import org.thingsboard.server.dao.queue.QueueService;
import org.thingsboard.server.dao.relation.RelationService;
import org.thingsboard.server.dao.resource.ResourceService;
import org.thingsboard.server.dao.rule.RuleChainService;
@ -161,8 +162,8 @@ class DefaultTbContext implements TbContext {
}
@Override
public void enqueue(TbMsg tbMsg, String queueName, Runnable onSuccess, Consumer<Throwable> onFailure) {
TopicPartitionInfo tpi = resolvePartition(tbMsg, queueName);
public void enqueue(TbMsg tbMsg, QueueId queueId, Runnable onSuccess, Consumer<Throwable> onFailure) {
TopicPartitionInfo tpi = resolvePartition(tbMsg, queueId);
enqueue(tpi, tbMsg, onFailure, onSuccess);
}
@ -213,33 +214,30 @@ class DefaultTbContext implements TbContext {
}
@Override
public void enqueueForTellNext(TbMsg tbMsg, String queueName, String relationType, Runnable onSuccess, Consumer<Throwable> onFailure) {
TopicPartitionInfo tpi = resolvePartition(tbMsg, queueName);
enqueueForTellNext(tpi, queueName, tbMsg, Collections.singleton(relationType), null, onSuccess, onFailure);
public void enqueueForTellNext(TbMsg tbMsg, QueueId queueId, String relationType, Runnable onSuccess, Consumer<Throwable> onFailure) {
TopicPartitionInfo tpi = resolvePartition(tbMsg, queueId);
enqueueForTellNext(tpi, queueId, tbMsg, Collections.singleton(relationType), null, onSuccess, onFailure);
}
@Override
public void enqueueForTellNext(TbMsg tbMsg, String queueName, Set<String> relationTypes, Runnable onSuccess, Consumer<Throwable> onFailure) {
TopicPartitionInfo tpi = resolvePartition(tbMsg, queueName);
enqueueForTellNext(tpi, queueName, tbMsg, relationTypes, null, onSuccess, onFailure);
public void enqueueForTellNext(TbMsg tbMsg, QueueId queueId, Set<String> relationTypes, Runnable onSuccess, Consumer<Throwable> onFailure) {
TopicPartitionInfo tpi = resolvePartition(tbMsg, queueId);
enqueueForTellNext(tpi, queueId, tbMsg, relationTypes, null, onSuccess, onFailure);
}
private TopicPartitionInfo resolvePartition(TbMsg tbMsg, String queueName) {
if (StringUtils.isEmpty(queueName)) {
queueName = ServiceQueue.MAIN;
}
return mainCtx.resolve(ServiceType.TB_RULE_ENGINE, queueName, getTenantId(), tbMsg.getOriginator());
private TopicPartitionInfo resolvePartition(TbMsg tbMsg, QueueId queueId) {
return mainCtx.resolve(ServiceType.TB_RULE_ENGINE, queueId, getTenantId(), tbMsg.getOriginator());
}
private TopicPartitionInfo resolvePartition(TbMsg tbMsg) {
return resolvePartition(tbMsg, tbMsg.getQueueName());
return resolvePartition(tbMsg, tbMsg.getQueueId());
}
private void enqueueForTellNext(TopicPartitionInfo tpi, TbMsg source, Set<String> relationTypes, String failureMessage, Runnable onSuccess, Consumer<Throwable> onFailure) {
enqueueForTellNext(tpi, source.getQueueName(), source, relationTypes, failureMessage, onSuccess, onFailure);
enqueueForTellNext(tpi, source.getQueueId(), source, relationTypes, failureMessage, onSuccess, onFailure);
}
private void enqueueForTellNext(TopicPartitionInfo tpi, String queueName, TbMsg source, Set<String> relationTypes, String failureMessage, Runnable onSuccess, Consumer<Throwable> onFailure) {
private void enqueueForTellNext(TopicPartitionInfo tpi, QueueId queueId, TbMsg source, Set<String> relationTypes, String failureMessage, Runnable onSuccess, Consumer<Throwable> onFailure) {
if (!source.isValid()) {
log.trace("[{}] Skip invalid message: {}", getTenantId(), source);
onFailure.accept(new IllegalArgumentException("Source message is no longer valid!"));
@ -247,7 +245,7 @@ class DefaultTbContext implements TbContext {
}
RuleChainId ruleChainId = nodeCtx.getSelf().getRuleChainId();
RuleNodeId ruleNodeId = nodeCtx.getSelf().getId();
TbMsg tbMsg = TbMsg.newMsg(source, queueName, ruleChainId, ruleNodeId);
TbMsg tbMsg = TbMsg.newMsg(source, queueId, ruleChainId, ruleNodeId);
TransportProtos.ToRuleEngineMsg.Builder msg = TransportProtos.ToRuleEngineMsg.newBuilder()
.setTenantIdMSB(getTenantId().getId().getMostSignificantBits())
.setTenantIdLSB(getTenantId().getId().getLeastSignificantBits())
@ -306,13 +304,13 @@ class DefaultTbContext implements TbContext {
}
@Override
public TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data) {
return newMsg(queueName, type, originator, null, metaData, data);
public TbMsg newMsg(QueueId queueId, String type, EntityId originator, TbMsgMetaData metaData, String data) {
return newMsg(queueId, type, originator, null, metaData, data);
}
@Override
public TbMsg newMsg(String queueName, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
return TbMsg.newMsg(queueName, type, originator, customerId, metaData, data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId());
public TbMsg newMsg(QueueId queueId, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
return TbMsg.newMsg(queueId, type, originator, customerId, metaData, data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId());
}
@Override
@ -326,20 +324,17 @@ class DefaultTbContext implements TbContext {
public TbMsg deviceCreatedMsg(Device device, RuleNodeId ruleNodeId) {
RuleChainId ruleChainId = null;
String queueName = ServiceQueue.MAIN;
QueueId queueId = null;
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;
queueId = deviceProfile.getDefaultQueueId();
}
}
return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, queueName, ruleChainId);
return entityActionMsg(device, device.getId(), ruleNodeId, DataConstants.ENTITY_CREATED, queueId, ruleChainId);
}
public TbMsg assetCreatedMsg(Asset asset, RuleNodeId ruleNodeId) {
@ -348,21 +343,18 @@ class DefaultTbContext implements TbContext {
public TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action) {
RuleChainId ruleChainId = null;
String queueName = ServiceQueue.MAIN;
QueueId queueId = null;
if (EntityType.DEVICE.equals(alarm.getOriginator().getEntityType())) {
DeviceId deviceId = new DeviceId(alarm.getOriginator().getId());
DeviceProfile deviceProfile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId);
if (deviceProfile == null) {
log.warn("[{}] Device profile is null!", deviceId);
ruleChainId = null;
queueName = ServiceQueue.MAIN;
} else {
ruleChainId = deviceProfile.getDefaultRuleChainId();
String defaultQueueName = deviceProfile.getDefaultQueueName();
queueName = defaultQueueName != null ? defaultQueueName : ServiceQueue.MAIN;
queueId = deviceProfile.getDefaultQueueId();
}
}
return entityActionMsg(alarm, alarm.getId(), ruleNodeId, action, queueName, ruleChainId);
return entityActionMsg(alarm, alarm.getId(), ruleNodeId, action, queueId, ruleChainId);
}
@Override
@ -371,12 +363,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);
return entityActionMsg(entity, id, ruleNodeId, action, null, null);
}
public <E, I extends EntityId> TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, String queueName, RuleChainId ruleChainId) {
public <E, I extends EntityId> TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, QueueId queueId, RuleChainId ruleChainId) {
try {
return TbMsg.newMsg(queueName, action, id, getActionMetaData(ruleNodeId), mapper.writeValueAsString(mapper.valueToTree(entity)), ruleChainId, null);
return TbMsg.newMsg(queueId, 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);
}
@ -548,6 +540,11 @@ class DefaultTbContext implements TbContext {
return mainCtx.getEdgeEventService();
}
@Override
public QueueService getQueueService() {
return mainCtx.getQueueService();
}
@Override
public EventLoopGroup getSharedEventLoop() {
return mainCtx.getSharedEventLoopGroupService().getSharedEventLoopGroup();

2
application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java

@ -293,7 +293,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
try {
checkComponentStateActive(msg);
EntityId entityId = msg.getOriginator();
TopicPartitionInfo tpi = systemContext.resolve(ServiceType.TB_RULE_ENGINE, msg.getQueueName(), tenantId, entityId);
TopicPartitionInfo tpi = systemContext.resolve(ServiceType.TB_RULE_ENGINE, msg.getQueueId(), tenantId, entityId);
List<RuleNodeRelation> ruleNodeRelations = nodeRoutes.get(originatorNodeId);
if (ruleNodeRelations == null) { // When unchecked, this will cause NullPointerException when rule node doesn't exist anymore

33
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java

@ -44,6 +44,7 @@ import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityViewId;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.UserId;
@ -51,7 +52,6 @@ import org.thingsboard.server.common.data.kv.AttributeKey;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.common.msg.queue.ServiceQueue;
import org.thingsboard.server.common.msg.session.SessionMsgType;
import org.thingsboard.server.common.transport.adaptor.JsonConverter;
import org.thingsboard.server.common.transport.util.JsonUtils;
@ -134,24 +134,23 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
return metaData;
}
private Pair<String, RuleChainId> getDefaultQueueNameAndRuleChainId(TenantId tenantId, EntityId entityId) {
private Pair<QueueId, RuleChainId> getDefaultQueueNameAndRuleChainId(TenantId tenantId, EntityId entityId) {
if (EntityType.DEVICE.equals(entityId.getEntityType())) {
DeviceProfile deviceProfile = deviceProfileCache.get(tenantId, new DeviceId(entityId.getId()));
RuleChainId ruleChainId;
String queueName;
QueueId queueId;
if (deviceProfile == null) {
log.warn("[{}] Device profile is null!", entityId);
ruleChainId = null;
queueName = ServiceQueue.MAIN;
queueId = null;
} else {
ruleChainId = deviceProfile.getDefaultRuleChainId();
String defaultQueueName = deviceProfile.getDefaultQueueName();
queueName = defaultQueueName != null ? defaultQueueName : ServiceQueue.MAIN;
queueId = deviceProfile.getDefaultQueueId();
}
return new ImmutablePair<>(queueName, ruleChainId);
return new ImmutablePair<>(queueId, ruleChainId);
} else {
return new ImmutablePair<>(ServiceQueue.MAIN, null);
return new ImmutablePair<>(null, null);
}
}
@ -160,10 +159,10 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
for (TransportProtos.TsKvListProto tsKv : msg.getTsKvListList()) {
JsonObject json = JsonUtils.getJsonObject(tsKv.getKvList());
metaData.putValue("ts", tsKv.getTs() + "");
Pair<String, RuleChainId> defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId);
String queueName = defaultQueueAndRuleChain.getKey();
Pair<QueueId, RuleChainId> defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId);
QueueId queueId = defaultQueueAndRuleChain.getKey();
RuleChainId ruleChainId = defaultQueueAndRuleChain.getValue();
TbMsg tbMsg = TbMsg.newMsg(queueName, SessionMsgType.POST_TELEMETRY_REQUEST.name(), entityId, customerId, metaData, gson.toJson(json), ruleChainId, null);
TbMsg tbMsg = TbMsg.newMsg(queueId, SessionMsgType.POST_TELEMETRY_REQUEST.name(), entityId, customerId, metaData, gson.toJson(json), ruleChainId, null);
tbClusterService.pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
@Override
public void onSuccess(TbQueueMsgMetadata metadata) {
@ -183,10 +182,10 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
private ListenableFuture<Void> processPostAttributes(TenantId tenantId, CustomerId customerId, EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) {
SettableFuture<Void> futureToSet = SettableFuture.create();
JsonObject json = JsonUtils.getJsonObject(msg.getKvList());
Pair<String, RuleChainId> defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId);
String queueName = defaultQueueAndRuleChain.getKey();
Pair<QueueId, RuleChainId> defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId);
QueueId queueId = defaultQueueAndRuleChain.getKey();
RuleChainId ruleChainId = defaultQueueAndRuleChain.getValue();
TbMsg tbMsg = TbMsg.newMsg(queueName, SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), entityId, customerId, metaData, gson.toJson(json), ruleChainId, null);
TbMsg tbMsg = TbMsg.newMsg(queueId, SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), entityId, customerId, metaData, gson.toJson(json), ruleChainId, null);
tbClusterService.pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
@Override
public void onSuccess(TbQueueMsgMetadata metadata) {
@ -210,10 +209,10 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
Futures.addCallback(future, new FutureCallback<List<Void>>() {
@Override
public void onSuccess(@Nullable List<Void> voids) {
Pair<String, RuleChainId> defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId);
String queueName = defaultQueueAndRuleChain.getKey();
Pair<QueueId, RuleChainId> defaultQueueAndRuleChain = getDefaultQueueNameAndRuleChainId(tenantId, entityId);
QueueId queueId = defaultQueueAndRuleChain.getKey();
RuleChainId ruleChainId = defaultQueueAndRuleChain.getValue();
TbMsg tbMsg = TbMsg.newMsg(queueName, DataConstants.ATTRIBUTES_UPDATED, entityId, customerId, metaData, gson.toJson(json), ruleChainId, null);
TbMsg tbMsg = TbMsg.newMsg(queueId, DataConstants.ATTRIBUTES_UPDATED, entityId, customerId, metaData, gson.toJson(json), ruleChainId, null);
tbClusterService.pushMsgToRuleEngine(tenantId, tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
@Override
public void onSuccess(TbQueueMsgMetadata metadata) {

10
application/src/main/java/org/thingsboard/server/service/queue/DefaultQueueRoutingInfoService.java

@ -18,7 +18,6 @@ package org.thingsboard.server.service.queue;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Service;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.dao.queue.QueueService;
import org.thingsboard.server.queue.discovery.QueueRoutingInfo;
import org.thingsboard.server.queue.discovery.QueueRoutingInfoService;
@ -42,13 +41,4 @@ public class DefaultQueueRoutingInfoService implements QueueRoutingInfoService {
return queueService.findAllQueues().stream().map(QueueRoutingInfo::new).collect(Collectors.toList());
}
@Override
public List<QueueRoutingInfo> getMainQueuesRoutingInfo() {
return queueService.findAllMainQueues().stream().map(QueueRoutingInfo::new).collect(Collectors.toList());
}
@Override
public List<QueueRoutingInfo> getQueuesRoutingInfo(TenantId tenantId) {
return queueService.findQueuesByTenantId(tenantId).stream().map(QueueRoutingInfo::new).collect(Collectors.toList());
}
}

15
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java

@ -29,6 +29,7 @@ import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.edge.EdgeEventActionType;
import org.thingsboard.server.common.data.edge.EdgeEventType;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.ToDeviceActorNotificationMsg;
import org.thingsboard.server.common.data.ApiUsageState;
@ -169,7 +170,7 @@ public class DefaultTbClusterService implements TbClusterService {
tbMsg = transformMsg(tbMsg, deviceProfileCache.get(tenantId, new DeviceProfileId(entityId.getId())));
}
}
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), tenantId, entityId);
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueId(), tenantId, entityId);
log.trace("PUSHING msg: {} to:{}", tbMsg, tpi);
ToRuleEngineMsg msg = ToRuleEngineMsg.newBuilder()
.setTenantIdMSB(tenantId.getId().getMostSignificantBits())
@ -182,16 +183,16 @@ public class DefaultTbClusterService implements TbClusterService {
private TbMsg transformMsg(TbMsg tbMsg, DeviceProfile deviceProfile) {
if (deviceProfile != null) {
RuleChainId targetRuleChainId = deviceProfile.getDefaultRuleChainId();
String targetQueueName = deviceProfile.getDefaultQueueName();
QueueId targetQueueId = deviceProfile.getDefaultQueueId();
boolean isRuleChainTransform = targetRuleChainId != null && !targetRuleChainId.equals(tbMsg.getRuleChainId());
boolean isQueueTransform = targetQueueName != null && !targetQueueName.equals(tbMsg.getQueueName());
boolean isQueueTransform = targetQueueId != null && !targetQueueId.equals(tbMsg.getQueueId());
if (isRuleChainTransform && isQueueTransform) {
tbMsg = TbMsg.transformMsg(tbMsg, targetRuleChainId, targetQueueName);
tbMsg = TbMsg.transformMsg(tbMsg, targetRuleChainId, targetQueueId);
} else if (isRuleChainTransform) {
tbMsg = TbMsg.transformMsg(tbMsg, targetRuleChainId);
} else if (isQueueTransform) {
tbMsg = TbMsg.transformMsg(tbMsg, targetQueueName);
tbMsg = TbMsg.transformMsg(tbMsg, targetQueueId);
}
}
return tbMsg;
@ -495,6 +496,8 @@ public class DefaultTbClusterService implements TbClusterService {
TransportProtos.QueueUpdateMsg queueUpdateMsg = TransportProtos.QueueUpdateMsg.newBuilder()
.setTenantIdMSB(queue.getTenantId().getId().getMostSignificantBits())
.setTenantIdLSB(queue.getTenantId().getId().getLeastSignificantBits())
.setQueueIdMSB(queue.getId().getId().getMostSignificantBits())
.setQueueIdLSB(queue.getId().getId().getLeastSignificantBits())
.setQueueName(queue.getName())
.setQueueTopic(queue.getTopic())
.setPartitions(queue.getPartitions())
@ -537,6 +540,8 @@ public class DefaultTbClusterService implements TbClusterService {
TransportProtos.QueueDeleteMsg queueDeleteMsg = TransportProtos.QueueDeleteMsg.newBuilder()
.setTenantIdMSB(queue.getTenantId().getId().getMostSignificantBits())
.setTenantIdLSB(queue.getTenantId().getId().getLeastSignificantBits())
.setQueueIdMSB(queue.getId().getId().getMostSignificantBits())
.setQueueIdLSB(queue.getId().getId().getLeastSignificantBits())
.setQueueName(queue.getName())
.build();

11
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java

@ -22,6 +22,7 @@ import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.data.rpc.RpcError;
@ -275,7 +276,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
final boolean timeout = !ctx.await(configuration.getPackProcessingTimeout(), TimeUnit.MILLISECONDS);
TbRuleEngineProcessingResult result = new TbRuleEngineProcessingResult(configuration.getName(), timeout, ctx);
TbRuleEngineProcessingResult result = new TbRuleEngineProcessingResult(configuration.getId(), timeout, ctx);
if (timeout) {
printFirstOrAll(configuration, ctx, ctx.getPendingMap(), "Timeout");
}
@ -340,7 +341,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
new TbMsgPackCallback(id, tenantId, ctx);
try {
if (toRuleEngineMsg.getTbMsg() != null && !toRuleEngineMsg.getTbMsg().isEmpty()) {
forwardToRuleEngineActor(configuration.getName(), tenantId, toRuleEngineMsg, callback);
forwardToRuleEngineActor(configuration.getId(), tenantId, toRuleEngineMsg, callback);
} else {
callback.onSuccess();
}
@ -354,7 +355,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
log.info("{} to process [{}] messages", prefix, map.size());
for (Map.Entry<UUID, TbProtoQueueMsg<ToRuleEngineMsg>> pending : map.entrySet()) {
ToRuleEngineMsg tmp = pending.getValue().getValue();
TbMsg tmpMsg = TbMsg.fromBytes(configuration.getName(), tmp.getTbMsg().toByteArray(), TbMsgCallback.EMPTY);
TbMsg tmpMsg = TbMsg.fromBytes(configuration.getId(), tmp.getTbMsg().toByteArray(), TbMsgCallback.EMPTY);
RuleNodeInfo ruleNodeInfo = ctx.getLastVisitedRuleNode(pending.getKey());
if (printAll) {
log.trace("[{}] {} to process message: {}, Last Rule Node: {}", TenantId.fromUUID(new UUID(tmp.getTenantIdMSB(), tmp.getTenantIdLSB())), prefix, tmpMsg, ruleNodeInfo);
@ -454,8 +455,8 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
partitionService.removeQueue(queueDeleteMsg);
}
private void forwardToRuleEngineActor(String queueName, TenantId tenantId, ToRuleEngineMsg toRuleEngineMsg, TbMsgCallback callback) {
TbMsg tbMsg = TbMsg.fromBytes(queueName, toRuleEngineMsg.getTbMsg().toByteArray(), callback);
private void forwardToRuleEngineActor(QueueId queueId, TenantId tenantId, ToRuleEngineMsg toRuleEngineMsg, TbMsgCallback callback) {
TbMsg tbMsg = TbMsg.fromBytes(queueId, toRuleEngineMsg.getTbMsg().toByteArray(), callback);
QueueToRuleEngineMsg msg;
ProtocolStringList relationTypesList = toRuleEngineMsg.getRelationTypesList();
Set<String> relationTypes = null;

7
application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingResult.java

@ -16,6 +16,7 @@
package org.thingsboard.server.service.queue.processing;
import lombok.Getter;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.queue.RuleEngineException;
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg;
@ -28,7 +29,7 @@ import java.util.concurrent.ConcurrentMap;
public class TbRuleEngineProcessingResult {
@Getter
private final String queueName;
private final QueueId queueId;
@Getter
private final boolean success;
@Getter
@ -36,8 +37,8 @@ public class TbRuleEngineProcessingResult {
@Getter
private final TbMsgPackProcessingContext ctx;
public TbRuleEngineProcessingResult(String queueName, boolean timeout, TbMsgPackProcessingContext ctx) {
this.queueName = queueName;
public TbRuleEngineProcessingResult(QueueId queueId, boolean timeout, TbMsgPackProcessingContext ctx) {
this.queueId = queueId;
this.timeout = timeout;
this.ctx = ctx;
this.success = !timeout && ctx.getPendingMap().isEmpty() && ctx.getFailedMap().isEmpty();

7
application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java

@ -23,7 +23,6 @@ import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.queue.TbMsgCallback;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.settings.TbRuleEngineQueueAckStrategyConfiguration;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@ -126,7 +125,7 @@ public class TbRuleEngineProcessingStrategyFactory {
}
log.debug("[{}] Going to reprocess {} messages", queueName, toReprocess.size());
if (log.isTraceEnabled()) {
toReprocess.forEach((id, msg) -> log.trace("Going to reprocess [{}]: {}", id, TbMsg.fromBytes(result.getQueueName(), msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY)));
toReprocess.forEach((id, msg) -> log.trace("Going to reprocess [{}]: {}", id, TbMsg.fromBytes(result.getQueueId(), msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY)));
}
if (pauseBetweenRetries > 0) {
try {
@ -165,10 +164,10 @@ public class TbRuleEngineProcessingStrategyFactory {
log.debug("[{}] Reprocessing skipped for {} failed and {} timeout messages", queueName, result.getFailedMap().size(), result.getPendingMap().size());
}
if (log.isTraceEnabled()) {
result.getFailedMap().forEach((id, msg) -> log.trace("Failed messages [{}]: {}", id, TbMsg.fromBytes(result.getQueueName(), msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY)));
result.getFailedMap().forEach((id, msg) -> log.trace("Failed messages [{}]: {}", id, TbMsg.fromBytes(result.getQueueId(), msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY)));
}
if (log.isTraceEnabled()) {
result.getPendingMap().forEach((id, msg) -> log.trace("Timeout messages [{}]: {}", id, TbMsg.fromBytes(result.getQueueName(), msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY)));
result.getPendingMap().forEach((id, msg) -> log.trace("Timeout messages [{}]: {}", id, TbMsg.fromBytes(result.getQueueId(), msg.getValue().getTbMsg().toByteArray(), TbMsgCallback.EMPTY)));
}
return new TbRuleEngineProcessingDecision(true, null);
}

15
application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java

@ -180,10 +180,6 @@ public class DefaultTransportApiService implements TransportApiService {
result = handle(transportApiRequestMsg.getDeviceCredentialsRequestMsg());
} else if (transportApiRequestMsg.hasOtaPackageRequestMsg()) {
result = handle(transportApiRequestMsg.getOtaPackageRequestMsg());
} else if (transportApiRequestMsg.hasGetAllMainQueueRoutingInfoRequestMsg()) {
return Futures.transform(handle(transportApiRequestMsg.getGetAllMainQueueRoutingInfoRequestMsg()), value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor());
} else if (transportApiRequestMsg.hasGetTenantQueueRoutingInfoRequestMsg()) {
return Futures.transform(handle(transportApiRequestMsg.getGetTenantQueueRoutingInfoRequestMsg()), value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor());
} else if (transportApiRequestMsg.hasGetAllQueueRoutingInfoRequestMsg()) {
return Futures.transform(handle(transportApiRequestMsg.getGetAllQueueRoutingInfoRequestMsg()), value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor());
}
@ -646,25 +642,18 @@ public class DefaultTransportApiService implements TransportApiService {
}
}
private ListenableFuture<TransportApiResponseMsg> handle(TransportProtos.GetAllMainQueueRoutingInfoRequestMsg requestMsg) {
return queuesToTransportApiResponseMsg(queueService.findAllMainQueues());
}
private ListenableFuture<TransportApiResponseMsg> handle(TransportProtos.GetAllQueueRoutingInfoRequestMsg requestMsg) {
return queuesToTransportApiResponseMsg(queueService.findAllQueues());
}
private ListenableFuture<TransportApiResponseMsg> handle(TransportProtos.GetTenantQueueRoutingInfoRequestMsg requestMsg) {
TenantId tenantId = new TenantId(new UUID(requestMsg.getTenantIdMSB(), requestMsg.getTenantIdLSB()));
return queuesToTransportApiResponseMsg(queueService.findQueuesByTenantId(tenantId));
}
private ListenableFuture<TransportApiResponseMsg> queuesToTransportApiResponseMsg(List<Queue> queues) {
return Futures.immediateFuture(TransportApiResponseMsg.newBuilder()
.addAllGetQueueRoutingInfoResponseMsgs(queues.stream()
.map(queue -> TransportProtos.GetQueueRoutingInfoResponseMsg.newBuilder()
.setTenantIdMSB(queue.getTenantId().getId().getMostSignificantBits())
.setTenantIdLSB(queue.getTenantId().getId().getLeastSignificantBits())
.setQueueIdMSB(queue.getId().getId().getMostSignificantBits())
.setQueueIdLSB(queue.getId().getId().getLeastSignificantBits())
.setQueueName(queue.getName())
.setQueueTopic(queue.getTopic())
.setPartitions(queue.getPartitions())

32
common/cluster-api/src/main/proto/queue.proto

@ -197,37 +197,35 @@ message GetEntityProfileRequestMsg {
int64 entityIdLSB = 3;
}
message GetAllMainQueueRoutingInfoRequestMsg {
}
message GetAllQueueRoutingInfoRequestMsg {
}
message GetTenantQueueRoutingInfoRequestMsg {
int64 tenantIdMSB = 1;
int64 tenantIdLSB = 2;
}
message GetQueueRoutingInfoResponseMsg {
int64 tenantIdMSB = 1;
int64 tenantIdLSB = 2;
string queueName = 3;
string queueTopic = 4;
int32 partitions = 5;
int64 queueIdMSB = 3;
int64 queueIdLSB = 4;
string queueName = 5;
string queueTopic = 6;
int32 partitions = 7;
}
message QueueUpdateMsg {
int64 tenantIdMSB = 1;
int64 tenantIdLSB = 2;
string queueName = 3;
string queueTopic = 4;
int32 partitions = 5;
int64 queueIdMSB = 3;
int64 queueIdLSB = 4;
string queueName = 5;
string queueTopic = 6;
int32 partitions = 7;
}
message QueueDeleteMsg {
int64 tenantIdMSB = 1;
int64 tenantIdLSB = 2;
string queueName = 3;
int64 queueIdMSB = 3;
int64 queueIdLSB = 4;
string queueName = 5;
}
message LwM2MRegistrationRequestMsg {
@ -706,9 +704,7 @@ message TransportApiRequestMsg {
GetSnmpDevicesRequestMsg snmpDevicesRequestMsg = 11;
GetDeviceRequestMsg deviceRequestMsg = 12;
GetDeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 13;
GetAllMainQueueRoutingInfoRequestMsg GetAllMainQueueRoutingInfoRequestMsg = 14;
GetTenantQueueRoutingInfoRequestMsg getTenantQueueRoutingInfoRequestMsg = 15;
GetAllQueueRoutingInfoRequestMsg getAllQueueRoutingInfoRequestMsg = 16;
GetAllQueueRoutingInfoRequestMsg getAllQueueRoutingInfoRequestMsg = 14;
}
/* Response from ThingsBoard Core Service to Transport Service */

5
common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java

@ -27,6 +27,7 @@ import org.thingsboard.server.common.data.device.profile.DeviceProfileData;
import org.thingsboard.server.common.data.id.DashboardId;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.id.OtaPackageId;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.validation.Length;
@ -76,7 +77,7 @@ public class DeviceProfile extends SearchTextBased<DeviceProfileId> implements H
@ApiModelProperty(position = 8, value = "Reference to the rule engine queue. " +
"If present, the specified queue will be used to store all unprocessed messages related to device, including telemetry, attribute updates, etc. " +
"Otherwise, the 'Main' queue will be used to store those messages.")
private String defaultQueueName;
private QueueId defaultQueueId;
@Valid
private transient DeviceProfileData profileData;
@JsonIgnore
@ -107,7 +108,7 @@ public class DeviceProfile extends SearchTextBased<DeviceProfileId> implements H
this.isDefault = deviceProfile.isDefault();
this.defaultRuleChainId = deviceProfile.getDefaultRuleChainId();
this.defaultDashboardId = deviceProfile.getDefaultDashboardId();
this.defaultQueueName = deviceProfile.getDefaultQueueName();
this.defaultQueueId = deviceProfile.getDefaultQueueId();
this.setProfileData(deviceProfile.getProfileData());
this.provisionDeviceKey = deviceProfile.getProvisionDeviceKey();
this.firmwareId = deviceProfile.getFirmwareId();

2
common/data/src/main/java/org/thingsboard/server/common/data/id/QueueId.java

@ -16,7 +16,6 @@
package org.thingsboard.server.common.data.id;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import org.thingsboard.server.common.data.EntityType;
@ -35,7 +34,6 @@ public class QueueId extends UUIDBased implements EntityId {
return new QueueId(UUID.fromString(queueId));
}
@JsonIgnore
@Override
public EntityType getEntityType() {
return EntityType.QUEUE;

63
common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java

@ -27,15 +27,14 @@ import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityIdFactory;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.RuleNodeId;
import org.thingsboard.server.common.msg.gen.MsgProtos;
import org.thingsboard.server.common.msg.queue.ServiceQueue;
import org.thingsboard.server.common.msg.queue.TbMsgCallback;
import java.io.Serializable;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Created by ashvayka on 13.01.18.
@ -44,7 +43,7 @@ import java.util.concurrent.atomic.AtomicInteger;
@Slf4j
public final class TbMsg implements Serializable {
private final String queueName;
private final QueueId queueId;
private final UUID id;
private final long ts;
private final String type;
@ -68,12 +67,12 @@ public final class TbMsg implements Serializable {
return ctx.getAndIncrementRuleNodeCounter();
}
public static TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
return newMsg(queueName, type, originator, null, metaData, data, ruleChainId, ruleNodeId);
public static TbMsg newMsg(QueueId queueId, String type, EntityId originator, TbMsgMetaData metaData, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
return newMsg(queueId, type, originator, null, metaData, data, ruleChainId, ruleNodeId);
}
public static TbMsg newMsg(String queueName, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
return new TbMsg(queueName, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
public static TbMsg newMsg(QueueId queueId, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
return new TbMsg(queueId, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
metaData.copy(), TbMsgDataType.JSON, data, ruleChainId, ruleNodeId, null, TbMsgCallback.EMPTY);
}
@ -82,23 +81,23 @@ public final class TbMsg implements Serializable {
}
public static TbMsg newMsg(String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
return new TbMsg(null, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
metaData.copy(), TbMsgDataType.JSON, data, null, null, null, TbMsgCallback.EMPTY);
}
// REALLY NEW MSG
public static TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data) {
return newMsg(queueName, type, originator, null, metaData, data);
public static TbMsg newMsg(QueueId queueId, String type, EntityId originator, TbMsgMetaData metaData, String data) {
return newMsg(queueId, type, originator, null, metaData, data);
}
public static TbMsg newMsg(String queueName, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
return new TbMsg(queueName, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
public static TbMsg newMsg(QueueId queueId, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
return new TbMsg(queueId, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
metaData.copy(), TbMsgDataType.JSON, data, null, null, null, TbMsgCallback.EMPTY);
}
public static TbMsg newMsg(String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data) {
return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
return new TbMsg(null, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
metaData.copy(), dataType, data, null, null, null, TbMsgCallback.EMPTY);
}
@ -109,50 +108,50 @@ public final class TbMsg implements Serializable {
// For Tests only
public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null,
return new TbMsg(null, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null,
metaData.copy(), dataType, data, ruleChainId, ruleNodeId, null, TbMsgCallback.EMPTY);
}
public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data, TbMsgCallback callback) {
return new TbMsg(ServiceQueue.MAIN, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null,
return new TbMsg(null, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null,
metaData.copy(), TbMsgDataType.JSON, data, null, null, null, callback);
}
public static TbMsg transformMsg(TbMsg tbMsg, String type, EntityId originator, TbMsgMetaData metaData, String data) {
return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, type, originator, tbMsg.customerId, metaData.copy(), tbMsg.dataType,
return new TbMsg(tbMsg.queueId, tbMsg.id, tbMsg.ts, type, originator, tbMsg.customerId, metaData.copy(), tbMsg.dataType,
data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.callback);
}
public static TbMsg transformMsg(TbMsg tbMsg, CustomerId customerId) {
return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, customerId, tbMsg.metaData, tbMsg.dataType,
return new TbMsg(tbMsg.queueId, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, customerId, tbMsg.metaData, tbMsg.dataType,
tbMsg.data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.getCallback());
}
public static TbMsg transformMsg(TbMsg tbMsg, RuleChainId ruleChainId) {
return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
return new TbMsg(tbMsg.queueId, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
tbMsg.data, ruleChainId, null, tbMsg.ctx.copy(), tbMsg.getCallback());
}
public static TbMsg transformMsg(TbMsg tbMsg, String queueName) {
return new TbMsg(queueName, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
public static TbMsg transformMsg(TbMsg tbMsg, QueueId queueId) {
return new TbMsg(queueId, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
tbMsg.data, tbMsg.getRuleChainId(), null, tbMsg.ctx.copy(), tbMsg.getCallback());
}
public static TbMsg transformMsg(TbMsg tbMsg, RuleChainId ruleChainId, String queueName) {
return new TbMsg(queueName, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
public static TbMsg transformMsg(TbMsg tbMsg, RuleChainId ruleChainId, QueueId queueId) {
return new TbMsg(queueId, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
tbMsg.data, ruleChainId, null, tbMsg.ctx.copy(), tbMsg.getCallback());
}
//used for enqueueForTellNext
public static TbMsg newMsg(TbMsg tbMsg, String queueName, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
return new TbMsg(queueName, UUID.randomUUID(), tbMsg.getTs(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.customerId, tbMsg.getMetaData().copy(),
public static TbMsg newMsg(TbMsg tbMsg, QueueId queueId, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
return new TbMsg(queueId, UUID.randomUUID(), tbMsg.getTs(), tbMsg.getType(), tbMsg.getOriginator(), tbMsg.customerId, tbMsg.getMetaData().copy(),
tbMsg.getDataType(), tbMsg.getData(), ruleChainId, ruleNodeId, tbMsg.ctx.copy(), TbMsgCallback.EMPTY);
}
private TbMsg(String queueName, UUID id, long ts, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data,
private TbMsg(QueueId queueId, UUID id, long ts, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data,
RuleChainId ruleChainId, RuleNodeId ruleNodeId, TbMsgProcessingCtx ctx, TbMsgCallback callback) {
this.id = id;
this.queueName = queueName != null ? queueName : ServiceQueue.MAIN;
this.queueId = queueId;
if (ts > 0) {
this.ts = ts;
} else {
@ -221,7 +220,7 @@ public final class TbMsg implements Serializable {
return builder.build().toByteArray();
}
public static TbMsg fromBytes(String queueName, byte[] data, TbMsgCallback callback) {
public static TbMsg fromBytes(QueueId queueId, byte[] data, TbMsgCallback callback) {
try {
MsgProtos.TbMsgProto proto = MsgProtos.TbMsgProto.parseFrom(data);
TbMsgMetaData metaData = new TbMsgMetaData(proto.getMetaData().getDataMap());
@ -248,7 +247,7 @@ public final class TbMsg implements Serializable {
}
TbMsgDataType dataType = TbMsgDataType.values()[proto.getDataType()];
return new TbMsg(queueName, UUID.fromString(proto.getId()), proto.getTs(), proto.getType(), entityId, customerId,
return new TbMsg(queueId, UUID.fromString(proto.getId()), proto.getTs(), proto.getType(), entityId, customerId,
metaData, dataType, proto.getData(), ruleChainId, ruleNodeId, ctx, callback);
} catch (InvalidProtocolBufferException e) {
throw new IllegalStateException("Could not parse protobuf for TbMsg", e);
@ -260,12 +259,12 @@ public final class TbMsg implements Serializable {
}
public TbMsg copyWithRuleChainId(RuleChainId ruleChainId, UUID msgId) {
return new TbMsg(this.queueName, msgId, this.ts, this.type, this.originator, this.customerId,
return new TbMsg(this.queueId, msgId, this.ts, this.type, this.originator, this.customerId,
this.metaData, this.dataType, this.data, ruleChainId, null, this.ctx, callback);
}
public TbMsg copyWithRuleNodeId(RuleChainId ruleChainId, RuleNodeId ruleNodeId, UUID msgId) {
return new TbMsg(this.queueName, msgId, this.ts, this.type, this.originator, this.customerId,
return new TbMsg(this.queueId, msgId, this.ts, this.type, this.originator, this.customerId,
this.metaData, this.dataType, this.data, ruleChainId, ruleNodeId, this.ctx, callback);
}
@ -278,10 +277,6 @@ public final class TbMsg implements Serializable {
}
}
public String getQueueName() {
return queueName != null ? queueName : ServiceQueue.MAIN;
}
public void pushToStack(RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
ctx.push(ruleChainId, ruleNodeId);
}

36
common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java

@ -22,6 +22,7 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.queue.ServiceQueue;
import org.thingsboard.server.common.msg.queue.ServiceQueueKey;
@ -62,6 +63,7 @@ public class HashPartitionService implements PartitionService {
private final TbServiceInfoProvider serviceInfoProvider;
private final TenantRoutingInfoService tenantRoutingInfoService;
private final QueueRoutingInfoService queueRoutingInfoService;
private final ConcurrentMap<QueueId, String> queueNames = new ConcurrentHashMap<>();
private final ConcurrentMap<TenantId, ConcurrentMap<ServiceQueue, String>> partitionTopicsMap = new ConcurrentHashMap<>();
private final ConcurrentMap<TenantId, ConcurrentMap<ServiceQueue, Integer>> partitionSizesMap = new ConcurrentHashMap<>();
private final ConcurrentMap<TenantId, TenantRoutingInfo> tenantRoutingInfoMap = new ConcurrentHashMap<>();
@ -91,27 +93,21 @@ public class HashPartitionService implements PartitionService {
}
private void partitionsInit() {
addPartitionSizeToMap(TenantId.SYS_TENANT_ID, new ServiceQueue(ServiceType.TB_CORE, "Main"), corePartitions);
addPartitionTopicToMap(TenantId.SYS_TENANT_ID, new ServiceQueue(ServiceType.TB_CORE, "Main"), coreTopic);
addPartitionSizeToMap(TenantId.SYS_TENANT_ID, new ServiceQueue(ServiceType.TB_CORE), corePartitions);
addPartitionTopicToMap(TenantId.SYS_TENANT_ID, new ServiceQueue(ServiceType.TB_CORE), coreTopic);
List<QueueRoutingInfo> queueRoutingInfoList;
String serviceType = serviceInfoProvider.getServiceType();
if ("tb-rule-engine".equals(serviceType)) {
queueRoutingInfoList = queueRoutingInfoService.getQueuesRoutingInfo(serviceInfoProvider.getIsolatedTenant().orElse(TenantId.SYS_TENANT_ID));
} else if ("monolith".equals(serviceType)) {
queueRoutingInfoList = queueRoutingInfoService.getAllQueuesRoutingInfo();
} else if ("tb-core".equals(serviceType)) {
queueRoutingInfoList = queueRoutingInfoService.getMainQueuesRoutingInfo();
} else {
if ("tb-transport".equals(serviceType)) {
//If transport started earlier than tb-core
int getQueuesRetries = 10;
while (true) {
if (getQueuesRetries > 0) {
log.info("Try to get queue routing info.");
try {
queueRoutingInfoList = queueRoutingInfoService.getMainQueuesRoutingInfo();
queueRoutingInfoList = queueRoutingInfoService.getAllQueuesRoutingInfo();
break;
} catch (Exception e) {
log.info("Failed to get queues routing info!");
@ -126,11 +122,14 @@ public class HashPartitionService implements PartitionService {
throw new RuntimeException("Failed to await queues routing info!");
}
}
} else {
queueRoutingInfoList = queueRoutingInfoService.getAllQueuesRoutingInfo();
}
queueRoutingInfoList.forEach(queue -> {
addPartitionTopicToMap(queue.getTenantId(), new ServiceQueue(ServiceType.TB_RULE_ENGINE, queue.getQueueName()), queue.getQueueTopic());
addPartitionSizeToMap(queue.getTenantId(), new ServiceQueue(ServiceType.TB_RULE_ENGINE, queue.getQueueName()), queue.getPartitions());
queueNames.put(queue.getQueueId(), queue.getQueueName());
});
}
@ -139,6 +138,8 @@ public class HashPartitionService implements PartitionService {
TenantId tenantId = new TenantId(new UUID(queueUpdateMsg.getTenantIdMSB(), queueUpdateMsg.getTenantIdLSB()));
addPartitionTopicToMap(tenantId, new ServiceQueue(ServiceType.TB_RULE_ENGINE, queueUpdateMsg.getQueueName()), queueUpdateMsg.getQueueTopic());
addPartitionSizeToMap(tenantId, new ServiceQueue(ServiceType.TB_RULE_ENGINE, queueUpdateMsg.getQueueName()), queueUpdateMsg.getPartitions());
queueNames.putIfAbsent(new QueueId(new UUID(queueUpdateMsg.getQueueIdMSB(), queueUpdateMsg.getQueueIdLSB())), queueUpdateMsg.getQueueName());
tpiCache.clear();
}
@Override
@ -148,6 +149,8 @@ public class HashPartitionService implements PartitionService {
partitionTopicsMap.get(tenantId).remove(serviceQueue);
partitionSizesMap.get(tenantId).remove(serviceQueue);
myPartitions.remove(new ServiceQueueKey(serviceQueue, tenantId));
queueNames.remove(new QueueId(new UUID(queueDeleteMsg.getQueueIdMSB(), queueDeleteMsg.getQueueIdLSB())));
tpiCache.clear();
}
private void addPartitionSizeToMap(TenantId tenantId, ServiceQueue serviceQueue, int partitions) {
@ -164,13 +167,14 @@ public class HashPartitionService implements PartitionService {
}
@Override
public TopicPartitionInfo resolve(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId) {
ServiceQueue serviceQueue = new ServiceQueue(serviceType, queueName);
ConcurrentMap<ServiceQueue, String> queues = partitionTopicsMap.get(getSystemIsolatedTenantId(serviceType, tenantId));
if (!queues.containsKey(serviceQueue)) {
serviceQueue = new ServiceQueue(serviceType);
public TopicPartitionInfo resolve(ServiceType serviceType, QueueId queueId, TenantId tenantId, EntityId entityId) {
String queueName;
if (queueId == null) {
queueName = ServiceQueue.MAIN;
} else {
queueName = queueNames.get(queueId);
}
ServiceQueue serviceQueue = new ServiceQueue(serviceType, queueName);
return resolve(serviceQueue, tenantId, entityId);
}

3
common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java

@ -16,6 +16,7 @@
package org.thingsboard.server.queue.discovery;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
@ -33,7 +34,7 @@ public interface PartitionService {
TopicPartitionInfo resolve(ServiceType serviceType, TenantId tenantId, EntityId entityId);
TopicPartitionInfo resolve(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId);
TopicPartitionInfo resolve(ServiceType serviceType, QueueId queueId, TenantId tenantId, EntityId entityId);
/**
* Received from the Discovery service when network topology is changed.

11
common/queue/src/main/java/org/thingsboard/server/queue/discovery/QueueRoutingInfo.java

@ -16,6 +16,7 @@
package org.thingsboard.server.queue.discovery;
import lombok.Data;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.gen.transport.TransportProtos.GetQueueRoutingInfoResponseMsg;
@ -27,19 +28,14 @@ import java.util.UUID;
public class QueueRoutingInfo {
private final TenantId tenantId;
private final QueueId queueId;
private final String queueName;
private final String queueTopic;
private final int partitions;
public QueueRoutingInfo(TenantId tenantId, String queueName, String queueTopic, int partitions) {
this.tenantId = tenantId;
this.queueName = queueName;
this.queueTopic = queueTopic;
this.partitions = partitions;
}
public QueueRoutingInfo(Queue queue) {
this.tenantId = queue.getTenantId();
this.queueId = queue.getId();
this.queueName = queue.getName();
this.queueTopic = queue.getTopic();
this.partitions = queue.getPartitions();
@ -47,6 +43,7 @@ public class QueueRoutingInfo {
public QueueRoutingInfo(GetQueueRoutingInfoResponseMsg routingInfo) {
this.tenantId = new TenantId(new UUID(routingInfo.getTenantIdMSB(), routingInfo.getTenantIdLSB()));
this.queueId = new QueueId(new UUID(routingInfo.getQueueIdMSB(), routingInfo.getQueueIdLSB()));
this.queueName = routingInfo.getQueueName();
this.queueTopic = routingInfo.getQueueTopic();
this.partitions = routingInfo.getPartitions();

5
common/queue/src/main/java/org/thingsboard/server/queue/discovery/QueueRoutingInfoService.java

@ -15,15 +15,10 @@
*/
package org.thingsboard.server.queue.discovery;
import org.thingsboard.server.common.data.id.TenantId;
import java.util.List;
public interface QueueRoutingInfoService {
List<QueueRoutingInfo> getAllQueuesRoutingInfo();
List<QueueRoutingInfo> getMainQueuesRoutingInfo();
List<QueueRoutingInfo> getQueuesRoutingInfo(TenantId tenantId);
}

4
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java

@ -68,10 +68,6 @@ public interface TransportService {
GetEntityProfileResponseMsg getEntityProfile(GetEntityProfileRequestMsg msg);
List<TransportProtos.GetQueueRoutingInfoResponseMsg> getQueueRoutingInfo(TransportProtos.GetAllMainQueueRoutingInfoRequestMsg msg);
List<TransportProtos.GetQueueRoutingInfoResponseMsg> getQueueRoutingInfo(TransportProtos.GetTenantQueueRoutingInfoRequestMsg msg);
List<TransportProtos.GetQueueRoutingInfoResponseMsg> getQueueRoutingInfo(TransportProtos.GetAllQueueRoutingInfoRequestMsg msg);
GetResourceResponseMsg getResource(GetResourceRequestMsg msg);

37
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java

@ -44,13 +44,13 @@ import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.TenantProfileId;
import org.thingsboard.server.common.data.rpc.RpcStatus;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.common.msg.queue.ServiceQueue;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.msg.session.SessionMsgType;
@ -322,30 +322,6 @@ public class DefaultTransportService implements TransportService {
}
}
@Override
public List<TransportProtos.GetQueueRoutingInfoResponseMsg> getQueueRoutingInfo(TransportProtos.GetAllMainQueueRoutingInfoRequestMsg msg) {
TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg> protoMsg =
new TbProtoQueueMsg<>(UUID.randomUUID(), TransportProtos.TransportApiRequestMsg.newBuilder().setGetAllMainQueueRoutingInfoRequestMsg(msg).build());
try {
TbProtoQueueMsg<TransportApiResponseMsg> response = transportApiRequestTemplate.send(protoMsg).get();
return response.getValue().getGetQueueRoutingInfoResponseMsgsList();
} catch (InterruptedException | ExecutionException e) {
throw new RuntimeException(e);
}
}
@Override
public List<TransportProtos.GetQueueRoutingInfoResponseMsg> getQueueRoutingInfo(TransportProtos.GetTenantQueueRoutingInfoRequestMsg msg) {
TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg> protoMsg =
new TbProtoQueueMsg<>(UUID.randomUUID(), TransportProtos.TransportApiRequestMsg.newBuilder().setGetTenantQueueRoutingInfoRequestMsg(msg).build());
try {
TbProtoQueueMsg<TransportApiResponseMsg> response = transportApiRequestTemplate.send(protoMsg).get();
return response.getValue().getGetQueueRoutingInfoResponseMsgsList();
} catch (InterruptedException | ExecutionException e) {
throw new RuntimeException(e);
}
}
@Override
public TransportProtos.GetResourceResponseMsg getResource(TransportProtos.GetResourceRequestMsg msg) {
TbProtoQueueMsg<TransportProtos.TransportApiRequestMsg> protoMsg =
@ -1112,7 +1088,7 @@ public class DefaultTransportService implements TransportService {
}
private void sendToRuleEngine(TenantId tenantId, TbMsg tbMsg, TbQueueCallback callback) {
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), tenantId, tbMsg.getOriginator());
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueId(), tenantId, tbMsg.getOriginator());
if (log.isTraceEnabled()) {
log.trace("[{}][{}] Pushing to topic {} message {}", tenantId, tbMsg.getOriginator(), tpi.getFullTopicName(), tbMsg);
}
@ -1129,19 +1105,18 @@ public class DefaultTransportService implements TransportService {
DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(sessionInfo.getDeviceProfileIdMSB(), sessionInfo.getDeviceProfileIdLSB()));
DeviceProfile deviceProfile = deviceProfileCache.get(deviceProfileId);
RuleChainId ruleChainId;
String queueName;
QueueId queueId;
if (deviceProfile == null) {
log.warn("[{}] Device profile is null!", deviceProfileId);
ruleChainId = null;
queueName = ServiceQueue.MAIN;
queueId = null;
} else {
ruleChainId = deviceProfile.getDefaultRuleChainId();
String defaultQueueName = deviceProfile.getDefaultQueueName();
queueName = defaultQueueName != null ? defaultQueueName : ServiceQueue.MAIN;
queueId = deviceProfile.getDefaultQueueId();
}
TbMsg tbMsg = TbMsg.newMsg(queueName, sessionMsgType.name(), deviceId, customerId, metaData, gson.toJson(json), ruleChainId, null);
TbMsg tbMsg = TbMsg.newMsg(queueId, sessionMsgType.name(), deviceId, customerId, metaData, gson.toJson(json), ruleChainId, null);
sendToRuleEngine(tenantId, tbMsg, callback);
}

15
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/TransportQueueRoutingInfoService.java

@ -20,11 +20,8 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.transport.TransportService;
import org.thingsboard.server.gen.transport.TransportProtos.GetAllQueueRoutingInfoRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.GetAllMainQueueRoutingInfoRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.GetTenantQueueRoutingInfoRequestMsg;
import org.thingsboard.server.queue.discovery.QueueRoutingInfo;
import org.thingsboard.server.queue.discovery.QueueRoutingInfoService;
@ -45,16 +42,4 @@ public class TransportQueueRoutingInfoService implements QueueRoutingInfoService
GetAllQueueRoutingInfoRequestMsg msg = GetAllQueueRoutingInfoRequestMsg.newBuilder().build();
return transportService.getQueueRoutingInfo(msg).stream().map(QueueRoutingInfo::new).collect(Collectors.toList());
}
@Override
public List<QueueRoutingInfo> getMainQueuesRoutingInfo() {
GetAllMainQueueRoutingInfoRequestMsg msg = GetAllMainQueueRoutingInfoRequestMsg.newBuilder().build();
return transportService.getQueueRoutingInfo(msg).stream().map(QueueRoutingInfo::new).collect(Collectors.toList());
}
@Override
public List<QueueRoutingInfo> getQueuesRoutingInfo(TenantId tenantId) {
GetTenantQueueRoutingInfoRequestMsg msg = GetTenantQueueRoutingInfoRequestMsg.newBuilder().build();
return transportService.getQueueRoutingInfo(msg).stream().map(QueueRoutingInfo::new).collect(Collectors.toList());
}
}

12
dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java

@ -114,7 +114,7 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D
return "[Transport Configuration] invalid " + schemaName + " provided!";
}
@Autowired(required = false)
@Autowired
private QueueService queueService;
@Autowired
@ -386,13 +386,9 @@ 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) {
//TODO: Yevhen replace queue name to queue id
if (!queueService.findQueuesByTenantId(tenantId)
.stream()
.map(Queue::getName)
.collect(Collectors.toSet())
.contains(deviceProfile.getDefaultQueueName())) {
if (deviceProfile.getDefaultQueueId() != null) {
Queue queue = queueService.findQueueById(tenantId, deviceProfile.getDefaultQueueId());
if (queue == null) {
throw new DataValidationException("Device profile is referencing to non-existent queue!");
}
}

2
dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java

@ -180,7 +180,7 @@ public class ModelConstants {
public static final String DEVICE_PROFILE_IS_DEFAULT_PROPERTY = "is_default";
public static final String DEVICE_PROFILE_DEFAULT_RULE_CHAIN_ID_PROPERTY = "default_rule_chain_id";
public static final String DEVICE_PROFILE_DEFAULT_DASHBOARD_ID_PROPERTY = "default_dashboard_id";
public static final String DEVICE_PROFILE_DEFAULT_QUEUE_NAME_PROPERTY = "default_queue_name";
public static final String DEVICE_PROFILE_DEFAULT_QUEUE_ID_PROPERTY = "default_queue_id";
public static final String DEVICE_PROFILE_PROVISION_DEVICE_KEY = "provision_device_key";
public static final String DEVICE_PROFILE_FIRMWARE_ID_PROPERTY = "firmware_id";
public static final String DEVICE_PROFILE_SOFTWARE_ID_PROPERTY = "software_id";

13
dao/src/main/java/org/thingsboard/server/dao/model/sql/DeviceProfileEntity.java

@ -30,6 +30,7 @@ import org.thingsboard.server.common.data.device.profile.DeviceProfileData;
import org.thingsboard.server.common.data.id.DashboardId;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.id.OtaPackageId;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.dao.model.BaseSqlEntity;
@ -87,8 +88,8 @@ public final class DeviceProfileEntity extends BaseSqlEntity<DeviceProfile> impl
@Column(name = ModelConstants.DEVICE_PROFILE_DEFAULT_DASHBOARD_ID_PROPERTY)
private UUID defaultDashboardId;
@Column(name = ModelConstants.DEVICE_PROFILE_DEFAULT_QUEUE_NAME_PROPERTY)
private String defaultQueueName;
@Column(name = ModelConstants.DEVICE_PROFILE_DEFAULT_QUEUE_ID_PROPERTY)
private UUID defaultQueueId;
@Type(type = "jsonb")
@Column(name = ModelConstants.DEVICE_PROFILE_PROFILE_DATA_PROPERTY, columnDefinition = "jsonb")
@ -129,7 +130,9 @@ public final class DeviceProfileEntity extends BaseSqlEntity<DeviceProfile> impl
if (deviceProfile.getDefaultDashboardId() != null) {
this.defaultDashboardId = deviceProfile.getDefaultDashboardId().getId();
}
this.defaultQueueName = deviceProfile.getDefaultQueueName();
if (deviceProfile.getDefaultQueueId() != null) {
this.defaultQueueId = deviceProfile.getDefaultQueueId().getId();
}
this.provisionDeviceKey = deviceProfile.getProvisionDeviceKey();
if (deviceProfile.getFirmwareId() != null) {
this.firmwareId = deviceProfile.getFirmwareId().getId();
@ -174,7 +177,9 @@ public final class DeviceProfileEntity extends BaseSqlEntity<DeviceProfile> impl
if (defaultDashboardId != null) {
deviceProfile.setDefaultDashboardId(new DashboardId(defaultDashboardId));
}
deviceProfile.setDefaultQueueName(defaultQueueName);
if (defaultQueueId != null) {
deviceProfile.setDefaultQueueId(new QueueId(defaultQueueId));
}
deviceProfile.setProvisionDeviceKey(provisionDeviceKey);
if (firmwareId != null) {

31
dao/src/main/resources/sql/schema-entities.sql

@ -216,6 +216,20 @@ CREATE TABLE IF NOT EXISTS ota_package (
-- CONSTRAINT fk_device_profile_firmware FOREIGN KEY (device_profile_id) REFERENCES device_profile(id) ON DELETE CASCADE
);
CREATE TABLE IF NOT EXISTS queue(
id uuid NOT NULL CONSTRAINT queue_pkey PRIMARY KEY,
created_time bigint NOT NULL,
tenant_id uuid,
name varchar(255),
topic varchar(255),
poll_interval int,
partitions int,
consumer_per_partition boolean,
pack_processing_timeout bigint,
submit_strategy varchar(255),
processing_strategy varchar(255)
);
CREATE TABLE IF NOT EXISTS device_profile (
id uuid NOT NULL CONSTRAINT device_profile_pkey PRIMARY KEY,
created_time bigint NOT NULL,
@ -233,12 +247,13 @@ CREATE TABLE IF NOT EXISTS device_profile (
software_id uuid,
default_rule_chain_id uuid,
default_dashboard_id uuid,
default_queue_name varchar(255),
default_queue_id uuid,
provision_device_key varchar,
CONSTRAINT device_profile_name_unq_key UNIQUE (tenant_id, name),
CONSTRAINT device_provision_key_unq_key UNIQUE (provision_device_key),
CONSTRAINT fk_default_rule_chain_device_profile FOREIGN KEY (default_rule_chain_id) REFERENCES rule_chain(id),
CONSTRAINT fk_default_dashboard_device_profile FOREIGN KEY (default_dashboard_id) REFERENCES dashboard(id),
CONSTRAINT fk_default_queue_device_profile FOREIGN KEY (default_queue_id) REFERENCES queue(id),
CONSTRAINT fk_firmware_device_profile FOREIGN KEY (firmware_id) REFERENCES ota_package(id),
CONSTRAINT fk_software_device_profile FOREIGN KEY (software_id) REFERENCES ota_package(id)
);
@ -634,20 +649,6 @@ CREATE TABLE IF NOT EXISTS rpc (
status varchar(255) NOT NULL
);
CREATE TABLE IF NOT EXISTS queue(
id uuid NOT NULL CONSTRAINT queue_pkey PRIMARY KEY,
created_time bigint NOT NULL,
tenant_id uuid,
name varchar(255),
topic varchar(255),
poll_interval int,
partitions int,
consumer_per_partition boolean,
pack_processing_timeout bigint,
submit_strategy varchar(255),
processing_strategy varchar(255)
);
CREATE OR REPLACE PROCEDURE cleanup_events_by_ttl(IN ttl bigint, IN debug_ttl bigint, INOUT deleted bigint)
LANGUAGE plpgsql AS
$$

6
dao/src/test/resources/sql/system-data.sql

@ -43,3 +43,9 @@ VALUES ( '6eaaefa6-4612-11e7-a919-92ebcb67fe33', 1592576748000, 'mail', '{
"username": "",
"password": ""
}' );
INSERT INTO queue ( id, created_time, tenant_id, name, topic, poll_interval, partitions, consumer_per_partition, pack_processing_timeout, submit_strategy, processing_strategy )
VALUES ( '6eaaefa6-4612-11e7-a919-92ebcb67fe33', 1592576748000 ,'13814000-1dd2-11b2-8080-808080808080', 'Main' ,'tb_rule_engine.main', 25, 10, true, 2000,
'{"type": "BURST", "batchSize": 1000}',
'{"type": "SKIP_ALL_FAILURES", "retries": 3, "failurePercentage": 0.0, "pauseBetweenRetries": 3, "maxPauseBetweenRetries": 3}'
);

14
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java

@ -29,6 +29,7 @@ import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.RuleNodeId;
import org.thingsboard.server.common.data.id.TenantId;
@ -50,6 +51,7 @@ import org.thingsboard.server.dao.entityview.EntityViewService;
import org.thingsboard.server.dao.nosql.CassandraStatementTask;
import org.thingsboard.server.dao.nosql.TbResultSetFuture;
import org.thingsboard.server.dao.ota.OtaPackageService;
import org.thingsboard.server.dao.queue.QueueService;
import org.thingsboard.server.dao.relation.RelationService;
import org.thingsboard.server.dao.resource.ResourceService;
import org.thingsboard.server.dao.rule.RuleChainService;
@ -145,7 +147,7 @@ public interface TbContext {
*
* @param msg - message
*/
void enqueue(TbMsg msg, String queueName, Runnable onSuccess, Consumer<Throwable> onFailure);
void enqueue(TbMsg msg, QueueId queueId, Runnable onSuccess, Consumer<Throwable> onFailure);
void enqueueForTellFailure(TbMsg msg, String failureMessage);
@ -157,15 +159,15 @@ public interface TbContext {
void enqueueForTellNext(TbMsg msg, Set<String> relationTypes, Runnable onSuccess, Consumer<Throwable> onFailure);
void enqueueForTellNext(TbMsg msg, String queueName, String relationType, Runnable onSuccess, Consumer<Throwable> onFailure);
void enqueueForTellNext(TbMsg msg, QueueId queueId, String relationType, Runnable onSuccess, Consumer<Throwable> onFailure);
void enqueueForTellNext(TbMsg msg, String queueName, Set<String> relationTypes, Runnable onSuccess, Consumer<Throwable> onFailure);
void enqueueForTellNext(TbMsg msg, QueueId queueId, Set<String> relationTypes, Runnable onSuccess, Consumer<Throwable> onFailure);
void ack(TbMsg tbMsg);
TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data);
TbMsg newMsg(QueueId queueId, String type, EntityId originator, TbMsgMetaData metaData, String data);
TbMsg newMsg(String queueName, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data);
TbMsg newMsg(QueueId queueId, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data);
TbMsg transformMsg(TbMsg origMsg, String type, EntityId originator, TbMsgMetaData metaData, String data);
@ -236,6 +238,8 @@ public interface TbContext {
EdgeEventService getEdgeEventService();
QueueService getQueueService();
ListeningExecutor getMailExecutor();
ListeningExecutor getSmsExecutor();

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java

@ -136,7 +136,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode {
}
private void transformAndTellNext(TbContext ctx, TbMsg msg, EntityView entityView) {
ctx.enqueueForTellNext(ctx.newMsg(msg.getQueueName(), msg.getType(), entityView.getId(), msg.getCustomerId(), msg.getMetaData(), msg.getData()), SUCCESS);
ctx.enqueueForTellNext(ctx.newMsg(msg.getQueueId(), msg.getType(), entityView.getId(), msg.getCustomerId(), msg.getMetaData(), msg.getData()), SUCCESS);
}
private boolean attributeContainsInEntityView(String scope, String attrKey, EntityView entityView) {

4
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java

@ -78,7 +78,7 @@ public class TbMsgCountNode implements TbNode {
TbMsgMetaData metaData = new TbMsgMetaData();
metaData.putValue("delta", Long.toString(System.currentTimeMillis() - lastScheduledTs + delay));
TbMsg tbMsg = TbMsg.newMsg(msg.getQueueName(), SessionMsgType.POST_TELEMETRY_REQUEST.name(), ctx.getTenantId(), msg.getCustomerId(), metaData, gson.toJson(telemetryJson));
TbMsg tbMsg = TbMsg.newMsg(msg.getQueueId(), SessionMsgType.POST_TELEMETRY_REQUEST.name(), ctx.getTenantId(), msg.getCustomerId(), metaData, gson.toJson(telemetryJson));
ctx.enqueueForTellNext(tbMsg, SUCCESS);
scheduleTickMsg(ctx, tbMsg);
} else {
@ -94,7 +94,7 @@ public class TbMsgCountNode implements TbNode {
}
lastScheduledTs = lastScheduledTs + delay;
long curDelay = Math.max(0L, (lastScheduledTs - curTs));
TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_COUNT_NODE_MSG, ctx.getSelfId(), msg != null ? msg.getCustomerId() : null, new TbMsgMetaData(), "");
TbMsg tickMsg = ctx.newMsg(null, TB_MSG_COUNT_NODE_MSG, ctx.getSelfId(), msg != null ? msg.getCustomerId() : null, new TbMsgMetaData(), "");
nextTickId = tickMsg.getId();
ctx.tellSelf(tickMsg, curDelay);
}

6
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java

@ -135,7 +135,7 @@ public class TbMsgGeneratorNode implements TbNode {
}
lastScheduledTs = lastScheduledTs + delay;
long curDelay = Math.max(0L, (lastScheduledTs - curTs));
TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_GENERATOR_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), "");
TbMsg tickMsg = ctx.newMsg(null, TB_MSG_GENERATOR_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), "");
nextTickId = tickMsg.getId();
ctx.tellSelf(tickMsg, curDelay);
}
@ -143,14 +143,14 @@ public class TbMsgGeneratorNode implements TbNode {
private ListenableFuture<TbMsg> generate(TbContext ctx, TbMsg msg) {
log.trace("generate, config {}", config);
if (prevMsg == null) {
prevMsg = ctx.newMsg(ServiceQueue.MAIN, "", originatorId, msg.getCustomerId(), new TbMsgMetaData(), "{}");
prevMsg = ctx.newMsg(null, "", originatorId, msg.getCustomerId(), new TbMsgMetaData(), "{}");
}
if (initialized.get()) {
ctx.logJsEvalRequest();
return Futures.transformAsync(jsEngine.executeGenerateAsync(prevMsg), generated -> {
log.trace("generate process response, generated {}, config {}", generated, config);
ctx.logJsEvalResponse();
prevMsg = ctx.newMsg(ServiceQueue.MAIN, generated.getType(), originatorId, msg.getCustomerId(), generated.getMetaData(), generated.getData());
prevMsg = ctx.newMsg(null, generated.getType(), originatorId, msg.getCustomerId(), generated.getMetaData(), generated.getData());
return Futures.immediateFuture(prevMsg);
}, MoreExecutors.directExecutor()); //usually it runs on js-executor-remote-callback thread pool
}

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java

@ -73,7 +73,7 @@ public class TbMsgDelayNode implements TbNode {
} else {
if (pendingMsgs.size() < config.getMaxPendingMsgs()) {
pendingMsgs.put(msg.getId(), msg);
TbMsg tickMsg = ctx.newMsg(ServiceQueue.MAIN, TB_MSG_DELAY_NODE_MSG, ctx.getSelfId(), msg.getCustomerId(), new TbMsgMetaData(), msg.getId().toString());
TbMsg tickMsg = ctx.newMsg(null, TB_MSG_DELAY_NODE_MSG, ctx.getSelfId(), msg.getCustomerId(), new TbMsgMetaData(), msg.getId().toString());
ctx.tellSelf(tickMsg, getDelay(msg));
ctx.ack(msg);
} else {

12
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNode.java

@ -17,7 +17,6 @@ package org.thingsboard.rule.engine.flow;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.ScriptEngine;
import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNode;
import org.thingsboard.rule.engine.api.TbNodeConfiguration;
@ -25,10 +24,9 @@ import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.TbRelationTypes;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.data.queue.Queue;
import org.thingsboard.server.common.msg.TbMsg;
import static org.thingsboard.common.util.DonAsynchron.withCallback;
@Slf4j
@RuleNode(
type = ComponentType.FLOW,
@ -46,11 +44,17 @@ public class TbCheckpointNode implements TbNode {
@Override
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
this.config = TbNodeUtils.convert(configuration, TbCheckpointNodeConfiguration.class);
if (config.getQueueId() == null) {
Queue foundQueue = ctx.getQueueService().findQueueByTenantIdAndName(ctx.getTenantId(), config.getQueueName());
if (foundQueue != null) {
config.setQueueId(foundQueue.getId());
}
}
}
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
ctx.enqueueForTellNext(msg, config.getQueueName(), TbRelationTypes.SUCCESS, () -> ctx.ack(msg), error -> ctx.tellFailure(msg, error));
ctx.enqueueForTellNext(msg, config.getQueueId(), TbRelationTypes.SUCCESS, () -> ctx.ack(msg), error -> ctx.tellFailure(msg, error));
}
@Override

3
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNodeConfiguration.java

@ -17,12 +17,15 @@ package org.thingsboard.rule.engine.flow;
import lombok.Data;
import org.thingsboard.rule.engine.api.NodeConfiguration;
import org.thingsboard.server.common.data.id.QueueId;
@Data
public class TbCheckpointNodeConfiguration implements NodeConfiguration<TbCheckpointNodeConfiguration> {
private String queueName;
private QueueId queueId;
@Override
public TbCheckpointNodeConfiguration defaultConfiguration() {
TbCheckpointNodeConfiguration configuration = new TbCheckpointNodeConfiguration();

8
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmState.java

@ -36,9 +36,9 @@ import org.thingsboard.server.common.data.device.profile.AlarmConditionSpecType;
import org.thingsboard.server.common.data.device.profile.DeviceProfileAlarm;
import org.thingsboard.server.common.data.id.DashboardId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.QueueId;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.common.msg.queue.ServiceQueue;
import org.thingsboard.server.dao.alarm.AlarmOperationResult;
import java.util.ArrayList;
@ -60,7 +60,7 @@ class AlarmState {
private volatile Alarm currentAlarm;
private volatile boolean initialFetchDone;
private volatile TbMsgMetaData lastMsgMetaData;
private volatile String lastMsgQueueName;
private volatile QueueId lastMsgQueueId;
private volatile DataSnapshot dataSnapshot;
private final DynamicPredicateValueCtx dynamicPredicateValueCtx;
@ -74,7 +74,7 @@ class AlarmState {
public boolean process(TbContext ctx, TbMsg msg, DataSnapshot data, SnapshotUpdate update) throws ExecutionException, InterruptedException {
initCurrentAlarm(ctx);
lastMsgMetaData = msg.getMetaData();
lastMsgQueueName = msg.getQueueName();
lastMsgQueueId = msg.getQueueId();
this.dataSnapshot = data;
try {
return createOrClearAlarms(ctx, msg, data, update, AlarmRuleState::eval);
@ -195,7 +195,7 @@ class AlarmState {
metaData.putValue(DataConstants.IS_CLEARED_ALARM, Boolean.TRUE.toString());
}
setAlarmConditionMetadata(ruleState, metaData);
TbMsg newMsg = ctx.newMsg(lastMsgQueueName != null ? lastMsgQueueName : ServiceQueue.MAIN, "ALARM",
TbMsg newMsg = ctx.newMsg(lastMsgQueueId != null ? lastMsgQueueId : null, "ALARM",
originator, msg != null ? msg.getCustomerId() : null, metaData, data);
ctx.enqueueForTellNext(newMsg, relationType);
}

4
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java

@ -116,10 +116,10 @@ public class TbSendRPCRequestNode implements TbNode {
ctx.getRpcService().sendRpcRequestToDevice(request, ruleEngineDeviceRpcResponse -> {
if (ruleEngineDeviceRpcResponse.getError().isEmpty()) {
TbMsg next = ctx.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getCustomerId(), msg.getMetaData(), ruleEngineDeviceRpcResponse.getResponse().orElse("{}"));
TbMsg next = ctx.newMsg(msg.getQueueId(), msg.getType(), msg.getOriginator(), msg.getCustomerId(), msg.getMetaData(), ruleEngineDeviceRpcResponse.getResponse().orElse("{}"));
ctx.enqueueForTellNext(next, TbRelationTypes.SUCCESS);
} else {
TbMsg next = ctx.newMsg(msg.getQueueName(), msg.getType(), msg.getOriginator(), msg.getCustomerId(), msg.getMetaData(), wrap("error", ruleEngineDeviceRpcResponse.getError().get().name()));
TbMsg next = ctx.newMsg(msg.getQueueId(), msg.getType(), msg.getOriginator(), msg.getCustomerId(), msg.getMetaData(), wrap("error", ruleEngineDeviceRpcResponse.getError().get().name()));
ctx.enqueueForTellFailure(next, ruleEngineDeviceRpcResponse.getError().get().name());
}
});

30
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java

@ -188,7 +188,7 @@ public class TbDeviceProfileNodeTest {
Mockito.when(alarmService.createOrUpdateAlarm(Mockito.any())).thenAnswer(AdditionalAnswers.returnsFirstArg());
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString())).thenReturn(theMsg);
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString())).thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
data.put("temperature", 42);
@ -200,7 +200,7 @@ public class TbDeviceProfileNodeTest {
verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any());
TbMsg theMsg2 = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "2");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString())).thenReturn(theMsg2);
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString())).thenReturn(theMsg2);
TbMsg msg2 = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, new TbMsgMetaData(),
@ -279,7 +279,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(attrListListenableFuture);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
@ -366,7 +366,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(attrListListenableFuture);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
@ -435,7 +435,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(listListenableFutureWithLess);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
@ -529,7 +529,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(listListenableFuture);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
@ -653,7 +653,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(listNoDurationAttribute);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
@ -762,7 +762,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(listListenableFuture);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
@ -878,7 +878,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(listNoDurationAttribute);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
@ -974,7 +974,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(listListenableFuture);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
@ -1072,7 +1072,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(listListenableFuture);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
@ -1153,7 +1153,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(optionalListenableFutureWithLess);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
@ -1227,7 +1227,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(optionalListenableFutureWithLess);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
@ -1311,7 +1311,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(optionalListenableFutureWithLess);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();
@ -1397,7 +1397,7 @@ public class TbDeviceProfileNodeTest {
.thenReturn(optionalListenableFutureWithLess);
TbMsg theMsg = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "");
Mockito.when(ctx.newMsg(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
Mockito.when(ctx.newMsg(Mockito.any(), Mockito.anyString(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyString()))
.thenReturn(theMsg);
ObjectNode data = mapper.createObjectNode();

15
ui-ngx/src/app/core/http/entity.service.ts

@ -83,15 +83,12 @@ import {
import { alarmFields } from '@shared/models/alarm.models';
import { OtaPackageService } from '@core/http/ota-package.service';
import { EdgeService } from '@core/http/edge.service';
import {
Edge,
EdgeEvent,
EdgeEventType,
bodyContentEdgeEventActionTypes
} from '@shared/models/edge.models';
import { bodyContentEdgeEventActionTypes, Edge, EdgeEvent, EdgeEventType } from '@shared/models/edge.models';
import { RuleChainMetaData, RuleChainType } from '@shared/models/rule-chain.models';
import { WidgetService } from '@core/http/widget.service';
import { DeviceProfileService } from '@core/http/device-profile.service';
import { QueueService } from "@core/http/queue.service";
import { ServiceType } from "@shared/models/queue.models";
@Injectable({
providedIn: 'root'
@ -115,7 +112,8 @@ export class EntityService {
private otaPackageService: OtaPackageService,
private widgetService: WidgetService,
private deviceProfileService: DeviceProfileService,
private utils: UtilsService
private utils: UtilsService,
private queueService: QueueService
) { }
private getEntityObservable(entityType: EntityType, entityId: string,
@ -156,6 +154,9 @@ export class EntityService {
case EntityType.OTA_PACKAGE:
observable = this.otaPackageService.getOtaPackageInfo(entityId, config);
break;
case EntityType.QUEUE:
observable = this.queueService.getQueueById(entityId, config);
break;
}
return observable;
}

4
ui-ngx/src/app/core/http/queue.service.ts

@ -36,8 +36,8 @@ export class QueueService {
defaultHttpOptionsFromConfig(config));
}
public getQueueById(queueId: string): Observable<QueueInfo> {
return this.http.get<QueueInfo>(`/api/tenant/queues/${queueId}`);
public getQueueById(queueId: string, config?: RequestConfig): Observable<QueueInfo> {
return this.http.get<QueueInfo>(`/api/tenant/queues/${queueId}`, defaultHttpOptionsFromConfig(config));
}
public getTenantQueuesByServiceType(pageLink: PageLink,

6
ui-ngx/src/app/modules/home/components/profile/add-device-profile-dialog.component.html

@ -54,10 +54,10 @@
formControlName="defaultDashboardId">
<div tb-hint>{{'device-profile.mobile-dashboard-hint' | translate}}</div>
</tb-dashboard-autocomplete>
<tb-queue-type-list
<tb-queue-autocomplete
[queueType]="serviceType"
formControlName="defaultQueueName">
</tb-queue-type-list>
formControlName="defaultQueueId">
</tb-queue-autocomplete>
<mat-form-field fxHide class="mat-block">
<mat-label translate>device-profile.type</mat-label>
<mat-select formControlName="type" required>

8
ui-ngx/src/app/modules/home/components/profile/add-device-profile-dialog.component.ts

@ -50,6 +50,7 @@ import { StepperSelectionEvent } from '@angular/cdk/stepper';
import { deepTrim } from '@core/utils';
import { ServiceType } from '@shared/models/queue.models';
import { DashboardId } from '@shared/models/id/dashboard-id';
import { QueueId } from "@shared/models/id/queue-id";
export interface AddDeviceProfileDialogData {
deviceProfileName: string;
@ -110,7 +111,7 @@ export class AddDeviceProfileDialogComponent extends
image: [null, []],
defaultRuleChainId: [null, []],
defaultDashboardId: [null, []],
defaultQueueName: ['', []],
defaultQueueId: [null, []],
description: ['', []]
}
);
@ -187,7 +188,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,
// defaultQueueId: this.deviceProfileDetailsFormGroup.get('defaultQueueId').value,
transportType: this.transportConfigFormGroup.get('transportType').value,
provisionType: deviceProvisionConfiguration.type,
provisionDeviceKey,
@ -205,6 +206,9 @@ export class AddDeviceProfileDialogComponent extends
if (this.deviceProfileDetailsFormGroup.get('defaultDashboardId').value) {
deviceProfile.defaultDashboardId = new DashboardId(this.deviceProfileDetailsFormGroup.get('defaultDashboardId').value);
}
if (this.deviceProfileDetailsFormGroup.get('defaultQueueId').value) {
deviceProfile.defaultQueueId = new QueueId(this.deviceProfileDetailsFormGroup.get('defaultQueueId').value);
}
this.deviceProfileService.saveDeviceProfile(deepTrim(deviceProfile)).subscribe(
(savedDeviceProfile) => {
this.dialogRef.close(savedDeviceProfile);

6
ui-ngx/src/app/modules/home/components/profile/device-profile.component.html

@ -73,10 +73,10 @@
formControlName="defaultDashboardId">
<div tb-hint>{{'device-profile.mobile-dashboard-hint' | translate}}</div>
</tb-dashboard-autocomplete>
<tb-queue-type-list
<tb-queue-autocomplete
[queueType]="serviceType"
formControlName="defaultQueueName">
</tb-queue-type-list>
formControlName="defaultQueueId">
</tb-queue-autocomplete>
<tb-ota-package-autocomplete
[useFullEntityId]="true"
[showDetailsPageLink]="true"

8
ui-ngx/src/app/modules/home/components/profile/device-profile.component.ts

@ -42,6 +42,7 @@ import { ServiceType } from '@shared/models/queue.models';
import { EntityId } from '@shared/models/id/entity-id';
import { OtaUpdateType } from '@shared/models/ota-package.models';
import { DashboardId } from '@shared/models/id/dashboard-id';
import { QueueId } from "@shared/models/id/queue-id";
@Component({
selector: 'tb-device-profile',
@ -117,7 +118,7 @@ export class DeviceProfileComponent extends EntityComponent<DeviceProfile> {
}),
defaultRuleChainId: [entity && entity.defaultRuleChainId ? entity.defaultRuleChainId.id : null, []],
defaultDashboardId: [entity && entity.defaultDashboardId ? entity.defaultDashboardId.id : null, []],
defaultQueueName: [entity ? entity.defaultQueueName : '', []],
defaultQueueId: [entity && entity.defaultQueueId ? entity.defaultQueueId.id : null, []],
firmwareId: [entity ? entity.firmwareId : null],
softwareId: [entity ? entity.softwareId : null],
description: [entity ? entity.description : '', []],
@ -197,7 +198,7 @@ export class DeviceProfileComponent extends EntityComponent<DeviceProfile> {
}}, {emitEvent: false});
this.entityForm.patchValue({defaultRuleChainId: entity.defaultRuleChainId ? entity.defaultRuleChainId.id : null}, {emitEvent: false});
this.entityForm.patchValue({defaultDashboardId: entity.defaultDashboardId ? entity.defaultDashboardId.id : null}, {emitEvent: false});
this.entityForm.patchValue({defaultQueueName: entity.defaultQueueName}, {emitEvent: false});
this.entityForm.patchValue({defaultQueueId: entity.defaultQueueId ? entity.defaultQueueId.id: null}, {emitEvent: false});
this.entityForm.patchValue({firmwareId: entity.firmwareId}, {emitEvent: false});
this.entityForm.patchValue({softwareId: entity.softwareId}, {emitEvent: false});
this.entityForm.patchValue({description: entity.description}, {emitEvent: false});
@ -210,6 +211,9 @@ export class DeviceProfileComponent extends EntityComponent<DeviceProfile> {
if (formValue.defaultDashboardId) {
formValue.defaultDashboardId = new DashboardId(formValue.defaultDashboardId);
}
if (formValue.defaultQueueId) {
formValue.defaultQueueId = new QueueId(formValue.defaultQueueId);
}
const deviceProvisionConfiguration: DeviceProvisionConfiguration = formValue.profileData.provisionConfiguration;
formValue.provisionType = deviceProvisionConfiguration.type;
formValue.provisionDeviceKey = deviceProvisionConfiguration.provisionDeviceKey;

6
ui-ngx/src/app/modules/home/components/wizard/device-wizard-dialog.component.html

@ -92,11 +92,11 @@
</tb-rule-chain-autocomplete>
</div>
<div fxLayout="column" fxLayoutAlign="flex-end start">
<tb-queue-type-list
<tb-queue-autocomplete
[ngClass]="{invisible: deviceWizardFormGroup.get('addProfileType').value !== 1}"
[queueType]="serviceType"
formControlName="defaultQueueName">
</tb-queue-type-list>
formControlName="defaultQueueId">
</tb-queue-autocomplete>
</div>
</div>
<div fxLayout="row" fxLayoutGap="16px" style="padding-bottom: 16px;">

2
ui-ngx/src/app/modules/home/components/wizard/device-wizard-dialog.component.scss

@ -64,7 +64,7 @@
}
}
tb-device-profile-autocomplete, tb-queue-type-list{
tb-device-profile-autocomplete, tb-queue-autocomplete{
.mat-form-field-wrapper{
width: 180px !important;
}

11
ui-ngx/src/app/modules/home/components/wizard/device-wizard-dialog.component.ts

@ -49,6 +49,7 @@ import { MediaBreakpoints } from '@shared/models/constants';
import { RuleChainId } from '@shared/models/id/rule-chain-id';
import { ServiceType } from '@shared/models/queue.models';
import { deepTrim } from '@core/utils';
import { QueueId } from "@shared/models/id/queue-id";
@Component({
selector: 'tb-device-wizard',
@ -113,7 +114,7 @@ export class DeviceWizardDialogComponent extends
deviceProfileId: [null, Validators.required],
newDeviceProfileTitle: [{value: null, disabled: true}],
defaultRuleChainId: [{value: null, disabled: true}],
defaultQueueName: [{value: null, disabled: true}],
defaultQueueId: [{value: null, disabled: true}],
description: ['']
}
);
@ -126,7 +127,7 @@ export class DeviceWizardDialogComponent extends
this.deviceWizardFormGroup.get('newDeviceProfileTitle').setValidators(null);
this.deviceWizardFormGroup.get('newDeviceProfileTitle').disable();
this.deviceWizardFormGroup.get('defaultRuleChainId').disable();
this.deviceWizardFormGroup.get('defaultQueueName').disable();
this.deviceWizardFormGroup.get('defaultQueueId').disable();
this.deviceWizardFormGroup.updateValueAndValidity();
this.createProfile = false;
} else {
@ -135,7 +136,7 @@ export class DeviceWizardDialogComponent extends
this.deviceWizardFormGroup.get('newDeviceProfileTitle').setValidators([Validators.required]);
this.deviceWizardFormGroup.get('newDeviceProfileTitle').enable();
this.deviceWizardFormGroup.get('defaultRuleChainId').enable();
this.deviceWizardFormGroup.get('defaultQueueName').enable();
this.deviceWizardFormGroup.get('defaultQueueId').enable();
this.deviceWizardFormGroup.updateValueAndValidity();
this.createProfile = true;
@ -297,7 +298,6 @@ export class DeviceWizardDialogComponent extends
const deviceProfile: DeviceProfile = {
name: this.deviceWizardFormGroup.get('newDeviceProfileTitle').value,
type: DeviceProfileType.DEFAULT,
defaultQueueName: this.deviceWizardFormGroup.get('defaultQueueName').value,
transportType: this.transportConfigFormGroup.get('transportType').value,
provisionType: deviceProvisionConfiguration.type,
provisionDeviceKey,
@ -311,6 +311,9 @@ export class DeviceWizardDialogComponent extends
if (this.deviceWizardFormGroup.get('defaultRuleChainId').value) {
deviceProfile.defaultRuleChainId = new RuleChainId(this.deviceWizardFormGroup.get('defaultRuleChainId').value);
}
if (this.deviceWizardFormGroup.get('defaultQueueId').value) {
deviceProfile.defaultQueueId = new QueueId(this.deviceWizardFormGroup.get('defaultQueueId').value);
}
return this.deviceProfileService.saveDeviceProfile(deepTrim(deviceProfile)).pipe(
tap((profile) => {
this.currentDeviceProfileTransportType = profile.transportType;

54
ui-ngx/src/app/shared/components/queue/queue-autocomplete.component.html

@ -0,0 +1,54 @@
<!--
Copyright © 2016-2022 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.
-->
<mat-form-field [formGroup]="selectQueueFormGroup" class="mat-block">
<input matInput type="text" placeholder="{{ 'queue.queue-name' | translate }}"
#queueInput
formControlName="queueId"
(focusin)="onFocus()"
[required]="required"
[matAutocomplete]="queueAutocomplete">
<button *ngIf="selectQueueFormGroup.get('queueId').value && !disabled"
type="button"
matSuffix mat-button mat-icon-button aria-label="Clear"
(click)="clear()">
<mat-icon class="material-icons">close</mat-icon>
</button>
<mat-autocomplete class="tb-autocomplete"
#queueAutocomplete="matAutocomplete"
[displayWith]="displayQueueFn">
<mat-option *ngFor="let queue of filteredQueues | async" [value]="queue">
<span [innerHTML]="queue.name | highlight:searchText"></span>
</mat-option>
<mat-option *ngIf="!(filteredQueues | async)?.length" [value]="null" class="tb-not-found">
<div class="tb-not-found-content" (click)="$event.stopPropagation()">
<div *ngIf="!textIsNotEmpty(searchText); else searchNotEmpty">
<span translate>device-profile.no-device-profiles-found</span>
</div>
<ng-template #searchNotEmpty>
<span>
{{ translate.get('queue.no-queues-matching',
{entity: truncate.transform(searchText, true, 6, &apos;...&apos;)}) | async }}
</span>
</ng-template>
</div>
</mat-option>
</mat-autocomplete>
<mat-error *ngIf="selectQueueFormGroup.get('queueId').hasError('required')">
{{ 'queue.queue-required' | translate }}
</mat-error>
</mat-form-field>

215
ui-ngx/src/app/shared/components/queue/queue-autocomplete.component.ts

@ -0,0 +1,215 @@
///
/// Copyright © 2016-2022 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.
///
import { Component, ElementRef, forwardRef, Input, OnInit, ViewChild } from '@angular/core';
import { ControlValueAccessor, FormBuilder, FormGroup, NG_VALUE_ACCESSOR } from '@angular/forms';
import { Observable, of } from 'rxjs';
import { catchError, debounceTime, distinctUntilChanged, map, share, switchMap, tap } from 'rxjs/operators';
import { Store } from '@ngrx/store';
import { AppState } from '@core/core.state';
import { TranslateService } from '@ngx-translate/core';
import { coerceBooleanProperty } from '@angular/cdk/coercion';
import { EntityId } from '@shared/models/id/entity-id';
import { EntityType } from '@shared/models/entity-type.models';
import { BaseData } from '@shared/models/base-data';
import { EntityService } from '@core/http/entity.service';
import { TruncatePipe } from '@shared/pipe/truncate.pipe';
import {QueueInfo, ServiceType} from "@shared/models/queue.models";
import { QueueService } from "@core/http/queue.service";
import { PageLink } from "@shared/models/page/page-link";
import { Direction } from "@shared/models/page/sort-order";
import { emptyPageData } from "@shared/models/page/page-data";
@Component({
selector: 'tb-queue-autocomplete',
templateUrl: './queue-autocomplete.component.html',
styleUrls: [],
providers: [{
provide: NG_VALUE_ACCESSOR,
useExisting: forwardRef(() => QueueAutocompleteComponent),
multi: true
}]
})
export class QueueAutocompleteComponent implements ControlValueAccessor, OnInit {
selectQueueFormGroup: FormGroup;
modelValue: string | null;
@Input()
labelText: string;
@Input()
requiredText: string;
private requiredValue: boolean;
get required(): boolean {
return this.requiredValue;
}
@Input()
set required(value: boolean) {
this.requiredValue = coerceBooleanProperty(value);
}
@Input()
queueType: ServiceType;
@Input()
disabled: boolean;
@ViewChild('queueInput', {static: true}) queueInput: ElementRef;
filteredQueues: Observable<Array<BaseData<EntityId>>>;
searchText = '';
private dirty = false;
private propagateChange = (v: any) => { };
constructor(private store: Store<AppState>,
public translate: TranslateService,
public truncate: TruncatePipe,
private entityService: EntityService,
private queueService: QueueService,
private fb: FormBuilder) {
this.selectQueueFormGroup = this.fb.group({
queueId: [null]
});
}
registerOnChange(fn: any): void {
this.propagateChange = fn;
}
registerOnTouched(fn: any): void {
}
ngOnInit() {
this.filteredQueues = this.selectQueueFormGroup.get('queueId').valueChanges
.pipe(
debounceTime(150),
tap(value => {
let modelValue;
if (typeof value === 'string' || !value) {
modelValue = null;
} else {
modelValue = value.id.id;
}
this.updateView(modelValue);
if (value === null) {
this.clear();
}
}),
map(value => value ? (typeof value === 'string' ? value : value.name) : ''),
distinctUntilChanged(),
switchMap(name => this.fetchQueue(name) ),
share()
);
}
ngAfterViewInit(): void {}
getCurrentEntity(): BaseData<EntityId> | null {
const currentRuleChain = this.selectQueueFormGroup.get('queueId').value;
if (currentRuleChain && typeof currentRuleChain !== 'string') {
return currentRuleChain as BaseData<EntityId>;
} else {
return null;
}
}
setDisabledState(isDisabled: boolean): void {
this.disabled = isDisabled;
if (this.disabled) {
this.selectQueueFormGroup.disable({emitEvent: false});
} else {
this.selectQueueFormGroup.enable({emitEvent: false});
}
}
textIsNotEmpty(text: string): boolean {
return (text && text.length > 0);
}
writeValue(value: string | null): void {
this.searchText = '';
if (value != null) {
const targetEntityType = EntityType.QUEUE;
this.entityService.getEntity(targetEntityType, value, {ignoreLoading: true, ignoreErrors: true}).subscribe(
(entity) => {
this.modelValue = entity.id.id;
this.selectQueueFormGroup.get('queueId').patchValue(entity, {emitEvent: false});
},
() => {
this.modelValue = null;
this.selectQueueFormGroup.get('queueId').patchValue('', {emitEvent: false});
if (value !== null) {
this.propagateChange(this.modelValue);
}
}
);
} else {
this.modelValue = null;
this.selectQueueFormGroup.get('queueId').patchValue('', {emitEvent: false});
}
this.dirty = true;
}
onFocus() {
if (this.dirty) {
this.selectQueueFormGroup.get('queueId').updateValueAndValidity({onlySelf: true, emitEvent: true});
this.dirty = false;
}
}
reset() {
this.selectQueueFormGroup.get('queueId').patchValue('', {emitEvent: false});
}
updateView(value: string | null) {
if (this.modelValue !== value) {
this.modelValue = value;
this.propagateChange(this.modelValue);
}
}
displayQueueFn(queue?: BaseData<EntityId>): string | undefined {
return queue ? queue.name : undefined;
}
fetchQueue(searchText?: string): Observable<Array<QueueInfo>> {
this.searchText = searchText;
const pageLink = new PageLink(10, 0, searchText, {
property: 'name',
direction: Direction.ASC
});
return this.queueService.getTenantQueuesByServiceType(pageLink, this.queueType, {ignoreLoading: true}).pipe(
catchError(() => of(emptyPageData<QueueInfo>())),
map(pageData => {
return pageData.data;
})
);
}
clear() {
this.selectQueueFormGroup.get('queueId').patchValue('', {emitEvent: true});
setTimeout(() => {
this.queueInput.nativeElement.blur();
this.queueInput.nativeElement.focus();
}, 0);
}
}

3
ui-ngx/src/app/shared/models/device.models.ts

@ -29,6 +29,7 @@ import * as _moment from 'moment';
import { AbstractControl, ValidationErrors } from '@angular/forms';
import { OtaPackageId } from '@shared/models/id/ota-package-id';
import { DashboardId } from '@shared/models/id/dashboard-id';
import { QueueId } from "@shared/models/id/queue-id";
import { DataType } from '@shared/models/constants';
import {
getDefaultProfileClientLwM2mSettingsConfig,
@ -569,7 +570,7 @@ export interface DeviceProfile extends BaseData<DeviceProfileId> {
provisionDeviceKey?: string;
defaultRuleChainId?: RuleChainId;
defaultDashboardId?: DashboardId;
defaultQueueName?: string;
defaultQueueId?: QueueId;
firmwareId?: OtaPackageId;
softwareId?: OtaPackageId;
profileData: DeviceProfileData;

3
ui-ngx/src/app/shared/shared.module.ts

@ -137,6 +137,7 @@ import { HistorySelectorComponent } from '@shared/components/time/history-select
import { EntityGatewaySelectComponent } from '@shared/components/entity/entity-gateway-select.component';
import { DndModule } from 'ngx-drag-drop';
import { QueueTypeListComponent } from '@shared/components/queue/queue-type-list.component';
import { QueueAutocompleteComponent } from '@shared/components/queue/queue-autocomplete.component';
import { ContactComponent } from '@shared/components/contact.component';
import { TimezoneSelectComponent } from '@shared/components/time/timezone-select.component';
import { FileSizePipe } from '@shared/pipe/file-size.pipe';
@ -232,6 +233,7 @@ export function MarkedOptionsFactory(markedOptionsService: MarkedOptionsService)
EntityListSelectComponent,
EntityTypeListComponent,
QueueTypeListComponent,
QueueAutocompleteComponent,
RelationTypeAutocompleteComponent,
SocialSharePanelComponent,
JsonObjectEditComponent,
@ -380,6 +382,7 @@ export function MarkedOptionsFactory(markedOptionsService: MarkedOptionsService)
EntityListSelectComponent,
EntityTypeListComponent,
QueueTypeListComponent,
QueueAutocompleteComponent,
RelationTypeAutocompleteComponent,
SocialSharePanelComponent,
JsonObjectEditComponent,

2
ui-ngx/src/assets/locale/locale.constant-en_US.json

@ -2677,7 +2677,7 @@
"browser-time": "Browser Time"
},
"queue": {
"queue-name": "Queue Name",
"queue-name": "Queue",
"no-queues-matching": "No queues matching '{{queue}}' were found.",
"select-name": "Select queue name",
"name": "Name",

Loading…
Cancel
Save