thingsboard-aplcache
Changes
application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java 81(+66 -15)
application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java 29(+18 -11)
application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java 147(+9 -138)
application/src/main/java/org/thingsboard/server/service/install/SystemDataLoaderService.java 4(+0 -4)
application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java 17(+17 -0)
application/src/test/java/org/thingsboard/server/controller/BaseComponentDescriptorControllerTest.java 21(+6 -15)
application/src/test/java/org/thingsboard/server/controller/BasePluginControllerTest.java 232(+0 -232)
application/src/test/java/org/thingsboard/server/controller/BaseRuleControllerTest.java 247(+0 -247)
application/src/test/java/org/thingsboard/server/controller/nosql/RuleControllerNoSqlTest.java 26(+0 -26)
application/src/test/java/org/thingsboard/server/controller/sql/PluginControllerSqlTest.java 26(+0 -26)
application/src/test/java/org/thingsboard/server/controller/sql/RuleControllerSqlTest.java 26(+0 -26)
application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java 2(+2 -0)
application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java 13(+7 -6)
application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java 5(+4 -1)
common/message/src/main/java/org/thingsboard/server/common/msg/core/AttributesUpdateRequest.java 2(+1 -1)
common/message/src/main/java/org/thingsboard/server/common/msg/core/BasicAttributesUpdateRequest.java 9(+4 -5)
common/message/src/main/java/org/thingsboard/server/common/msg/core/RuleEngineErrorMsg.java 16(+3 -13)
common/transport/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java 6(+3 -3)
extensions/extension-kafka/src/main/java/org/thingsboard/server/extensions/kafka/action/KafkaPluginAction.java 2(+1 -1)
extensions/extension-kafka/src/main/java/org/thingsboard/server/extensions/kafka/plugin/KafkaMsgHandler.java 4(+2 -2)
extensions/extension-mqtt/src/main/java/org/thingsboard/server/extensions/mqtt/action/MqttPluginAction.java 2(+1 -1)
extensions/extension-mqtt/src/main/java/org/thingsboard/server/extensions/mqtt/plugin/MqttMsgHandler.java 4(+2 -2)
extensions/extension-rabbitmq/src/main/java/org/thingsboard/server/extensions/rabbitmq/action/RabbitMqPluginAction.java 2(+1 -1)
extensions/extension-rabbitmq/src/main/java/org/thingsboard/server/extensions/rabbitmq/plugin/RabbitMqMsgHandler.java 2(+1 -1)
extensions/extension-rest-api-call/src/main/java/org/thingsboard/server/extensions/rest/action/RestApiCallPluginAction.java 2(+1 -1)
extensions/extension-rest-api-call/src/main/java/org/thingsboard/server/extensions/rest/plugin/RestApiCallMsgHandler.java 2(+1 -1)
extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginAction.java 2(+1 -1)
extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsMessageHandler.java 2(+1 -1)
extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginAction.java 2(+1 -1)
extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginAction.java 2(+1 -1)
extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsMessageHandler.java 2(+1 -1)
extensions-api/src/main/java/org/thingsboard/server/extensions/api/plugins/msg/UpdateAttributesRequestRuleToPluginMsg.java 6(+3 -3)
extensions-core/src/main/java/org/thingsboard/server/extensions/core/action/telemetry/TelemetryPluginAction.java 5(+2 -3)
extensions-core/src/main/java/org/thingsboard/server/extensions/core/filter/DeviceAttributesFilter.java 4(+2 -2)
extensions-core/src/main/java/org/thingsboard/server/extensions/core/filter/NashornJsEvaluator.java 4(+2 -2)
extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/handlers/TelemetryRuleMsgHandler.java 4(+2 -2)
extensions-core/src/main/java/org/thingsboard/server/extensions/core/processor/AlarmProcessor.java 4(+2 -2)
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java 76(+76 -0)
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNodeConfiguration.java 23(+15 -8)
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java 15(+6 -9)
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java 8(+3 -5)
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java 31(+31 -0)
transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java 3(+1 -2)
Details
diff --git a/application/src/main/conf/thingsboard.conf b/application/src/main/conf/thingsboard.conf
index a6e404d..2baee7d 100644
--- a/application/src/main/conf/thingsboard.conf
+++ b/application/src/main/conf/thingsboard.conf
@@ -14,7 +14,7 @@
# limitations under the License.
#
-export JAVA_OPTS="$JAVA_OPTS -Dplatform=@pkg.platform@"
+export JAVA_OPTS="$JAVA_OPTS -Dplatform=@pkg.platform@ -Dinstall.data_dir=@pkg.installFolder@"
export LOG_FILENAME=${pkg.name}.out
export LOADER_PATH=${pkg.installFolder}/conf,${pkg.installFolder}/extensions
export SQL_DATA_FOLDER=${pkg.installFolder}/data/sql
diff --git a/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json b/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json
new file mode 100644
index 0000000..a5573f8
--- /dev/null
+++ b/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json
@@ -0,0 +1,102 @@
+{
+ "ruleChain": {
+ "additionalInfo": null,
+ "name": "Root Rule Chain",
+ "firstRuleNodeId": null,
+ "root": true,
+ "debugMode": false,
+ "configuration": null
+ },
+ "metadata": {
+ "firstNodeIndex": 2,
+ "nodes": [
+ {
+ "additionalInfo": {
+ "layoutX": 639,
+ "layoutY": 113
+ },
+ "type": "org.thingsboard.rule.engine.filter.TbMsgTypeFilterNode",
+ "name": "PostAttributes",
+ "debugMode": true,
+ "configuration": {
+ "messageTypes": [
+ "POST_ATTRIBUTES_REQUEST"
+ ]
+ }
+ },
+ {
+ "additionalInfo": {
+ "layoutX": 638,
+ "layoutY": 206
+ },
+ "type": "org.thingsboard.rule.engine.filter.TbMsgTypeFilterNode",
+ "name": "PostTelemetry",
+ "debugMode": true,
+ "configuration": {
+ "messageTypes": [
+ "POST_TELEMETRY_REQUEST"
+ ]
+ }
+ },
+ {
+ "additionalInfo": {
+ "layoutX": 297,
+ "layoutY": 148
+ },
+ "type": "org.thingsboard.rule.engine.action.TbLogNode",
+ "name": "Log",
+ "debugMode": false,
+ "configuration": {
+ "jsScript": "return 'incoming message = ' + msg;"
+ }
+ },
+ {
+ "additionalInfo": {
+ "layoutX": 905,
+ "layoutY": 203
+ },
+ "type": "org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNode",
+ "name": "SaveTS",
+ "debugMode": true,
+ "configuration": {
+ "defaultTTL": 0
+ }
+ },
+ {
+ "additionalInfo": {
+ "layoutX": 904,
+ "layoutY": 110
+ },
+ "type": "org.thingsboard.rule.engine.telemetry.TbMsgAttributesNode",
+ "name": "save client attributes",
+ "debugMode": true,
+ "configuration": {
+ "scope": "CLIENT_SCOPE"
+ }
+ }
+ ],
+ "connections": [
+ {
+ "fromIndex": 0,
+ "toIndex": 4,
+ "type": "True"
+ },
+ {
+ "fromIndex": 1,
+ "toIndex": 3,
+ "type": "True"
+ },
+ {
+ "fromIndex": 2,
+ "toIndex": 0,
+ "type": "Success"
+ },
+ {
+ "fromIndex": 2,
+ "toIndex": 1,
+ "type": "Success"
+ }
+ ],
+ "ruleChainConnections": null
+ }
+}
diff --git a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
index 07ff89e..f6433e8 100644
--- a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
@@ -103,7 +103,7 @@ public class AppActor extends RuleChainManagerActor {
case DEVICE_CREDENTIALS_UPDATE_TO_DEVICE_ACTOR_MSG:
case DEVICE_NAME_OR_TYPE_UPDATE_TO_DEVICE_ACTOR_MSG:
case DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG:
- onToDeviceActorMsg((DeviceToDeviceActorMsg) msg);
+ onToDeviceActorMsg((TenantAwareMsg) msg);
break;
default:
return false;
diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
index e289f13..26c0a28 100644
--- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
@@ -19,6 +19,9 @@ import akka.actor.ActorContext;
import akka.actor.ActorRef;
import akka.event.LoggingAdapter;
import com.datastax.driver.core.utils.UUIDs;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
@@ -39,14 +42,18 @@ import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.common.msg.cluster.ClusterEventMsg;
import org.thingsboard.server.common.msg.cluster.ServerAddress;
import org.thingsboard.server.common.msg.core.AttributesUpdateNotification;
+import org.thingsboard.server.common.msg.core.AttributesUpdateRequest;
import org.thingsboard.server.common.msg.core.BasicCommandAckResponse;
+import org.thingsboard.server.common.msg.core.BasicGetAttributesResponse;
import org.thingsboard.server.common.msg.core.BasicStatusCodeResponse;
import org.thingsboard.server.common.msg.core.BasicToDeviceSessionActorMsg;
+import org.thingsboard.server.common.msg.core.GetAttributesRequest;
import org.thingsboard.server.common.msg.core.RuleEngineError;
import org.thingsboard.server.common.msg.core.RuleEngineErrorMsg;
import org.thingsboard.server.common.msg.core.SessionCloseMsg;
import org.thingsboard.server.common.msg.core.SessionCloseNotification;
import org.thingsboard.server.common.msg.core.SessionOpenMsg;
+import org.thingsboard.server.common.msg.core.StatusCodeResponse;
import org.thingsboard.server.common.msg.core.TelemetryUploadRequest;
import org.thingsboard.server.common.msg.core.ToDeviceRpcRequestMsg;
import org.thingsboard.server.common.msg.core.ToDeviceRpcResponseMsg;
@@ -64,10 +71,15 @@ import org.thingsboard.server.common.msg.timeout.DeviceActorQueueTimeoutMsg;
import org.thingsboard.server.common.msg.timeout.DeviceActorRpcTimeoutMsg;
import org.thingsboard.server.extensions.api.device.DeviceAttributesEventNotificationMsg;
import org.thingsboard.server.extensions.api.device.DeviceNameOrTypeUpdateMsg;
+import org.thingsboard.server.extensions.api.plugins.PluginCallback;
+import org.thingsboard.server.extensions.api.plugins.PluginContext;
import org.thingsboard.server.extensions.api.plugins.msg.FromDeviceRpcResponse;
import org.thingsboard.server.extensions.api.plugins.msg.RpcError;
+import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -114,7 +126,6 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
}
private void initAttributes() {
- //TODO: add invalidation of deviceType cache.
Device device = systemContext.getDeviceService().findDeviceById(deviceId);
this.deviceName = device.getName();
this.deviceType = device.getType();
@@ -238,6 +249,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
processSubscriptionCommands(context, msg);
processRpcResponses(context, msg);
processSessionStateMsgs(msg);
+
SessionMsgType sessionMsgType = msg.getPayload().getMsgType();
if (sessionMsgType.requiresRulesProcessing()) {
switch (sessionMsgType) {
@@ -245,6 +257,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
handleGetAttributesRequest(msg);
break;
case POST_ATTRIBUTES_REQUEST:
+ handlePostAttributesRequest(context, msg);
break;
case POST_TELEMETRY_REQUEST:
handlePostTelemetryRequest(context, msg);
@@ -256,14 +269,62 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
}
}
- private void handleGetAttributesRequest(DeviceToDeviceActorMsg msg) {
+ private void handleGetAttributesRequest(DeviceToDeviceActorMsg src) {
+ GetAttributesRequest request = (GetAttributesRequest) src.getPayload();
+ ListenableFuture<List<AttributeKvEntry>> clientAttributesFuture = getAttributeKvEntries(deviceId, DataConstants.CLIENT_SCOPE, request.getClientAttributeNames());
+ ListenableFuture<List<AttributeKvEntry>> sharedAttributesFuture = getAttributeKvEntries(deviceId, DataConstants.SHARED_SCOPE, request.getClientAttributeNames());
+
+ Futures.addCallback(Futures.allAsList(Arrays.asList(clientAttributesFuture, sharedAttributesFuture)), new FutureCallback<List<List<AttributeKvEntry>>>() {
+ @Override
+ public void onSuccess(@Nullable List<List<AttributeKvEntry>> result) {
+ BasicGetAttributesResponse response = BasicGetAttributesResponse.onSuccess(request.getMsgType(),
+ request.getRequestId(), BasicAttributeKVMsg.from(result.get(0), result.get(1)));
+ sendMsgToSessionActor(new BasicToDeviceSessionActorMsg(response, src.getSessionId()), src.getServerAddress());
+ }
+ @Override
+ public void onFailure(Throwable t) {
+ if (t instanceof Exception) {
+ ToDeviceMsg toDeviceMsg = BasicStatusCodeResponse.onError(SessionMsgType.GET_ATTRIBUTES_REQUEST, request.getRequestId(), (Exception) t);
+ sendMsgToSessionActor(new BasicToDeviceSessionActorMsg(toDeviceMsg, src.getSessionId()), src.getServerAddress());
+ } else {
+ logger.error("[{}] Failed to process attributes request", deviceId, t);
+ }
+ }
+ });
+ }
+
+ private ListenableFuture<List<AttributeKvEntry>> getAttributeKvEntries(DeviceId deviceId, String scope, Optional<Set<String>> names) {
+ if (names.isPresent()) {
+ if (!names.get().isEmpty()) {
+ return systemContext.getAttributesService().find(deviceId, scope, names.get());
+ } else {
+ return systemContext.getAttributesService().findAll(deviceId, scope);
+ }
+ } else {
+ return Futures.immediateFuture(Collections.emptyList());
+ }
+ }
+
+ private void handlePostAttributesRequest(ActorContext context, DeviceToDeviceActorMsg src) {
+ AttributesUpdateRequest request = (AttributesUpdateRequest) src.getPayload();
+
+ JsonObject json = new JsonObject();
+ for (AttributeKvEntry kv : request.getAttributes()) {
+ kv.getBooleanValue().ifPresent(v -> json.addProperty(kv.getKey(), v));
+ kv.getLongValue().ifPresent(v -> json.addProperty(kv.getKey(), v));
+ kv.getDoubleValue().ifPresent(v -> json.addProperty(kv.getKey(), v));
+ kv.getStrValue().ifPresent(v -> json.addProperty(kv.getKey(), v));
+ }
+
+ TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), deviceId, defaultMetaData, TbMsgDataType.JSON, gson.toJson(json));
+ pushToRuleEngineWithTimeout(context, tbMsg, src, request);
}
private void handlePostTelemetryRequest(ActorContext context, DeviceToDeviceActorMsg src) {
- TelemetryUploadRequest telemetry = (TelemetryUploadRequest) src.getPayload();
+ TelemetryUploadRequest request = (TelemetryUploadRequest) src.getPayload();
- Map<Long, List<KvEntry>> tsData = telemetry.getData();
+ Map<Long, List<KvEntry>> tsData = request.getData();
JsonArray json = new JsonArray();
for (Map.Entry<Long, List<KvEntry>> entry : tsData.entrySet()) {
@@ -281,7 +342,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
}
TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, defaultMetaData, TbMsgDataType.JSON, gson.toJson(json));
- pushToRuleEngineWithTimeout(context, tbMsg, src, telemetry);
+ pushToRuleEngineWithTimeout(context, tbMsg, src, request);
}
private void pushToRuleEngineWithTimeout(ActorContext context, TbMsg tbMsg, DeviceToDeviceActorMsg src, FromDeviceRequestMsg fromDeviceRequestMsg) {
@@ -403,16 +464,6 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
}
}
- private List<AttributeKvEntry> fetchAttributes(String scope) {
- try {
- //TODO: replace this with async operation. Happens only during actor creation, but is still criticla for performance,
- return systemContext.getAttributesService().findAll(this.deviceId, scope).get();
- } catch (InterruptedException | ExecutionException e) {
- logger.warning("[{}] Failed to fetch attributes for scope: {}", deviceId, scope);
- throw new RuntimeException(e);
- }
- }
-
void processCredentialsUpdate() {
sessions.forEach((k, v) -> {
sendMsgToSessionActor(new BasicToDeviceSessionActorMsg(new SessionCloseNotification(), k), v.getServer());
diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java
index f60e7dd..55525c6 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java
@@ -1,12 +1,12 @@
/**
* Copyright © 2016-2018 The Thingsboard Authors
- * <p>
+ *
* 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ * 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.
diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
index e703cb7..f05b9a8 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
@@ -1,12 +1,12 @@
/**
* Copyright © 2016-2018 The Thingsboard Authors
- * <p>
+ *
* 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ * 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.
@@ -60,6 +60,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
private RuleNodeId firstId;
private RuleNodeCtx firstNode;
+ private boolean started;
RuleChainActorMessageProcessor(TenantId tenantId, RuleChainId ruleChainId, ActorSystemContext systemContext
, LoggingAdapter logger, ActorRef parent, ActorRef self) {
@@ -73,14 +74,19 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
@Override
public void start(ActorContext context) throws Exception {
- RuleChain ruleChain = service.findRuleChainById(entityId);
- List<RuleNode> ruleNodeList = service.getRuleChainNodes(entityId);
- // Creating and starting the actors;
- for (RuleNode ruleNode : ruleNodeList) {
- ActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode);
- nodeActors.put(ruleNode.getId(), new RuleNodeCtx(tenantId, self, ruleNodeActor, ruleNode));
+ if (!started) {
+ RuleChain ruleChain = service.findRuleChainById(entityId);
+ List<RuleNode> ruleNodeList = service.getRuleChainNodes(entityId);
+ // Creating and starting the actors;
+ for (RuleNode ruleNode : ruleNodeList) {
+ ActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode);
+ nodeActors.put(ruleNode.getId(), new RuleNodeCtx(tenantId, self, ruleNodeActor, ruleNode));
+ }
+ initRoutes(ruleChain, ruleNodeList);
+ started = true;
+ } else {
+ onUpdate(context);
}
- initRoutes(ruleChain, ruleNodeList);
}
@Override
@@ -115,6 +121,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
nodeActors.clear();
nodeRoutes.clear();
context.stop(self);
+ started = false;
}
@Override
diff --git a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
index fb38134..0963a6d 100644
--- a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
@@ -81,7 +81,7 @@ public class TenantActor extends RuleChainManagerActor {
case DEVICE_CREDENTIALS_UPDATE_TO_DEVICE_ACTOR_MSG:
case DEVICE_NAME_OR_TYPE_UPDATE_TO_DEVICE_ACTOR_MSG:
case DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG:
- onToDeviceActorMsg((DeviceToDeviceActorMsg) msg);
+ onToDeviceActorMsg((DeviceAwareMsg) msg);
break;
default:
return false;
diff --git a/application/src/main/java/org/thingsboard/server/controller/RpcController.java b/application/src/main/java/org/thingsboard/server/controller/RpcController.java
index 3ca73f6..2c7d2e6 100644
--- a/application/src/main/java/org/thingsboard/server/controller/RpcController.java
+++ b/application/src/main/java/org/thingsboard/server/controller/RpcController.java
@@ -114,9 +114,10 @@ public class RpcController extends BaseController {
final DeferredResult<ResponseEntity> response = new DeferredResult<>();
long timeout = System.currentTimeMillis() + (cmd.getTimeout() != null ? cmd.getTimeout() : DEFAULT_TIMEOUT);
ToDeviceRpcRequestBody body = new ToDeviceRpcRequestBody(cmd.getMethodName(), cmd.getRequestData());
- accessValidator.validate(currentUser, deviceId, new FutureCallback<ValidationResult>() {
+ accessValidator.validate(currentUser, deviceId, new HttpValidationCallback(response, new FutureCallback<DeferredResult<ResponseEntity>>() {
@Override
- public void onSuccess(@Nullable ValidationResult result) {
+ public void onSuccess(@Nullable DeferredResult<ResponseEntity> result) {
+
ToDeviceRpcRequest rpcRequest = new ToDeviceRpcRequest(UUID.randomUUID(),
tenantId,
deviceId,
@@ -124,7 +125,7 @@ public class RpcController extends BaseController {
timeout,
body
);
- deviceRpcService.process(rpcRequest, new LocalRequestMetaData(rpcRequest, currentUser, response));
+ deviceRpcService.process(rpcRequest, new LocalRequestMetaData(rpcRequest, currentUser, result));
}
@Override
@@ -138,7 +139,7 @@ public class RpcController extends BaseController {
deviceRpcService.logRpcCall(currentUser, deviceId, body, oneWay, Optional.empty(), e);
response.setResult(entity);
}
- });
+ }));
return response;
} catch (IOException ioe) {
throw new ThingsboardException("Invalid request body", ioe, ThingsboardErrorCode.BAD_REQUEST_PARAMS);
diff --git a/application/src/main/java/org/thingsboard/server/controller/TenantController.java b/application/src/main/java/org/thingsboard/server/controller/TenantController.java
index bf49074..1a7c116 100644
--- a/application/src/main/java/org/thingsboard/server/controller/TenantController.java
+++ b/application/src/main/java/org/thingsboard/server/controller/TenantController.java
@@ -15,21 +15,34 @@
*/
package org.thingsboard.server.controller;
+import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.security.access.prepost.PreAuthorize;
-import org.springframework.web.bind.annotation.*;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
+import org.springframework.web.bind.annotation.ResponseStatus;
+import org.springframework.web.bind.annotation.RestController;
import org.thingsboard.server.common.data.Tenant;
+import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.TextPageData;
import org.thingsboard.server.common.data.page.TextPageLink;
import org.thingsboard.server.dao.tenant.TenantService;
-import org.thingsboard.server.common.data.exception.ThingsboardException;
+import org.thingsboard.server.service.install.InstallScripts;
@RestController
@RequestMapping("/api")
+@Slf4j
public class TenantController extends BaseController {
-
+
+ @Autowired
+ private InstallScripts installScripts;
+
@Autowired
private TenantService tenantService;
@@ -49,10 +62,15 @@ public class TenantController extends BaseController {
@PreAuthorize("hasAuthority('SYS_ADMIN')")
@RequestMapping(value = "/tenant", method = RequestMethod.POST)
- @ResponseBody
+ @ResponseBody
public Tenant saveTenant(@RequestBody Tenant tenant) throws ThingsboardException {
try {
- return checkNotNull(tenantService.saveTenant(tenant));
+ boolean newTenant = tenant.getId() == null;
+ tenant = checkNotNull(tenantService.saveTenant(tenant));
+ if (newTenant) {
+ installScripts.createDefaultRuleChains(tenant.getId());
+ }
+ return tenant;
} catch (Exception e) {
throw handleException(e);
}
@@ -72,7 +90,7 @@ public class TenantController extends BaseController {
}
@PreAuthorize("hasAuthority('SYS_ADMIN')")
- @RequestMapping(value = "/tenants", params = { "limit" }, method = RequestMethod.GET)
+ @RequestMapping(value = "/tenants", params = {"limit"}, method = RequestMethod.GET)
@ResponseBody
public TextPageData<Tenant> getTenants(@RequestParam int limit,
@RequestParam(required = false) String textSearch,
@@ -85,5 +103,5 @@ public class TenantController extends BaseController {
throw handleException(e);
}
}
-
+
}
diff --git a/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java b/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java
index 9f7f044..e33c2e3 100644
--- a/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/DefaultSystemDataLoaderService.java
@@ -60,21 +60,12 @@ import java.nio.file.Paths;
@Slf4j
public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
- private static final String JSON_DIR = "json";
- private static final String SYSTEM_DIR = "system";
- private static final String DEMO_DIR = "demo";
- private static final String WIDGET_BUNDLES_DIR = "widget_bundles";
- private static final String PLUGINS_DIR = "plugins";
- private static final String RULES_DIR = "rules";
- private static final String DASHBOARDS_DIR = "dashboards";
-
private static final ObjectMapper objectMapper = new ObjectMapper();
- public static final String JSON_EXT = ".json";
public static final String CUSTOMER_CRED = "customer";
public static final String DEFAULT_DEVICE_TYPE = "default";
- @Value("${install.data_dir}")
- private String dataDir;
+ @Autowired
+ private InstallScripts installScripts;
@Autowired
private BCryptPasswordEncoder passwordEncoder;
@@ -89,15 +80,6 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
private WidgetsBundleService widgetsBundleService;
@Autowired
- private WidgetTypeService widgetTypeService;
-
- @Autowired
- private PluginService pluginService;
-
- @Autowired
- private RuleService ruleService;
-
- @Autowired
private TenantService tenantService;
@Autowired
@@ -109,9 +91,6 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
@Autowired
private DeviceCredentialsService deviceCredentialsService;
- @Autowired
- private DashboardService dashboardService;
-
@Bean
protected BCryptPasswordEncoder passwordEncoder() {
return new BCryptPasswordEncoder();
@@ -147,55 +126,12 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
}
@Override
- public void loadSystemWidgets() throws Exception {
- Path widgetBundlesDir = Paths.get(dataDir, JSON_DIR, SYSTEM_DIR, WIDGET_BUNDLES_DIR);
- try (DirectoryStream<Path> dirStream = Files.newDirectoryStream(widgetBundlesDir, path -> path.toString().endsWith(JSON_EXT))) {
- dirStream.forEach(
- path -> {
- try {
- JsonNode widgetsBundleDescriptorJson = objectMapper.readTree(path.toFile());
- JsonNode widgetsBundleJson = widgetsBundleDescriptorJson.get("widgetsBundle");
- WidgetsBundle widgetsBundle = objectMapper.treeToValue(widgetsBundleJson, WidgetsBundle.class);
- WidgetsBundle savedWidgetsBundle = widgetsBundleService.saveWidgetsBundle(widgetsBundle);
- JsonNode widgetTypesArrayJson = widgetsBundleDescriptorJson.get("widgetTypes");
- widgetTypesArrayJson.forEach(
- widgetTypeJson -> {
- try {
- WidgetType widgetType = objectMapper.treeToValue(widgetTypeJson, WidgetType.class);
- widgetType.setBundleAlias(savedWidgetsBundle.getAlias());
- widgetTypeService.saveWidgetType(widgetType);
- } catch (Exception e) {
- log.error("Unable to load widget type from json: [{}]", path.toString());
- throw new RuntimeException("Unable to load widget type from json", e);
- }
- }
- );
- } catch (Exception e) {
- log.error("Unable to load widgets bundle from json: [{}]", path.toString());
- throw new RuntimeException("Unable to load widgets bundle from json", e);
- }
- }
- );
- }
- }
-
- @Override
- public void loadSystemPlugins() throws Exception {
-// loadPlugins(Paths.get(dataDir, JSON_DIR, SYSTEM_DIR, PLUGINS_DIR), null);
- }
-
-
- @Override
- public void loadSystemRules() throws Exception {
-// loadRules(Paths.get(dataDir, JSON_DIR, SYSTEM_DIR, RULES_DIR), null);
- }
-
- @Override
public void loadDemoData() throws Exception {
Tenant demoTenant = new Tenant();
demoTenant.setRegion("Global");
demoTenant.setTitle("Tenant");
demoTenant = tenantService.saveTenant(demoTenant);
+ installScripts.createDefaultRuleChains(demoTenant.getId());
createUser(Authority.TENANT_ADMIN, demoTenant.getId(), null, "tenant@thingsboard.org", "tenant");
Customer customerA = new Customer();
@@ -227,9 +163,7 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
createDevice(demoTenant.getId(), null, DEFAULT_DEVICE_TYPE, "Raspberry Pi Demo Device", "RASPBERRY_PI_DEMO_TOKEN", "Demo device that is used in " +
"Raspberry Pi GPIO control sample application");
-// loadPlugins(Paths.get(dataDir, JSON_DIR, DEMO_DIR, PLUGINS_DIR), demoTenant.getId());
-// loadRules(Paths.get(dataDir, JSON_DIR, DEMO_DIR, RULES_DIR), demoTenant.getId());
- loadDashboards(Paths.get(dataDir, JSON_DIR, DEMO_DIR, DASHBOARDS_DIR), demoTenant.getId(), null);
+ installScripts.loadDashboards(demoTenant.getId(), null);
}
@Override
@@ -240,6 +174,11 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
}
}
+ @Override
+ public void loadSystemWidgets() throws Exception {
+ installScripts.loadSystemWidgets();
+ }
+
private User createUser(Authority authority,
TenantId tenantId,
CustomerId customerId,
@@ -282,72 +221,4 @@ public class DefaultSystemDataLoaderService implements SystemDataLoaderService {
return device;
}
- private void loadPlugins(Path pluginsDir, TenantId tenantId) throws Exception{
- try (DirectoryStream<Path> dirStream = Files.newDirectoryStream(pluginsDir, path -> path.toString().endsWith(JSON_EXT))) {
- dirStream.forEach(
- path -> {
- try {
- JsonNode pluginJson = objectMapper.readTree(path.toFile());
- PluginMetaData plugin = objectMapper.treeToValue(pluginJson, PluginMetaData.class);
- plugin.setTenantId(tenantId);
- if (plugin.getState() == ComponentLifecycleState.ACTIVE) {
- plugin.setState(ComponentLifecycleState.SUSPENDED);
- PluginMetaData savedPlugin = pluginService.savePlugin(plugin);
- pluginService.activatePluginById(savedPlugin.getId());
- } else {
- pluginService.savePlugin(plugin);
- }
- } catch (Exception e) {
- log.error("Unable to load plugin from json: [{}]", path.toString());
- throw new RuntimeException("Unable to load plugin from json", e);
- }
- }
- );
- }
- }
-
- private void loadRules(Path rulesDir, TenantId tenantId) throws Exception {
- try (DirectoryStream<Path> dirStream = Files.newDirectoryStream(rulesDir, path -> path.toString().endsWith(JSON_EXT))) {
- dirStream.forEach(
- path -> {
- try {
- JsonNode ruleJson = objectMapper.readTree(path.toFile());
- RuleMetaData rule = objectMapper.treeToValue(ruleJson, RuleMetaData.class);
- rule.setTenantId(tenantId);
- if (rule.getState() == ComponentLifecycleState.ACTIVE) {
- rule.setState(ComponentLifecycleState.SUSPENDED);
- RuleMetaData savedRule = ruleService.saveRule(rule);
- ruleService.activateRuleById(savedRule.getId());
- } else {
- ruleService.saveRule(rule);
- }
- } catch (Exception e) {
- log.error("Unable to load rule from json: [{}]", path.toString());
- throw new RuntimeException("Unable to load rule from json", e);
- }
- }
- );
- }
- }
-
- private void loadDashboards(Path dashboardsDir, TenantId tenantId, CustomerId customerId) throws Exception {
- try (DirectoryStream<Path> dirStream = Files.newDirectoryStream(dashboardsDir, path -> path.toString().endsWith(JSON_EXT))) {
- dirStream.forEach(
- path -> {
- try {
- JsonNode dashboardJson = objectMapper.readTree(path.toFile());
- Dashboard dashboard = objectMapper.treeToValue(dashboardJson, Dashboard.class);
- dashboard.setTenantId(tenantId);
- Dashboard savedDashboard = dashboardService.saveDashboard(dashboard);
- if (customerId != null && !customerId.isNullUid()) {
- dashboardService.assignDashboardToCustomer(savedDashboard.getId(), customerId);
- }
- } catch (Exception e) {
- log.error("Unable to load dashboard from json: [{}]", path.toString());
- throw new RuntimeException("Unable to load dashboard from json", e);
- }
- }
- );
- }
- }
}
diff --git a/application/src/main/java/org/thingsboard/server/service/install/InstallScripts.java b/application/src/main/java/org/thingsboard/server/service/install/InstallScripts.java
new file mode 100644
index 0000000..5376e76
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/service/install/InstallScripts.java
@@ -0,0 +1,182 @@
+/**
+ * Copyright © 2016-2018 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.service.install;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+import org.thingsboard.server.common.data.Dashboard;
+import org.thingsboard.server.common.data.id.CustomerId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.rule.RuleChain;
+import org.thingsboard.server.common.data.rule.RuleChainMetaData;
+import org.thingsboard.server.common.data.widget.WidgetType;
+import org.thingsboard.server.common.data.widget.WidgetsBundle;
+import org.thingsboard.server.dao.dashboard.DashboardService;
+import org.thingsboard.server.dao.rule.RuleChainService;
+import org.thingsboard.server.dao.widget.WidgetTypeService;
+import org.thingsboard.server.dao.widget.WidgetsBundleService;
+
+import java.io.IOException;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import static org.thingsboard.server.service.install.DatabaseHelper.objectMapper;
+
+/**
+ * Created by ashvayka on 18.04.18.
+ */
+@Component
+@Slf4j
+public class InstallScripts {
+
+ public static final String APP_DIR = "application";
+ public static final String SRC_DIR = "src";
+ public static final String MAIN_DIR = "main";
+ public static final String DATA_DIR = "data";
+ public static final String JSON_DIR = "json";
+ public static final String SYSTEM_DIR = "system";
+ public static final String TENANT_DIR = "tenant";
+ public static final String DEMO_DIR = "demo";
+ public static final String RULE_CHAINS_DIR = "rule_chains";
+ public static final String WIDGET_BUNDLES_DIR = "widget_bundles";
+ public static final String DASHBOARDS_DIR = "dashboards";
+
+ public static final String JSON_EXT = ".json";
+
+ @Value("${install.data_dir:}")
+ private String dataDir;
+
+ @Autowired
+ private RuleChainService ruleChainService;
+
+ @Autowired
+ private DashboardService dashboardService;
+
+ @Autowired
+ private WidgetTypeService widgetTypeService;
+
+ @Autowired
+ private WidgetsBundleService widgetsBundleService;
+
+ public Path getTenantRuleChainsDir() {
+ return Paths.get(getDataDir(), JSON_DIR, TENANT_DIR, RULE_CHAINS_DIR);
+ }
+
+ public String getDataDir() {
+ if (!StringUtils.isEmpty(dataDir)) {
+ return dataDir;
+ } else {
+ String workDir = System.getProperty("user.dir");
+ if (workDir.endsWith("application")) {
+ return Paths.get(workDir, SRC_DIR, MAIN_DIR, DATA_DIR).toString();
+ } else {
+ Path dataDirPath = Paths.get(workDir, APP_DIR, SRC_DIR, MAIN_DIR, DATA_DIR);
+ if (Files.exists(dataDirPath)) {
+ return dataDirPath.toString();
+ } else {
+ throw new RuntimeException("Not valid working directory: " + workDir + ". Please use either root project directory, application module directory or specify valid \"install.data_dir\" ENV variable to avoid automatic data directory lookup!");
+ }
+ }
+ }
+ }
+
+ public void createDefaultRuleChains(TenantId tenantId) throws IOException {
+ Path tenantChainsDir = getTenantRuleChainsDir();
+ try (DirectoryStream<Path> dirStream = Files.newDirectoryStream(tenantChainsDir, path -> path.toString().endsWith(InstallScripts.JSON_EXT))) {
+ dirStream.forEach(
+ path -> {
+ try {
+ JsonNode ruleChainJson = objectMapper.readTree(path.toFile());
+ RuleChain ruleChain = objectMapper.treeToValue(ruleChainJson.get("ruleChain"), RuleChain.class);
+ RuleChainMetaData ruleChainMetaData = objectMapper.treeToValue(ruleChainJson.get("metadata"), RuleChainMetaData.class);
+
+ ruleChain.setTenantId(tenantId);
+ ruleChain = ruleChainService.saveRuleChain(ruleChain);
+
+ ruleChainMetaData.setRuleChainId(ruleChain.getId());
+ ruleChainService.saveRuleChainMetaData(ruleChainMetaData);
+ } catch (Exception e) {
+ log.error("Unable to load rule chain from json: [{}]", path.toString());
+ throw new RuntimeException("Unable to load rule chain from json", e);
+ }
+ }
+ );
+ }
+ }
+
+ public void loadSystemWidgets() throws Exception {
+ Path widgetBundlesDir = Paths.get(getDataDir(), JSON_DIR, SYSTEM_DIR, WIDGET_BUNDLES_DIR);
+ try (DirectoryStream<Path> dirStream = Files.newDirectoryStream(widgetBundlesDir, path -> path.toString().endsWith(JSON_EXT))) {
+ dirStream.forEach(
+ path -> {
+ try {
+ JsonNode widgetsBundleDescriptorJson = objectMapper.readTree(path.toFile());
+ JsonNode widgetsBundleJson = widgetsBundleDescriptorJson.get("widgetsBundle");
+ WidgetsBundle widgetsBundle = objectMapper.treeToValue(widgetsBundleJson, WidgetsBundle.class);
+ WidgetsBundle savedWidgetsBundle = widgetsBundleService.saveWidgetsBundle(widgetsBundle);
+ JsonNode widgetTypesArrayJson = widgetsBundleDescriptorJson.get("widgetTypes");
+ widgetTypesArrayJson.forEach(
+ widgetTypeJson -> {
+ try {
+ WidgetType widgetType = objectMapper.treeToValue(widgetTypeJson, WidgetType.class);
+ widgetType.setBundleAlias(savedWidgetsBundle.getAlias());
+ widgetTypeService.saveWidgetType(widgetType);
+ } catch (Exception e) {
+ log.error("Unable to load widget type from json: [{}]", path.toString());
+ throw new RuntimeException("Unable to load widget type from json", e);
+ }
+ }
+ );
+ } catch (Exception e) {
+ log.error("Unable to load widgets bundle from json: [{}]", path.toString());
+ throw new RuntimeException("Unable to load widgets bundle from json", e);
+ }
+ }
+ );
+ }
+ }
+
+ public void loadDashboards(TenantId tenantId, CustomerId customerId) throws Exception {
+ Path dashboardsDir = Paths.get(getDataDir(), JSON_DIR, DEMO_DIR, DASHBOARDS_DIR);
+ try (DirectoryStream<Path> dirStream = Files.newDirectoryStream(dashboardsDir, path -> path.toString().endsWith(JSON_EXT))) {
+ dirStream.forEach(
+ path -> {
+ try {
+ JsonNode dashboardJson = objectMapper.readTree(path.toFile());
+ Dashboard dashboard = objectMapper.treeToValue(dashboardJson, Dashboard.class);
+ dashboard.setTenantId(tenantId);
+ Dashboard savedDashboard = dashboardService.saveDashboard(dashboard);
+ if (customerId != null && !customerId.isNullUid()) {
+ dashboardService.assignDashboardToCustomer(savedDashboard.getId(), customerId);
+ }
+ } catch (Exception e) {
+ log.error("Unable to load dashboard from json: [{}]", path.toString());
+ throw new RuntimeException("Unable to load dashboard from json", e);
+ }
+ }
+ );
+ }
+ }
+
+
+}
diff --git a/application/src/main/java/org/thingsboard/server/service/install/SystemDataLoaderService.java b/application/src/main/java/org/thingsboard/server/service/install/SystemDataLoaderService.java
index a3dcb68..f3a6af4 100644
--- a/application/src/main/java/org/thingsboard/server/service/install/SystemDataLoaderService.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/SystemDataLoaderService.java
@@ -23,10 +23,6 @@ public interface SystemDataLoaderService {
void loadSystemWidgets() throws Exception;
- void loadSystemPlugins() throws Exception;
-
- void loadSystemRules() throws Exception;
-
void loadDemoData() throws Exception;
void deleteSystemWidgetBundle(String bundleAlias) throws Exception;
diff --git a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultDeviceRpcService.java b/application/src/main/java/org/thingsboard/server/service/rpc/DefaultDeviceRpcService.java
index 018f415..4a730db 100644
--- a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultDeviceRpcService.java
+++ b/application/src/main/java/org/thingsboard/server/service/rpc/DefaultDeviceRpcService.java
@@ -96,8 +96,10 @@ public class DefaultDeviceRpcService implements DeviceRpcService {
sendRpcRequest(request);
UUID requestId = request.getId();
localRpcRequests.put(requestId, metaData);
- long timeout = Math.max(0, System.currentTimeMillis() - request.getExpirationTime());
+ long timeout = Math.max(0, request.getExpirationTime() - System.currentTimeMillis());
+ log.error("[{}] processing the request: [{}]", this.hashCode(), requestId);
rpcCallBackExecutor.schedule(() -> {
+ log.error("[{}] timeout the request: [{}]", this.hashCode(), requestId);
LocalRequestMetaData localMetaData = localRpcRequests.remove(requestId);
if (localMetaData != null) {
reply(localMetaData, new FromDeviceRpcResponse(requestId, null, RpcError.TIMEOUT));
@@ -118,6 +120,7 @@ public class DefaultDeviceRpcService implements DeviceRpcService {
@Override
public void process(FromDeviceRpcResponse response) {
+ log.error("[{}] response the request: [{}]", this.hashCode(), response.getId());
//TODO: send to another server if needed.
UUID requestId = response.getId();
LocalRequestMetaData md = localRpcRequests.remove(requestId);
diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java
index 93fe767..30c7dc3 100644
--- a/application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/AbstractRuleEngineControllerTest.java
@@ -16,6 +16,8 @@
package org.thingsboard.server.controller;
import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import org.springframework.beans.factory.annotation.Autowired;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.Event;
import org.thingsboard.server.common.data.id.EntityId;
@@ -25,12 +27,18 @@ import org.thingsboard.server.common.data.page.TimePageData;
import org.thingsboard.server.common.data.page.TimePageLink;
import org.thingsboard.server.common.data.rule.RuleChain;
import org.thingsboard.server.common.data.rule.RuleChainMetaData;
+import org.thingsboard.server.dao.rule.RuleChainService;
+
+import java.io.IOException;
/**
* Created by ashvayka on 20.03.18.
*/
public class AbstractRuleEngineControllerTest extends AbstractControllerTest {
+ @Autowired
+ protected RuleChainService ruleChainService;
+
protected RuleChain saveRuleChain(RuleChain ruleChain) throws Exception {
return doPost("/api/ruleChain", ruleChain, RuleChain.class);
}
@@ -53,4 +61,13 @@ public class AbstractRuleEngineControllerTest extends AbstractControllerTest {
new TypeReference<TimePageData<Event>>() {
}, pageLink, entityId.getEntityType(), entityId.getId(), DataConstants.DEBUG_RULE_NODE, tenantId.getId());
}
+
+ protected JsonNode getMetadata(Event outEvent) {
+ String metaDataStr = outEvent.getBody().get("metadata").asText();
+ try {
+ return mapper.readTree(metaDataStr);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
}
diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseComponentDescriptorControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseComponentDescriptorControllerTest.java
index 4346538..d04fb64 100644
--- a/application/src/test/java/org/thingsboard/server/controller/BaseComponentDescriptorControllerTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/BaseComponentDescriptorControllerTest.java
@@ -20,6 +20,7 @@ import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.thingsboard.rule.engine.filter.TbJsFilterNode;
import org.thingsboard.server.common.data.Tenant;
import org.thingsboard.server.common.data.User;
import org.thingsboard.server.common.data.plugin.ComponentDescriptor;
@@ -35,7 +36,7 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
public abstract class BaseComponentDescriptorControllerTest extends AbstractControllerTest {
- private static final int AMOUNT_OF_DEFAULT_PLUGINS_DESCRIPTORS = 5;
+ private static final int AMOUNT_OF_DEFAULT_FILTER_NODES = 3;
private Tenant savedTenant;
private User tenantAdmin;
@@ -69,38 +70,28 @@ public abstract class BaseComponentDescriptorControllerTest extends AbstractCont
@Test
public void testGetByClazz() throws Exception {
ComponentDescriptor descriptor =
- doGet("/api/component/" + TelemetryStoragePlugin.class.getName(), ComponentDescriptor.class);
+ doGet("/api/component/" + TbJsFilterNode.class.getName(), ComponentDescriptor.class);
Assert.assertNotNull(descriptor);
Assert.assertNotNull(descriptor.getId());
Assert.assertNotNull(descriptor.getName());
Assert.assertEquals(ComponentScope.TENANT, descriptor.getScope());
- Assert.assertEquals(ComponentType.PLUGIN, descriptor.getType());
+ Assert.assertEquals(ComponentType.FILTER, descriptor.getType());
Assert.assertEquals(descriptor.getClazz(), descriptor.getClazz());
}
@Test
public void testGetByType() throws Exception {
List<ComponentDescriptor> descriptors = readResponse(
- doGet("/api/components/" + ComponentType.PLUGIN).andExpect(status().isOk()), new TypeReference<List<ComponentDescriptor>>() {
+ doGet("/api/components/" + ComponentType.FILTER).andExpect(status().isOk()), new TypeReference<List<ComponentDescriptor>>() {
});
Assert.assertNotNull(descriptors);
- Assert.assertEquals(AMOUNT_OF_DEFAULT_PLUGINS_DESCRIPTORS, descriptors.size());
+ Assert.assertEquals(AMOUNT_OF_DEFAULT_FILTER_NODES, descriptors.size());
for (ComponentType type : ComponentType.values()) {
doGet("/api/components/" + type).andExpect(status().isOk());
}
}
- @Test
- public void testGetActionsByType() throws Exception {
- List<ComponentDescriptor> descriptors = readResponse(
- doGet("/api/components/actions/" + TelemetryStoragePlugin.class.getName()).andExpect(status().isOk()), new TypeReference<List<ComponentDescriptor>>() {
- });
-
- Assert.assertNotNull(descriptors);
- Assert.assertEquals(1, descriptors.size());
- Assert.assertEquals(TelemetryPluginAction.class.getName(), descriptors.get(0).getClazz());
- }
}
diff --git a/application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java b/application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java
index b732e66..af47764 100644
--- a/application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java
@@ -80,6 +80,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractC
}
}
+ @Ignore
@Test
public void testServerMqttOneWayRpc() throws Exception {
Device device = new Device();
@@ -106,6 +107,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractC
Assert.assertTrue(StringUtils.isEmpty(result));
}
+ @Ignore
@Test
public void testServerMqttOneWayRpcDeviceOffline() throws Exception {
Device device = new Device();
diff --git a/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java
index f88eb24..f45e303 100644
--- a/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java
@@ -16,6 +16,7 @@
package org.thingsboard.server.rules.flow;
import com.datastax.driver.core.utils.UUIDs;
+import com.fasterxml.jackson.databind.JsonNode;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.junit.After;
@@ -28,6 +29,7 @@ import org.thingsboard.server.actors.service.ActorService;
import org.thingsboard.server.common.data.*;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.StringDataEntry;
+import org.thingsboard.server.common.data.page.TextPageLink;
import org.thingsboard.server.common.data.page.TimePageData;
import org.thingsboard.server.common.data.rule.RuleChain;
import org.thingsboard.server.common.data.rule.RuleChainMetaData;
@@ -40,6 +42,7 @@ import org.thingsboard.server.controller.AbstractRuleEngineControllerTest;
import org.thingsboard.server.dao.attributes.AttributesService;
import org.thingsboard.server.dao.rule.RuleChainService;
+import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
@@ -60,9 +63,6 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
@Autowired
protected AttributesService attributesService;
- @Autowired
- protected RuleChainService ruleChainService;
-
@Before
public void beforeTest() throws Exception {
loginSysAdmin();
@@ -71,6 +71,7 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
tenant.setTitle("My tenant");
savedTenant = doPost("/api/tenant", tenant, Tenant.class);
Assert.assertNotNull(savedTenant);
+ ruleChainService.deleteRuleChainsByTenantId(savedTenant.getId());
tenantAdmin = new User();
tenantAdmin.setAuthority(Authority.TENANT_ADMIN);
@@ -166,7 +167,7 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
Assert.assertEquals(ruleChain.getFirstRuleNodeId(), outEvent.getEntityId());
Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
- Assert.assertEquals("serverAttributeValue1", outEvent.getBody().get("metadata").get("ss.serverAttributeKey1").asText());
+ Assert.assertEquals("serverAttributeValue1", getMetadata(outEvent).get("ss_serverAttributeKey1").asText());
RuleChain finalRuleChain = ruleChain;
RuleNode lastRuleNode = metaData.getNodes().stream().filter(node -> !node.getId().equals(finalRuleChain.getFirstRuleNodeId())).findFirst().get();
@@ -183,8 +184,8 @@ public abstract class AbstractRuleEngineFlowIntegrationTest extends AbstractRule
Assert.assertEquals(lastRuleNode.getId(), outEvent.getEntityId());
Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
- Assert.assertEquals("serverAttributeValue1", outEvent.getBody().get("metadata").get("ss.serverAttributeKey1").asText());
- Assert.assertEquals("serverAttributeValue2", outEvent.getBody().get("metadata").get("ss.serverAttributeKey2").asText());
+ Assert.assertEquals("serverAttributeValue1", getMetadata(outEvent).get("ss_serverAttributeKey1").asText());
+ Assert.assertEquals("serverAttributeValue2", getMetadata(outEvent).get("ss_serverAttributeKey2").asText());
}
}
diff --git a/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java
index 22d79f0..29e8b73 100644
--- a/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java
@@ -16,6 +16,7 @@
package org.thingsboard.server.rules.lifecycle;
import com.datastax.driver.core.utils.UUIDs;
+import com.fasterxml.jackson.databind.JsonNode;
import lombok.extern.slf4j.Slf4j;
import org.junit.After;
import org.junit.Assert;
@@ -42,6 +43,7 @@ import org.thingsboard.server.common.msg.system.ServiceToRuleEngineMsg;
import org.thingsboard.server.controller.AbstractRuleEngineControllerTest;
import org.thingsboard.server.dao.attributes.AttributesService;
+import java.io.IOException;
import java.util.Collections;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
@@ -69,6 +71,7 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac
tenant.setTitle("My tenant");
savedTenant = doPost("/api/tenant", tenant, Tenant.class);
Assert.assertNotNull(savedTenant);
+ ruleChainService.deleteRuleChainsByTenantId(savedTenant.getId());
tenantAdmin = new User();
tenantAdmin.setAuthority(Authority.TENANT_ADMIN);
@@ -152,7 +155,7 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac
Assert.assertEquals(ruleChain.getFirstRuleNodeId(), outEvent.getEntityId());
Assert.assertEquals(device.getId().getId().toString(), outEvent.getBody().get("entityId").asText());
- Assert.assertEquals("serverAttributeValue", outEvent.getBody().get("metadata").get("ss.serverAttributeKey").asText());
+ Assert.assertEquals("serverAttributeValue", getMetadata(outEvent).get("ss_serverAttributeKey").asText());
}
}
diff --git a/application/src/test/java/org/thingsboard/server/rules/RuleEngineSqlTestSuite.java b/application/src/test/java/org/thingsboard/server/rules/RuleEngineSqlTestSuite.java
index 65b4293..c438826 100644
--- a/application/src/test/java/org/thingsboard/server/rules/RuleEngineSqlTestSuite.java
+++ b/application/src/test/java/org/thingsboard/server/rules/RuleEngineSqlTestSuite.java
@@ -24,7 +24,8 @@ import java.util.Arrays;
@RunWith(ClasspathSuite.class)
@ClasspathSuite.ClassnameFilters({
- "org.thingsboard.server.rules.flow.*Test"})
+ "org.thingsboard.server.rules.flow.*Test",
+ "org.thingsboard.server.rules.lifecycle.*Test"})
public class RuleEngineSqlTestSuite {
@ClassRule
diff --git a/application/src/test/java/org/thingsboard/server/system/SystemSqlTestSuite.java b/application/src/test/java/org/thingsboard/server/system/SystemSqlTestSuite.java
index cfa0c58..97c6749 100644
--- a/application/src/test/java/org/thingsboard/server/system/SystemSqlTestSuite.java
+++ b/application/src/test/java/org/thingsboard/server/system/SystemSqlTestSuite.java
@@ -35,5 +35,4 @@ public class SystemSqlTestSuite {
"sql/drop-all-tables.sql",
"sql-test.properties");
-
}
diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/core/RuleEngineError.java b/common/message/src/main/java/org/thingsboard/server/common/msg/core/RuleEngineError.java
index 978d585..f5e249c 100644
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/core/RuleEngineError.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/core/RuleEngineError.java
@@ -21,7 +21,7 @@ package org.thingsboard.server.common.msg.core;
public enum RuleEngineError {
- NO_RULES, NO_ACTIVE_RULES, NO_FILTERS_MATCHED, NO_REQUEST_FROM_ACTIONS, NO_TWO_WAY_ACTIONS, NO_RESPONSE_FROM_ACTIONS, QUEUE_PUT_TIMEOUT(true);
+ QUEUE_PUT_TIMEOUT(true), SERVER_ERROR(true);
private final boolean critical;
diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/core/RuleEngineErrorMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/core/RuleEngineErrorMsg.java
index 540f629..61e5cf5 100644
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/core/RuleEngineErrorMsg.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/core/RuleEngineErrorMsg.java
@@ -40,20 +40,10 @@ public class RuleEngineErrorMsg implements ToDeviceMsg {
public String getErrorMsg() {
switch (error) {
- case NO_RULES:
- return "No rules configured!";
- case NO_ACTIVE_RULES:
- return "No active rules!";
- case NO_FILTERS_MATCHED:
- return "No rules that match current message!";
- case NO_REQUEST_FROM_ACTIONS:
- return "Rule filters match, but no plugin message produced by rule action!";
- case NO_TWO_WAY_ACTIONS:
- return "Rule filters match, but no rule with two-way action configured!";
- case NO_RESPONSE_FROM_ACTIONS:
- return "Rule filters match, message processed by plugin, but no response produced by rule action!";
case QUEUE_PUT_TIMEOUT:
- return "Timeout during processing of message by plugin!";
+ return "Timeout during persistence of the message to the queue!";
+ case SERVER_ERROR:
+ return "Error during processing of message by the server!";
default:
throw new RuntimeException("Error " + error + " is not supported!");
}
diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java
index 390266a..b99c450 100644
--- a/common/transport/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java
+++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java
@@ -118,13 +118,13 @@ public class JsonConverter {
}
}
- public static UpdateAttributesRequest convertToAttributes(JsonElement element) {
+ public static AttributesUpdateRequest convertToAttributes(JsonElement element) {
return convertToAttributes(element, BasicRequest.DEFAULT_REQUEST_ID);
}
- public static UpdateAttributesRequest convertToAttributes(JsonElement element, int requestId) {
+ public static AttributesUpdateRequest convertToAttributes(JsonElement element, int requestId) {
if (element.isJsonObject()) {
- BasicUpdateAttributesRequest request = new BasicUpdateAttributesRequest(requestId);
+ BasicAttributesUpdateRequest request = new BasicAttributesUpdateRequest(requestId);
long ts = System.currentTimeMillis();
request.add(parseValues(element.getAsJsonObject()).stream().map(kv -> new BaseAttributeKvEntry(kv, ts)).collect(Collectors.toList()));
return request;
diff --git a/dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java b/dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java
index a7e7362..9ce1fbe 100644
--- a/dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java
+++ b/dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java
@@ -75,8 +75,6 @@ public class BaseRuleChainService extends AbstractEntityService implements RuleC
log.trace("Save system rule chain with predefined id {}", SYSTEM_TENANT);
ruleChain.setTenantId(SYSTEM_TENANT);
}
- //TODO: Temporary Hack to continue tests;
- ruleChain.setRoot(true);
RuleChain savedRuleChain = ruleChainDao.save(ruleChain);
if (ruleChain.isRoot() && ruleChain.getTenantId() != null && ruleChain.getId() == null) {
try {
diff --git a/extensions/extension-kafka/src/main/java/org/thingsboard/server/extensions/kafka/action/KafkaPluginAction.java b/extensions/extension-kafka/src/main/java/org/thingsboard/server/extensions/kafka/action/KafkaPluginAction.java
index f6fdf28..d294c86 100644
--- a/extensions/extension-kafka/src/main/java/org/thingsboard/server/extensions/kafka/action/KafkaPluginAction.java
+++ b/extensions/extension-kafka/src/main/java/org/thingsboard/server/extensions/kafka/action/KafkaPluginAction.java
@@ -32,7 +32,7 @@ public class KafkaPluginAction extends AbstractTemplatePluginAction<KafkaPluginA
@Override
protected Optional<RuleToPluginMsg> buildRuleToPluginMsg(RuleContext ctx, DeviceToDeviceActorMsg msg, FromDeviceRequestMsg payload) {
KafkaActionPayload.KafkaActionPayloadBuilder builder = KafkaActionPayload.builder();
- builder.msgType(payload.getMsgType());
+ builder.sessionMsgType(payload.getMsgType());
builder.requestId(payload.getRequestId());
builder.sync(configuration.isSync());
builder.topic(configuration.getTopic());
diff --git a/extensions/extension-kafka/src/main/java/org/thingsboard/server/extensions/kafka/plugin/KafkaMsgHandler.java b/extensions/extension-kafka/src/main/java/org/thingsboard/server/extensions/kafka/plugin/KafkaMsgHandler.java
index 3e87507..4000c22 100644
--- a/extensions/extension-kafka/src/main/java/org/thingsboard/server/extensions/kafka/plugin/KafkaMsgHandler.java
+++ b/extensions/extension-kafka/src/main/java/org/thingsboard/server/extensions/kafka/plugin/KafkaMsgHandler.java
@@ -49,10 +49,10 @@ public class KafkaMsgHandler implements RuleMsgHandler {
if (payload.isSync()) {
if (metadata != null) {
ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId,
- BasicStatusCodeResponse.onSuccess(payload.getMsgType(), payload.getRequestId())));
+ BasicStatusCodeResponse.onSuccess(payload.getSessionMsgType(), payload.getRequestId())));
} else {
ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId,
- BasicStatusCodeResponse.onError(payload.getMsgType(), payload.getRequestId(), e)));
+ BasicStatusCodeResponse.onError(payload.getSessionMsgType(), payload.getRequestId(), e)));
}
}
});
diff --git a/extensions/extension-mqtt/src/main/java/org/thingsboard/server/extensions/mqtt/action/MqttPluginAction.java b/extensions/extension-mqtt/src/main/java/org/thingsboard/server/extensions/mqtt/action/MqttPluginAction.java
index 2a0e78a..694fa9f 100644
--- a/extensions/extension-mqtt/src/main/java/org/thingsboard/server/extensions/mqtt/action/MqttPluginAction.java
+++ b/extensions/extension-mqtt/src/main/java/org/thingsboard/server/extensions/mqtt/action/MqttPluginAction.java
@@ -31,7 +31,7 @@ public class MqttPluginAction extends AbstractTemplatePluginAction<MqttPluginAct
protected Optional<RuleToPluginMsg> buildRuleToPluginMsg(RuleContext ctx, DeviceToDeviceActorMsg msg, FromDeviceRequestMsg payload) {
MqttActionPayload.MqttActionPayloadBuilder builder = MqttActionPayload.builder();
builder.sync(configuration.isSync());
- builder.msgType(payload.getMsgType());
+ builder.sessionMsgType(payload.getMsgType());
builder.requestId(payload.getRequestId());
builder.topic(configuration.getTopic());
builder.msgBody(getMsgBody(ctx, msg));
diff --git a/extensions/extension-mqtt/src/main/java/org/thingsboard/server/extensions/mqtt/plugin/MqttMsgHandler.java b/extensions/extension-mqtt/src/main/java/org/thingsboard/server/extensions/mqtt/plugin/MqttMsgHandler.java
index e6ac32a..4083729 100644
--- a/extensions/extension-mqtt/src/main/java/org/thingsboard/server/extensions/mqtt/plugin/MqttMsgHandler.java
+++ b/extensions/extension-mqtt/src/main/java/org/thingsboard/server/extensions/mqtt/plugin/MqttMsgHandler.java
@@ -51,7 +51,7 @@ public class MqttMsgHandler implements RuleMsgHandler {
log.debug("Message [{}] was successfully delivered to topic [{}]!", msg.toString(), payload.getTopic());
if (payload.isSync()) {
ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId,
- BasicStatusCodeResponse.onSuccess(payload.getMsgType(), payload.getRequestId())));
+ BasicStatusCodeResponse.onSuccess(payload.getSessionMsgType(), payload.getRequestId())));
}
}
@Override
@@ -59,7 +59,7 @@ public class MqttMsgHandler implements RuleMsgHandler {
log.warn("Failed to deliver message [{}] to topic [{}]!", msg.toString(), payload.getTopic());
if (payload.isSync()) {
ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId,
- BasicStatusCodeResponse.onError(payload.getMsgType(), payload.getRequestId(), new Exception(e))));
+ BasicStatusCodeResponse.onError(payload.getSessionMsgType(), payload.getRequestId(), new Exception(e))));
}
}
});
diff --git a/extensions/extension-rabbitmq/src/main/java/org/thingsboard/server/extensions/rabbitmq/action/RabbitMqPluginAction.java b/extensions/extension-rabbitmq/src/main/java/org/thingsboard/server/extensions/rabbitmq/action/RabbitMqPluginAction.java
index 97d4c63..0ae65b7 100644
--- a/extensions/extension-rabbitmq/src/main/java/org/thingsboard/server/extensions/rabbitmq/action/RabbitMqPluginAction.java
+++ b/extensions/extension-rabbitmq/src/main/java/org/thingsboard/server/extensions/rabbitmq/action/RabbitMqPluginAction.java
@@ -38,7 +38,7 @@ public class RabbitMqPluginAction extends AbstractTemplatePluginAction<RabbitMqP
builder.exchange(configuration.getExchange());
builder.queueName(configuration.getQueueName());
builder.messageProperties(configuration.getMessageProperties());
- builder.msgType(payload.getMsgType());
+ builder.sessionMsgType(payload.getMsgType());
builder.requestId(payload.getRequestId());
builder.payload(getMsgBody(ctx, msg));
return Optional.of(new RabbitMqActionMsg(msg.getTenantId(),
diff --git a/extensions/extension-rabbitmq/src/main/java/org/thingsboard/server/extensions/rabbitmq/plugin/RabbitMqMsgHandler.java b/extensions/extension-rabbitmq/src/main/java/org/thingsboard/server/extensions/rabbitmq/plugin/RabbitMqMsgHandler.java
index 764dd0d..b964ee1 100644
--- a/extensions/extension-rabbitmq/src/main/java/org/thingsboard/server/extensions/rabbitmq/plugin/RabbitMqMsgHandler.java
+++ b/extensions/extension-rabbitmq/src/main/java/org/thingsboard/server/extensions/rabbitmq/plugin/RabbitMqMsgHandler.java
@@ -57,7 +57,7 @@ public class RabbitMqMsgHandler implements RuleMsgHandler {
payload.getPayload().getBytes(UTF8));
if (payload.isSync()) {
ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId,
- BasicStatusCodeResponse.onSuccess(payload.getMsgType(), payload.getRequestId())));
+ BasicStatusCodeResponse.onSuccess(payload.getSessionMsgType(), payload.getRequestId())));
}
} catch (IOException e) {
throw new RuleException(e.getMessage(), e);
diff --git a/extensions/extension-rest-api-call/src/main/java/org/thingsboard/server/extensions/rest/action/RestApiCallPluginAction.java b/extensions/extension-rest-api-call/src/main/java/org/thingsboard/server/extensions/rest/action/RestApiCallPluginAction.java
index e68607a..6e60078 100644
--- a/extensions/extension-rest-api-call/src/main/java/org/thingsboard/server/extensions/rest/action/RestApiCallPluginAction.java
+++ b/extensions/extension-rest-api-call/src/main/java/org/thingsboard/server/extensions/rest/action/RestApiCallPluginAction.java
@@ -35,7 +35,7 @@ public class RestApiCallPluginAction extends AbstractTemplatePluginAction<RestAp
@Override
protected Optional<RuleToPluginMsg> buildRuleToPluginMsg(RuleContext ctx, DeviceToDeviceActorMsg msg, FromDeviceRequestMsg payload) {
RestApiCallActionPayload.RestApiCallActionPayloadBuilder builder = RestApiCallActionPayload.builder();
- builder.msgType(payload.getMsgType());
+ builder.sessionMsgType(payload.getMsgType());
builder.requestId(payload.getRequestId());
builder.sync(configuration.isSync());
builder.actionPath(configuration.getActionPath());
diff --git a/extensions/extension-rest-api-call/src/main/java/org/thingsboard/server/extensions/rest/plugin/RestApiCallMsgHandler.java b/extensions/extension-rest-api-call/src/main/java/org/thingsboard/server/extensions/rest/plugin/RestApiCallMsgHandler.java
index bc7cc71..7c04a43 100644
--- a/extensions/extension-rest-api-call/src/main/java/org/thingsboard/server/extensions/rest/plugin/RestApiCallMsgHandler.java
+++ b/extensions/extension-rest-api-call/src/main/java/org/thingsboard/server/extensions/rest/plugin/RestApiCallMsgHandler.java
@@ -52,7 +52,7 @@ public class RestApiCallMsgHandler implements RuleMsgHandler {
String.class);
if (exchangeResponse.getStatusCode().equals(payload.getExpectedResultCode()) && payload.isSync()) {
ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId,
- BasicStatusCodeResponse.onSuccess(payload.getMsgType(), payload.getRequestId())));
+ BasicStatusCodeResponse.onSuccess(payload.getSessionMsgType(), payload.getRequestId())));
} else if(!exchangeResponse.getStatusCode().equals(payload.getExpectedResultCode())) {
throw new RuntimeException("Response Status Code '"
+ exchangeResponse.getStatusCode()
diff --git a/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginAction.java b/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginAction.java
index 212b279..c65ed99 100644
--- a/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginAction.java
+++ b/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/action/SnsTopicPluginAction.java
@@ -33,7 +33,7 @@ public class SnsTopicPluginAction extends AbstractTemplatePluginAction<SnsTopicP
@Override
protected Optional<RuleToPluginMsg> buildRuleToPluginMsg(RuleContext ctx, DeviceToDeviceActorMsg msg, FromDeviceRequestMsg payload) {
SnsTopicActionPayload.SnsTopicActionPayloadBuilder builder = SnsTopicActionPayload.builder();
- builder.msgType(payload.getMsgType());
+ builder.sessionMsgType(payload.getMsgType());
builder.requestId(payload.getRequestId());
builder.topicArn(configuration.getTopicArn());
builder.msgBody(getMsgBody(ctx, msg));
diff --git a/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsMessageHandler.java b/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsMessageHandler.java
index d84208a..b90bb78 100644
--- a/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsMessageHandler.java
+++ b/extensions/extension-sns/src/main/java/org/thingsboard/server/extensions/sns/plugin/SnsMessageHandler.java
@@ -52,7 +52,7 @@ public class SnsMessageHandler implements RuleMsgHandler {
sns.publish(publishRequest);
if (payload.isSync()) {
ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId,
- BasicStatusCodeResponse.onSuccess(payload.getMsgType(), payload.getRequestId())));
+ BasicStatusCodeResponse.onSuccess(payload.getSessionMsgType(), payload.getRequestId())));
}
return;
}
diff --git a/extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginAction.java b/extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginAction.java
index a2db8ac..29c14e1 100644
--- a/extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginAction.java
+++ b/extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/fifo/SqsFifoQueuePluginAction.java
@@ -33,7 +33,7 @@ public class SqsFifoQueuePluginAction extends AbstractTemplatePluginAction<SqsFi
@Override
protected Optional<RuleToPluginMsg> buildRuleToPluginMsg(RuleContext ctx, DeviceToDeviceActorMsg msg, FromDeviceRequestMsg payload) {
SqsFifoQueueActionPayload.SqsFifoQueueActionPayloadBuilder builder = SqsFifoQueueActionPayload.builder();
- builder.msgType(payload.getMsgType());
+ builder.sessionMsgType(payload.getMsgType());
builder.requestId(payload.getRequestId());
builder.queue(configuration.getQueue());
builder.deviceId(msg.getDeviceId().toString());
diff --git a/extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginAction.java b/extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginAction.java
index ef74910..4ebf7bb 100644
--- a/extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginAction.java
+++ b/extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/action/standard/SqsStandardQueuePluginAction.java
@@ -33,7 +33,7 @@ public class SqsStandardQueuePluginAction extends AbstractTemplatePluginAction<S
@Override
protected Optional<RuleToPluginMsg> buildRuleToPluginMsg(RuleContext ctx, DeviceToDeviceActorMsg msg, FromDeviceRequestMsg payload) {
SqsStandardQueueActionPayload.SqsStandardQueueActionPayloadBuilder builder = SqsStandardQueueActionPayload.builder();
- builder.msgType(payload.getMsgType());
+ builder.sessionMsgType(payload.getMsgType());
builder.requestId(payload.getRequestId());
builder.queue(configuration.getQueue());
builder.delaySeconds(configuration.getDelaySeconds());
diff --git a/extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsMessageHandler.java b/extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsMessageHandler.java
index 4128013..248996a 100644
--- a/extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsMessageHandler.java
+++ b/extensions/extension-sqs/src/main/java/org/thingsboard/server/extensions/sqs/plugin/SqsMessageHandler.java
@@ -78,7 +78,7 @@ public class SqsMessageHandler implements RuleMsgHandler {
sqs.sendMessage(sendMsgRequest);
if (payload.isSync()) {
ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId,
- BasicStatusCodeResponse.onSuccess(payload.getMsgType(), payload.getRequestId())));
+ BasicStatusCodeResponse.onSuccess(payload.getSessionMsgType(), payload.getRequestId())));
}
}
}
diff --git a/extensions-api/src/main/java/org/thingsboard/server/extensions/api/plugins/msg/UpdateAttributesRequestRuleToPluginMsg.java b/extensions-api/src/main/java/org/thingsboard/server/extensions/api/plugins/msg/UpdateAttributesRequestRuleToPluginMsg.java
index 991ad8d..a6d40ff 100644
--- a/extensions-api/src/main/java/org/thingsboard/server/extensions/api/plugins/msg/UpdateAttributesRequestRuleToPluginMsg.java
+++ b/extensions-api/src/main/java/org/thingsboard/server/extensions/api/plugins/msg/UpdateAttributesRequestRuleToPluginMsg.java
@@ -18,13 +18,13 @@ package org.thingsboard.server.extensions.api.plugins.msg;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.common.msg.core.UpdateAttributesRequest;
+import org.thingsboard.server.common.msg.core.AttributesUpdateRequest;
-public class UpdateAttributesRequestRuleToPluginMsg extends AbstractRuleToPluginMsg<UpdateAttributesRequest> {
+public class UpdateAttributesRequestRuleToPluginMsg extends AbstractRuleToPluginMsg<AttributesUpdateRequest> {
private static final long serialVersionUID = 1L;
- public UpdateAttributesRequestRuleToPluginMsg(TenantId tenantId, CustomerId customerId, DeviceId deviceId, UpdateAttributesRequest payload) {
+ public UpdateAttributesRequestRuleToPluginMsg(TenantId tenantId, CustomerId customerId, DeviceId deviceId, AttributesUpdateRequest payload) {
super(tenantId, customerId, deviceId, payload);
}
diff --git a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/action/telemetry/TelemetryPluginAction.java b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/action/telemetry/TelemetryPluginAction.java
index 514eb0b..b32137e 100644
--- a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/action/telemetry/TelemetryPluginAction.java
+++ b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/action/telemetry/TelemetryPluginAction.java
@@ -18,11 +18,10 @@ package org.thingsboard.server.extensions.core.action.telemetry;
import org.springframework.util.StringUtils;
import org.thingsboard.server.common.msg.core.GetAttributesRequest;
import org.thingsboard.server.common.msg.core.TelemetryUploadRequest;
-import org.thingsboard.server.common.msg.core.UpdateAttributesRequest;
+import org.thingsboard.server.common.msg.core.AttributesUpdateRequest;
import org.thingsboard.server.common.msg.device.DeviceToDeviceActorMsg;
import org.thingsboard.server.common.msg.session.FromDeviceMsg;
import org.thingsboard.server.common.msg.session.SessionMsgType;
-import org.thingsboard.server.common.msg.session.SessionMsgType;
import org.thingsboard.server.common.msg.session.ToDeviceMsg;
import org.thingsboard.server.extensions.api.component.Action;
import org.thingsboard.server.extensions.api.plugins.PluginAction;
@@ -58,7 +57,7 @@ public class TelemetryPluginAction extends SimpleRuleLifecycleComponent implemen
return Optional.of(new TelemetryUploadRequestRuleToPluginMsg(deviceToDeviceActorMsg.getTenantId(), deviceToDeviceActorMsg.getCustomerId(),
deviceToDeviceActorMsg.getDeviceId(), payload, ttl));
} else if (msg.getMsgType() == SessionMsgType.POST_ATTRIBUTES_REQUEST) {
- UpdateAttributesRequest payload = (UpdateAttributesRequest) msg;
+ AttributesUpdateRequest payload = (AttributesUpdateRequest) msg;
return Optional.of(new UpdateAttributesRequestRuleToPluginMsg(deviceToDeviceActorMsg.getTenantId(), deviceToDeviceActorMsg.getCustomerId(),
deviceToDeviceActorMsg.getDeviceId(), payload));
} else if (msg.getMsgType() == SessionMsgType.GET_ATTRIBUTES_REQUEST) {
diff --git a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/filter/DeviceAttributesFilter.java b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/filter/DeviceAttributesFilter.java
index 468fde2..5865c1d 100644
--- a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/filter/DeviceAttributesFilter.java
+++ b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/filter/DeviceAttributesFilter.java
@@ -16,7 +16,7 @@
package org.thingsboard.server.extensions.core.filter;
import lombok.extern.slf4j.Slf4j;
-import org.thingsboard.server.common.msg.core.UpdateAttributesRequest;
+import org.thingsboard.server.common.msg.core.AttributesUpdateRequest;
import org.thingsboard.server.common.msg.device.DeviceToDeviceActorMsg;
import org.thingsboard.server.common.msg.session.FromDeviceMsg;
import org.thingsboard.server.extensions.api.component.Filter;
@@ -44,7 +44,7 @@ public class DeviceAttributesFilter extends BasicJsFilter {
if (msg != null) {
switch (msg.getMsgType()) {
case POST_ATTRIBUTES_REQUEST:
- bindings = NashornJsEvaluator.updateBindings(bindings, (UpdateAttributesRequest) msg);
+ bindings = NashornJsEvaluator.updateBindings(bindings, (AttributesUpdateRequest) msg);
break;
default:
break;
diff --git a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/filter/NashornJsEvaluator.java b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/filter/NashornJsEvaluator.java
index 20cb397..6e9fe3f 100644
--- a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/filter/NashornJsEvaluator.java
+++ b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/filter/NashornJsEvaluator.java
@@ -19,7 +19,7 @@ import jdk.nashorn.api.scripting.NashornScriptEngineFactory;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
-import org.thingsboard.server.common.msg.core.UpdateAttributesRequest;
+import org.thingsboard.server.common.msg.core.AttributesUpdateRequest;
import org.thingsboard.server.extensions.api.device.DeviceAttributes;
import javax.script.*;
@@ -69,7 +69,7 @@ public class NashornJsEvaluator {
return bindings;
}
- public static Bindings updateBindings(Bindings bindings, UpdateAttributesRequest msg) {
+ public static Bindings updateBindings(Bindings bindings, AttributesUpdateRequest msg) {
Map<String, Object> attrMap = (Map<String, Object>) bindings.get(CLIENT_SIDE);
for (AttributeKvEntry attr : msg.getAttributes()) {
if (!CLIENT_SIDE.equalsIgnoreCase(attr.getKey()) && !SERVER_SIDE.equalsIgnoreCase(attr.getKey())
diff --git a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/handlers/TelemetryRuleMsgHandler.java b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/handlers/TelemetryRuleMsgHandler.java
index 9cb67fd..014d149 100644
--- a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/handlers/TelemetryRuleMsgHandler.java
+++ b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/handlers/TelemetryRuleMsgHandler.java
@@ -28,7 +28,7 @@ import org.thingsboard.server.common.msg.core.BasicGetAttributesResponse;
import org.thingsboard.server.common.msg.core.BasicStatusCodeResponse;
import org.thingsboard.server.common.msg.core.GetAttributesRequest;
import org.thingsboard.server.common.msg.core.TelemetryUploadRequest;
-import org.thingsboard.server.common.msg.core.UpdateAttributesRequest;
+import org.thingsboard.server.common.msg.core.AttributesUpdateRequest;
import org.thingsboard.server.common.msg.kv.BasicAttributeKVMsg;
import org.thingsboard.server.extensions.api.plugins.PluginCallback;
import org.thingsboard.server.extensions.api.plugins.PluginContext;
@@ -132,7 +132,7 @@ public class TelemetryRuleMsgHandler extends DefaultRuleMsgHandler {
@Override
public void handleUpdateAttributesRequest(PluginContext ctx, TenantId tenantId, RuleId ruleId, UpdateAttributesRequestRuleToPluginMsg msg) {
- UpdateAttributesRequest request = msg.getPayload();
+ AttributesUpdateRequest request = msg.getPayload();
ctx.saveAttributes(msg.getTenantId(), msg.getDeviceId(), DataConstants.CLIENT_SCOPE, request.getAttributes().stream().collect(Collectors.toList()),
new PluginCallback<Void>() {
@Override
diff --git a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/processor/AlarmProcessor.java b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/processor/AlarmProcessor.java
index 711d86a..df6e45a 100644
--- a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/processor/AlarmProcessor.java
+++ b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/processor/AlarmProcessor.java
@@ -26,7 +26,7 @@ import org.thingsboard.server.common.data.alarm.AlarmSeverity;
import org.thingsboard.server.common.data.alarm.AlarmStatus;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.msg.core.TelemetryUploadRequest;
-import org.thingsboard.server.common.msg.core.UpdateAttributesRequest;
+import org.thingsboard.server.common.msg.core.AttributesUpdateRequest;
import org.thingsboard.server.common.msg.device.DeviceToDeviceActorMsg;
import org.thingsboard.server.common.msg.session.FromDeviceMsg;
import org.thingsboard.server.extensions.api.component.Processor;
@@ -219,7 +219,7 @@ public class AlarmProcessor implements RuleProcessor<AlarmProcessorConfiguration
if (msg != null) {
switch (msg.getMsgType()) {
case POST_ATTRIBUTES_REQUEST:
- bindings = NashornJsEvaluator.updateBindings(bindings, (UpdateAttributesRequest) msg);
+ bindings = NashornJsEvaluator.updateBindings(bindings, (AttributesUpdateRequest) msg);
break;
case POST_TELEMETRY_REQUEST:
TelemetryUploadRequest telemetryMsg = (TelemetryUploadRequest) msg;
diff --git a/extensions-core/src/main/resources/MsgTypeFilterDescriptor.json b/extensions-core/src/main/resources/MsgTypeFilterDescriptor.json
index a0339c9..6f40df9 100644
--- a/extensions-core/src/main/resources/MsgTypeFilterDescriptor.json
+++ b/extensions-core/src/main/resources/MsgTypeFilterDescriptor.json
@@ -9,19 +9,19 @@
"minItems" : 1,
"items": [
{
- "value": "GET_ATTRIBUTES",
+ "value": "GET_ATTRIBUTES_REQUEST",
"label": "Get attributes"
},
{
- "value": "POST_ATTRIBUTES",
+ "value": "POST_ATTRIBUTES_REQUEST",
"label": "Post attributes"
},
{
- "value": "POST_TELEMETRY",
+ "value": "POST_TELEMETRY_REQUEST",
"label": "Post telemetry"
},
{
- "value": "RPC_REQUEST",
+ "value": "RPC_REQUEST_REQUEST",
"label": "RPC Request"
}
],
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java
new file mode 100644
index 0000000..6db4a5e
--- /dev/null
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgAttributesNode.java
@@ -0,0 +1,76 @@
+/**
+ * Copyright © 2016-2018 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.rule.engine.telemetry;
+
+import com.google.gson.JsonParser;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.util.StringUtils;
+import org.thingsboard.rule.engine.TbNodeUtils;
+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.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
+import org.thingsboard.server.common.data.kv.KvEntry;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.data.plugin.ComponentType;
+import org.thingsboard.server.common.msg.TbMsg;
+import org.thingsboard.server.common.msg.core.AttributesUpdateRequest;
+import org.thingsboard.server.common.msg.core.TelemetryUploadRequest;
+import org.thingsboard.server.common.msg.session.SessionMsgType;
+import org.thingsboard.server.common.transport.adaptor.JsonConverter;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+@Slf4j
+@RuleNode(
+ type = ComponentType.ACTION,
+ name = "save attributes",
+ configClazz = TbMsgAttributesNodeConfiguration.class,
+ nodeDescription = "Saves attributes data",
+ nodeDetails = "Saves entity attributes based on configurable scope parameter. Expects messages with 'POST_ATTRIBUTES_REQUEST' message type"
+)
+public class TbMsgAttributesNode implements TbNode {
+
+ private TbMsgAttributesNodeConfiguration config;
+
+ @Override
+ public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
+ this.config = TbNodeUtils.convert(configuration, TbMsgAttributesNodeConfiguration.class);
+ }
+
+ @Override
+ public void onMsg(TbContext ctx, TbMsg msg) {
+ if (!msg.getType().equals(SessionMsgType.POST_ATTRIBUTES_REQUEST.name())) {
+ ctx.tellError(msg, new IllegalArgumentException("Unsupported msg type: " + msg.getType()));
+ return;
+ }
+
+ String src = msg.getData();
+ Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(src)).getAttributes();
+ ctx.getTelemetryService().saveAndNotify(msg.getOriginator(), config.getScope(), new ArrayList<>(attributes), new TelemetryNodeCallback(ctx, msg));
+ }
+
+ @Override
+ public void destroy() {
+ }
+
+}
diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java
index 69c3aca..c0aec73 100644
--- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java
+++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/action/TbAlarmNodeTest.java
@@ -17,9 +17,14 @@ package org.thingsboard.rule.engine.action;
import com.datastax.driver.core.utils.UUIDs;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.util.concurrent.AbstractListeningExecutorService;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
import org.apache.commons.lang3.NotImplementedException;
+import org.junit.After;
+import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
@@ -38,6 +43,8 @@ import org.thingsboard.server.dao.alarm.AlarmService;
import javax.script.ScriptException;
import java.io.IOException;
import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import static org.junit.Assert.*;
import static org.mockito.Matchers.any;
@@ -66,11 +73,32 @@ public class TbAlarmNodeTest {
@Mock
private ScriptEngine detailsJs;
+ private ListeningExecutor dbExecutor;
+
private EntityId originator = new DeviceId(UUIDs.timeBased());
private TenantId tenantId = new TenantId(UUIDs.timeBased());
private TbMsgMetaData metaData = new TbMsgMetaData();
private String rawJson = "{\"name\": \"Vit\", \"passed\": 5}";
+ @Before
+ public void before() {
+ dbExecutor = new ListeningExecutor() {
+ @Override
+ public <T> ListenableFuture<T> executeAsync(Callable<T> task) {
+ try {
+ return Futures.immediateFuture(task.call());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void execute(Runnable command) {
+ command.run();
+ }
+ };
+ }
+
@Test
public void newAlarmCanBeCreated() throws ScriptException, IOException {
initWithScript();
@@ -128,6 +156,7 @@ public class TbAlarmNodeTest {
verify(ctx).createJsScriptEngine("CLEAR", "isCleared");
verify(ctx).createJsScriptEngine("DETAILS", "Details");
verify(ctx).getJsExecutor();
+ verify(ctx).getDbCallbackExecutor();
verifyNoMoreInteractions(ctx, alarmService, clearJs, detailsJs);
}
@@ -151,6 +180,7 @@ public class TbAlarmNodeTest {
verify(ctx).createJsScriptEngine("DETAILS", "Details");
verify(ctx, times(2)).getJsExecutor();
verify(ctx).getAlarmService();
+ verify(ctx, times(3)).getDbCallbackExecutor();
verify(ctx).getTenantId();
verify(alarmService).findLatestByOriginatorAndType(tenantId, originator, "SomeType");
@@ -307,6 +337,7 @@ public class TbAlarmNodeTest {
when(ctx.getTenantId()).thenReturn(tenantId);
when(ctx.getJsExecutor()).thenReturn(executor);
when(ctx.getAlarmService()).thenReturn(alarmService);
+ when(ctx.getDbCallbackExecutor()).thenReturn(dbExecutor);
mockJsExecutor();
diff --git a/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java b/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java
index 6472dfb..1c96311 100644
--- a/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java
+++ b/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java
@@ -31,7 +31,6 @@ import org.thingsboard.server.common.msg.session.FromDeviceMsg;
import org.thingsboard.server.common.msg.session.SessionMsgType;
import org.thingsboard.server.common.msg.session.SessionActorToAdaptorMsg;
import org.thingsboard.server.common.msg.session.SessionContext;
-import org.thingsboard.server.common.msg.session.SessionMsgType;
import org.thingsboard.server.common.msg.session.ToDeviceMsg;
import org.thingsboard.server.common.msg.session.ex.ProcessingTimeoutException;
import org.thingsboard.server.common.transport.adaptor.AdaptorException;
@@ -157,7 +156,7 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor {
return response;
}
- private UpdateAttributesRequest convertToUpdateAttributesRequest(SessionContext ctx, Request inbound) throws AdaptorException {
+ private AttributesUpdateRequest convertToUpdateAttributesRequest(SessionContext ctx, Request inbound) throws AdaptorException {
String payload = validatePayload(ctx, inbound);
try {
return JsonConverter.convertToAttributes(new JsonParser().parse(payload));
diff --git a/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java b/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java
index e35cfd5..f0b29cb 100644
--- a/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java
+++ b/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/adaptors/JsonMqttAdaptor.java
@@ -219,7 +219,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
}
}
- private UpdateAttributesRequest convertToUpdateAttributesRequest(SessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
+ private AttributesUpdateRequest convertToUpdateAttributesRequest(SessionContext ctx, MqttPublishMessage inbound) throws AdaptorException {
String payload = validatePayload(ctx.getSessionId(), inbound.payload());
try {
return JsonConverter.convertToAttributes(new JsonParser().parse(payload), inbound.variableHeader().messageId());
diff --git a/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java b/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java
index b4dd8db..f35434a 100644
--- a/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java
+++ b/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionCtx.java
@@ -179,7 +179,7 @@ public class GatewaySessionCtx {
throw new JsonSyntaxException(CAN_T_PARSE_VALUE + json);
}
long ts = System.currentTimeMillis();
- BasicUpdateAttributesRequest request = new BasicUpdateAttributesRequest(requestId);
+ BasicAttributesUpdateRequest request = new BasicAttributesUpdateRequest(requestId);
JsonObject deviceData = deviceEntry.getValue().getAsJsonObject();
request.add(JsonConverter.parseValues(deviceData).stream().map(kv -> new BaseAttributeKvEntry(kv, ts)).collect(Collectors.toList()));
GatewayDeviceSessionCtx deviceSessionCtx = devices.get(deviceName);