Browse Source

Merge remote-tracking branch 'origin/master' into tbel_new_ver_Unmodif_FunctionDate

pull/12804/head
nick 1 year ago
parent
commit
7cff794f03
  1. 2
      application/src/main/data/json/edge/rule_chains/edge_root_rule_chain.json
  2. 12
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
  3. 4
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/edge/EdgeEntityProcessor.java
  4. 82
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/rule/BaseRuleChainProcessor.java
  5. 98
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/rule/RuleChainEdgeProcessor.java
  6. 5
      application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java
  7. 10
      application/src/main/resources/thingsboard.yml
  8. 49
      application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java
  9. 27
      application/src/test/java/org/thingsboard/server/edge/AssetEdgeTest.java
  10. 1
      application/src/test/java/org/thingsboard/server/edge/DashboardEdgeTest.java
  11. 24
      application/src/test/java/org/thingsboard/server/edge/DeviceEdgeTest.java
  12. 84
      application/src/test/java/org/thingsboard/server/edge/RuleChainEdgeTest.java
  13. 2
      common/dao-api/src/main/java/org/thingsboard/server/dao/rule/RuleChainService.java
  14. 26
      common/edge-api/src/main/proto/edge.proto
  15. 56
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicService.java
  16. 11
      dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java
  17. 2
      monitoring/src/main/java/org/thingsboard/monitoring/service/transport/TransportHealthChecker.java
  18. 6
      monitoring/src/main/resources/lwm2m/device_profile.json
  19. 324
      monitoring/src/main/resources/lwm2m/models/test-model.xml
  20. 2
      monitoring/src/main/resources/lwm2m/resource.json
  21. 2
      msa/vc-executor/src/main/resources/tb-vc-executor.yml
  22. 2
      pom.xml
  23. 4
      transport/coap/src/main/resources/tb-coap-transport.yml
  24. 4
      transport/http/src/main/resources/tb-http-transport.yml
  25. 4
      transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml
  26. 4
      transport/mqtt/src/main/resources/tb-mqtt-transport.yml
  27. 4
      transport/snmp/src/main/resources/tb-snmp-transport.yml

2
application/src/main/data/json/edge/rule_chains/edge_root_rule_chain.json

@ -119,7 +119,7 @@
"type": "org.thingsboard.rule.engine.edge.TbMsgPushToCloudNode",
"name": "Push to cloud",
"configuration": {
"scope": "SERVER_SCOPE"
"scope": "CLIENT_SCOPE"
},
"externalId": null
},

12
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java

@ -74,6 +74,8 @@ import org.thingsboard.server.gen.edge.v1.RequestMsgType;
import org.thingsboard.server.gen.edge.v1.ResourceUpdateMsg;
import org.thingsboard.server.gen.edge.v1.ResponseMsg;
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataRequestMsg;
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataUpdateMsg;
import org.thingsboard.server.gen.edge.v1.RuleChainUpdateMsg;
import org.thingsboard.server.gen.edge.v1.SyncCompletedMsg;
import org.thingsboard.server.gen.edge.v1.UplinkMsg;
import org.thingsboard.server.gen.edge.v1.UplinkResponseMsg;
@ -820,6 +822,16 @@ public abstract class EdgeGrpcSession implements Closeable {
result.add(ctx.getAssetProcessor().processAssetMsgFromEdge(edge.getTenantId(), edge, assetUpdateMsg));
}
}
if (uplinkMsg.getRuleChainUpdateMsgCount() > 0) {
for (RuleChainUpdateMsg ruleChainUpdateMsg : uplinkMsg.getRuleChainUpdateMsgList()) {
result.add(ctx.getRuleChainProcessor().processRuleChainMsgFromEdge(edge.getTenantId(), edge, ruleChainUpdateMsg));
}
}
if (uplinkMsg.getRuleChainMetadataUpdateMsgCount() > 0) {
for (RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg : uplinkMsg.getRuleChainMetadataUpdateMsgList()) {
result.add(ctx.getRuleChainProcessor().processRuleChainMetadataMsgFromEdge(edge.getTenantId(), edge, ruleChainMetadataUpdateMsg));
}
}
if (uplinkMsg.getEntityViewUpdateMsgCount() > 0) {
for (EntityViewUpdateMsg entityViewUpdateMsg : uplinkMsg.getEntityViewUpdateMsgList()) {
result.add(ctx.getEntityViewProcessor().processEntityViewMsgFromEdge(edge.getTenantId(), edge, entityViewUpdateMsg));

4
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/edge/EdgeEntityProcessor.java

@ -49,8 +49,12 @@ public class EdgeEntityProcessor extends BaseEdgeProcessor {
@Override
public ListenableFuture<Void> processEntityNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) {
try {
EdgeId originatorEdgeId = safeGetEdgeId(edgeNotificationMsg.getOriginatorEdgeIdMSB(), edgeNotificationMsg.getOriginatorEdgeIdLSB());
EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction());
EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB()));
if (edgeId.equals(originatorEdgeId)) {
return Futures.immediateFuture(null);
}
switch (actionType) {
case ASSIGNED_TO_CUSTOMER: {
CustomerId customerId = JacksonUtil.fromString(edgeNotificationMsg.getBody(), CustomerId.class);

82
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/rule/BaseRuleChainProcessor.java

@ -0,0 +1,82 @@
/**
* Copyright © 2016-2025 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.service.edge.rpc.processor.rule;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.util.Pair;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.rule.RuleChain;
import org.thingsboard.server.common.data.rule.RuleChainMetaData;
import org.thingsboard.server.common.data.rule.RuleChainType;
import org.thingsboard.server.common.data.rule.RuleNode;
import org.thingsboard.server.dao.service.DataValidator;
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataUpdateMsg;
import org.thingsboard.server.gen.edge.v1.RuleChainUpdateMsg;
import org.thingsboard.server.service.edge.rpc.processor.BaseEdgeProcessor;
import java.util.function.Function;
@Slf4j
public class BaseRuleChainProcessor extends BaseEdgeProcessor {
@Autowired
private DataValidator<RuleChain> ruleChainValidator;
protected Pair<Boolean, Boolean> saveOrUpdateRuleChain(TenantId tenantId, RuleChainId ruleChainId, RuleChainUpdateMsg ruleChainUpdateMsg, RuleChainType ruleChainType) {
boolean created = false;
RuleChain ruleChainFromDb = edgeCtx.getRuleChainService().findRuleChainById(tenantId, ruleChainId);
if (ruleChainFromDb == null) {
created = true;
}
RuleChain ruleChain = JacksonUtil.fromString(ruleChainUpdateMsg.getEntity(), RuleChain.class, true);
if (ruleChain == null) {
throw new RuntimeException("[{" + tenantId + "}] ruleChainUpdateMsg {" + ruleChainUpdateMsg + "} cannot be converted to rule chain");
}
boolean isRoot = ruleChain.isRoot();
if (RuleChainType.CORE.equals(ruleChainType)) {
ruleChain.setRoot(false);
} else {
ruleChain.setRoot(ruleChainFromDb == null ? false : ruleChainFromDb.isRoot());
}
ruleChain.setType(ruleChainType);
ruleChainValidator.validate(ruleChain, RuleChain::getTenantId);
if (created) {
ruleChain.setId(ruleChainId);
}
edgeCtx.getRuleChainService().saveRuleChain(ruleChain, true, false);
return Pair.of(created, isRoot);
}
protected void saveOrUpdateRuleChainMetadata(TenantId tenantId, RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg) {
RuleChainMetaData ruleChainMetadata = JacksonUtil.fromString(ruleChainMetadataUpdateMsg.getEntity(), RuleChainMetaData.class, true);
if (ruleChainMetadata == null) {
throw new RuntimeException("[{" + tenantId + "}] ruleChainMetadataUpdateMsg {" + ruleChainMetadataUpdateMsg + "} cannot be converted to rule chain metadata");
}
if (!ruleChainMetadata.getNodes().isEmpty()) {
ruleChainMetadata.setVersion(null);
for (RuleNode ruleNode : ruleChainMetadata.getNodes()) {
ruleNode.setRuleChainId(null);
ruleNode.setId(null);
}
edgeCtx.getRuleChainService().saveRuleChainMetaData(tenantId, ruleChainMetadata, Function.identity(), true);
}
}
}

98
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/rule/RuleChainEdgeProcessor.java

@ -15,29 +15,123 @@
*/
package org.thingsboard.server.service.edge.rpc.processor.rule;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.util.Pair;
import org.springframework.stereotype.Component;
import org.thingsboard.server.common.data.EdgeUtils;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.edge.EdgeEvent;
import org.thingsboard.server.common.data.edge.EdgeEventType;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.data.rule.RuleChain;
import org.thingsboard.server.common.data.rule.RuleChainMetaData;
import org.thingsboard.server.common.data.rule.RuleChainType;
import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.dao.exception.DataValidationException;
import org.thingsboard.server.gen.edge.v1.DownlinkMsg;
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataUpdateMsg;
import org.thingsboard.server.gen.edge.v1.RuleChainUpdateMsg;
import org.thingsboard.server.gen.edge.v1.UpdateMsgType;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.edge.EdgeMsgConstructorUtils;
import org.thingsboard.server.service.edge.rpc.processor.BaseEdgeProcessor;
import java.util.UUID;
import static org.thingsboard.server.dao.edge.EdgeServiceImpl.EDGE_IS_ROOT_BODY_KEY;
@Slf4j
@Component
@TbCoreComponent
public class RuleChainEdgeProcessor extends BaseEdgeProcessor {
public class RuleChainEdgeProcessor extends BaseRuleChainProcessor {
public ListenableFuture<Void> processRuleChainMsgFromEdge(TenantId tenantId, Edge edge, RuleChainUpdateMsg ruleChainUpdateMsg) {
log.trace("[{}] executing processRuleChainMsgFromEdge [{}] from edge [{}]", tenantId, ruleChainUpdateMsg, edge.getName());
RuleChainId ruleChainId = new RuleChainId(new UUID(ruleChainUpdateMsg.getIdMSB(), ruleChainUpdateMsg.getIdLSB()));
try {
edgeSynchronizationManager.getEdgeId().set(edge.getId());
switch (ruleChainUpdateMsg.getMsgType()) {
case ENTITY_CREATED_RPC_MESSAGE:
case ENTITY_UPDATED_RPC_MESSAGE:
return saveOrUpdateRuleChain(tenantId, ruleChainId, ruleChainUpdateMsg, edge);
case ENTITY_DELETED_RPC_MESSAGE:
RuleChain ruleChainToDelete = edgeCtx.getRuleChainService().findRuleChainById(tenantId, ruleChainId);
if (ruleChainToDelete != null) {
edgeCtx.getRuleChainService().unassignRuleChainFromEdge(tenantId, ruleChainId, edge.getId(), false);
}
return Futures.immediateFuture(null);
case UNRECOGNIZED:
default:
return handleUnsupportedMsgType(ruleChainUpdateMsg.getMsgType());
}
} catch (DataValidationException e) {
if (e.getMessage().contains("limit reached")) {
log.warn("[{}] Number of allowed rule chains violated {}", tenantId, ruleChainUpdateMsg, e);
return Futures.immediateFuture(null);
} else {
return Futures.immediateFailedFuture(e);
}
} finally {
edgeSynchronizationManager.getEdgeId().remove();
}
}
private ListenableFuture<Void> saveOrUpdateRuleChain(TenantId tenantId, RuleChainId ruleChainId, RuleChainUpdateMsg ruleChainUpdateMsg, Edge edge) {
try {
Pair<Boolean, Boolean> resultPair = super.saveOrUpdateRuleChain(tenantId, ruleChainId, ruleChainUpdateMsg, RuleChainType.EDGE);
Boolean created = resultPair.getFirst();
if (created) {
createRelationFromEdge(tenantId, edge.getId(), ruleChainId);
pushRuleChainCreatedEventToRuleEngine(tenantId, edge, ruleChainId, ruleChainUpdateMsg.getEntity());
edgeCtx.getRuleChainService().assignRuleChainToEdge(tenantId, ruleChainId, edge.getId());
}
Boolean isRoot = resultPair.getSecond();
if (isRoot) {
edge = edgeCtx.getEdgeService().findEdgeById(tenantId, edge.getId());
edgeCtx.getEdgeService().setEdgeRootRuleChain(tenantId, edge, ruleChainId);
}
} catch (Exception e) {
log.error("Failed to save or update rule chain", e);
return Futures.immediateFailedFuture(e);
}
return Futures.immediateFuture(null);
}
private void pushRuleChainCreatedEventToRuleEngine(TenantId tenantId, Edge edge, RuleChainId ruleChainId, String ruleChainAsString) {
try {
TbMsgMetaData msgMetaData = getEdgeActionTbMsgMetaData(edge, null);
pushEntityEventToRuleEngine(tenantId, ruleChainId, null, TbMsgType.ENTITY_CREATED, ruleChainAsString, msgMetaData);
} catch (Exception e) {
log.warn("[{}][{}] Failed to push rule chain action to rule engine: {}", tenantId, ruleChainId, TbMsgType.ENTITY_CREATED.name(), e);
}
}
public ListenableFuture<Void> processRuleChainMetadataMsgFromEdge(TenantId tenantId, Edge edge, RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg) {
log.trace("[{}] executing processRuleChainMetadataMsgFromEdge [{}] from edge [{}]", tenantId, ruleChainMetadataUpdateMsg, edge.getName());
try {
edgeSynchronizationManager.getEdgeId().set(edge.getId());
switch (ruleChainMetadataUpdateMsg.getMsgType()) {
case ENTITY_CREATED_RPC_MESSAGE:
case ENTITY_UPDATED_RPC_MESSAGE:
saveOrUpdateRuleChainMetadata(tenantId, ruleChainMetadataUpdateMsg);
return Futures.immediateFuture(null);
case UNRECOGNIZED:
default:
return handleUnsupportedMsgType(ruleChainMetadataUpdateMsg.getMsgType());
}
} catch (Exception e) {
String errMsg = String.format("Can't process rule chain metadata update msg %s", ruleChainMetadataUpdateMsg);
log.error(errMsg, e);
return Futures.immediateFailedFuture(new RuntimeException(errMsg, e));
} finally {
edgeSynchronizationManager.getEdgeId().remove();
}
}
@Override
public DownlinkMsg convertEdgeEventToDownlink(EdgeEvent edgeEvent) {

5
application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java

@ -52,6 +52,7 @@ import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.common.msg.edge.EdgeEventUpdateMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.rule.engine.DeviceCredentialsUpdateNotificationMsg;
import org.thingsboard.server.dao.edge.EdgeSynchronizationManager;
import org.thingsboard.server.dao.eventsourcing.ActionEntityEvent;
import org.thingsboard.server.dao.eventsourcing.DeleteEntityEvent;
import org.thingsboard.server.dao.eventsourcing.SaveEntityEvent;
@ -67,6 +68,7 @@ public class EntityStateSourcingListener {
private final TenantService tenantService;
private final TbClusterService tbClusterService;
private final EdgeSynchronizationManager edgeSynchronizationManager;
@PostConstruct
public void init() {
@ -270,6 +272,9 @@ public class EntityStateSourcingListener {
private void onEdgeEvent(TenantId tenantId, EntityId entityId, Object entity, ComponentLifecycleEvent lifecycleEvent) {
if (entity instanceof Edge) {
if (entityId.equals(edgeSynchronizationManager.getEdgeId().get())) {
return;
}
tbClusterService.onEdgeStateChangeEvent(new ComponentLifecycleMsg(tenantId, entityId, lifecycleEvent));
} else if (entity instanceof EdgeEvent edgeEvent) {
tbClusterService.onEdgeEventUpdate(new EdgeEventUpdateMsg(tenantId, edgeEvent.getEdgeId()));

10
application/src/main/resources/thingsboard.yml

@ -1659,6 +1659,8 @@ queue:
core:
# Default topic name
topic: "${TB_QUEUE_CORE_TOPIC:tb_core}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_CORE_NOTIFICATIONS_TOPIC:tb_core.notifications}"
# Interval in milliseconds to poll messages by Core microservices
poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}"
# Amount of partitions used by Core microservices
@ -1734,6 +1736,8 @@ queue:
rule-engine:
# Deprecated. It will be removed in the nearest releases
topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb_rule_engine}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_RULE_ENGINE_NOTIFICATIONS_TOPIC:tb_rule_engine.notifications}"
# Interval in milliseconds to poll messages by Rule Engine
poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}"
# Timeout for processing a message pack of Rule Engine
@ -1754,6 +1758,8 @@ queue:
event_topic: "${TB_QUEUE_CF_EVENT_TOPIC:tb_cf_event}"
# Topic name for Calculated Field (CF) compacted states
state_topic: "${TB_QUEUE_CF_STATE_TOPIC:tb_cf_state}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_CF_NOTIFICATIONS_TOPIC:calculated_field.notifications}"
# Interval in milliseconds to poll messages by CF (Rule Engine) microservices
poll_interval: "${TB_QUEUE_CF_POLL_INTERVAL_MS:25}"
# Amount of partitions used by CF microservices
@ -1772,6 +1778,10 @@ queue:
edge:
# Default topic name
topic: "${TB_QUEUE_EDGE_TOPIC:tb_edge}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_EDGE_NOTIFICATIONS_TOPIC:tb_edge.notifications}"
# For edge events messages
event_notifications_topic: "${TB_QUEUE_EDGE_EVENT_NOTIFICATIONS_TOPIC:tb_edge_event.notifications}"
# Amount of partitions used by Edge services
partitions: "${TB_QUEUE_EDGE_PARTITIONS:10}"
# Poll interval for topics related to Edge services

49
application/src/test/java/org/thingsboard/server/edge/AbstractEdgeTest.java

@ -90,14 +90,12 @@ import org.thingsboard.server.gen.edge.v1.EdgeConfiguration;
import org.thingsboard.server.gen.edge.v1.OAuth2ClientUpdateMsg;
import org.thingsboard.server.gen.edge.v1.OAuth2DomainUpdateMsg;
import org.thingsboard.server.gen.edge.v1.QueueUpdateMsg;
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataRequestMsg;
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataUpdateMsg;
import org.thingsboard.server.gen.edge.v1.RuleChainUpdateMsg;
import org.thingsboard.server.gen.edge.v1.SyncCompletedMsg;
import org.thingsboard.server.gen.edge.v1.TenantProfileUpdateMsg;
import org.thingsboard.server.gen.edge.v1.TenantUpdateMsg;
import org.thingsboard.server.gen.edge.v1.UpdateMsgType;
import org.thingsboard.server.gen.edge.v1.UplinkMsg;
import org.thingsboard.server.gen.edge.v1.UserCredentialsUpdateMsg;
import org.thingsboard.server.gen.edge.v1.UserUpdateMsg;
@ -142,35 +140,14 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest {
installation();
edgeImitator = new EdgeImitator("localhost", 7070, edge.getRoutingKey(), edge.getSecret());
edgeImitator.expectMessageAmount(25);
edgeImitator.ignoreType(OAuth2ClientUpdateMsg.class);
edgeImitator.ignoreType(OAuth2DomainUpdateMsg.class);
edgeImitator.expectMessageAmount(26);
edgeImitator.connect();
requestEdgeRuleChainMetadata();
verifyEdgeConnectionAndInitialData();
}
private void requestEdgeRuleChainMetadata() throws Exception {
RuleChainId rootRuleChainId = getEdgeRootRuleChainId();
RuleChainMetadataRequestMsg.Builder builder = RuleChainMetadataRequestMsg.newBuilder()
.setRuleChainIdMSB(rootRuleChainId.getId().getMostSignificantBits())
.setRuleChainIdLSB(rootRuleChainId.getId().getLeastSignificantBits());
testAutoGeneratedCodeByProtobuf(builder);
UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder()
.addRuleChainMetadataRequestMsg(builder.build());
edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());
}
private RuleChainId getEdgeRootRuleChainId() throws Exception {
return doGetTypedWithPageLink("/api/ruleChains?type={type}&", new TypeReference<PageData<RuleChain>>() {
},
new PageLink(100, 0, "Edge Root Rule Chain"),
"EDGE")
.getData().get(0).getId();
}
@After
public void teardownEdgeTest() {
try {
@ -213,6 +190,19 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest {
doPost("/api/ruleChain/metadata", rootRuleChainMetadata, RuleChainMetaData.class);
}
private RuleChainId getEdgeRootRuleChainId() throws Exception {
List<RuleChain> edgeRuleChains = doGetTypedWithPageLink("/api/ruleChains?type={type}&",
new TypeReference<PageData<RuleChain>>() {},
new PageLink(100, 0, "Edge Root Rule Chain"),
"EDGE").getData();
for (RuleChain edgeRuleChain : edgeRuleChains) {
if (edgeRuleChain.isRoot()) {
return edgeRuleChain.getId();
}
}
throw new RuntimeException("Root rule chain not found");
}
protected void extendDeviceProfileData(DeviceProfile deviceProfile) {
DeviceProfileData profileData = deviceProfile.getProfileData();
List<DeviceProfileAlarm> alarms = new ArrayList<>();
@ -255,8 +245,8 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest {
validateMsgsCnt(RuleChainUpdateMsg.class, 1);
UUID ruleChainUUID = validateRuleChains();
// 1 from request message
validateMsgsCnt(RuleChainMetadataUpdateMsg.class, 2);
// 1 from rule chain fetcher
validateMsgsCnt(RuleChainMetadataUpdateMsg.class, 1);
validateRuleChainMetadataUpdates(ruleChainUUID);
// 4 messages ('general', 'mail', 'connectivity', 'jwt')
@ -438,12 +428,11 @@ abstract public class AbstractEdgeTest extends AbstractControllerTest {
}
private void validateRuleChainMetadataUpdates(UUID expectedRuleChainUUID) {
Optional<RuleChainMetadataUpdateMsg> ruleChainMetadataUpdateOpt = edgeImitator.findMessageByType(RuleChainMetadataUpdateMsg.class);
Assert.assertTrue(ruleChainMetadataUpdateOpt.isPresent());
RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = ruleChainMetadataUpdateOpt.get();
Optional<RuleChainMetadataUpdateMsg> ruleChainMetadataUpdateMsgOpt = edgeImitator.findMessageByType(RuleChainMetadataUpdateMsg.class);
Assert.assertTrue(ruleChainMetadataUpdateMsgOpt.isPresent());
RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = ruleChainMetadataUpdateMsgOpt.get();
Assert.assertEquals(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE, ruleChainMetadataUpdateMsg.getMsgType());
RuleChainMetaData ruleChainMetaData = JacksonUtil.fromString(ruleChainMetadataUpdateMsg.getEntity(), RuleChainMetaData.class, true);
Assert.assertNotNull(ruleChainMetaData);
Assert.assertEquals(expectedRuleChainUUID, ruleChainMetaData.getRuleChainId().getId());
}

27
application/src/test/java/org/thingsboard/server/edge/AssetEdgeTest.java

@ -169,6 +169,7 @@ public class AssetEdgeTest extends AbstractEdgeTest {
public void testSendAssetToCloud() throws Exception {
Asset asset = buildAssetForUplinkMsg("Asset Edge 2");
// created asset on edge
UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder();
AssetUpdateMsg.Builder assetUpdateMsgBuilder = AssetUpdateMsg.newBuilder();
assetUpdateMsgBuilder.setIdMSB(asset.getUuidId().getMostSignificantBits());
@ -191,6 +192,32 @@ public class AssetEdgeTest extends AbstractEdgeTest {
Asset foundAsset = doGet("/api/asset/" + asset.getUuidId(), Asset.class);
Assert.assertNotNull(foundAsset);
Assert.assertEquals("Asset Edge 2", foundAsset.getName());
// update asset on edge
asset.setName("Asset Edge 2 Updated");
uplinkMsgBuilder = UplinkMsg.newBuilder();
assetUpdateMsgBuilder = AssetUpdateMsg.newBuilder();
assetUpdateMsgBuilder.setIdMSB(asset.getUuidId().getMostSignificantBits());
assetUpdateMsgBuilder.setIdLSB(asset.getUuidId().getLeastSignificantBits());
assetUpdateMsgBuilder.setEntity(JacksonUtil.toString(asset));
assetUpdateMsgBuilder.setMsgType(UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE);
testAutoGeneratedCodeByProtobuf(assetUpdateMsgBuilder);
uplinkMsgBuilder.addAssetUpdateMsg(assetUpdateMsgBuilder.build());
testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder);
edgeImitator.expectResponsesAmount(1);
edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());
Assert.assertTrue(edgeImitator.waitForResponses());
latestResponseMsg = edgeImitator.getLatestResponseMsg();
Assert.assertTrue(latestResponseMsg.getSuccess());
foundAsset = doGet("/api/asset/" + asset.getUuidId(), Asset.class);
Assert.assertNotNull(foundAsset);
Assert.assertEquals("Asset Edge 2 Updated", foundAsset.getName());
}
@Test

1
application/src/test/java/org/thingsboard/server/edge/DashboardEdgeTest.java

@ -184,6 +184,7 @@ public class DashboardEdgeTest extends AbstractEdgeTest {
Dashboard dashboard = buildDashboardForUplinkMsg(savedCustomer);
// create dashboard on edge
UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder();
DashboardUpdateMsg.Builder dashboardUpdateMsgBuilder = DashboardUpdateMsg.newBuilder();
dashboardUpdateMsgBuilder.setIdMSB(dashboard.getUuidId().getMostSignificantBits());

24
application/src/test/java/org/thingsboard/server/edge/DeviceEdgeTest.java

@ -593,8 +593,10 @@ public class DeviceEdgeTest extends AbstractEdgeTest {
@Test
public void testSendDeviceToCloud() throws Exception {
Device deviceMsg = buildDeviceForUplinkMsg("Edge Device 2", "test");
String deviceName = "Edge Device 2";
Device deviceMsg = buildDeviceForUplinkMsg(deviceName, "test");
// create device on edge
UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder();
DeviceUpdateMsg.Builder deviceUpdateMsgBuilder = DeviceUpdateMsg.newBuilder();
deviceUpdateMsgBuilder.setIdMSB(deviceMsg.getUuidId().getMostSignificantBits());
@ -609,7 +611,25 @@ public class DeviceEdgeTest extends AbstractEdgeTest {
Device device = doGet("/api/device/" + deviceMsg.getId().getId(), Device.class);
Assert.assertNotNull(device);
Assert.assertEquals("Edge Device 2", device.getName());
Assert.assertEquals(deviceName, device.getName());
// update device on edge
deviceMsg.setName(deviceName + " Updated");
uplinkMsgBuilder = UplinkMsg.newBuilder();
deviceUpdateMsgBuilder = DeviceUpdateMsg.newBuilder();
deviceUpdateMsgBuilder.setIdMSB(deviceMsg.getUuidId().getMostSignificantBits());
deviceUpdateMsgBuilder.setIdLSB(deviceMsg.getUuidId().getLeastSignificantBits());
deviceUpdateMsgBuilder.setEntity(JacksonUtil.toString(deviceMsg));
deviceUpdateMsgBuilder.setMsgType(UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE);
uplinkMsgBuilder.addDeviceUpdateMsg(deviceUpdateMsgBuilder.build());
edgeImitator.expectResponsesAmount(1);
edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());
Assert.assertTrue(edgeImitator.waitForResponses());
device = doGet("/api/device/" + deviceMsg.getId().getId(), Device.class);
Assert.assertNotNull(device);
Assert.assertEquals(deviceName + " Updated", device.getName());
}
@Test

84
application/src/test/java/org/thingsboard/server/edge/RuleChainEdgeTest.java

@ -15,7 +15,7 @@
*/
package org.thingsboard.server.edge;
import com.google.protobuf.AbstractMessage;
import com.datastax.oss.driver.api.core.uuid.Uuids;
import org.junit.Assert;
import org.junit.Test;
import org.thingsboard.common.util.JacksonUtil;
@ -29,16 +29,17 @@ import org.thingsboard.server.common.data.rule.RuleChainMetaData;
import org.thingsboard.server.common.data.rule.RuleChainType;
import org.thingsboard.server.common.data.rule.RuleNode;
import org.thingsboard.server.dao.service.DaoSqlTest;
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataRequestMsg;
import org.thingsboard.server.gen.edge.v1.RuleChainMetadataUpdateMsg;
import org.thingsboard.server.gen.edge.v1.RuleChainUpdateMsg;
import org.thingsboard.server.gen.edge.v1.UpdateMsgType;
import org.thingsboard.server.gen.edge.v1.UplinkMsg;
import org.thingsboard.server.gen.edge.v1.UplinkResponseMsg;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
@ -74,8 +75,9 @@ public class RuleChainEdgeTest extends AbstractEdgeTest {
RuleChainMetaData ruleChainMetaData = JacksonUtil.fromString(ruleChainMetadataUpdateMsg.getEntity(), RuleChainMetaData.class, true);
Assert.assertNotNull(ruleChainMetaData);
Assert.assertEquals(ruleChainMetaData.getRuleChainId(), savedRuleChain.getId());
testRuleChainMetadataRequestMsg(savedRuleChain.getId());
for (RuleNode ruleNode : ruleChainMetaData.getNodes()) {
Assert.assertEquals(CONFIGURATION_VERSION, ruleNode.getConfigurationVersion());
}
// unassign rule chain from edge
edgeImitator.expectMessageAmount(1);
@ -97,60 +99,62 @@ public class RuleChainEdgeTest extends AbstractEdgeTest {
}
@Test
public void testSendRuleChainMetadataRequestToCloud() throws Exception {
RuleChainId edgeRootRuleChainId = edge.getRootRuleChainId();
public void testRuleChainToCloud() throws Exception {
String ruleChainName = "Rule Chain Edge";
UUID uuid = Uuids.timeBased();
// create rule chain on edge
RuleChain edgeRuleChain = new RuleChain();
edgeRuleChain.setTenantId(tenantId);
edgeRuleChain.setId(new RuleChainId(uuid));
edgeRuleChain.setName(ruleChainName);
UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder();
RuleChainMetadataRequestMsg.Builder ruleChainMetadataRequestMsgBuilder = RuleChainMetadataRequestMsg.newBuilder();
ruleChainMetadataRequestMsgBuilder.setRuleChainIdMSB(edgeRootRuleChainId.getId().getMostSignificantBits());
ruleChainMetadataRequestMsgBuilder.setRuleChainIdLSB(edgeRootRuleChainId.getId().getLeastSignificantBits());
testAutoGeneratedCodeByProtobuf(ruleChainMetadataRequestMsgBuilder);
uplinkMsgBuilder.addRuleChainMetadataRequestMsg(ruleChainMetadataRequestMsgBuilder.build());
RuleChainUpdateMsg.Builder ruleChainUpdateMsgBuilder = RuleChainUpdateMsg.newBuilder();
ruleChainUpdateMsgBuilder.setIdMSB(uuid.getMostSignificantBits());
ruleChainUpdateMsgBuilder.setIdLSB(uuid.getLeastSignificantBits());
ruleChainUpdateMsgBuilder.setEntity(JacksonUtil.toString(edgeRuleChain));
ruleChainUpdateMsgBuilder.setMsgType(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE);
testAutoGeneratedCodeByProtobuf(ruleChainUpdateMsgBuilder);
uplinkMsgBuilder.addRuleChainUpdateMsg(ruleChainUpdateMsgBuilder.build());
testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder);
edgeImitator.expectResponsesAmount(1);
edgeImitator.expectMessageAmount(1);
edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());
Assert.assertTrue(edgeImitator.waitForResponses());
Assert.assertTrue(edgeImitator.waitForMessages());
AbstractMessage latestMessage = edgeImitator.getLatestMessage();
Assert.assertTrue(latestMessage instanceof RuleChainMetadataUpdateMsg);
RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = (RuleChainMetadataUpdateMsg) latestMessage;
RuleChainMetaData ruleChainMetadataMsg = JacksonUtil.fromString(ruleChainMetadataUpdateMsg.getEntity(), RuleChainMetaData.class, true);
Assert.assertNotNull(ruleChainMetadataMsg);
Assert.assertEquals(edgeRootRuleChainId, ruleChainMetadataMsg.getRuleChainId());
UplinkResponseMsg latestResponseMsg = edgeImitator.getLatestResponseMsg();
Assert.assertTrue(latestResponseMsg.getSuccess());
testAutoGeneratedCodeByProtobuf(ruleChainMetadataUpdateMsg);
}
RuleChain ruleChain = doGet("/api/ruleChain/" + uuid, RuleChain.class);
Assert.assertNotNull(ruleChain);
Assert.assertEquals("Rule Chain Edge", ruleChain.getName());
private void testRuleChainMetadataRequestMsg(RuleChainId ruleChainId) throws Exception {
RuleChainMetadataRequestMsg.Builder ruleChainMetadataRequestMsgBuilder = RuleChainMetadataRequestMsg.newBuilder()
.setRuleChainIdMSB(ruleChainId.getId().getMostSignificantBits())
.setRuleChainIdLSB(ruleChainId.getId().getLeastSignificantBits());
testAutoGeneratedCodeByProtobuf(ruleChainMetadataRequestMsgBuilder);
// update rule chain on edge
edgeRuleChain.setName(ruleChainName + " Updated");
uplinkMsgBuilder = UplinkMsg.newBuilder();
ruleChainUpdateMsgBuilder = RuleChainUpdateMsg.newBuilder();
ruleChainUpdateMsgBuilder.setIdMSB(uuid.getMostSignificantBits());
ruleChainUpdateMsgBuilder.setIdLSB(uuid.getLeastSignificantBits());
ruleChainUpdateMsgBuilder.setEntity(JacksonUtil.toString(edgeRuleChain));
ruleChainUpdateMsgBuilder.setMsgType(UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE);
testAutoGeneratedCodeByProtobuf(ruleChainUpdateMsgBuilder);
uplinkMsgBuilder.addRuleChainUpdateMsg(ruleChainUpdateMsgBuilder.build());
UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder()
.addRuleChainMetadataRequestMsg(ruleChainMetadataRequestMsgBuilder.build());
testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder);
edgeImitator.expectResponsesAmount(1);
edgeImitator.expectMessageAmount(1);
edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());
Assert.assertTrue(edgeImitator.waitForResponses());
Assert.assertTrue(edgeImitator.waitForMessages());
AbstractMessage latestMessage = edgeImitator.getLatestMessage();
Assert.assertTrue(latestMessage instanceof RuleChainMetadataUpdateMsg);
RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = (RuleChainMetadataUpdateMsg) latestMessage;
RuleChainMetaData ruleChainMetadataMsg = JacksonUtil.fromString(ruleChainMetadataUpdateMsg.getEntity(), RuleChainMetaData.class, true);
Assert.assertNotNull(ruleChainMetadataMsg);
Assert.assertEquals(ruleChainId, ruleChainMetadataMsg.getRuleChainId());
latestResponseMsg = edgeImitator.getLatestResponseMsg();
Assert.assertTrue(latestResponseMsg.getSuccess());
for (RuleNode ruleNode : ruleChainMetadataMsg.getNodes()) {
Assert.assertEquals(CONFIGURATION_VERSION, ruleNode.getConfigurationVersion());
}
ruleChain = doGet("/api/ruleChain/" + uuid, RuleChain.class);
Assert.assertNotNull(ruleChain);
Assert.assertEquals(ruleChainName + " Updated", ruleChain.getName());
}
private RuleChainMetaData createRuleChainMetadata(RuleChain ruleChain) {

2
common/dao-api/src/main/java/org/thingsboard/server/dao/rule/RuleChainService.java

@ -46,6 +46,8 @@ public interface RuleChainService extends EntityDaoService {
RuleChain saveRuleChain(RuleChain ruleChain, boolean publishSaveEvent);
RuleChain saveRuleChain(RuleChain ruleChain, boolean publishSaveEvent, boolean doValidate);
boolean setRootRuleChain(TenantId tenantId, RuleChainId ruleChainId);
RuleChainUpdateResult saveRuleChainMetaData(TenantId tenantId, RuleChainMetaData ruleChainMetaData, Function<RuleNode, RuleNode> ruleNodeUpdater);

26
common/edge-api/src/main/proto/edge.proto

@ -42,6 +42,8 @@ enum EdgeVersion {
V_3_8_0 = 8;
V_3_9_0 = 9;
V_4_0_0 = 10;
V_LATEST = 999;
}
/**
@ -303,7 +305,9 @@ message NotificationTemplateUpdateMsg {
optional string entity = 4;
}
// DEPRECATED. FOR REMOVAL
message RuleChainMetadataRequestMsg {
option deprecated = true;
int64 ruleChainIdMSB = 1;
int64 ruleChainIdLSB = 2;
}
@ -321,22 +325,30 @@ message RelationRequestMsg {
string entityType = 3;
}
// DEPRECATED. FOR REMOVAL
message UserCredentialsRequestMsg {
option deprecated = true;
int64 userIdMSB = 1;
int64 userIdLSB = 2;
}
// DEPRECATED. FOR REMOVAL
message DeviceCredentialsRequestMsg {
option deprecated = true;
int64 deviceIdMSB = 1;
int64 deviceIdLSB = 2;
}
// DEPRECATED. FOR REMOVAL
message WidgetBundleTypesRequestMsg {
option deprecated = true;
int64 widgetBundleIdMSB = 1;
int64 widgetBundleIdLSB = 2;
}
// DEPRECATED. FOR REMOVAL
message EntityViewsRequestMsg {
option deprecated = true;
int64 entityIdMSB = 1;
int64 entityIdLSB = 2;
string entityType = 3;
@ -394,14 +406,14 @@ message UplinkMsg {
repeated DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg = 4;
repeated AlarmUpdateMsg alarmUpdateMsg = 5;
repeated RelationUpdateMsg relationUpdateMsg = 6;
repeated RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg = 7;
repeated RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg = 7 [deprecated = true];
repeated AttributesRequestMsg attributesRequestMsg = 8;
repeated RelationRequestMsg relationRequestMsg = 9;
repeated UserCredentialsRequestMsg userCredentialsRequestMsg = 10;
repeated DeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 11;
repeated UserCredentialsRequestMsg userCredentialsRequestMsg = 10 [deprecated = true];
repeated DeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 11 [deprecated = true];
repeated DeviceRpcCallMsg deviceRpcCallMsg = 12;
repeated WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg = 14;
repeated EntityViewsRequestMsg entityViewsRequestMsg = 15;
repeated WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg = 14 [deprecated = true];
repeated EntityViewsRequestMsg entityViewsRequestMsg = 15 [deprecated = true];
repeated AssetUpdateMsg assetUpdateMsg = 16;
repeated DashboardUpdateMsg dashboardUpdateMsg = 17;
repeated EntityViewUpdateMsg entityViewUpdateMsg = 18;
@ -409,6 +421,8 @@ message UplinkMsg {
repeated DeviceProfileUpdateMsg deviceProfileUpdateMsg = 20;
repeated ResourceUpdateMsg resourceUpdateMsg = 21;
repeated AlarmCommentUpdateMsg alarmCommentUpdateMsg = 22;
repeated RuleChainUpdateMsg ruleChainUpdateMsg = 23;
repeated RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = 24;
}
message UplinkResponseMsg {
@ -427,7 +441,7 @@ message DownlinkMsg {
int32 downlinkMsgId = 1;
SyncCompletedMsg syncCompletedMsg = 2;
repeated EntityDataProto entityData = 3;
repeated DeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 4;
repeated DeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 4 [deprecated = true];
repeated DeviceUpdateMsg deviceUpdateMsg = 5;
repeated DeviceProfileUpdateMsg deviceProfileUpdateMsg = 6;
repeated DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg = 7;

56
common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicService.java

@ -32,6 +32,24 @@ public class TopicService {
@Value("${queue.prefix:}")
private String prefix;
@Value("${queue.core.notifications-topic:tb_core.notifications}")
private String tbCoreNotificationsTopic;
@Value("${queue.rule-engine.notifications-topic:tb_rule_engine.notifications}")
private String tbRuleEngineNotificationsTopic;
@Value("${queue.transport.notifications-topics:tb_transport.notifications}")
private String tbTransportNotificationsTopic;
@Value("${queue.edge.notifications-topic:tb_edge.notifications}")
private String tbEdgeNotificationsTopic;
@Value("${queue.edge.event-notifications-topic:tb_edge_event.notifications}")
private String tbEdgeEventNotificationsTopic;
@Value("${queue.calculated_fields.notifications-topic:calculated_field.notifications}")
private String tbCalculatedFieldNotificationsTopic;
private final ConcurrentMap<String, TopicPartitionInfo> tbCoreNotificationTopics = new ConcurrentHashMap<>();
private final ConcurrentMap<String, TopicPartitionInfo> tbRuleEngineNotificationTopics = new ConcurrentHashMap<>();
private final ConcurrentMap<String, TopicPartitionInfo> tbEdgeNotificationTopics = new ConcurrentHashMap<>();
@ -48,24 +66,32 @@ public class TopicService {
public TopicPartitionInfo getNotificationsTopic(ServiceType serviceType, String serviceId) {
return switch (serviceType) {
case TB_CORE -> tbCoreNotificationTopics.computeIfAbsent(serviceId,
id -> buildNotificationsTopicPartitionInfo(serviceType, serviceId));
id -> buildNotificationsTopicPartitionInfo(tbCoreNotificationsTopic, serviceId));
case TB_RULE_ENGINE -> tbRuleEngineNotificationTopics.computeIfAbsent(serviceId,
id -> buildNotificationsTopicPartitionInfo(serviceType, serviceId));
default -> buildNotificationsTopicPartitionInfo(serviceType, serviceId);
id -> buildNotificationsTopicPartitionInfo(tbRuleEngineNotificationsTopic, serviceId));
case TB_TRANSPORT -> buildNotificationsTopicPartitionInfo(tbTransportNotificationsTopic, serviceId);
default -> throw new IllegalStateException("Unexpected service type: " + serviceType);
};
}
private TopicPartitionInfo buildNotificationsTopicPartitionInfo(String topic, String serviceId) {
return buildTopicPartitionInfo(buildNotificationTopicName(topic, serviceId), null, null, false);
}
public TopicPartitionInfo buildTopicPartitionInfo(String topic, TenantId tenantId, Integer partition, boolean myPartition) {
return new TopicPartitionInfo(buildTopicName(topic), tenantId, partition, myPartition);
}
public TopicPartitionInfo getEdgeNotificationsTopic(String serviceId) {
return tbEdgeNotificationTopics.computeIfAbsent(serviceId, id -> buildEdgeNotificationsTopicPartitionInfo(serviceId));
}
private TopicPartitionInfo buildEdgeNotificationsTopicPartitionInfo(String serviceId) {
return buildTopicPartitionInfo("tb_edge.notifications." + serviceId, null, null, false);
return buildTopicPartitionInfo(buildNotificationTopicName(tbEdgeNotificationsTopic, serviceId), null, null, false);
}
public TopicPartitionInfo getCalculatedFieldNotificationsTopic(String serviceId) {
return tbCalculatedFieldNotificationTopics.computeIfAbsent(serviceId,
id -> buildNotificationsTopicPartitionInfo("calculated_field", serviceId));
return tbCalculatedFieldNotificationTopics.computeIfAbsent(serviceId, id -> buildNotificationsTopicPartitionInfo(tbCalculatedFieldNotificationsTopic, serviceId));
}
public TopicPartitionInfo getEdgeEventNotificationsTopic(TenantId tenantId, EdgeId edgeId) {
@ -73,25 +99,17 @@ public class TopicService {
}
public TopicPartitionInfo buildEdgeEventNotificationsTopicPartitionInfo(TenantId tenantId, EdgeId edgeId) {
return buildTopicPartitionInfo("tb_edge_event.notifications." + tenantId + "." + edgeId, null, null, false);
}
private TopicPartitionInfo buildNotificationsTopicPartitionInfo(ServiceType serviceType, String serviceId) {
return buildNotificationsTopicPartitionInfo(serviceType.name().toLowerCase(), serviceId);
}
private TopicPartitionInfo buildNotificationsTopicPartitionInfo(String serviceType, String serviceId) {
return buildTopicPartitionInfo(serviceType + ".notifications." + serviceId, null, null, false);
}
public TopicPartitionInfo buildTopicPartitionInfo(String topic, TenantId tenantId, Integer partition, boolean myPartition) {
return new TopicPartitionInfo(buildTopicName(topic), tenantId, partition, myPartition);
return buildTopicPartitionInfo(tbEdgeEventNotificationsTopic + "." + tenantId + "." + edgeId, null, null, false);
}
public String buildTopicName(String topic) {
return prefix.isBlank() ? topic : prefix + "." + topic;
}
private String buildNotificationTopicName(String topic, String serviceId) {
return topic + "." + serviceId;
}
public String buildConsumerGroupId(String servicePrefix, TenantId tenantId, String queueName, Integer partitionId) {
return this.buildTopicName(
servicePrefix + queueName

11
dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java

@ -118,7 +118,16 @@ public class BaseRuleChainService extends AbstractEntityService implements RuleC
@Override
@Transactional
public RuleChain saveRuleChain(RuleChain ruleChain, boolean publishSaveEvent) {
ruleChainValidator.validate(ruleChain, RuleChain::getTenantId);
return saveRuleChain(ruleChain, publishSaveEvent, true);
}
@Override
@Transactional
public RuleChain saveRuleChain(RuleChain ruleChain, boolean publishSaveEvent, boolean doValidate) {
log.trace("Executing doSaveRuleChain [{}]", ruleChain);
if (doValidate) {
ruleChainValidator.validate(ruleChain, RuleChain::getTenantId);
}
try {
RuleChain savedRuleChain = ruleChainDao.saveAndFlush(ruleChain.getTenantId(), ruleChain);
if (ruleChain.getId() == null) {

2
monitoring/src/main/java/org/thingsboard/monitoring/service/transport/TransportHealthChecker.java

@ -145,7 +145,7 @@ public abstract class TransportHealthChecker<C extends TransportMonitoringConfig
profileData.setTransportConfiguration(new DefaultDeviceProfileTransportConfiguration());
deviceProfile.setProfileData(profileData);
} else {
tbClient.getResources(new PageLink(1, 0, "LwM2M Monitoring id=3 v1.0")).getData()
tbClient.getResources(new PageLink(1, 0, "LwM2M Monitoring")).getData()
.stream().findFirst()
.orElseGet(() -> {
TbResource newResource = ResourceUtils.getResource("lwm2m/resource.json", TbResource.class);

6
monitoring/src/main/resources/lwm2m/device_profile.json

@ -12,14 +12,14 @@
"transportConfiguration": {
"observeAttr": {
"observe": [
"/3_1.0/0/0"
"/3_1.1/0/0"
],
"attribute": [],
"telemetry": [
"/3_1.0/0/0"
"/3_1.1/0/0"
],
"keyName": {
"/3_1.0/0/0": "testData"
"/3_1.1/0/0": "testData"
},
"attributeLwm2m": {}
},

324
monitoring/src/main/resources/lwm2m/models/test-model.xml

@ -1,45 +1,331 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
FILE INFORMATION
OMA Permanent Document
File: OMA-SUP-XML_3-V1_2-20201110-A.xml
Path: http://www.openmobilealliance.org/release/ObjLwM2M_Device/
OMNA LwM2M Registry
Path: https://github.com/OpenMobileAlliance/lwm2m-registry
Name: 3.xml
NORMATIVE INFORMATION
Information about this file can be found in the latest revision of
OMA-TS-LightweightM2M_Core-V1_2
This is available at http://www.openmobilealliance.org/release/LightweightM2M/
Send comments to https://github.com/OpenMobileAlliance/OMA_LwM2M_for_Developers/issues
LEGAL DISCLAIMER
Copyright © 2016-2025 The Thingsboard Authors
Copyright 2020 Open Mobile Alliance.
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
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:
http://www.apache.org/licenses/LICENSE-2.0
1. Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.
3. Neither the name of the copyright holder nor the names of its
contributors may be used to endorse or promote products derived
from this software without specific prior written permission.
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.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.
The above license is used as a license under copyright only. Please
reference the OMA IPR Policy for patent licensing terms:
https://www.omaspecworks.org/about/intellectual-property-rights/
-->
<LWM2M xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:noNamespaceSchemaLocation="http://www.openmobilealliance.org/tech/profiles/LWM2M-v1_1.xsd">
<LWM2M xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="http://www.openmobilealliance.org/tech/profiles/LWM2M-v1_1.xsd">
<Object ObjectType="MODefinition">
<Name>LwM2M Monitoring</Name>
<Description1>
<![CDATA[]]></Description1>
<Description1><![CDATA[This LwM2M Object provides a range of device related information which can be queried by the LwM2M Server, and a device reboot and factory reset function.]]></Description1>
<ObjectID>3</ObjectID>
<ObjectURN>urn:oma:lwm2m:oma:3</ObjectURN>
<ObjectURN>urn:oma:lwm2m:oma:3:1.1</ObjectURN>
<LWM2MVersion>1.1</LWM2MVersion>
<ObjectVersion>1.0</ObjectVersion>
<ObjectVersion>1.1</ObjectVersion>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Mandatory</Mandatory>
<Resources>
<Item ID="0">
<Name>Test data</Name>
<Name>Manufacturer</Name>
<Operations>R</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>String</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Human readable manufacturer name]]></Description>
</Item>
<Item ID="1">
<Name>Model Number</Name>
<Operations>R</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>String</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Test data]]></Description>
<Description><![CDATA[A model identifier (manufacturer specified string)]]></Description>
</Item>
</Resources>
<Item ID="2">
<Name>Serial Number</Name>
<Operations>R</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>String</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Serial Number]]></Description>
</Item>
<Item ID="3">
<Name>Firmware Version</Name>
<Operations>R</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>String</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Current firmware version of the Device.The Firmware Management function could rely on this resource.]]></Description>
</Item>
<Item ID="4">
<Name>Reboot</Name>
<Operations>E</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Mandatory</Mandatory>
<Type></Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Reboot the LwM2M Device to restore the Device from unexpected firmware failure.]]></Description>
</Item>
<Item ID="5">
<Name>Factory Reset</Name>
<Operations>E</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type></Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Perform factory reset of the LwM2M Device to make the LwM2M Device to go through initial deployment sequence where provisioning and bootstrap sequence is performed. This requires client ensuring post factory reset to have minimal information to allow it to carry out one of the bootstrap methods specified in section 5.2.3.
When this Resource is executed, "De-register" operation MAY be sent to the LwM2M Server(s) before factory reset of the LwM2M Device.]]></Description>
</Item>
<Item ID="6">
<Name>Available Power Sources</Name>
<Operations>R</Operations>
<MultipleInstances>Multiple</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>Integer</Type>
<RangeEnumeration>0..7</RangeEnumeration>
<Units></Units>
<Description><![CDATA[0: DC power
1: Internal Battery
2: External Battery
3: Fuel Cell
4: Power over Ethernet
5: USB
6: AC (Mains) power
7: Solar
The same Resource Instance ID MUST be used to associate a given Power Source (Resource ID:6) with its Present Voltage (Resource ID:7) and its Present Current (Resource ID:8)]]></Description>
</Item>
<Item ID="7">
<Name>Power Source Voltage</Name>
<Operations>R</Operations>
<MultipleInstances>Multiple</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>Integer</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Present voltage for each Available Power Sources Resource Instance. The unit used for this resource is in mV.]]></Description>
</Item>
<Item ID="8">
<Name>Power Source Current</Name>
<Operations>R</Operations>
<MultipleInstances>Multiple</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>Integer</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Present current for each Available Power Source. The unit used for this resource is in mA.]]></Description>
</Item>
<Item ID="9">
<Name>Battery Level</Name>
<Operations>R</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>Integer</Type>
<RangeEnumeration>0..100</RangeEnumeration>
<Units>/100</Units>
<Description><![CDATA[Contains the current battery level as a percentage (with a range from 0 to 100). This value is only valid for the Device internal Battery if present (one Available Power Sources Resource Instance is 1).]]></Description>
</Item>
<Item ID="10">
<Name>Memory Free</Name>
<Operations>R</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>Integer</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Estimated current available amount of storage space which can store data and software in the LwM2M Device (expressed in kilobytes). Note: 1 kilobyte corresponds to 1000 bytes.]]></Description>
</Item>
<Item ID="11">
<Name>Error Code</Name>
<Operations>R</Operations>
<MultipleInstances>Multiple</MultipleInstances>
<Mandatory>Mandatory</Mandatory>
<Type>Integer</Type>
<RangeEnumeration>0..32</RangeEnumeration>
<Units></Units>
<Description><![CDATA[0=No error
1=Low battery power
2=External power supply off
3=GPS module failure
4=Low received signal strength
5=Out of memory
6=SMS failure
7=IP connectivity failure
8=Peripheral malfunction
9..15=Reserved for future use
16..32=Device specific error codes
When the single Device Object Instance is initiated, there is only one error code Resource Instance whose value is equal to 0 that means no error. When the first error happens, the LwM2M Client changes error code Resource Instance to any non-zero value to indicate the error type. When any other error happens, a new error code Resource Instance is created. When an error associated with a Resource Instance is no longer present, that Resource Instance is deleted. When the single existing error is no longer present, the LwM2M Client returns to the original no error state where Instance 0 has value 0.
This error code Resource MAY be observed by the LwM2M Server. How to deal with LwM2M Client’s error report depends on the policy of the LwM2M Server. Error codes in between 16 and 32 are specific to the Device and may have different meanings among implementations.]]></Description>
</Item>
<Item ID="12">
<Name>Reset Error Code</Name>
<Operations>E</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type></Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Delete all error code Resource Instances and create only one zero-value error code that implies no error, then re-evaluate all error conditions and update and create Resources Instances to capture all current error conditions.]]></Description>
</Item>
<Item ID="13">
<Name>Current Time</Name>
<Operations>RW</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>Time</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Current UNIX time of the LwM2M Client.
The LwM2M Client should be responsible to increase this time value as every second elapses.
The LwM2M Server is able to write this Resource to make the LwM2M Client synchronized with the LwM2M Server.]]></Description>
</Item>
<Item ID="14">
<Name>UTC Offset</Name>
<Operations>RW</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>String</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Indicates the UTC offset currently in effect for this LwM2M Device. UTC+X [ISO 8601].]]></Description>
</Item>
<Item ID="15">
<Name>Timezone</Name>
<Operations>RW</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>String</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Indicates in which time zone the LwM2M Device is located, in IANA Timezone (TZ) database format.]]></Description>
</Item>
<Item ID="16">
<Name>Supported Binding and Modes</Name>
<Operations>R</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Mandatory</Mandatory>
<Type>String</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Indicates which bindings and modes are supported in the LwM2M Client. The possible values are those listed in the LwM2M Core Specification.]]></Description>
</Item>
<Item ID="17"><Name>Device Type</Name>
<Operations>R</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>String</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Type of the device (manufacturer specified string: e.g. smart meters / dev Class / ...)]]></Description>
</Item>
<Item ID="18"><Name>Hardware Version</Name>
<Operations>R</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>String</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Current hardware version of the device]]></Description>
</Item>
<Item ID="19"><Name>Software Version</Name>
<Operations>R</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>String</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Current software version of the device (manufacturer specified string). On elaborated LwM2M device, SW could be split in 2 parts: a firmware one and a higher level software on top.
Both pieces of Software are together managed by LwM2M Firmware Update Object (Object ID 5)]]></Description>
</Item>
<Item ID="20"><Name>Battery Status</Name>
<Operations>R</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>Integer</Type>
<RangeEnumeration>0..6</RangeEnumeration>
<Units></Units>
<Description><![CDATA[This value is only valid for the Device Internal Battery if present (one Available Power Sources Resource Instance value is 1).
Battery
Status Meaning Description
0 Normal The battery is operating normally and not on power.
1 Charging The battery is currently charging.
2 Charge Complete The battery is fully charged and still on power.
3 Damaged The battery has some problem.
4 Low Battery The battery is low on charge.
5 Not Installed The battery is not installed.
6 Unknown The battery information is not available.]]></Description>
</Item>
<Item ID="21"><Name>Memory Total</Name>
<Operations>R</Operations>
<MultipleInstances>Single</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>Integer</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Total amount of storage space which can store data and software in the LwM2M Device (expressed in kilobytes). Note: 1 kilobyte corresponds to 1000 bytes.]]></Description>
</Item>
<Item ID="22"><Name>ExtDevInfo</Name>
<Operations>R</Operations>
<MultipleInstances>Multiple</MultipleInstances>
<Mandatory>Optional</Mandatory>
<Type>Objlnk</Type>
<RangeEnumeration></RangeEnumeration>
<Units></Units>
<Description><![CDATA[Reference to external "Device" object instance containing information. For example, such an external device can be a Host Device, which is a device into which the Device containing the LwM2M client is embedded. This Resource may be used to retrieve information about the Host Device.]]></Description>
</Item></Resources>
<Description2></Description2>
</Object>
</LWM2M>

2
monitoring/src/main/resources/lwm2m/resource.json

File diff suppressed because one or more lines are too long

2
msa/vc-executor/src/main/resources/tb-vc-executor.yml

@ -151,6 +151,8 @@ queue:
core:
# Default topic name
topic: "${TB_QUEUE_CORE_TOPIC:tb_core}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_CORE_NOTIFICATIONS_TOPIC:tb_core.notifications}"
# Interval in milliseconds to poll messages by Core microservices
poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}"
# Amount of partitions used by Core microservices

2
pom.xml

@ -850,6 +850,8 @@
<exclude>.run/**</exclude>
<exclude>**/NetworkReceive.java</exclude>
<exclude>**/lwm2m-registry/**</exclude>
<exclude>**/test/resources/lwm2m/**</exclude>
<exclude>**/resources/lwm2m/models/**</exclude>
<exclude>src/main/data/resources/**</exclude>
</excludes>
<mapping>

4
transport/coap/src/main/resources/tb-coap-transport.yml

@ -347,6 +347,8 @@ queue:
core:
# Default topic name
topic: "${TB_QUEUE_CORE_TOPIC:tb_core}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_CORE_NOTIFICATIONS_TOPIC:tb_core.notifications}"
# Interval in milliseconds to poll messages by Core microservices
poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}"
# Amount of partitions used by Core microservices
@ -377,6 +379,8 @@ queue:
rule-engine:
# Deprecated. It will be removed in the nearest releases
topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb_rule_engine}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_RULE_ENGINE_NOTIFICATIONS_TOPIC:tb_rule_engine.notifications}"
# Interval in milliseconds to poll messages by Rule Engine
poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}"
# Timeout for processing a message pack of Rule Engine

4
transport/http/src/main/resources/tb-http-transport.yml

@ -296,6 +296,8 @@ queue:
core:
# Default topic name
topic: "${TB_QUEUE_CORE_TOPIC:tb_core}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_CORE_NOTIFICATIONS_TOPIC:tb_core.notifications}"
# Interval in milliseconds to poll messages by Core microservices
poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}"
# Amount of partitions used by Core microservices
@ -326,6 +328,8 @@ queue:
rule-engine:
# Deprecated. It will be removed in the nearest releases
topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb_rule_engine}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_RULE_ENGINE_NOTIFICATIONS_TOPIC:tb_rule_engine.notifications}"
# Interval in milliseconds to poll messages by Rule Engine
poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}"
# Timeout for processing a message pack of Rule Engine

4
transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml

@ -397,6 +397,8 @@ queue:
core:
# Default topic name
topic: "${TB_QUEUE_CORE_TOPIC:tb_core}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_CORE_NOTIFICATIONS_TOPIC:tb_core.notifications}"
# Interval in milliseconds to poll messages by Core microservices
poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}"
# Amount of partitions used by Core microservices
@ -427,6 +429,8 @@ queue:
rule-engine:
# Deprecated. It will be removed in the nearest releases
topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb_rule_engine}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_RULE_ENGINE_NOTIFICATIONS_TOPIC:tb_rule_engine.notifications}"
# Interval in milliseconds to poll messages by Rule Engine
poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}"
# Timeout for processing a message pack of Rule Engine

4
transport/mqtt/src/main/resources/tb-mqtt-transport.yml

@ -330,6 +330,8 @@ queue:
core:
# Default topic name
topic: "${TB_QUEUE_CORE_TOPIC:tb_core}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_CORE_NOTIFICATIONS_TOPIC:tb_core.notifications}"
# Interval in milliseconds to poll messages by Core microservices
poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}"
# Amount of partitions used by Core microservices
@ -360,6 +362,8 @@ queue:
rule-engine:
# Deprecated. It will be removed in the nearest releases
topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb_rule_engine}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_RULE_ENGINE_NOTIFICATIONS_TOPIC:tb_rule_engine.notifications}"
# Interval in milliseconds to poll messages by Rule Engine
poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}"
# Timeout for processing a message pack of Rule Engine

4
transport/snmp/src/main/resources/tb-snmp-transport.yml

@ -283,6 +283,8 @@ queue:
core:
# Default topic name
topic: "${TB_QUEUE_CORE_TOPIC:tb_core}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_CORE_NOTIFICATIONS_TOPIC:tb_core.notifications}"
# Interval in milliseconds to poll messages by Core microservices
poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}"
# Amount of partitions used by Core microservices
@ -313,6 +315,8 @@ queue:
rule-engine:
# Deprecated. It will be removed in the nearest releases
topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb_rule_engine}"
# For high-priority notifications that require minimum latency and processing time
notifications_topic: "${TB_QUEUE_RULE_ENGINE_NOTIFICATIONS_TOPIC:tb_rule_engine.notifications}"
# Interval in milliseconds to poll messages by Rule Engine
poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}"
# Timeout for processing a message pack of Rule Engine

Loading…
Cancel
Save