typesWithNullRuleNodeConnection = List.of(
+ ALARM,
+ ALARM_DELETE,
+ ENTITY_ASSIGNED_TO_EDGE,
+ ENTITY_UNASSIGNED_FROM_EDGE,
+ PROVISION_FAILURE,
+ PROVISION_SUCCESS,
+ SEND_EMAIL,
+ GENERATOR_NODE_SELF_MSG,
+ DEVICE_PROFILE_PERIODIC_SELF_MSG,
+ DEVICE_PROFILE_UPDATE_SELF_MSG,
+ DEVICE_UPDATE_SELF_MSG,
+ DEDUPLICATION_TIMEOUT_SELF_MSG,
+ DELAY_TIMEOUT_SELF_MSG,
+ MSG_COUNT_SELF_MSG
+ );
+
+ // backward-compatibility tests
+
+ @Test
+ void getRuleNodeConnectionsTest() {
+ var tbMsgTypes = TbMsgType.values();
+ for (var type : tbMsgTypes) {
+ if (typesWithNullRuleNodeConnection.contains(type)) {
+ assertThat(type.getRuleNodeConnection()).isNull();
+ } else {
+ assertThat(type.getRuleNodeConnection()).isNotNull();
+ }
+ }
+ }
+
+ @Test
+ void getRuleNodeConnectionOrElseOtherTest() {
+ assertThat(TbMsgType.getRuleNodeConnectionOrElseOther(null))
+ .isEqualTo(TbNodeConnectionType.OTHER);
+ var tbMsgTypes = TbMsgType.values();
+ for (var type : tbMsgTypes) {
+ if (typesWithNullRuleNodeConnection.contains(type)) {
+ assertThat(TbMsgType.getRuleNodeConnectionOrElseOther(type.name()))
+ .isEqualTo(TbNodeConnectionType.OTHER);
+ } else {
+ assertThat(TbMsgType.getRuleNodeConnectionOrElseOther(type.name())).isNotNull()
+ .isNotEqualTo(TbNodeConnectionType.OTHER);
+ }
+ }
+ }
+
+}
diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
index 17bc7c0a8f..a987a4a253 100644
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java
@@ -29,10 +29,12 @@ import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityIdFactory;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.RuleNodeId;
+import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.msg.gen.MsgProtos;
import org.thingsboard.server.common.msg.queue.TbMsgCallback;
import java.io.Serializable;
+import java.util.Objects;
import java.util.UUID;
/**
@@ -42,6 +44,10 @@ import java.util.UUID;
@Slf4j
public final class TbMsg implements Serializable {
+ public static final String EMPTY_JSON_OBJECT = "{}";
+ public static final String EMPTY_JSON_ARRAY = "[]";
+ public static final String EMPTY_STRING = "";
+
private final String queueName;
private final UUID id;
private final long ts;
@@ -66,82 +72,234 @@ public final class TbMsg implements Serializable {
return ctx.getAndIncrementRuleNodeCounter();
}
+ @Deprecated(since = "3.5.2", forRemoval = true)
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);
}
+ /**
+ * Creates a new TbMsg instance with the specified parameters.
+ *
+ * Deprecated: This method is deprecated since version 3.5.2 and should only be used when you need to
+ * specify a custom message type that doesn't exist in the {@link TbMsgType} enum. For standard message types,
+ * it is recommended to use the {@link #newMsg(String, TbMsgType, EntityId, CustomerId, TbMsgMetaData, String, RuleChainId, RuleNodeId)}
+ * method instead.
+ *
+ * @param queueName the name of the queue where the message will be sent
+ * @param type the type of the message
+ * @param originator the originator of the message
+ * @param customerId the ID of the customer associated with the message
+ * @param metaData the metadata of the message
+ * @param data the data of the message
+ * @param ruleChainId the ID of the rule chain associated with the message
+ * @param ruleNodeId the ID of the rule node associated with the message
+ * @return new TbMsg instance
+ */
+ @Deprecated(since = "3.5.2")
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,
metaData.copy(), TbMsgDataType.JSON, data, ruleChainId, ruleNodeId, null, TbMsgCallback.EMPTY);
}
+ @Deprecated(since = "3.5.2", forRemoval = true)
public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data) {
return newMsg(type, originator, null, metaData, data);
}
+ @Deprecated(since = "3.5.2", forRemoval = true)
public static TbMsg newMsg(String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
return new TbMsg(null, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
metaData.copy(), TbMsgDataType.JSON, data, null, null, null, TbMsgCallback.EMPTY);
}
+ public static TbMsg newMsg(String queueName, TbMsgType 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(String queueName, TbMsgType type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
+ return new TbMsg(queueName, UUID.randomUUID(), System.currentTimeMillis(), type.name(), originator, customerId,
+ metaData.copy(), TbMsgDataType.JSON, data, ruleChainId, ruleNodeId, null, TbMsgCallback.EMPTY);
+ }
+
+ public static TbMsg newMsg(TbMsgType type, EntityId originator, TbMsgMetaData metaData, String data) {
+ return newMsg(type, originator, null, metaData, data);
+ }
+
+ public static TbMsg newMsg(TbMsgType type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
+ return new TbMsg(null, UUID.randomUUID(), System.currentTimeMillis(), type.name(), originator, customerId,
+ metaData.copy(), TbMsgDataType.JSON, data, null, null, null, TbMsgCallback.EMPTY);
+ }
+
// REALLY NEW MSG
+ /**
+ * Creates a new TbMsg instance with the specified parameters.
+ *
+ * Deprecated: This method is deprecated since version 3.5.2 and should only be used when you need to
+ * specify a custom message type that doesn't exist in the {@link TbMsgType} enum. For standard message types,
+ * it is recommended to use the {@link #newMsg(String, TbMsgType, EntityId, TbMsgMetaData, String)}
+ * method instead.
+ *
+ * @param queueName the name of the queue where the message will be sent
+ * @param type the type of the message
+ * @param originator the originator of the message
+ * @param metaData the metadata of the message
+ * @param data the data of the message
+ * @return new TbMsg instance
+ */
+ @Deprecated(since = "3.5.2")
public static TbMsg newMsg(String queueName, String type, EntityId originator, TbMsgMetaData metaData, String data) {
return newMsg(queueName, type, originator, null, metaData, data);
}
+ /**
+ * Creates a new TbMsg instance with the specified parameters.
+ *
+ * Deprecated: This method is deprecated since version 3.5.2 and should only be used when you need to
+ * specify a custom message type that doesn't exist in the {@link TbMsgType} enum. For standard message types,
+ * it is recommended to use the {@link #newMsg(String, TbMsgType, EntityId, CustomerId, TbMsgMetaData, String)}
+ * method instead.
+ *
+ * @param queueName the name of the queue where the message will be sent
+ * @param type the type of the message
+ * @param originator the originator of the message
+ * @param customerId the ID of the customer associated with the message
+ * @param metaData the metadata of the message
+ * @param data the data of the message
+ * @return new TbMsg instance
+ */
+ @Deprecated(since = "3.5.2")
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,
metaData.copy(), TbMsgDataType.JSON, data, null, null, null, TbMsgCallback.EMPTY);
}
+ @Deprecated(since = "3.5.2", forRemoval = true)
public static TbMsg newMsg(String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data) {
return new TbMsg(null, UUID.randomUUID(), System.currentTimeMillis(), type, originator, customerId,
metaData.copy(), dataType, data, null, null, null, TbMsgCallback.EMPTY);
}
+ /**
+ * Creates a new TbMsg instance with the specified parameters.
+ *
+ * Deprecated: This method is deprecated since version 3.5.2 and should only be used when you need to
+ * specify a custom message type that doesn't exist in the {@link TbMsgType} enum. For standard message types,
+ * it is recommended to use the {@link #newMsg(TbMsgType, EntityId, TbMsgMetaData, TbMsgDataType, String)}
+ * method instead.
+ *
+ * @param type the type of the message
+ * @param originator the originator of the message
+ * @param metaData the metadata of the message
+ * @param dataType the dataType of the message
+ * @param data the data of the message
+ * @return new TbMsg instance
+ */
+ @Deprecated(since = "3.5.2")
public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data) {
return newMsg(type, originator, null, metaData, dataType, data);
}
+ public static TbMsg newMsg(String queueName, TbMsgType type, EntityId originator, TbMsgMetaData metaData, String data) {
+ return newMsg(queueName, type, originator, null, metaData, data);
+ }
+
+ public static TbMsg newMsg(String queueName, TbMsgType type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data) {
+ return new TbMsg(queueName, UUID.randomUUID(), System.currentTimeMillis(), type.name(), originator, customerId,
+ metaData.copy(), TbMsgDataType.JSON, data, null, null, null, TbMsgCallback.EMPTY);
+ }
+
+ public static TbMsg newMsg(TbMsgType type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, TbMsgDataType dataType, String data) {
+ return new TbMsg(null, UUID.randomUUID(), System.currentTimeMillis(), type.name(), originator, customerId,
+ metaData.copy(), dataType, data, null, null, null, TbMsgCallback.EMPTY);
+ }
+
+ public static TbMsg newMsg(TbMsgType type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data) {
+ return newMsg(type, originator, null, metaData, dataType, data);
+ }
+
// For Tests only
+ @Deprecated(since = "3.5.2", forRemoval = true)
public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
return new TbMsg(null, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null,
metaData.copy(), dataType, data, ruleChainId, ruleNodeId, null, TbMsgCallback.EMPTY);
}
+ @Deprecated(since = "3.5.2", forRemoval = true)
public static TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data, TbMsgCallback callback) {
return new TbMsg(null, UUID.randomUUID(), System.currentTimeMillis(), type, originator, null,
metaData.copy(), TbMsgDataType.JSON, data, null, null, null, callback);
}
+ /**
+ * Transforms an existing TbMsg instance by changing its message type, originator, metadata, and data.
+ *
+ * Deprecated: This method is deprecated since version 3.5.2 and should only be used when you need to
+ * specify a custom message type that doesn't exist in the {@link TbMsgType} enum. For standard message types,
+ * it is recommended to use the {@link #transformMsg(TbMsg, TbMsgType, EntityId, TbMsgMetaData, String)}
+ * method instead.
+ *
+ *
+ * @param tbMsg the TbMsg instance to transform
+ * @param type the new message type
+ * @param originator the new originator
+ * @param metaData the new metadata
+ * @param data the new data
+ * @return the transformed TbMsg instance
+ */
+ @Deprecated(since = "3.5.2")
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,
data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.callback);
}
+ public static TbMsg newMsg(TbMsgType type, EntityId originator, TbMsgMetaData metaData, TbMsgDataType dataType, String data, RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
+ return new TbMsg(null, UUID.randomUUID(), System.currentTimeMillis(), type.name(), originator, null,
+ metaData.copy(), dataType, data, ruleChainId, ruleNodeId, null, TbMsgCallback.EMPTY);
+ }
+
+ public static TbMsg newMsg(TbMsgType type, EntityId originator, TbMsgMetaData metaData, String data, TbMsgCallback callback) {
+ return new TbMsg(null, UUID.randomUUID(), System.currentTimeMillis(), type.name(), originator, null,
+ metaData.copy(), TbMsgDataType.JSON, data, null, null, null, callback);
+ }
+
+ public static TbMsg transformMsg(TbMsg tbMsg, TbMsgType type, EntityId originator, TbMsgMetaData metaData, String data) {
+ return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, type.name(), originator, tbMsg.customerId, metaData.copy(), tbMsg.dataType,
+ data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.callback);
+ }
+
+ public static TbMsg transformMsgOriginator(TbMsg tbMsg, EntityId originatorId) {
+ return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.type, originatorId, tbMsg.getCustomerId(), tbMsg.metaData, tbMsg.dataType,
+ tbMsg.data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.getCallback());
+ }
+
public static TbMsg transformMsgData(TbMsg tbMsg, String data) {
return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.getCallback());
}
- public static TbMsg transformMsg(TbMsg tbMsg, TbMsgMetaData metadata) {
+ public static TbMsg transformMsgMetadata(TbMsg tbMsg, TbMsgMetaData metadata) {
return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, metadata.copy(), tbMsg.dataType,
tbMsg.data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.getCallback());
}
- public static TbMsg transformMsg(TbMsg tbMsg, CustomerId customerId) {
+ public static TbMsg transformMsg(TbMsg tbMsg, TbMsgMetaData metadata, String data) {
+ return new TbMsg(tbMsg.queueName, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, metadata, tbMsg.dataType,
+ data, tbMsg.ruleChainId, tbMsg.ruleNodeId, tbMsg.ctx.copy(), tbMsg.getCallback());
+ }
+
+ public static TbMsg transformMsgCustomerId(TbMsg tbMsg, CustomerId customerId) {
return new TbMsg(tbMsg.queueName, 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) {
+ public static TbMsg transformMsgRuleChainId(TbMsg tbMsg, RuleChainId ruleChainId) {
return new TbMsg(tbMsg.queueName, 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) {
+ public static TbMsg transformMsgQueueName(TbMsg tbMsg, String queueName) {
return new TbMsg(queueName, tbMsg.id, tbMsg.ts, tbMsg.type, tbMsg.originator, tbMsg.customerId, tbMsg.metaData, tbMsg.dataType,
tbMsg.data, tbMsg.getRuleChainId(), null, tbMsg.ctx.copy(), tbMsg.getCallback());
}
@@ -183,11 +341,7 @@ public final class TbMsg implements Serializable {
this.ruleChainId = ruleChainId;
this.ruleNodeId = ruleNodeId;
this.ctx = ctx != null ? ctx : new TbMsgProcessingCtx();
- if (callback != null) {
- this.callback = callback;
- } else {
- this.callback = TbMsgCallback.EMPTY;
- }
+ this.callback = Objects.requireNonNullElse(callback, TbMsgCallback.EMPTY);
}
public static ByteString toByteString(TbMsg msg) {
@@ -284,11 +438,7 @@ public final class TbMsg implements Serializable {
public TbMsgCallback getCallback() {
// May be null in case of deserialization;
- if (callback != null) {
- return callback;
- } else {
- return TbMsgCallback.EMPTY;
- }
+ return Objects.requireNonNullElse(callback, TbMsgCallback.EMPTY);
}
public void pushToStack(RuleChainId ruleChainId, RuleNodeId ruleNodeId) {
@@ -317,4 +467,18 @@ public final class TbMsg implements Serializable {
}
return ts;
}
+
+ public boolean isTypeOf(TbMsgType tbMsgType) {
+ return tbMsgType != null && tbMsgType.name().equals(this.type);
+ }
+
+ public boolean isTypeOneOf(TbMsgType... types) {
+ for (TbMsgType type : types) {
+ if (isTypeOf(type)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
}
diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/session/SessionMsgType.java b/common/message/src/main/java/org/thingsboard/server/common/msg/session/SessionMsgType.java
index dfcf178473..ca7c94e9ce 100644
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/session/SessionMsgType.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/session/SessionMsgType.java
@@ -15,6 +15,12 @@
*/
package org.thingsboard.server.common.msg.session;
+/**
+ * @deprecated This enum is deprecated and will be removed in a future version.
+ * Note: This enum was originally part of the public API but is now specific to CoAP transport only.
+ * Please use {@link org.thingsboard.server.transport.coap.CoapSessionMsgType} instead.
+ */
+@Deprecated(since="3.5.2", forRemoval = true)
public enum SessionMsgType {
GET_ATTRIBUTES_REQUEST(true), POST_ATTRIBUTES_REQUEST(true), GET_ATTRIBUTES_RESPONSE,
SUBSCRIBE_ATTRIBUTES_REQUEST, UNSUBSCRIBE_ATTRIBUTES_REQUEST, ATTRIBUTES_UPDATE_NOTIFICATION,
diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ZkDiscoveryService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ZkDiscoveryService.java
index fcf80bcf3d..44999d016a 100644
--- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ZkDiscoveryService.java
+++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ZkDiscoveryService.java
@@ -16,6 +16,7 @@
package org.thingsboard.server.queue.discovery;
import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.ProtocolStringList;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
@@ -44,8 +45,10 @@ import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.NoSuchElementException;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -66,6 +69,10 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi
private Integer zkSessionTimeout;
@Value("${zk.zk_dir}")
private String zkDir;
+ @Value("${zk.recalculate_delay:60000}")
+ private Long recalculateDelay;
+
+ protected final ConcurrentHashMap> delayedTasks;
private final TbServiceInfoProvider serviceInfoProvider;
private final PartitionService partitionService;
@@ -82,6 +89,7 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi
PartitionService partitionService) {
this.serviceInfoProvider = serviceInfoProvider;
this.partitionService = partitionService;
+ delayedTasks = new ConcurrentHashMap<>();
}
@PostConstruct
@@ -287,11 +295,39 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi
log.error("Failed to decode server instance for node {}", data.getPath(), e);
throw e;
}
- log.debug("Processing [{}] event for [{}]", pathChildrenCacheEvent.getType(), instance.getServiceId());
+
+ String serviceId = instance.getServiceId();
+ ProtocolStringList serviceTypesList = instance.getServiceTypesList();
+
+ log.trace("Processing [{}] event for [{}]", pathChildrenCacheEvent.getType(), serviceId);
switch (pathChildrenCacheEvent.getType()) {
case CHILD_ADDED:
+ ScheduledFuture task = delayedTasks.remove(serviceId);
+ if (task != null) {
+ if (task.cancel(false)) {
+ log.debug("[{}] Recalculate partitions ignored. Service was restarted in time [{}].",
+ serviceId, serviceTypesList);
+ } else {
+ log.debug("[{}] Going to recalculate partitions. Service was not restarted in time [{}]!",
+ serviceId, serviceTypesList);
+ recalculatePartitions();
+ }
+ } else {
+ log.trace("[{}] Going to recalculate partitions due to adding new node [{}].",
+ serviceId, serviceTypesList);
+ recalculatePartitions();
+ }
+ break;
case CHILD_REMOVED:
- recalculatePartitions();
+ ScheduledFuture future = zkExecutorService.schedule(() -> {
+ log.debug("[{}] Going to recalculate partitions due to removed node [{}]",
+ serviceId, serviceTypesList);
+ ScheduledFuture removedTask = delayedTasks.remove(serviceId);
+ if (removedTask != null) {
+ recalculatePartitions();
+ }
+ }, recalculateDelay, TimeUnit.MILLISECONDS);
+ delayedTasks.put(serviceId, future);
break;
default:
break;
@@ -303,6 +339,8 @@ public class ZkDiscoveryService implements DiscoveryService, PathChildrenCacheLi
* Synchronized to ensure that other servers info is up to date
* */
synchronized void recalculatePartitions() {
+ delayedTasks.values().forEach(future -> future.cancel(false));
+ delayedTasks.clear();
partitionService.recalculatePartitions(serviceInfoProvider.getServiceInfo(), getOtherServers());
}
diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/discovery/ZkDiscoveryServiceTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/discovery/ZkDiscoveryServiceTest.java
new file mode 100644
index 0000000000..a8810efd0e
--- /dev/null
+++ b/common/queue/src/test/java/org/thingsboard/server/queue/discovery/ZkDiscoveryServiceTest.java
@@ -0,0 +1,189 @@
+/**
+ * Copyright © 2016-2023 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.queue.discovery;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.imps.CuratorFrameworkState;
+import org.apache.curator.framework.recipes.cache.ChildData;
+import org.apache.curator.framework.recipes.cache.PathChildrenCache;
+import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.gen.transport.TransportProtos;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type.CHILD_ADDED;
+import static org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type.CHILD_REMOVED;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ZkDiscoveryServiceTest {
+
+ @Mock
+ private TbServiceInfoProvider serviceInfoProvider;
+
+ @Mock
+ private PartitionService partitionService;
+
+ @Mock
+ private CuratorFramework client;
+
+ @Mock
+ private PathChildrenCache cache;
+
+ @Mock
+ private CuratorFramework curatorFramework;
+
+ private ZkDiscoveryService zkDiscoveryService;
+
+ private static final long RECALCULATE_DELAY = 100L;
+
+ final TransportProtos.ServiceInfo currentInfo = TransportProtos.ServiceInfo.newBuilder().setServiceId("tb-rule-engine-0").build();
+ final ChildData currentData = new ChildData("/thingsboard/nodes/0000000010", null, currentInfo.toByteArray());
+ final TransportProtos.ServiceInfo childInfo = TransportProtos.ServiceInfo.newBuilder().setServiceId("tb-rule-engine-1").build();
+ final ChildData childData = new ChildData("/thingsboard/nodes/0000000020", null, childInfo.toByteArray());
+
+ @Before
+ public void setup() {
+ zkDiscoveryService = Mockito.spy(new ZkDiscoveryService(serviceInfoProvider, partitionService));
+ ScheduledExecutorService zkExecutorService = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("zk-discovery"));
+ when(client.getState()).thenReturn(CuratorFrameworkState.STARTED);
+ ReflectionTestUtils.setField(zkDiscoveryService, "stopped", false);
+ ReflectionTestUtils.setField(zkDiscoveryService, "client", client);
+ ReflectionTestUtils.setField(zkDiscoveryService, "cache", cache);
+ ReflectionTestUtils.setField(zkDiscoveryService, "nodePath", "/thingsboard/nodes/0000000010");
+ ReflectionTestUtils.setField(zkDiscoveryService, "zkExecutorService", zkExecutorService);
+ ReflectionTestUtils.setField(zkDiscoveryService, "recalculateDelay", RECALCULATE_DELAY);
+ ReflectionTestUtils.setField(zkDiscoveryService, "zkDir", "/thingsboard");
+
+ when(serviceInfoProvider.getServiceInfo()).thenReturn(currentInfo);
+
+ List dataList = new ArrayList<>();
+ dataList.add(currentData);
+ when(cache.getCurrentData()).thenReturn(dataList);
+ }
+
+ @Test
+ public void restartNodeInTimeTest() throws Exception {
+ startNode(childData);
+
+ verify(partitionService, times(1)).recalculatePartitions(eq(currentInfo), eq(List.of(childInfo)));
+
+ reset(partitionService);
+
+ stopNode(childData);
+
+ assertEquals(1, zkDiscoveryService.delayedTasks.size());
+
+ verify(partitionService, never()).recalculatePartitions(any(), any());
+
+ startNode(childData);
+
+ verify(partitionService, never()).recalculatePartitions(any(), any());
+
+ Thread.sleep(RECALCULATE_DELAY * 2);
+
+ verify(partitionService, never()).recalculatePartitions(any(), any());
+
+ assertTrue(zkDiscoveryService.delayedTasks.isEmpty());
+ }
+
+ @Test
+ public void restartNodeNotInTimeTest() throws Exception {
+ startNode(childData);
+
+ verify(partitionService, times(1)).recalculatePartitions(eq(currentInfo), eq(List.of(childInfo)));
+
+ reset(partitionService);
+
+ stopNode(childData);
+
+ assertEquals(1, zkDiscoveryService.delayedTasks.size());
+
+ Thread.sleep(RECALCULATE_DELAY * 2);
+
+ assertTrue(zkDiscoveryService.delayedTasks.isEmpty());
+
+ startNode(childData);
+
+ verify(partitionService, times(1)).recalculatePartitions(eq(currentInfo), eq(Collections.emptyList()));
+
+ verify(partitionService, times(1)).recalculatePartitions(eq(currentInfo), eq(List.of(childInfo)));
+
+ reset(partitionService);
+ }
+
+ @Test
+ public void startAnotherNodeDuringRestartTest() throws Exception {
+ var anotherInfo = TransportProtos.ServiceInfo.newBuilder().setServiceId("tb-transport").build();
+ var anotherData = new ChildData("/thingsboard/nodes/0000000030", null, anotherInfo.toByteArray());
+
+ startNode(childData);
+
+ verify(partitionService, times(1)).recalculatePartitions(eq(currentInfo), eq(List.of(childInfo)));
+
+ reset(partitionService);
+
+ stopNode(childData);
+
+ assertEquals(1, zkDiscoveryService.delayedTasks.size());
+
+ startNode(anotherData);
+
+ assertTrue(zkDiscoveryService.delayedTasks.isEmpty());
+
+ verify(partitionService, times(1)).recalculatePartitions(eq(currentInfo), eq(List.of(anotherInfo)));
+ reset(partitionService);
+
+ Thread.sleep(RECALCULATE_DELAY * 2);
+
+ verify(partitionService, never()).recalculatePartitions(any(), any());
+
+ startNode(childData);
+
+ verify(partitionService, times(1)).recalculatePartitions(eq(currentInfo), eq(List.of(anotherInfo, childInfo)));
+ }
+
+ private void startNode(ChildData data) throws Exception {
+ cache.getCurrentData().add(data);
+ zkDiscoveryService.childEvent(curatorFramework, new PathChildrenCacheEvent(CHILD_ADDED, data));
+ }
+
+ private void stopNode(ChildData data) throws Exception {
+ cache.getCurrentData().remove(data);
+ zkDiscoveryService.childEvent(curatorFramework, new PathChildrenCacheEvent(CHILD_REMOVED, data));
+ }
+
+}
diff --git a/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/DefaultTbelInvokeService.java b/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/DefaultTbelInvokeService.java
index bbf441a659..2a60980f84 100644
--- a/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/DefaultTbelInvokeService.java
+++ b/common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/DefaultTbelInvokeService.java
@@ -66,8 +66,6 @@ public class DefaultTbelInvokeService extends AbstractScriptInvokeService implem
protected final Map scriptIdToHash = new ConcurrentHashMap<>();
protected final Map scriptMap = new ConcurrentHashMap<>();
- private final String tbelSwitch = "switch";
- private final String tbelSwitchErrorMsg = "TBEL does not support the 'switch'.";
protected Cache compiledScriptsCache;
private SandboxedParserConfiguration parserConfig;
@@ -183,11 +181,6 @@ public class DefaultTbelInvokeService extends AbstractScriptInvokeService implem
lock.unlock();
}
return scriptId;
- } catch (CompileException ce) {
- if ( ce.getExpr() != null && new String(ce.getExpr()).contains(tbelSwitch)) {
- ce = new CompileException(tbelSwitchErrorMsg, ce.getExpr(), ce.getCursor(), ce.getCause());
- }
- throw new TbScriptException(scriptId, TbScriptException.ErrorCode.COMPILATION, scriptBody, ce);
} catch (Exception e) {
throw new TbScriptException(scriptId, TbScriptException.ErrorCode.COMPILATION, scriptBody, e);
}
diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapSessionMsgType.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapSessionMsgType.java
new file mode 100644
index 0000000000..fcf33ea037
--- /dev/null
+++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapSessionMsgType.java
@@ -0,0 +1,31 @@
+/**
+ * Copyright © 2016-2023 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.transport.coap;
+
+public enum CoapSessionMsgType {
+
+ GET_ATTRIBUTES_REQUEST,
+ POST_ATTRIBUTES_REQUEST,
+ SUBSCRIBE_ATTRIBUTES_REQUEST,
+ UNSUBSCRIBE_ATTRIBUTES_REQUEST,
+ POST_TELEMETRY_REQUEST,
+ SUBSCRIBE_RPC_COMMANDS_REQUEST,
+ UNSUBSCRIBE_RPC_COMMANDS_REQUEST,
+ TO_DEVICE_RPC_RESPONSE,
+ TO_SERVER_RPC_REQUEST,
+ CLAIM_REQUEST;
+
+}
diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java
index 7dde25bfd0..d6958137c7 100644
--- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java
+++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java
@@ -33,7 +33,6 @@ import org.thingsboard.server.common.data.DeviceTransportType;
import org.thingsboard.server.common.data.TransportPayloadType;
import org.thingsboard.server.common.data.security.DeviceTokenCredentials;
import org.thingsboard.server.common.msg.session.FeatureType;
-import org.thingsboard.server.common.msg.session.SessionMsgType;
import org.thingsboard.server.common.transport.TransportServiceCallback;
import org.thingsboard.server.common.transport.adaptor.AdaptorException;
import org.thingsboard.server.common.transport.adaptor.JsonConverter;
@@ -120,7 +119,7 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
} else if (exchange.getRequestOptions().hasObserve()) {
processExchangeGetRequest(exchange, featureType.get());
} else if (featureType.get() == FeatureType.ATTRIBUTES) {
- processRequest(exchange, SessionMsgType.GET_ATTRIBUTES_REQUEST);
+ processRequest(exchange, CoapSessionMsgType.GET_ATTRIBUTES_REQUEST);
} else {
log.trace("Invalid feature type parameter");
exchange.respond(CoAP.ResponseCode.BAD_REQUEST);
@@ -129,13 +128,13 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
private void processExchangeGetRequest(CoapExchange exchange, FeatureType featureType) {
boolean unsubscribe = exchange.getRequestOptions().getObserve() == 1;
- SessionMsgType sessionMsgType;
+ CoapSessionMsgType coapSessionMsgType;
if (featureType == FeatureType.RPC) {
- sessionMsgType = unsubscribe ? SessionMsgType.UNSUBSCRIBE_RPC_COMMANDS_REQUEST : SessionMsgType.SUBSCRIBE_RPC_COMMANDS_REQUEST;
+ coapSessionMsgType = unsubscribe ? CoapSessionMsgType.UNSUBSCRIBE_RPC_COMMANDS_REQUEST : CoapSessionMsgType.SUBSCRIBE_RPC_COMMANDS_REQUEST;
} else {
- sessionMsgType = unsubscribe ? SessionMsgType.UNSUBSCRIBE_ATTRIBUTES_REQUEST : SessionMsgType.SUBSCRIBE_ATTRIBUTES_REQUEST;
+ coapSessionMsgType = unsubscribe ? CoapSessionMsgType.UNSUBSCRIBE_ATTRIBUTES_REQUEST : CoapSessionMsgType.SUBSCRIBE_ATTRIBUTES_REQUEST;
}
- processRequest(exchange, sessionMsgType);
+ processRequest(exchange, coapSessionMsgType);
}
@Override
@@ -147,21 +146,21 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
} else {
switch (featureType.get()) {
case ATTRIBUTES:
- processRequest(exchange, SessionMsgType.POST_ATTRIBUTES_REQUEST);
+ processRequest(exchange, CoapSessionMsgType.POST_ATTRIBUTES_REQUEST);
break;
case TELEMETRY:
- processRequest(exchange, SessionMsgType.POST_TELEMETRY_REQUEST);
+ processRequest(exchange, CoapSessionMsgType.POST_TELEMETRY_REQUEST);
break;
case RPC:
Optional requestId = getRequestId(exchange.advanced().getRequest());
if (requestId.isPresent()) {
- processRequest(exchange, SessionMsgType.TO_DEVICE_RPC_RESPONSE);
+ processRequest(exchange, CoapSessionMsgType.TO_DEVICE_RPC_RESPONSE);
} else {
- processRequest(exchange, SessionMsgType.TO_SERVER_RPC_REQUEST);
+ processRequest(exchange, CoapSessionMsgType.TO_SERVER_RPC_REQUEST);
}
break;
case CLAIM:
- processRequest(exchange, SessionMsgType.CLAIM_REQUEST);
+ processRequest(exchange, CoapSessionMsgType.CLAIM_REQUEST);
break;
case PROVISION:
processProvision(exchange);
@@ -195,7 +194,7 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
}
}
- private void processRequest(CoapExchange exchange, SessionMsgType type) {
+ private void processRequest(CoapExchange exchange, CoapSessionMsgType type) {
log.trace("Processing {}", exchange.advanced().getRequest());
deferAccept(exchange);
Exchange advanced = exchange.advanced();
@@ -218,7 +217,7 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
}
}
- private void processAccessTokenRequest(CoapExchange exchange, SessionMsgType type, Request request) {
+ private void processAccessTokenRequest(CoapExchange exchange, CoapSessionMsgType type, Request request) {
Optional credentials = decodeCredentials(request);
if (credentials.isEmpty()) {
exchange.respond(CoAP.ResponseCode.UNAUTHORIZED);
@@ -228,7 +227,7 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
new CoapDeviceAuthCallback(exchange, (deviceCredentials, deviceProfile) -> processRequest(exchange, type, request, deviceCredentials, deviceProfile)));
}
- private void processRequest(CoapExchange exchange, SessionMsgType type, Request request, ValidateDeviceCredentialsResponse deviceCredentials, DeviceProfile deviceProfile) {
+ private void processRequest(CoapExchange exchange, CoapSessionMsgType type, Request request, ValidateDeviceCredentialsResponse deviceCredentials, DeviceProfile deviceProfile) {
TbCoapClientState clientState = null;
try {
clientState = clients.getOrCreateClient(type, deviceCredentials, deviceProfile);
diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/CoapClientContext.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/CoapClientContext.java
index 36ef8472de..a8531a7fba 100644
--- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/CoapClientContext.java
+++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/CoapClientContext.java
@@ -18,7 +18,7 @@ package org.thingsboard.server.transport.coap.client;
import org.eclipse.californium.core.observe.ObserveRelation;
import org.eclipse.californium.core.server.resources.CoapExchange;
import org.thingsboard.server.common.data.DeviceProfile;
-import org.thingsboard.server.common.msg.session.SessionMsgType;
+import org.thingsboard.server.transport.coap.CoapSessionMsgType;
import org.thingsboard.server.common.transport.adaptor.AdaptorException;
import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
import org.thingsboard.server.gen.transport.TransportProtos;
@@ -33,7 +33,7 @@ public interface CoapClientContext {
AtomicInteger getNotificationCounterByToken(String token);
- TbCoapClientState getOrCreateClient(SessionMsgType type, ValidateDeviceCredentialsResponse deviceCredentials, DeviceProfile deviceProfile) throws AdaptorException;
+ TbCoapClientState getOrCreateClient(CoapSessionMsgType type, ValidateDeviceCredentialsResponse deviceCredentials, DeviceProfile deviceProfile) throws AdaptorException;
TransportProtos.SessionInfoProto getNewSyncSession(TbCoapClientState clientState);
diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java
index f70967b72d..26effe7058 100644
--- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java
+++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java
@@ -45,7 +45,7 @@ import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.rpc.RpcStatus;
import org.thingsboard.server.common.msg.session.FeatureType;
-import org.thingsboard.server.common.msg.session.SessionMsgType;
+import org.thingsboard.server.transport.coap.CoapSessionMsgType;
import org.thingsboard.server.common.transport.DeviceDeletedEvent;
import org.thingsboard.server.common.transport.DeviceProfileUpdatedEvent;
import org.thingsboard.server.common.transport.DeviceUpdatedEvent;
@@ -388,7 +388,7 @@ public class DefaultCoapClientContext implements CoapClientContext {
}
@Override
- public TbCoapClientState getOrCreateClient(SessionMsgType type, ValidateDeviceCredentialsResponse deviceCredentials, DeviceProfile deviceProfile) throws AdaptorException {
+ public TbCoapClientState getOrCreateClient(CoapSessionMsgType type, ValidateDeviceCredentialsResponse deviceCredentials, DeviceProfile deviceProfile) throws AdaptorException {
DeviceId deviceId = deviceCredentials.getDeviceInfo().getDeviceId();
TbCoapClientState state = getClientState(deviceId);
state.lock();
diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/model/LwM2MModelConfigServiceImpl.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/model/LwM2MModelConfigServiceImpl.java
index 302bd20c8b..eef9a53024 100644
--- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/model/LwM2MModelConfigServiceImpl.java
+++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/model/LwM2MModelConfigServiceImpl.java
@@ -52,7 +52,7 @@ import java.util.stream.Collectors;
public class LwM2MModelConfigServiceImpl implements LwM2MModelConfigService {
@Autowired
- private TbLwM2MModelConfigStore modelStore;
+ TbLwM2MModelConfigStore modelStore;
@Autowired
@Lazy
@@ -67,14 +67,14 @@ public class LwM2MModelConfigServiceImpl implements LwM2MModelConfigService {
@Autowired
private LwM2MTelemetryLogService logService;
- private ConcurrentMap currentModelConfigs;
+ ConcurrentMap currentModelConfigs;
@AfterStartUp(order = AfterStartUp.BEFORE_TRANSPORT_SERVICE)
- private void init() {
+ public void init() {
List models = modelStore.getAll();
log.debug("Fetched model configs: {}", models);
currentModelConfigs = models.stream()
- .collect(Collectors.toConcurrentMap(LwM2MModelConfig::getEndpoint, m -> m));
+ .collect(Collectors.toConcurrentMap(LwM2MModelConfig::getEndpoint, m -> m, (existing, replacement) -> existing));
}
@Override
diff --git a/common/transport/lwm2m/src/test/java/org/thingsboard/server/transport/lwm2m/server/model/LwM2MModelConfigServiceImplTest.java b/common/transport/lwm2m/src/test/java/org/thingsboard/server/transport/lwm2m/server/model/LwM2MModelConfigServiceImplTest.java
new file mode 100644
index 0000000000..fc54ca9e0b
--- /dev/null
+++ b/common/transport/lwm2m/src/test/java/org/thingsboard/server/transport/lwm2m/server/model/LwM2MModelConfigServiceImplTest.java
@@ -0,0 +1,73 @@
+/**
+ * Copyright © 2016-2023 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.transport.lwm2m.server.model;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.thingsboard.server.transport.lwm2m.server.store.TbLwM2MModelConfigStore;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.willReturn;
+import static org.mockito.Mockito.mock;
+
+class LwM2MModelConfigServiceImplTest {
+
+ LwM2MModelConfigServiceImpl service;
+ TbLwM2MModelConfigStore modelStore;
+
+ @BeforeEach
+ void setUp() {
+ service = new LwM2MModelConfigServiceImpl();
+ modelStore = mock(TbLwM2MModelConfigStore.class);
+ service.modelStore = modelStore;
+ }
+
+ @Test
+ void testInitWithDuplicatedModels() {
+ LwM2MModelConfig config = new LwM2MModelConfig("urn:imei:951358811362976");
+ List models = List.of(config, config);
+ willReturn(models).given(modelStore).getAll();
+ service.init();
+ assertThat(service.currentModelConfigs).containsExactlyEntriesOf(Map.of(config.getEndpoint(), config));
+ }
+
+ @Test
+ void testInitWithNonUniqueEndpoints() {
+ LwM2MModelConfig configAlfa = new LwM2MModelConfig("urn:imei:951358811362976");
+ LwM2MModelConfig configBravo = new LwM2MModelConfig("urn:imei:151358811362976");
+ LwM2MModelConfig configDelta = new LwM2MModelConfig("urn:imei:151358811362976");
+ assertThat(configBravo.getEndpoint()).as("non-unique endpoints provided").isEqualTo(configDelta.getEndpoint());
+ List models = List.of(configAlfa, configBravo, configDelta);
+ willReturn(models).given(modelStore).getAll();
+ service.init();
+ assertThat(service.currentModelConfigs).containsExactlyInAnyOrderEntriesOf(Map.of(
+ configAlfa.getEndpoint(), configAlfa,
+ configBravo.getEndpoint(), configBravo
+ ));
+ }
+
+ @Test
+ void testInitWithEmptyModels() {
+ willReturn(Collections.emptyList()).given(modelStore).getAll();
+ service.init();
+ assertThat(service.currentModelConfigs).isEmpty();
+ }
+
+}
diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
index 95603e751c..5cd6ba9145 100644
--- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
+++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
@@ -71,6 +71,7 @@ import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import static org.springframework.util.ConcurrentReferenceHashMap.ReferenceType;
+import static org.thingsboard.server.common.data.DataConstants.DEFAULT_DEVICE_TYPE;
import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_CLOSED;
import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_OPEN;
import static org.thingsboard.server.common.transport.service.DefaultTransportService.SUBSCRIBE_TO_ATTRIBUTE_UPDATES_ASYNC_MSG;
@@ -85,7 +86,6 @@ import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMess
@Slf4j
public abstract class AbstractGatewaySessionHandler {
- protected static final String DEFAULT_DEVICE_TYPE = "default";
private static final String CAN_T_PARSE_VALUE = "Can't parse value: ";
private static final String DEVICE_PROPERTY = "device";
diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
index 3b827edbae..497540ee44 100644
--- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
+++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
@@ -49,14 +49,14 @@ 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.limit.LimitedApi;
+import org.thingsboard.server.common.data.msg.TbMsgType;
+import org.thingsboard.server.common.data.notification.rule.trigger.RateLimitsTrigger;
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.notification.NotificationRuleProcessor;
-import org.thingsboard.server.common.data.notification.rule.trigger.RateLimitsTrigger;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
-import org.thingsboard.server.common.msg.session.SessionMsgType;
import org.thingsboard.server.common.msg.tools.TbRateLimitsException;
import org.thingsboard.server.common.stats.MessagesStats;
import org.thingsboard.server.common.stats.StatsFactory;
@@ -583,7 +583,7 @@ public class DefaultTransportService implements TransportService {
metaData.putValue("deviceType", sessionInfo.getDeviceType());
metaData.putValue("ts", tsKv.getTs() + "");
JsonObject json = JsonUtils.getJsonObject(tsKv.getKvList());
- sendToRuleEngine(tenantId, deviceId, customerId, sessionInfo, json, metaData, SessionMsgType.POST_TELEMETRY_REQUEST, packCallback);
+ sendToRuleEngine(tenantId, deviceId, customerId, sessionInfo, json, metaData, TbMsgType.POST_TELEMETRY_REQUEST, packCallback);
}
}
}
@@ -603,7 +603,7 @@ public class DefaultTransportService implements TransportService {
}
metaData.putValue(DataConstants.NOTIFY_DEVICE_METADATA_KEY, "false");
CustomerId customerId = getCustomerId(sessionInfo);
- sendToRuleEngine(tenantId, deviceId, customerId, sessionInfo, json, metaData, SessionMsgType.POST_ATTRIBUTES_REQUEST,
+ sendToRuleEngine(tenantId, deviceId, customerId, sessionInfo, json, metaData, TbMsgType.POST_ATTRIBUTES_REQUEST,
new TransportTbQueueCallback(new ApiStatsProxyCallback<>(tenantId, customerId, msg.getKvList().size(), callback)));
}
}
@@ -723,7 +723,7 @@ public class DefaultTransportService implements TransportService {
metaData.putValue("serviceId", serviceInfoProvider.getServiceId());
metaData.putValue("sessionId", sessionId.toString());
sendToRuleEngine(tenantId, deviceId, getCustomerId(sessionInfo), sessionInfo, json, metaData,
- SessionMsgType.TO_SERVER_RPC_REQUEST, new TransportTbQueueCallback(callback));
+ TbMsgType.TO_SERVER_RPC_REQUEST, new TransportTbQueueCallback(callback));
String requestId = sessionId + "-" + msg.getRequestId();
toServerRpcPendingMap.put(requestId, new RpcRequestMetadata(sessionId, msg.getRequestId()));
scheduler.schedule(() -> processTimeout(requestId), clientSideRpcTimeout, TimeUnit.MILLISECONDS);
@@ -1136,7 +1136,7 @@ public class DefaultTransportService implements TransportService {
}
private void sendToRuleEngine(TenantId tenantId, DeviceId deviceId, CustomerId customerId, TransportProtos.SessionInfoProto sessionInfo, JsonObject json,
- TbMsgMetaData metaData, SessionMsgType sessionMsgType, TbQueueCallback callback) {
+ TbMsgMetaData metaData, TbMsgType tbMsgType, TbQueueCallback callback) {
DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(sessionInfo.getDeviceProfileIdMSB(), sessionInfo.getDeviceProfileIdLSB()));
DeviceProfile deviceProfile = deviceProfileCache.get(deviceProfileId);
RuleChainId ruleChainId;
@@ -1151,7 +1151,7 @@ public class DefaultTransportService implements TransportService {
queueName = deviceProfile.getDefaultQueueName();
}
- TbMsg tbMsg = TbMsg.newMsg(queueName, sessionMsgType.name(), deviceId, customerId, metaData, gson.toJson(json), ruleChainId, null);
+ TbMsg tbMsg = TbMsg.newMsg(queueName, tbMsgType, deviceId, customerId, metaData, gson.toJson(json), ruleChainId, null);
sendToRuleEngine(tenantId, tbMsg, callback);
}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceConnectivityConfiguration.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceConnectivityConfiguration.java
new file mode 100644
index 0000000000..033aa4e0bc
--- /dev/null
+++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceConnectivityConfiguration.java
@@ -0,0 +1,34 @@
+/**
+ * Copyright © 2016-2023 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.dao.device;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.Map;
+
+@Configuration
+@ConfigurationProperties(prefix = "device")
+@Data
+public class DeviceConnectivityConfiguration {
+ private Map connectivity;
+
+ public boolean isEnabled(String protocol) {
+ var info = connectivity.get(protocol);
+ return info != null && info.isEnabled();
+ }
+}
diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbRelationTypes.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceConnectivityInfo.java
similarity index 73%
rename from rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbRelationTypes.java
rename to dao/src/main/java/org/thingsboard/server/dao/device/DeviceConnectivityInfo.java
index 2535762ad5..b243be9995 100644
--- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbRelationTypes.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceConnectivityInfo.java
@@ -13,14 +13,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.thingsboard.rule.engine.api;
+package org.thingsboard.server.dao.device;
-/**
- * Created by ashvayka on 19.01.18.
- */
-public final class TbRelationTypes {
-
- public static String SUCCESS = "Success";
- public static String FAILURE = "Failure";
+import lombok.Data;
+@Data
+public class DeviceConnectivityInfo {
+ private boolean enabled;
+ private String host;
+ private String port;
+ private String pemCertFile;
}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceConnectivityServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceConnectivityServiceImpl.java
new file mode 100644
index 0000000000..c06103d8f3
--- /dev/null
+++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceConnectivityServiceImpl.java
@@ -0,0 +1,268 @@
+/**
+ * Copyright © 2016-2023 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.dao.device;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.server.common.data.Device;
+import org.thingsboard.server.common.data.DeviceProfile;
+import org.thingsboard.server.common.data.DeviceTransportType;
+import org.thingsboard.server.common.data.ResourceUtils;
+import org.thingsboard.server.common.data.StringUtils;
+import org.thingsboard.server.common.data.device.profile.MqttDeviceProfileTransportConfiguration;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.security.DeviceCredentials;
+import org.thingsboard.server.common.data.security.DeviceCredentialsType;
+import org.thingsboard.server.dao.util.DeviceConnectivityUtil;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+import static org.thingsboard.server.dao.service.Validator.validateId;
+import static org.thingsboard.server.dao.util.DeviceConnectivityUtil.CHECK_DOCUMENTATION;
+import static org.thingsboard.server.dao.util.DeviceConnectivityUtil.COAP;
+import static org.thingsboard.server.dao.util.DeviceConnectivityUtil.COAPS;
+import static org.thingsboard.server.dao.util.DeviceConnectivityUtil.DOCKER;
+import static org.thingsboard.server.dao.util.DeviceConnectivityUtil.HTTP;
+import static org.thingsboard.server.dao.util.DeviceConnectivityUtil.HTTPS;
+import static org.thingsboard.server.dao.util.DeviceConnectivityUtil.MQTT;
+import static org.thingsboard.server.dao.util.DeviceConnectivityUtil.MQTTS;
+
+@Service("DeviceConnectivityDaoService")
+@Slf4j
+public class DeviceConnectivityServiceImpl implements DeviceConnectivityService {
+
+ public static final String INCORRECT_TENANT_ID = "Incorrect tenantId ";
+ public static final String INCORRECT_DEVICE_ID = "Incorrect deviceId ";
+ public static final String DEFAULT_DEVICE_TELEMETRY_TOPIC = "v1/devices/me/telemetry";
+
+ @Autowired
+ private DeviceCredentialsService deviceCredentialsService;
+
+ @Autowired
+ private DeviceProfileService deviceProfileService;
+
+ @Autowired
+ private DeviceConnectivityConfiguration deviceConnectivityConfiguration;
+
+ @Override
+ public JsonNode findDevicePublishTelemetryCommands(String baseUrl, Device device) throws URISyntaxException {
+ DeviceId deviceId = device.getId();
+ log.trace("Executing findDevicePublishTelemetryCommands [{}]", deviceId);
+ validateId(deviceId, INCORRECT_DEVICE_ID + deviceId);
+
+ DeviceCredentials creds = deviceCredentialsService.findDeviceCredentialsByDeviceId(device.getTenantId(), deviceId);
+ DeviceProfile deviceProfile = deviceProfileService.findDeviceProfileById(device.getTenantId(), device.getDeviceProfileId());
+ DeviceTransportType transportType = deviceProfile.getTransportType();
+
+ ObjectNode commands = JacksonUtil.newObjectNode();
+ switch (transportType) {
+ case DEFAULT:
+ Optional.ofNullable(getHttpTransportPublishCommands(baseUrl, creds))
+ .ifPresent(v -> commands.set(HTTP, v));
+ Optional.ofNullable(getMqttTransportPublishCommands(baseUrl, creds))
+ .ifPresent(v -> commands.set(MQTT, v));
+ Optional.ofNullable(getCoapTransportPublishCommands(baseUrl, creds))
+ .ifPresent(v -> commands.set(COAP, v));
+ break;
+ case MQTT:
+ MqttDeviceProfileTransportConfiguration transportConfiguration =
+ (MqttDeviceProfileTransportConfiguration) deviceProfile.getProfileData().getTransportConfiguration();
+ //TODO: add sparkplug command with emulator (check SSL)
+ if (transportConfiguration.isSparkplug()) {
+ ObjectNode sparkplug = JacksonUtil.newObjectNode();
+ sparkplug.put("sparkplug", CHECK_DOCUMENTATION);
+ commands.set(MQTT, sparkplug);
+ } else {
+ String topicName = transportConfiguration.getDeviceTelemetryTopic();
+
+ Optional.ofNullable(getMqttTransportPublishCommands(baseUrl, topicName, creds))
+ .ifPresent(v -> commands.set(MQTT, v));
+ }
+ break;
+ case COAP:
+ Optional.ofNullable(getCoapTransportPublishCommands(baseUrl, creds))
+ .ifPresent(v -> commands.set(COAP, v));
+ break;
+ default:
+ commands.put(transportType.name(), CHECK_DOCUMENTATION);
+ }
+ return commands;
+ }
+
+ @Override
+ public Resource getPemCertFile(String protocol) {
+ String certFilePath = deviceConnectivityConfiguration.getConnectivity()
+ .get(protocol)
+ .getPemCertFile();
+
+ if (StringUtils.isNotBlank(certFilePath) && ResourceUtils.resourceExists(this, certFilePath)) {
+ return new ClassPathResource(certFilePath);
+ } else {
+ return null;
+ }
+ }
+
+ private JsonNode getHttpTransportPublishCommands(String defaultHostname, DeviceCredentials deviceCredentials) throws URISyntaxException {
+ ObjectNode httpCommands = JacksonUtil.newObjectNode();
+ Optional.ofNullable(getHttpPublishCommand(HTTP, defaultHostname, deviceCredentials))
+ .ifPresent(v -> httpCommands.put(HTTP, v));
+ Optional.ofNullable(getHttpPublishCommand(HTTPS, defaultHostname, deviceCredentials))
+ .ifPresent(v -> httpCommands.put(HTTPS, v));
+ return httpCommands.isEmpty() ? null : httpCommands;
+ }
+
+ private String getHttpPublishCommand(String protocol, String baseUrl, DeviceCredentials deviceCredentials) throws URISyntaxException {
+ DeviceConnectivityInfo properties = deviceConnectivityConfiguration.getConnectivity().get(protocol);
+ if (properties == null || !properties.isEnabled() ||
+ deviceCredentials.getCredentialsType() != DeviceCredentialsType.ACCESS_TOKEN) {
+ return null;
+ }
+ String hostName = getHost(baseUrl, properties);
+ String port = properties.getPort().isEmpty() ? "" : ":" + properties.getPort();
+
+ return DeviceConnectivityUtil.getHttpPublishCommand(protocol, hostName, port, deviceCredentials);
+ }
+
+ private JsonNode getMqttTransportPublishCommands(String baseUrl, DeviceCredentials deviceCredentials) throws URISyntaxException {
+ return getMqttTransportPublishCommands(baseUrl, DEFAULT_DEVICE_TELEMETRY_TOPIC, deviceCredentials);
+ }
+
+ private JsonNode getMqttTransportPublishCommands(String baseUrl, String topic, DeviceCredentials deviceCredentials) throws URISyntaxException {
+ ObjectNode mqttCommands = JacksonUtil.newObjectNode();
+
+ if (deviceCredentials.getCredentialsType() == DeviceCredentialsType.X509_CERTIFICATE) {
+ mqttCommands.put(MQTTS, CHECK_DOCUMENTATION);
+ return mqttCommands;
+ }
+
+ ObjectNode dockerMqttCommands = JacksonUtil.newObjectNode();
+
+ if (deviceConnectivityConfiguration.isEnabled(MQTT)) {
+ Optional.ofNullable(getMqttPublishCommand(baseUrl, topic, deviceCredentials)).
+ ifPresent(v -> mqttCommands.put(MQTT, v));
+
+ Optional.ofNullable(getDockerMqttPublishCommand(MQTT, baseUrl, topic, deviceCredentials))
+ .ifPresent(v -> dockerMqttCommands.put(MQTT, v));
+ }
+
+ if (deviceConnectivityConfiguration.isEnabled(MQTTS)) {
+ List mqttsPublishCommand = getMqttsPublishCommand(baseUrl, topic, deviceCredentials);
+ if (mqttsPublishCommand != null) {
+ ArrayNode arrayNode = mqttCommands.putArray(MQTTS);
+ mqttsPublishCommand.forEach(arrayNode::add);
+ }
+
+ Optional.ofNullable(getDockerMqttPublishCommand(MQTTS, baseUrl, topic, deviceCredentials))
+ .ifPresent(v -> dockerMqttCommands.put(MQTTS, v));
+ }
+
+ if (!dockerMqttCommands.isEmpty()) {
+ mqttCommands.set(DOCKER, dockerMqttCommands);
+ }
+ return mqttCommands.isEmpty() ? null : mqttCommands;
+ }
+
+ private String getMqttPublishCommand(String baseUrl, String deviceTelemetryTopic, DeviceCredentials deviceCredentials) throws URISyntaxException {
+ DeviceConnectivityInfo properties = deviceConnectivityConfiguration.getConnectivity().get(MQTT);
+ String mqttHost = getHost(baseUrl, properties);
+ String mqttPort = properties.getPort().isEmpty() ? null : properties.getPort();
+ return DeviceConnectivityUtil.getMqttPublishCommand(MQTT, mqttHost, mqttPort, deviceTelemetryTopic, deviceCredentials);
+ }
+
+ private List getMqttsPublishCommand(String baseUrl, String deviceTelemetryTopic, DeviceCredentials deviceCredentials) throws URISyntaxException {
+ DeviceConnectivityInfo properties = deviceConnectivityConfiguration.getConnectivity().get(MQTTS);
+ String mqttHost = getHost(baseUrl, properties);
+ String mqttPort = properties.getPort().isEmpty() ? null : properties.getPort();
+ String pubCommand = DeviceConnectivityUtil.getMqttPublishCommand(MQTTS, mqttHost, mqttPort, deviceTelemetryTopic, deviceCredentials);
+
+ ArrayList commands = new ArrayList<>();
+ if (pubCommand != null) {
+ commands.add(DeviceConnectivityUtil.getCurlPemCertCommand(baseUrl, MQTTS));
+ commands.add(pubCommand);
+ return commands;
+ }
+ return null;
+ }
+
+ private String getDockerMqttPublishCommand(String protocol, String baseUrl, String deviceTelemetryTopic, DeviceCredentials deviceCredentials) throws URISyntaxException {
+ DeviceConnectivityInfo properties = deviceConnectivityConfiguration.getConnectivity().get(protocol);
+ String mqttHost = getHost(baseUrl, properties);
+ String mqttPort = properties.getPort().isEmpty() ? null : properties.getPort();
+ return DeviceConnectivityUtil.getDockerMqttPublishCommand(protocol, baseUrl, mqttHost, mqttPort, deviceTelemetryTopic, deviceCredentials);
+ }
+
+ private JsonNode getCoapTransportPublishCommands(String baseUrl, DeviceCredentials deviceCredentials) throws URISyntaxException {
+ ObjectNode coapCommands = JacksonUtil.newObjectNode();
+
+ if (deviceCredentials.getCredentialsType() == DeviceCredentialsType.X509_CERTIFICATE) {
+ coapCommands.put(COAPS, CHECK_DOCUMENTATION);
+ return coapCommands;
+ }
+
+ ObjectNode dockerCoapCommands = JacksonUtil.newObjectNode();
+
+ if (deviceConnectivityConfiguration.isEnabled(COAP)) {
+ Optional.ofNullable(getCoapPublishCommand(COAP, baseUrl, deviceCredentials))
+ .ifPresent(v -> coapCommands.put(COAP, v));
+
+ Optional.ofNullable(getDockerCoapPublishCommand(COAP, baseUrl, deviceCredentials))
+ .ifPresent(v -> dockerCoapCommands.put(COAP, v));
+ }
+
+ if (deviceConnectivityConfiguration.isEnabled(COAPS)) {
+ Optional.ofNullable(getCoapPublishCommand(COAPS, baseUrl, deviceCredentials))
+ .ifPresent(v -> coapCommands.put(COAPS, v));
+
+ Optional.ofNullable(getDockerCoapPublishCommand(COAPS, baseUrl, deviceCredentials))
+ .ifPresent(v -> dockerCoapCommands.put(COAPS, v));
+ }
+
+ if (!dockerCoapCommands.isEmpty()) {
+ coapCommands.set(DOCKER, dockerCoapCommands);
+ }
+
+ return coapCommands.isEmpty() ? null : coapCommands;
+ }
+
+ private String getCoapPublishCommand(String protocol, String baseUrl, DeviceCredentials deviceCredentials) throws URISyntaxException {
+ DeviceConnectivityInfo properties = deviceConnectivityConfiguration.getConnectivity().get(protocol);
+ String hostName = getHost(baseUrl, properties);
+ String port = properties.getPort().isEmpty() ? "" : ":" + properties.getPort();
+ return DeviceConnectivityUtil.getCoapPublishCommand(protocol, hostName, port, deviceCredentials);
+ }
+
+ private String getDockerCoapPublishCommand(String protocol, String baseUrl, DeviceCredentials deviceCredentials) throws URISyntaxException {
+ DeviceConnectivityInfo properties = deviceConnectivityConfiguration.getConnectivity().get(protocol);
+ String host = getHost(baseUrl, properties);
+ String port = properties.getPort().isEmpty() ? "" : ":" + properties.getPort();
+ return DeviceConnectivityUtil.getDockerCoapPublishCommand(protocol, host, port, deviceCredentials);
+ }
+
+ private String getHost(String baseUrl, DeviceConnectivityInfo properties) throws URISyntaxException {
+ return properties.getHost().isEmpty() ? new URI(baseUrl).getHost() : properties.getHost();
+ }
+}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraAbstractAsyncDao.java b/dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraAbstractAsyncDao.java
index 425724bf2c..92289b7178 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraAbstractAsyncDao.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/nosql/CassandraAbstractAsyncDao.java
@@ -19,13 +19,13 @@ import com.google.common.base.Function;
import com.google.common.util.concurrent.AsyncFunction;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.thingsboard.common.util.ThingsBoardExecutors;
import javax.annotation.Nullable;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
/**
* Created by ashvayka on 21.02.17.
@@ -34,9 +34,12 @@ public abstract class CassandraAbstractAsyncDao extends CassandraAbstractDao {
protected ExecutorService readResultsProcessingExecutor;
+ @Value("${cassandra.query.result_processing_threads:50}")
+ private int threadPoolSize;
+
@PostConstruct
public void startExecutor() {
- readResultsProcessingExecutor = Executors.newCachedThreadPool(ThingsBoardThreadFactory.forName("cassandra-callback"));
+ readResultsProcessingExecutor = ThingsBoardExecutors.newWorkStealingPool(threadPoolSize, "cassandra-callback");
}
@PreDestroy
diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java
index fb86e09ab3..f7ae4a1596 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java
@@ -220,17 +220,36 @@ public class BaseRelationService implements RelationService {
return future;
}
+ @Transactional
+ @Override
+ public void deleteEntityCommonRelations(TenantId tenantId, EntityId entityId) {
+ deleteEntityRelations(tenantId, entityId, RelationTypeGroup.COMMON);
+ }
+
@Transactional
@Override
public void deleteEntityRelations(TenantId tenantId, EntityId entityId) {
+ deleteEntityRelations(tenantId, entityId, null);
+ }
+
+ @Transactional
+ public void deleteEntityRelations(TenantId tenantId, EntityId entityId, RelationTypeGroup relationTypeGroup) {
log.trace("Executing deleteEntityRelations [{}]", entityId);
validate(entityId);
- List inboundRelations = new ArrayList<>(relationDao.findAllByTo(tenantId, entityId));
- List outboundRelations = new ArrayList<>(relationDao.findAllByFrom(tenantId, entityId));
+ List inboundRelations = relationTypeGroup == null
+ ? relationDao.findAllByTo(tenantId, entityId)
+ : relationDao.findAllByTo(tenantId, entityId, relationTypeGroup);
+ List outboundRelations = relationTypeGroup == null
+ ? relationDao.findAllByFrom(tenantId, entityId)
+ : relationDao.findAllByFrom(tenantId, entityId, relationTypeGroup);
if (!inboundRelations.isEmpty()) {
try {
- relationDao.deleteInboundRelations(tenantId, entityId);
+ if (relationTypeGroup == null) {
+ relationDao.deleteInboundRelations(tenantId, entityId);
+ } else {
+ relationDao.deleteInboundRelations(tenantId, entityId, relationTypeGroup);
+ }
} catch (ConcurrencyFailureException e) {
log.debug("Concurrency exception while deleting relations [{}]", inboundRelations, e);
}
@@ -241,7 +260,11 @@ public class BaseRelationService implements RelationService {
}
if (!outboundRelations.isEmpty()) {
- relationDao.deleteOutboundRelations(tenantId, entityId);
+ if (relationTypeGroup == null) {
+ relationDao.deleteOutboundRelations(tenantId, entityId);
+ } else {
+ relationDao.deleteOutboundRelations(tenantId, entityId, relationTypeGroup);
+ }
for (EntityRelation relation : outboundRelations) {
eventPublisher.publishEvent(EntityRelationEvent.from(relation));
diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/RelationDao.java b/dao/src/main/java/org/thingsboard/server/dao/relation/RelationDao.java
index 7fee4a31ff..250a0c6105 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/relation/RelationDao.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/relation/RelationDao.java
@@ -64,8 +64,12 @@ public interface RelationDao {
void deleteOutboundRelations(TenantId tenantId, EntityId entity);
+ void deleteOutboundRelations(TenantId tenantId, EntityId entity, RelationTypeGroup relationTypeGroup);
+
void deleteInboundRelations(TenantId tenantId, EntityId entity);
+ void deleteInboundRelations(TenantId tenantId, EntityId entity, RelationTypeGroup relationTypeGroup);
+
ListenableFuture deleteOutboundRelationsAsync(TenantId tenantId, EntityId entity);
List findRuleNodeToRuleChainRelations(RuleChainType ruleChainType, int limit);
diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java
index f35e7e608f..cf2ff1e7a0 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java
@@ -31,6 +31,7 @@ import org.thingsboard.server.common.data.event.RuleChainDebugEventFilter;
import org.thingsboard.server.common.data.event.RuleNodeDebugEventFilter;
import org.thingsboard.server.common.data.event.StatisticsEventFilter;
import org.thingsboard.server.common.data.id.EventId;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.TimePageLink;
import org.thingsboard.server.common.stats.StatsFactory;
@@ -280,7 +281,7 @@ public class JpaBaseEventDao implements EventDao {
private PageData findEventByFilter(UUID tenantId, UUID entityId, LifeCycleEventFilter eventFilter, TimePageLink pageLink) {
boolean statusFilterEnabled = !StringUtils.isEmpty(eventFilter.getStatus());
- boolean statusFilter = statusFilterEnabled && eventFilter.getStatus().equalsIgnoreCase("Success");
+ boolean statusFilter = statusFilterEnabled && eventFilter.getStatus().equalsIgnoreCase(TbNodeConnectionType.SUCCESS);
return DaoUtil.toPageData(
lcEventRepository.findEvents(
tenantId,
@@ -359,7 +360,7 @@ public class JpaBaseEventDao implements EventDao {
private void removeEventsByFilter(UUID tenantId, UUID entityId, LifeCycleEventFilter eventFilter, Long startTime, Long endTime) {
boolean statusFilterEnabled = !StringUtils.isEmpty(eventFilter.getStatus());
- boolean statusFilter = statusFilterEnabled && eventFilter.getStatus().equalsIgnoreCase("Success");
+ boolean statusFilter = statusFilterEnabled && eventFilter.getStatus().equalsIgnoreCase(TbNodeConnectionType.SUCCESS);
lcEventRepository.removeEvents(
tenantId,
entityId,
diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java
index c1b17f160e..31125a0791 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java
@@ -37,6 +37,7 @@ import org.thingsboard.server.dao.util.SqlDao;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
@@ -205,6 +206,15 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple
}
}
+ @Override
+ public void deleteOutboundRelations(TenantId tenantId, EntityId entity, RelationTypeGroup relationTypeGroup) {
+ try {
+ relationRepository.deleteByFromIdAndFromTypeAndRelationTypeGroupIn(entity.getId(), entity.getEntityType().name(), Collections.singletonList(relationTypeGroup.name()));
+ } catch (ConcurrencyFailureException e) {
+ log.debug("Concurrency exception while deleting relations [{}]", entity, e);
+ }
+ }
+
@Override
public void deleteInboundRelations(TenantId tenantId, EntityId entity) {
try {
@@ -214,6 +224,15 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple
}
}
+ @Override
+ public void deleteInboundRelations(TenantId tenantId, EntityId entity, RelationTypeGroup relationTypeGroup) {
+ try {
+ relationRepository.deleteByToIdAndToTypeAndRelationTypeGroupIn(entity.getId(), entity.getEntityType().name(), Collections.singletonList(relationTypeGroup.name()));
+ } catch (ConcurrencyFailureException e) {
+ log.debug("Concurrency exception while deleting relations [{}]", entity, e);
+ }
+ }
+
@Override
public ListenableFuture deleteOutboundRelationsAsync(TenantId tenantId, EntityId entity) {
return service.submit(
diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/relation/RelationRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/relation/RelationRepository.java
index a3d6d8570d..10c8c826eb 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/sql/relation/RelationRepository.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/sql/relation/RelationRepository.java
@@ -82,4 +82,9 @@ public interface RelationRepository
@Query("DELETE FROM RelationEntity r where r.toId = :toId and r.toType = :toType and r.relationTypeGroup in :relationTypeGroups")
void deleteByToIdAndToTypeAndRelationTypeGroupIn(@Param("toId") UUID toId, @Param("toType") String toType, @Param("relationTypeGroups") List relationTypeGroups);
+ @Transactional
+ @Modifying
+ @Query("DELETE FROM RelationEntity r where r.fromId = :fromId and r.fromType = :fromType and r.relationTypeGroup in :relationTypeGroups")
+ void deleteByFromIdAndFromTypeAndRelationTypeGroupIn(@Param("fromId") UUID fromId, @Param("fromType") String fromType, @Param("relationTypeGroups") List relationTypeGroups);
+
}
diff --git a/dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java b/dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java
index 4c8d0c5afe..d6c3a8efdd 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java
@@ -121,7 +121,7 @@ public abstract class AbstractBufferedRateExecutor taskCtx = null;
@@ -169,7 +169,7 @@ public abstract class AbstractBufferedRateExecutor= printQueriesFreq) {
printQueriesIdx.set(0);
String query = queryToString(finalTaskCtx);
- log.info("[{}] Cassandra query: {}", taskCtx.getId(), query);
+ log.info("[{}][{}] Cassandra query: {}", getBufferName(), taskCtx.getId(), query);
}
}
logTask("Processing", finalTaskCtx);
@@ -222,7 +222,7 @@ public abstract class AbstractBufferedRateExecutor taskCtx) {
@@ -298,7 +298,7 @@ public abstract class AbstractBufferedRateExecutor3.8.1
3.21.9
1.42.1
- 1.0.6
+ 1.0.7
1.18.18
1.2.4
1.2.5
diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/EmptyNodeConfiguration.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/EmptyNodeConfiguration.java
index 5c54687205..22ffe34769 100644
--- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/EmptyNodeConfiguration.java
+++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/EmptyNodeConfiguration.java
@@ -24,7 +24,6 @@ public class EmptyNodeConfiguration implements NodeConfigurationDeprecated: This method is deprecated since version 3.5.2 and should only be used when you need to
+ * specify a custom message type that doesn't exist in the {@link TbMsgType} enum. For standard message types,
+ * it is recommended to use the {@link #newMsg(String, TbMsgType, EntityId, CustomerId, TbMsgMetaData, String)}
+ * method instead.
+ *
+ * @param queueName the name of the queue where the message will be sent
+ * @param type the type of the message
+ * @param originator the originator of the message
+ * @param customerId the ID of the customer associated with the message
+ * @param metaData the metadata of the message
+ * @param data the data of the message
+ * @return new TbMsg instance
+ */
+ @Deprecated(since = "3.5.2")
TbMsg newMsg(String queueName, String type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data);
+ @Deprecated(since = "3.5.2", forRemoval = true)
TbMsg transformMsg(TbMsg origMsg, String type, EntityId originator, TbMsgMetaData metaData, String data);
+ TbMsg newMsg(String queueName, TbMsgType type, EntityId originator, TbMsgMetaData metaData, String data);
+
+ TbMsg newMsg(String queueName, TbMsgType type, EntityId originator, CustomerId customerId, TbMsgMetaData metaData, String data);
+
+ TbMsg transformMsg(TbMsg origMsg, TbMsgType type, EntityId originator, TbMsgMetaData metaData, String data);
+
+ TbMsg transformMsg(TbMsg origMsg, TbMsgMetaData metaData, String data);
+
+ TbMsg transformMsgOriginator(TbMsg origMsg, EntityId originator);
+
TbMsg customerCreatedMsg(Customer customer, RuleNodeId ruleNodeId);
TbMsg deviceCreatedMsg(Device device, RuleNodeId ruleNodeId);
TbMsg assetCreatedMsg(Asset asset, RuleNodeId ruleNodeId);
- // TODO: Does this changes the message?
+ @Deprecated(since = "3.5.2", forRemoval = true)
TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action);
+ TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, TbMsgType actionMsgType);
+
TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List attributes);
TbMsg attributesDeletedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List keys);
diff --git a/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/util/TbNodeUtilsTest.java b/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/util/TbNodeUtilsTest.java
index 956cfaf40c..cb5514a82b 100644
--- a/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/util/TbNodeUtilsTest.java
+++ b/rule-engine/rule-engine-api/src/test/java/org/thingsboard/rule/engine/api/util/TbNodeUtilsTest.java
@@ -22,6 +22,7 @@ import org.junit.runner.RunWith;
import org.mockito.junit.MockitoJUnitRunner;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;
@@ -43,7 +44,7 @@ public class TbNodeUtilsTest {
ObjectNode node = JacksonUtil.newObjectNode();
node.put("data_key", "data_value");
- TbMsg msg = TbMsg.newMsg("CUSTOM", TenantId.SYS_TENANT_ID, md, JacksonUtil.toString(node));
+ TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, TenantId.SYS_TENANT_ID, md, JacksonUtil.toString(node));
String result = TbNodeUtils.processPattern(pattern, msg);
Assert.assertEquals("ABC metadata_value data_value", result);
}
@@ -57,7 +58,7 @@ public class TbNodeUtilsTest {
ObjectNode node = JacksonUtil.newObjectNode();
node.put("key", "data_value");
- TbMsg msg = TbMsg.newMsg("CUSTOM", TenantId.SYS_TENANT_ID, md, JacksonUtil.toString(node));
+ TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, TenantId.SYS_TENANT_ID, md, JacksonUtil.toString(node));
String result = TbNodeUtils.processPattern(pattern, msg);
Assert.assertEquals(pattern, result);
}
@@ -71,7 +72,7 @@ public class TbNodeUtilsTest {
ObjectNode node = JacksonUtil.newObjectNode();
node.put("key", "data_value");
- TbMsg msg = TbMsg.newMsg("CUSTOM", TenantId.SYS_TENANT_ID, md, JacksonUtil.toString(node));
+ TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, TenantId.SYS_TENANT_ID, md, JacksonUtil.toString(node));
String result = TbNodeUtils.processPattern(pattern, msg);
Assert.assertEquals("ABC metadata_value data_value", result);
}
@@ -92,7 +93,7 @@ public class TbNodeUtilsTest {
ObjectNode node = JacksonUtil.newObjectNode();
node.set("key1", key1Node);
- TbMsg msg = TbMsg.newMsg("CUSTOM", TenantId.SYS_TENANT_ID, md, JacksonUtil.toString(node));
+ TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, TenantId.SYS_TENANT_ID, md, JacksonUtil.toString(node));
String result = TbNodeUtils.processPattern(pattern, msg);
Assert.assertEquals("ABC metadata_value value3", result);
}
@@ -113,7 +114,7 @@ public class TbNodeUtilsTest {
ObjectNode node = JacksonUtil.newObjectNode();
node.set("key1", key1Node);
- TbMsg msg = TbMsg.newMsg("CUSTOM", TenantId.SYS_TENANT_ID, md, JacksonUtil.toString(node));
+ TbMsg msg = TbMsg.newMsg(TbMsgType.POST_TELEMETRY_REQUEST, TenantId.SYS_TENANT_ID, md, JacksonUtil.toString(node));
String result = TbNodeUtils.processPattern(pattern, msg);
Assert.assertEquals("ABC metadata_value $[key1.key2[0].key3]", result);
}
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java
index 3ca55f5db9..57871c91ff 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java
@@ -26,6 +26,8 @@ import org.thingsboard.rule.engine.api.TbNode;
import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.server.common.data.DataConstants;
+import org.thingsboard.server.common.data.msg.TbMsgType;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
import org.thingsboard.server.common.data.script.ScriptLanguage;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;
@@ -55,13 +57,13 @@ public abstract class TbAbstractAlarmNode {
if (alarmResult.alarm == null) {
- ctx.tellNext(msg, "False");
+ ctx.tellNext(msg, TbNodeConnectionType.FALSE);
} else if (alarmResult.isCreated) {
- tellNext(ctx, msg, alarmResult, DataConstants.ENTITY_CREATED, "Created");
+ tellNext(ctx, msg, alarmResult, TbMsgType.ENTITY_CREATED, "Created");
} else if (alarmResult.isUpdated) {
- tellNext(ctx, msg, alarmResult, DataConstants.ENTITY_UPDATED, "Updated");
+ tellNext(ctx, msg, alarmResult, TbMsgType.ENTITY_UPDATED, "Updated");
} else if (alarmResult.isCleared) {
- tellNext(ctx, msg, alarmResult, DataConstants.ALARM_CLEAR, "Cleared");
+ tellNext(ctx, msg, alarmResult, TbMsgType.ALARM_CLEAR, "Cleared");
} else {
ctx.tellSuccess(msg);
}
@@ -71,13 +73,13 @@ public abstract class TbAbstractAlarmNode processAlarm(TbContext ctx, TbMsg msg);
- protected ListenableFuture buildAlarmDetails(TbContext ctx, TbMsg msg, JsonNode previousDetails) {
+ protected ListenableFuture buildAlarmDetails(TbMsg msg, JsonNode previousDetails) {
try {
TbMsg dummyMsg = msg;
if (previousDetails != null) {
TbMsgMetaData metaData = msg.getMetaData().copy();
metaData.putValue(PREV_ALARM_DETAILS, JacksonUtil.toString(previousDetails));
- dummyMsg = ctx.transformMsg(msg, msg.getType(), msg.getOriginator(), metaData, msg.getData());
+ dummyMsg = TbMsg.transformMsgMetadata(msg, metaData);
}
return scriptEngine.executeJsonAsync(dummyMsg);
} catch (Exception e) {
@@ -96,7 +98,7 @@ public abstract class TbAbstractAlarmNode ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), alarmAction),
throwable -> ctx.tellFailure(toAlarmMsg(ctx, alarmResult, msg), throwable));
}
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractRelationActionNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractRelationActionNode.java
index 1ed7689bd6..ea53138d10 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractRelationActionNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractRelationActionNode.java
@@ -57,8 +57,8 @@ import java.util.Optional;
import java.util.concurrent.TimeUnit;
import static org.thingsboard.common.util.DonAsynchron.withCallback;
-import static org.thingsboard.rule.engine.api.TbRelationTypes.FAILURE;
-import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
+import static org.thingsboard.server.common.data.msg.TbNodeConnectionType.FAILURE;
+import static org.thingsboard.server.common.data.msg.TbNodeConnectionType.SUCCESS;
@Slf4j
public abstract class TbAbstractRelationActionNode implements TbNode {
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbClearAlarmNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbClearAlarmNode.java
index a8d7d4985f..d0215a441f 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbClearAlarmNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbClearAlarmNode.java
@@ -71,7 +71,7 @@ public class TbClearAlarmNode extends TbAbstractAlarmNode clearAlarm(TbContext ctx, TbMsg msg, Alarm alarm) {
ctx.logJsEvalRequest();
- ListenableFuture asyncDetails = buildAlarmDetails(ctx, msg, alarm.getDetails());
+ ListenableFuture asyncDetails = buildAlarmDetails(msg, alarm.getDetails());
return Futures.transform(asyncDetails, details -> {
ctx.logJsEvalResponse();
AlarmApiCallResult result = ctx.getAlarmService().clearAlarm(ctx.getTenantId(), alarm.getId(), System.currentTimeMillis(), details);
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java
index 6f7bb7cf8d..c103dd4006 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCopyAttributesToEntityViewNode.java
@@ -36,7 +36,6 @@ import org.thingsboard.server.common.data.objects.AttributesEntityView;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.data.util.CollectionsUtil;
import org.thingsboard.server.common.msg.TbMsg;
-import org.thingsboard.server.common.msg.session.SessionMsgType;
import org.thingsboard.server.common.transport.adaptor.JsonConverter;
import javax.annotation.Nullable;
@@ -45,7 +44,12 @@ import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
-import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
+import static org.thingsboard.server.common.data.msg.TbMsgType.ACTIVITY_EVENT;
+import static org.thingsboard.server.common.data.msg.TbMsgType.ATTRIBUTES_DELETED;
+import static org.thingsboard.server.common.data.msg.TbMsgType.ATTRIBUTES_UPDATED;
+import static org.thingsboard.server.common.data.msg.TbMsgType.INACTIVITY_EVENT;
+import static org.thingsboard.server.common.data.msg.TbMsgType.POST_ATTRIBUTES_REQUEST;
+import static org.thingsboard.server.common.data.msg.TbNodeConnectionType.SUCCESS;
@Slf4j
@RuleNode(
@@ -71,14 +75,11 @@ public class TbCopyAttributesToEntityViewNode implements TbNode {
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
- if (DataConstants.ATTRIBUTES_UPDATED.equals(msg.getType()) ||
- DataConstants.ATTRIBUTES_DELETED.equals(msg.getType()) ||
- DataConstants.ACTIVITY_EVENT.equals(msg.getType()) ||
- DataConstants.INACTIVITY_EVENT.equals(msg.getType()) ||
- SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msg.getType())) {
+ if (msg.isTypeOneOf(ATTRIBUTES_UPDATED, ATTRIBUTES_DELETED,
+ ACTIVITY_EVENT, INACTIVITY_EVENT, POST_ATTRIBUTES_REQUEST)) {
if (!msg.getMetaData().getData().isEmpty()) {
long now = System.currentTimeMillis();
- String scope = msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name()) ?
+ String scope = msg.isTypeOf(POST_ATTRIBUTES_REQUEST) ?
DataConstants.CLIENT_SCOPE : msg.getMetaData().getValue(DataConstants.SCOPE);
ListenableFuture> entityViewsFuture =
@@ -90,9 +91,9 @@ public class TbCopyAttributesToEntityViewNode implements TbNode {
long startTime = entityView.getStartTimeMs();
long endTime = entityView.getEndTimeMs();
if ((endTime != 0 && endTime > now && startTime < now) || (endTime == 0 && startTime < now)) {
- if (DataConstants.ATTRIBUTES_DELETED.equals(msg.getType())) {
+ if (msg.isTypeOf(ATTRIBUTES_DELETED)) {
List attributes = new ArrayList<>();
- for (JsonElement element : new JsonParser().parse(msg.getData()).getAsJsonObject().get("attributes").getAsJsonArray()) {
+ for (JsonElement element : JsonParser.parseString(msg.getData()).getAsJsonObject().get("attributes").getAsJsonArray()) {
if (element.isJsonPrimitive()) {
JsonPrimitive value = element.getAsJsonPrimitive();
if (value.isString()) {
@@ -107,7 +108,7 @@ public class TbCopyAttributesToEntityViewNode implements TbNode {
getFutureCallback(ctx, msg, entityView));
}
} else {
- Set attributes = JsonConverter.convertToAttributes(new JsonParser().parse(msg.getData()));
+ Set attributes = JsonConverter.convertToAttributes(JsonParser.parseString(msg.getData()));
List filteredAttributes =
attributes.stream().filter(attr -> attributeContainsInEntityView(scope, attr.getKey(), entityView)).collect(Collectors.toList());
ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), entityView.getId(), scope, filteredAttributes,
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCreateAlarmNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCreateAlarmNode.java
index 84e82b76e8..b08fd1a562 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCreateAlarmNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCreateAlarmNode.java
@@ -120,7 +120,7 @@ public class TbCreateAlarmNode extends TbAbstractAlarmNode future = createRelationIfAbsent(ctx, msg, entity, relationType);
return Futures.transform(future, result -> {
if (result && config.isChangeOriginatorToRelatedEntity()) {
- TbMsg tbMsg = ctx.transformMsg(msg, msg.getType(), entity.getEntityId(), msg.getMetaData(), msg.getData());
+ TbMsg tbMsg = ctx.transformMsgOriginator(msg, entity.getEntityId());
return new RelationContainer(tbMsg, result);
}
return new RelationContainer(msg, result);
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java
index 88f2be46a8..021123a308 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbMsgCountNode.java
@@ -24,17 +24,16 @@ import org.thingsboard.rule.engine.api.TbNode;
import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
+import org.thingsboard.server.common.data.msg.TbMsgType;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;
-import org.thingsboard.server.common.msg.session.SessionMsgType;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
-import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
-
@Slf4j
@RuleNode(
type = ComponentType.ACTION,
@@ -48,8 +47,6 @@ import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
)
public class TbMsgCountNode implements TbNode {
- private static final String TB_MSG_COUNT_NODE_MSG = "TbMsgCountNodeMsg";
-
private AtomicLong messagesProcessed = new AtomicLong(0);
private final Gson gson = new Gson();
private UUID nextTickId;
@@ -68,7 +65,7 @@ public class TbMsgCountNode implements TbNode {
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
- if (msg.getType().equals(TB_MSG_COUNT_NODE_MSG) && msg.getId().equals(nextTickId)) {
+ if (msg.isTypeOf(TbMsgType.MSG_COUNT_SELF_MSG) && msg.getId().equals(nextTickId)) {
JsonObject telemetryJson = new JsonObject();
telemetryJson.addProperty(this.telemetryPrefix + "_" + ctx.getServiceId(), messagesProcessed.longValue());
@@ -77,8 +74,8 @@ 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));
- ctx.enqueueForTellNext(tbMsg, SUCCESS);
+ TbMsg tbMsg = TbMsg.newMsg(msg.getQueueName(), TbMsgType.POST_TELEMETRY_REQUEST, ctx.getTenantId(), msg.getCustomerId(), metaData, gson.toJson(telemetryJson));
+ ctx.enqueueForTellNext(tbMsg, TbNodeConnectionType.SUCCESS);
scheduleTickMsg(ctx, tbMsg);
} else {
messagesProcessed.incrementAndGet();
@@ -93,7 +90,7 @@ public class TbMsgCountNode implements TbNode {
}
lastScheduledTs = lastScheduledTs + delay;
long curDelay = Math.max(0L, (lastScheduledTs - curTs));
- TbMsg tickMsg = ctx.newMsg(null, TB_MSG_COUNT_NODE_MSG, ctx.getSelfId(), msg != null ? msg.getCustomerId() : null, new TbMsgMetaData(), "");
+ TbMsg tickMsg = ctx.newMsg(null, TbMsgType.MSG_COUNT_SELF_MSG, ctx.getSelfId(), msg != null ? msg.getCustomerId() : null, TbMsgMetaData.EMPTY, TbMsg.EMPTY_STRING);
nextTickId = tickMsg.getId();
ctx.tellSelf(tickMsg, curDelay);
}
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sns/TbSnsNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sns/TbSnsNode.java
index a8a1137a98..e24c73f9d3 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sns/TbSnsNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sns/TbSnsNode.java
@@ -29,7 +29,6 @@ import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNodeConfiguration;
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.rule.engine.external.TbAbstractExternalNode;
import org.thingsboard.server.common.data.plugin.ComponentType;
@@ -106,13 +105,13 @@ public class TbSnsNode extends TbAbstractExternalNode {
TbMsgMetaData metaData = origMsg.getMetaData().copy();
metaData.putValue(MESSAGE_ID, result.getMessageId());
metaData.putValue(REQUEST_ID, result.getSdkResponseMetadata().getRequestId());
- return ctx.transformMsg(origMsg, origMsg.getType(), origMsg.getOriginator(), metaData, origMsg.getData());
+ return TbMsg.transformMsgMetadata(origMsg, metaData);
}
private TbMsg processException(TbContext ctx, TbMsg origMsg, Throwable t) {
TbMsgMetaData metaData = origMsg.getMetaData().copy();
metaData.putValue(ERROR, t.getClass() + ": " + t.getMessage());
- return ctx.transformMsg(origMsg, origMsg.getType(), origMsg.getOriginator(), metaData, origMsg.getData());
+ return TbMsg.transformMsgMetadata(origMsg, metaData);
}
@Override
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNode.java
index 2b9f8916cf..34072fbb8a 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/aws/sqs/TbSqsNode.java
@@ -28,7 +28,6 @@ import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.TbContext;
-import org.thingsboard.rule.engine.api.TbNode;
import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
@@ -93,14 +92,14 @@ public class TbSqsNode extends TbAbstractExternalNode {
var tbMsg = ackIfNeeded(ctx, msg);
withCallback(publishMessageAsync(ctx, tbMsg),
m -> tellSuccess(ctx, m),
- t -> tellFailure(ctx, processException(ctx, tbMsg, t), t));
+ t -> tellFailure(ctx, processException(tbMsg, t), t));
}
private ListenableFuture publishMessageAsync(TbContext ctx, TbMsg msg) {
- return ctx.getExternalCallExecutor().executeAsync(() -> publishMessage(ctx, msg));
+ return ctx.getExternalCallExecutor().executeAsync(() -> publishMessage(msg));
}
- private TbMsg publishMessage(TbContext ctx, TbMsg msg) {
+ private TbMsg publishMessage(TbMsg msg) {
String queueUrl = TbNodeUtils.processPattern(this.config.getQueueUrlPattern(), msg);
SendMessageRequest sendMsgRequest = new SendMessageRequest();
sendMsgRequest.withQueueUrl(queueUrl);
@@ -119,10 +118,10 @@ public class TbSqsNode extends TbAbstractExternalNode {
sendMsgRequest.withMessageGroupId(msg.getOriginator().toString());
}
SendMessageResult result = this.sqsClient.sendMessage(sendMsgRequest);
- return processSendMessageResult(ctx, msg, result);
+ return processSendMessageResult(msg, result);
}
- private TbMsg processSendMessageResult(TbContext ctx, TbMsg origMsg, SendMessageResult result) {
+ private TbMsg processSendMessageResult(TbMsg origMsg, SendMessageResult result) {
TbMsgMetaData metaData = origMsg.getMetaData().copy();
metaData.putValue(MESSAGE_ID, result.getMessageId());
metaData.putValue(REQUEST_ID, result.getSdkResponseMetadata().getRequestId());
@@ -135,13 +134,13 @@ public class TbSqsNode extends TbAbstractExternalNode {
if (!StringUtils.isEmpty(result.getSequenceNumber())) {
metaData.putValue(SEQUENCE_NUMBER, result.getSequenceNumber());
}
- return ctx.transformMsg(origMsg, origMsg.getType(), origMsg.getOriginator(), metaData, origMsg.getData());
+ return TbMsg.transformMsgMetadata(origMsg, metaData);
}
- private TbMsg processException(TbContext ctx, TbMsg origMsg, Throwable t) {
+ private TbMsg processException(TbMsg origMsg, Throwable t) {
TbMsgMetaData metaData = origMsg.getMetaData().copy();
metaData.putValue(ERROR, t.getClass() + ": " + t.getMessage());
- return ctx.transformMsg(origMsg, origMsg.getType(), origMsg.getOriginator(), metaData, origMsg.getData());
+ return TbMsg.transformMsgMetadata(origMsg, metaData);
}
@Override
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java
index 51effe02ef..b31c98bc0f 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNode.java
@@ -28,8 +28,11 @@ import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.server.common.data.StringUtils;
+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.msg.TbMsgType;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.data.script.ScriptLanguage;
import org.thingsboard.server.common.msg.TbMsg;
@@ -41,7 +44,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.thingsboard.common.util.DonAsynchron.withCallback;
-import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
@Slf4j
@RuleNode(
@@ -58,8 +60,6 @@ import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
public class TbMsgGeneratorNode implements TbNode {
- private static final String TB_MSG_GENERATOR_NODE_MSG = "TbMsgGeneratorNodeMsg";
-
private TbMsgGeneratorNodeConfiguration config;
private ScriptEngine scriptEngine;
private long delay;
@@ -97,7 +97,7 @@ public class TbMsgGeneratorNode implements TbNode {
if (initialized.compareAndSet(false, true)) {
this.scriptEngine = ctx.createScriptEngine(config.getScriptLang(),
ScriptLanguage.TBEL.equals(config.getScriptLang()) ? config.getTbelScript() : config.getJsScript(), "prevMsg", "prevMetadata", "prevMsgType");
- scheduleTickMsg(ctx);
+ scheduleTickMsg(ctx, null);
}
} else if (initialized.compareAndSet(true, false)) {
destroy();
@@ -107,14 +107,14 @@ public class TbMsgGeneratorNode implements TbNode {
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
log.trace("onMsg, config {}, msg {}", config, msg);
- if (initialized.get() && msg.getType().equals(TB_MSG_GENERATOR_NODE_MSG) && msg.getId().equals(nextTickId)) {
+ if (initialized.get() && msg.isTypeOf(TbMsgType.GENERATOR_NODE_SELF_MSG) && msg.getId().equals(nextTickId)) {
TbStopWatch sw = TbStopWatch.create();
withCallback(generate(ctx, msg),
m -> {
log.trace("onMsg onSuccess callback, took {}ms, config {}, msg {}", sw.stopAndGetTotalTimeMillis(), config, msg);
if (initialized.get() && (config.getMsgCount() == TbMsgGeneratorNodeConfiguration.UNLIMITED_MSG_COUNT || currentMsgCount < config.getMsgCount())) {
- ctx.enqueueForTellNext(m, SUCCESS);
- scheduleTickMsg(ctx);
+ ctx.enqueueForTellNext(m, TbNodeConnectionType.SUCCESS);
+ scheduleTickMsg(ctx, msg);
currentMsgCount++;
}
},
@@ -122,14 +122,14 @@ public class TbMsgGeneratorNode implements TbNode {
log.trace("onMsg onFailure callback, took {}ms, config {}, msg {}", sw.stopAndGetTotalTimeMillis(), config, msg, t);
if (initialized.get() && (config.getMsgCount() == TbMsgGeneratorNodeConfiguration.UNLIMITED_MSG_COUNT || currentMsgCount < config.getMsgCount())) {
ctx.tellFailure(msg, t);
- scheduleTickMsg(ctx);
+ scheduleTickMsg(ctx, msg);
currentMsgCount++;
}
});
}
}
- private void scheduleTickMsg(TbContext ctx) {
+ private void scheduleTickMsg(TbContext ctx, TbMsg msg) {
log.trace("scheduleTickMsg, config {}", config);
long curTs = System.currentTimeMillis();
if (lastScheduledTs == 0L) {
@@ -137,7 +137,8 @@ public class TbMsgGeneratorNode implements TbNode {
}
lastScheduledTs = lastScheduledTs + delay;
long curDelay = Math.max(0L, (lastScheduledTs - curTs));
- TbMsg tickMsg = ctx.newMsg(config.getQueueName(), TB_MSG_GENERATOR_NODE_MSG, ctx.getSelfId(), new TbMsgMetaData(), "");
+ TbMsg tickMsg = ctx.newMsg(config.getQueueName(), TbMsgType.GENERATOR_NODE_SELF_MSG, ctx.getSelfId(),
+ getCustomerIdFromMsg(msg), TbMsgMetaData.EMPTY, TbMsg.EMPTY_STRING);
nextTickId = tickMsg.getId();
ctx.tellSelf(tickMsg, curDelay);
}
@@ -145,7 +146,7 @@ public class TbMsgGeneratorNode implements TbNode {
private ListenableFuture generate(TbContext ctx, TbMsg msg) {
log.trace("generate, config {}", config);
if (prevMsg == null) {
- prevMsg = ctx.newMsg(config.getQueueName(), "", originatorId, msg.getCustomerId(), new TbMsgMetaData(), "{}");
+ prevMsg = ctx.newMsg(config.getQueueName(), TbMsg.EMPTY_STRING, originatorId, msg.getCustomerId(), TbMsgMetaData.EMPTY, TbMsg.EMPTY_JSON_OBJECT);
}
if (initialized.get()) {
ctx.logJsEvalRequest();
@@ -160,6 +161,10 @@ public class TbMsgGeneratorNode implements TbNode {
}
+ private CustomerId getCustomerIdFromMsg(TbMsg msg) {
+ return msg != null ? msg.getCustomerId() : null;
+ }
+
@Override
public void destroy() {
log.trace("destroy, config {}", config);
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java
index 6890aa6bf4..81bb3d6772 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/deduplication/TbMsgDeduplicationNode.java
@@ -24,9 +24,10 @@ import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNode;
import org.thingsboard.rule.engine.api.TbNodeConfiguration;
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.id.EntityId;
+import org.thingsboard.server.common.data.msg.TbMsgType;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.data.util.TbPair;
import org.thingsboard.server.common.msg.TbMsg;
@@ -60,10 +61,7 @@ import java.util.concurrent.TimeUnit;
@Slf4j
public class TbMsgDeduplicationNode implements TbNode {
- private static final String TB_MSG_DEDUPLICATION_TIMEOUT_MSG = "TbMsgDeduplicationNodeMsg";
public static final int TB_MSG_DEDUPLICATION_RETRY_DELAY = 10;
- private static final String EMPTY_DATA = "";
- private static final TbMsgMetaData EMPTY_META_DATA = new TbMsgMetaData();
private TbMsgDeduplicationNodeConfiguration config;
@@ -82,7 +80,7 @@ public class TbMsgDeduplicationNode implements TbNode {
@Override
public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException {
- if (TB_MSG_DEDUPLICATION_TIMEOUT_MSG.equals(msg.getType())) {
+ if (msg.isTypeOf(TbMsgType.DEDUPLICATION_TIMEOUT_SELF_MSG)) {
processDeduplication(ctx, msg.getOriginator());
} else {
processOnRegularMsg(ctx, msg);
@@ -196,7 +194,7 @@ public class TbMsgDeduplicationNode implements TbNode {
private void enqueueForTellNextWithRetry(TbContext ctx, TbMsg msg, int retryAttempt) {
if (config.getMaxRetries() > retryAttempt) {
- ctx.enqueueForTellNext(msg, TbRelationTypes.SUCCESS,
+ ctx.enqueueForTellNext(msg, TbNodeConnectionType.SUCCESS,
() -> {
log.trace("[{}][{}][{}] Successfully enqueue deduplication result message!", ctx.getSelfId(), msg.getOriginator(), retryAttempt);
},
@@ -210,7 +208,7 @@ public class TbMsgDeduplicationNode implements TbNode {
}
private void scheduleTickMsg(TbContext ctx, EntityId deduplicationId) {
- ctx.tellSelf(ctx.newMsg(null, TB_MSG_DEDUPLICATION_TIMEOUT_MSG, deduplicationId, EMPTY_META_DATA, EMPTY_DATA), deduplicationInterval + 1);
+ ctx.tellSelf(ctx.newMsg(null, TbMsgType.DEDUPLICATION_TIMEOUT_SELF_MSG, deduplicationId, TbMsgMetaData.EMPTY, TbMsg.EMPTY_STRING), deduplicationInterval + 1);
}
private String getMergedData(List msgs) {
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
index 1f18aea86e..5414cc2bbe 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/delay/TbMsgDelayNode.java
@@ -23,6 +23,8 @@ import org.thingsboard.rule.engine.api.TbNode;
import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
+import org.thingsboard.server.common.data.msg.TbMsgType;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;
@@ -32,8 +34,6 @@ import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
-import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
-
@Slf4j
@RuleNode(
type = ComponentType.ACTION,
@@ -50,8 +50,6 @@ import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
)
public class TbMsgDelayNode implements TbNode {
- private static final String TB_MSG_DELAY_NODE_MSG = "TbMsgDelayNodeMsg";
-
private TbMsgDelayNodeConfiguration config;
private Map pendingMsgs;
@@ -63,7 +61,7 @@ public class TbMsgDelayNode implements TbNode {
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
- if (msg.getType().equals(TB_MSG_DELAY_NODE_MSG)) {
+ if (msg.isTypeOf(TbMsgType.DELAY_TIMEOUT_SELF_MSG)) {
TbMsg pendingMsg = pendingMsgs.remove(UUID.fromString(msg.getData()));
if (pendingMsg != null) {
ctx.enqueueForTellNext(
@@ -75,13 +73,13 @@ public class TbMsgDelayNode implements TbNode {
pendingMsg.getMetaData(),
pendingMsg.getData()
),
- SUCCESS
+ TbNodeConnectionType.SUCCESS
);
}
} else {
if (pendingMsgs.size() < config.getMaxPendingMsgs()) {
pendingMsgs.put(msg.getId(), msg);
- TbMsg tickMsg = ctx.newMsg(null, TB_MSG_DELAY_NODE_MSG, ctx.getSelfId(), msg.getCustomerId(), new TbMsgMetaData(), msg.getId().toString());
+ TbMsg tickMsg = ctx.newMsg(null, TbMsgType.DELAY_TIMEOUT_SELF_MSG, ctx.getSelfId(), msg.getCustomerId(), TbMsgMetaData.EMPTY, msg.getId().toString());
ctx.tellSelf(tickMsg, getDelay(msg));
ctx.ack(msg);
} else {
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java
index 318b38aa39..7888922dc6 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java
@@ -30,13 +30,23 @@ import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.edge.EdgeEventActionType;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.TbMsg;
-import org.thingsboard.server.common.msg.session.SessionMsgType;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import static org.thingsboard.server.common.data.msg.TbMsgType.ACTIVITY_EVENT;
+import static org.thingsboard.server.common.data.msg.TbMsgType.ALARM;
+import static org.thingsboard.server.common.data.msg.TbMsgType.ATTRIBUTES_DELETED;
+import static org.thingsboard.server.common.data.msg.TbMsgType.ATTRIBUTES_UPDATED;
+import static org.thingsboard.server.common.data.msg.TbMsgType.CONNECT_EVENT;
+import static org.thingsboard.server.common.data.msg.TbMsgType.DISCONNECT_EVENT;
+import static org.thingsboard.server.common.data.msg.TbMsgType.INACTIVITY_EVENT;
+import static org.thingsboard.server.common.data.msg.TbMsgType.POST_ATTRIBUTES_REQUEST;
+import static org.thingsboard.server.common.data.msg.TbMsgType.POST_TELEMETRY_REQUEST;
+import static org.thingsboard.server.common.data.msg.TbMsgType.TIMESERIES_UPDATED;
+
@Slf4j
public abstract class AbstractTbMsgPushNode implements TbNode {
@@ -57,7 +67,7 @@ public abstract class AbstractTbMsgPushNode metadata = msg.getMetaData().getData();
- EdgeEventActionType actionType = getEdgeEventActionTypeByMsgType(msgType, metadata);
+ EdgeEventActionType actionType = getEdgeEventActionTypeByMsgType(msg);
Map entityBody = new HashMap<>();
JsonNode dataJson = JacksonUtil.toJsonNode(msg.getData());
switch (actionType) {
@@ -148,45 +157,31 @@ public abstract class AbstractTbMsgPushNode metadata) {
+ protected EdgeEventActionType getEdgeEventActionTypeByMsgType(TbMsg msg) {
EdgeEventActionType actionType;
- if (SessionMsgType.POST_TELEMETRY_REQUEST.name().equals(msgType)
- || DataConstants.TIMESERIES_UPDATED.equals(msgType)) {
+ if (msg.isTypeOneOf(POST_TELEMETRY_REQUEST, TIMESERIES_UPDATED)) {
actionType = EdgeEventActionType.TIMESERIES_UPDATED;
- } else if (DataConstants.ATTRIBUTES_UPDATED.equals(msgType)) {
+ } else if (msg.isTypeOf(ATTRIBUTES_UPDATED)) {
actionType = EdgeEventActionType.ATTRIBUTES_UPDATED;
- } else if (SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msgType)) {
+ } else if (msg.isTypeOf(POST_ATTRIBUTES_REQUEST)) {
actionType = EdgeEventActionType.POST_ATTRIBUTES;
- } else if (DataConstants.ATTRIBUTES_DELETED.equals(msgType)) {
+ } else if (msg.isTypeOf(ATTRIBUTES_DELETED)) {
actionType = EdgeEventActionType.ATTRIBUTES_DELETED;
- } else if (DataConstants.CONNECT_EVENT.equals(msgType)
- || DataConstants.DISCONNECT_EVENT.equals(msgType)
- || DataConstants.ACTIVITY_EVENT.equals(msgType)
- || DataConstants.INACTIVITY_EVENT.equals(msgType)) {
- String scope = metadata.get(SCOPE);
- if ( StringUtils.isEmpty(scope)) {
- actionType = EdgeEventActionType.TIMESERIES_UPDATED;
- } else {
- actionType = EdgeEventActionType.ATTRIBUTES_UPDATED;
- }
+ } else if (msg.isTypeOneOf(CONNECT_EVENT, DISCONNECT_EVENT, ACTIVITY_EVENT, INACTIVITY_EVENT)) {
+ String scope = msg.getMetaData().getValue(SCOPE);
+ actionType = StringUtils.isEmpty(scope) ?
+ EdgeEventActionType.TIMESERIES_UPDATED : EdgeEventActionType.ATTRIBUTES_UPDATED;
} else {
- log.warn("Unsupported msg type [{}]", msgType);
- throw new IllegalArgumentException("Unsupported msg type: " + msgType);
+ String type = msg.getType();
+ log.warn("Unsupported msg type [{}]", type);
+ throw new IllegalArgumentException("Unsupported msg type: " + type);
}
return actionType;
}
- protected boolean isSupportedMsgType(String msgType) {
- return SessionMsgType.POST_TELEMETRY_REQUEST.name().equals(msgType)
- || SessionMsgType.POST_ATTRIBUTES_REQUEST.name().equals(msgType)
- || DataConstants.ATTRIBUTES_UPDATED.equals(msgType)
- || DataConstants.ATTRIBUTES_DELETED.equals(msgType)
- || DataConstants.TIMESERIES_UPDATED.equals(msgType)
- || DataConstants.ALARM.equals(msgType)
- || DataConstants.CONNECT_EVENT.equals(msgType)
- || DataConstants.DISCONNECT_EVENT.equals(msgType)
- || DataConstants.ACTIVITY_EVENT.equals(msgType)
- || DataConstants.INACTIVITY_EVENT.equals(msgType);
+ protected boolean isSupportedMsgType(TbMsg msg) {
+ return msg.isTypeOneOf(POST_TELEMETRY_REQUEST, POST_ATTRIBUTES_REQUEST, ATTRIBUTES_UPDATED,
+ ATTRIBUTES_DELETED, TIMESERIES_UPDATED, ALARM, CONNECT_EVENT, DISCONNECT_EVENT, ACTIVITY_EVENT, INACTIVITY_EVENT);
}
protected boolean isSupportedOriginator(EntityType entityType) {
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/external/TbAbstractExternalNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/external/TbAbstractExternalNode.java
index 834dbe2390..1fb000d709 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/external/TbAbstractExternalNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/external/TbAbstractExternalNode.java
@@ -17,7 +17,7 @@ package org.thingsboard.rule.engine.external;
import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNode;
-import org.thingsboard.rule.engine.api.TbRelationTypes;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
import org.thingsboard.server.common.msg.TbMsg;
public abstract class TbAbstractExternalNode implements TbNode {
@@ -30,7 +30,7 @@ public abstract class TbAbstractExternalNode implements TbNode {
protected void tellSuccess(TbContext ctx, TbMsg tbMsg) {
if (forceAck) {
- ctx.enqueueForTellNext(tbMsg.copyWithNewCtx(), TbRelationTypes.SUCCESS);
+ ctx.enqueueForTellNext(tbMsg.copyWithNewCtx(), TbNodeConnectionType.SUCCESS);
} else {
ctx.tellSuccess(tbMsg);
}
@@ -39,13 +39,13 @@ public abstract class TbAbstractExternalNode implements TbNode {
protected void tellFailure(TbContext ctx, TbMsg tbMsg, Throwable t) {
if (forceAck) {
if (t == null) {
- ctx.enqueueForTellNext(tbMsg.copyWithNewCtx(), TbRelationTypes.FAILURE);
+ ctx.enqueueForTellNext(tbMsg.copyWithNewCtx(), TbNodeConnectionType.FAILURE);
} else {
ctx.enqueueForTellFailure(tbMsg.copyWithNewCtx(), t);
}
} else {
if (t == null) {
- ctx.tellNext(tbMsg, TbRelationTypes.FAILURE);
+ ctx.tellNext(tbMsg, TbNodeConnectionType.FAILURE);
} else {
ctx.tellFailure(tbMsg, t);
}
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbAssetTypeSwitchNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbAssetTypeSwitchNode.java
index 21e109e8f3..39c37f5c81 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbAssetTypeSwitchNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbAssetTypeSwitchNode.java
@@ -31,10 +31,11 @@ import org.thingsboard.server.common.data.plugin.ComponentType;
type = ComponentType.FILTER,
name = "asset profile switch",
customRelations = true,
- relationTypes = {},
+ relationTypes = {"default"},
configClazz = EmptyNodeConfiguration.class,
nodeDescription = "Route incoming messages based on the name of the asset profile",
- nodeDetails = "Route incoming messages based on the name of the asset profile. The asset profile name is case-sensitive",
+ nodeDetails = "Route incoming messages based on the name of the asset profile. The asset profile name is case-sensitive.
" +
+ "Output connections: Asset profile name or Failure",
uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective = "tbNodeEmptyConfig")
public class TbAssetTypeSwitchNode extends TbAbstractTypeSwitchNode {
@@ -42,7 +43,8 @@ public class TbAssetTypeSwitchNode extends TbAbstractTypeSwitchNode {
@Override
protected String getRelationType(TbContext ctx, EntityId originator) throws TbNodeException {
if (!EntityType.ASSET.equals(originator.getEntityType())) {
- throw new TbNodeException("Unsupported originator type: " + originator.getEntityType() + "! Only 'ASSET' type is allowed.");
+ throw new TbNodeException("Unsupported originator type: " + originator.getEntityType().getNormalName() + "!" +
+ " Only " + EntityType.ASSET.getNormalName() + " type is allowed.");
}
AssetProfile assetProfile = ctx.getAssetProfileCache().get(ctx.getTenantId(), (AssetId) originator);
if (assetProfile == null) {
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckAlarmStatusNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckAlarmStatusNode.java
index dcaf6698cd..ca58e47554 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckAlarmStatusNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckAlarmStatusNode.java
@@ -18,7 +18,6 @@ package org.thingsboard.rule.engine.filter;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.rule.engine.api.RuleNode;
@@ -28,24 +27,25 @@ import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.server.common.data.alarm.Alarm;
-import org.thingsboard.server.common.data.alarm.AlarmStatus;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg;
import javax.annotation.Nullable;
-import java.io.IOException;
@Slf4j
@RuleNode(
type = ComponentType.FILTER,
- name = "check alarm status",
+ name = "alarm status filter",
configClazz = TbCheckAlarmStatusNodeConfig.class,
- relationTypes = {"True", "False"},
+ relationTypes = {TbNodeConnectionType.TRUE, TbNodeConnectionType.FALSE},
nodeDescription = "Checks alarm status.",
- nodeDetails = "Checks the alarm status to match one of the specified statuses.",
+ nodeDetails = "Checks the alarm status to match one of the specified statuses.
" +
+ "Output connections: True, False, Failure.",
uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective = "tbFilterNodeCheckAlarmStatusConfig")
public class TbCheckAlarmStatusNode implements TbNode {
+
private TbCheckAlarmStatusNodeConfig config;
@Override
@@ -60,33 +60,24 @@ public class TbCheckAlarmStatusNode implements TbNode {
ListenableFuture latest = ctx.getAlarmService().findAlarmByIdAsync(ctx.getTenantId(), alarm.getId());
- Futures.addCallback(latest, new FutureCallback() {
+ Futures.addCallback(latest, new FutureCallback<>() {
@Override
public void onSuccess(@Nullable Alarm result) {
- if (result != null) {
- boolean isPresent = false;
- for (AlarmStatus alarmStatus : config.getAlarmStatusList()) {
- if (result.getStatus() == alarmStatus) {
- isPresent = true;
- break;
- }
- }
- if (isPresent) {
- ctx.tellNext(msg, "True");
- } else {
- ctx.tellNext(msg, "False");
- }
- } else {
+ if (result == null) {
ctx.tellFailure(msg, new TbNodeException("No such alarm found."));
+ return;
}
+ boolean isPresent = config.getAlarmStatusList().stream()
+ .anyMatch(alarmStatus -> result.getStatus() == alarmStatus);
+ ctx.tellNext(msg, isPresent ? TbNodeConnectionType.TRUE : TbNodeConnectionType.FALSE);
}
@Override
public void onFailure(Throwable t) {
ctx.tellFailure(msg, t);
}
- }, MoreExecutors.directExecutor());
- } catch (IllegalArgumentException e) {
+ }, ctx.getDbCallbackExecutor());
+ } catch (Exception e) {
log.error("Failed to parse alarm: [{}]", msg.getData());
throw new TbNodeException(e);
}
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckAlarmStatusNodeConfig.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckAlarmStatusNodeConfig.java
index d979eb10ca..4c15d635b8 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckAlarmStatusNodeConfig.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckAlarmStatusNodeConfig.java
@@ -24,12 +24,14 @@ import java.util.List;
@Data
public class TbCheckAlarmStatusNodeConfig implements NodeConfiguration {
+
private List alarmStatusList;
@Override
public TbCheckAlarmStatusNodeConfig defaultConfiguration() {
- TbCheckAlarmStatusNodeConfig config = new TbCheckAlarmStatusNodeConfig();
+ var config = new TbCheckAlarmStatusNodeConfig();
config.setAlarmStatusList(Arrays.asList(AlarmStatus.ACTIVE_ACK, AlarmStatus.ACTIVE_UNACK));
return config;
}
+
}
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckMessageNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckMessageNode.java
index 3853b3a62a..a5ee3b969b 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckMessageNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckMessageNode.java
@@ -22,6 +22,7 @@ import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNode;
import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg;
@@ -33,12 +34,12 @@ import java.util.Map;
@RuleNode(
type = ComponentType.FILTER,
name = "check fields presence",
- relationTypes = {"True", "False"},
+ relationTypes = {TbNodeConnectionType.TRUE, TbNodeConnectionType.FALSE},
configClazz = TbCheckMessageNodeConfiguration.class,
nodeDescription = "Checks the presence of the specified fields in the message and/or metadata.",
- nodeDetails = "Checks the presence of the specified fields in the message and/or metadata. " +
- "By default, the rule node checks that all specified fields need to be present. " +
- "Uncheck the 'Check that all specified fields are present' if the presence of at least one field is sufficient.",
+ nodeDetails = "By default, the rule node checks that all specified fields are present. " +
+ "Uncheck the 'Check that all selected fields are present' if the presence of at least one field is sufficient.
" +
+ "Output connections: True, False, Failure",
uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective = "tbFilterNodeCheckMessageConfig")
public class TbCheckMessageNode implements TbNode {
@@ -59,11 +60,10 @@ public class TbCheckMessageNode implements TbNode {
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
try {
- if (config.isCheckAllKeys()) {
- ctx.tellNext(msg, allKeysData(msg) && allKeysMetadata(msg) ? "True" : "False");
- } else {
- ctx.tellNext(msg, atLeastOneData(msg) || atLeastOneMetadata(msg) ? "True" : "False");
- }
+ String relationType = config.isCheckAllKeys() ?
+ allKeysData(msg) && allKeysMetadata(msg) ? TbNodeConnectionType.TRUE : TbNodeConnectionType.FALSE :
+ atLeastOneData(msg) || atLeastOneMetadata(msg) ? TbNodeConnectionType.TRUE : TbNodeConnectionType.FALSE;
+ ctx.tellNext(msg, relationType);
} catch (Exception e) {
ctx.tellFailure(msg, e);
}
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckMessageNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckMessageNodeConfiguration.java
index f145be3fd2..25a6efa417 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckMessageNodeConfiguration.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckMessageNodeConfiguration.java
@@ -22,7 +22,7 @@ import java.util.Collections;
import java.util.List;
@Data
-public class TbCheckMessageNodeConfiguration implements NodeConfiguration {
+public class TbCheckMessageNodeConfiguration implements NodeConfiguration {
private List messageNames;
private List metadataNames;
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckRelationNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckRelationNode.java
index 4a0d6b05e7..3af5c9bfe1 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckRelationNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckRelationNode.java
@@ -15,22 +15,26 @@
*/
package org.thingsboard.rule.engine.filter;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.rule.engine.api.RuleNode;
import org.thingsboard.rule.engine.api.TbContext;
-import org.thingsboard.rule.engine.api.TbNode;
import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException;
+import org.thingsboard.rule.engine.api.TbVersionedNode;
import org.thingsboard.rule.engine.api.util.TbNodeUtils;
+import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.EntityIdFactory;
+import org.thingsboard.server.common.data.msg.TbNodeConnectionType;
import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.data.relation.EntityRelation;
import org.thingsboard.server.common.data.relation.EntitySearchDirection;
import org.thingsboard.server.common.data.relation.RelationTypeGroup;
+import org.thingsboard.server.common.data.util.TbPair;
import org.thingsboard.server.common.msg.TbMsg;
import java.util.List;
@@ -43,15 +47,20 @@ import static org.thingsboard.common.util.DonAsynchron.withCallback;
@Slf4j
@RuleNode(
type = ComponentType.FILTER,
- name = "check relation",
+ name = "check relation presence",
configClazz = TbCheckRelationNodeConfiguration.class,
- relationTypes = {"True", "False"},
+ version = 1,
+ relationTypes = {TbNodeConnectionType.TRUE, TbNodeConnectionType.FALSE},
nodeDescription = "Checks the presence of the relation between the originator of the message and other entities.",
- nodeDetails = "If 'check relation to specific entity' is selected, one must specify a related entity. " +
- "Otherwise, the rule node checks the presence of a relation to any entity that matches the direction and relation type criteria.",
+ nodeDetails = "If 'check relation to specific entity' is selected, you should specify a related entity. " +
+ "Otherwise, the rule node checks the presence of a relation to any entity. " +
+ "In both cases, relation lookup is based on configured direction and type.
" +
+ "Output connections: True, False, Failure",
uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective = "tbFilterNodeCheckRelationConfig")
-public class TbCheckRelationNode implements TbNode {
+public class TbCheckRelationNode implements TbVersionedNode {
+
+ private static final String DIRECTION_PROPERTY_NAME = "direction";
private TbCheckRelationNodeConfiguration config;
private EntityId singleEntityId;
@@ -60,6 +69,9 @@ public class TbCheckRelationNode implements TbNode {
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
this.config = TbNodeUtils.convert(configuration, TbCheckRelationNodeConfiguration.class);
if (config.isCheckForSingleEntity()) {
+ if (StringUtils.isEmpty(config.getEntityType()) || StringUtils.isEmpty(config.getEntityId())) {
+ throw new TbNodeException("Entity should be specified!");
+ }
this.singleEntityId = EntityIdFactory.getByTypeAndId(config.getEntityType(), config.getEntityId());
ctx.checkTenantEntity(singleEntityId);
}
@@ -67,44 +79,56 @@ public class TbCheckRelationNode implements TbNode {
@Override
public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException {
- ListenableFuture checkRelationFuture;
- if (config.isCheckForSingleEntity()) {
- checkRelationFuture = processSingle(ctx, msg);
- } else {
- checkRelationFuture = processList(ctx, msg);
- }
- withCallback(checkRelationFuture, filterResult -> ctx.tellNext(msg, filterResult ? "True" : "False"), t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
+ ListenableFuture checkRelationFuture = config.isCheckForSingleEntity() ?
+ processSingle(ctx, msg) : processList(ctx, msg);
+ withCallback(checkRelationFuture,
+ filterResult -> ctx.tellNext(msg, filterResult ? TbNodeConnectionType.TRUE : TbNodeConnectionType.FALSE),
+ t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
}
private ListenableFuture processSingle(TbContext ctx, TbMsg msg) {
EntityId from;
EntityId to;
if (EntitySearchDirection.FROM.name().equals(config.getDirection())) {
- from = singleEntityId;
- to = msg.getOriginator();
- } else {
to = singleEntityId;
from = msg.getOriginator();
+ } else {
+ from = singleEntityId;
+ to = msg.getOriginator();
}
return ctx.getRelationService().checkRelationAsync(ctx.getTenantId(), from, to, config.getRelationType(), RelationTypeGroup.COMMON);
}
private ListenableFuture processList(TbContext ctx, TbMsg msg) {
- if (EntitySearchDirection.FROM.name().equals(config.getDirection())) {
- return Futures.transformAsync(ctx.getRelationService()
- .findByToAndTypeAsync(ctx.getTenantId(), msg.getOriginator(), config.getRelationType(), RelationTypeGroup.COMMON), this::isEmptyList, MoreExecutors.directExecutor());
- } else {
- return Futures.transformAsync(ctx.getRelationService()
- .findByFromAndTypeAsync(ctx.getTenantId(), msg.getOriginator(), config.getRelationType(), RelationTypeGroup.COMMON), this::isEmptyList, MoreExecutors.directExecutor());
- }
+ ListenableFuture> relationListFuture = EntitySearchDirection.FROM.name().equals(config.getDirection()) ?
+ ctx.getRelationService().findByFromAndTypeAsync(ctx.getTenantId(), msg.getOriginator(), config.getRelationType(), RelationTypeGroup.COMMON) :
+ ctx.getRelationService().findByToAndTypeAsync(ctx.getTenantId(), msg.getOriginator(), config.getRelationType(), RelationTypeGroup.COMMON);
+ return Futures.transformAsync(relationListFuture, this::isEmptyList, ctx.getDbCallbackExecutor());
}
private ListenableFuture isEmptyList(List entityRelations) {
- if (entityRelations.isEmpty()) {
- return Futures.immediateFuture(false);
- } else {
- return Futures.immediateFuture(true);
+ return entityRelations.isEmpty() ? Futures.immediateFuture(false) : Futures.immediateFuture(true);
+ }
+
+ @Override
+ public TbPair upgrade(int fromVersion, JsonNode oldConfiguration) throws TbNodeException {
+ if (fromVersion == 0) {
+ var newConfigObjectNode = (ObjectNode) oldConfiguration;
+ if (!newConfigObjectNode.has(DIRECTION_PROPERTY_NAME)) {
+ throw new TbNodeException("property to update: '" + DIRECTION_PROPERTY_NAME + "' doesn't exists in configuration!");
+ }
+ String direction = newConfigObjectNode.get(DIRECTION_PROPERTY_NAME).asText();
+ if (EntitySearchDirection.TO.name().equals(direction)) {
+ newConfigObjectNode.put(DIRECTION_PROPERTY_NAME, EntitySearchDirection.FROM.name());
+ return new TbPair<>(true, newConfigObjectNode);
+ }
+ if (EntitySearchDirection.FROM.name().equals(direction)) {
+ newConfigObjectNode.put(DIRECTION_PROPERTY_NAME, EntitySearchDirection.TO.name());
+ return new TbPair<>(true, newConfigObjectNode);
+ }
+ throw new TbNodeException("property to update: '" + DIRECTION_PROPERTY_NAME + "' has invalid value!");
}
+ return new TbPair<>(false, oldConfiguration);
}
}
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckRelationNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckRelationNodeConfiguration.java
index aeda247583..1a8ab7068d 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckRelationNodeConfiguration.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbCheckRelationNodeConfiguration.java
@@ -17,6 +17,7 @@ package org.thingsboard.rule.engine.filter;
import lombok.Data;
import org.thingsboard.rule.engine.api.NodeConfiguration;
+import org.thingsboard.server.common.data.relation.EntityRelation;
import org.thingsboard.server.common.data.relation.EntitySearchDirection;
/**
@@ -33,9 +34,9 @@ public class TbCheckRelationNodeConfiguration implements NodeConfigurationDevice profile name