> /etc/tb-edge/conf/tb-edge.conf
@@ -110,7 +97,7 @@ EOL'
{:copy-code}
```
-##### [Optional] Configure PostgreSQL
+##### [Optional] Configure PostgreSQL Connection
If you changed PostgreSQL default datasource settings, use the following command:
```bash
@@ -122,10 +109,10 @@ EOL'
{:copy-code}
```
-PUT_YOUR_POSTGRESQL_PASSWORD_HERE: Replace with your actual PostgreSQL user password.
+* **PUT_YOUR_POSTGRESQL_PASSWORD_HERE**: Replace with your actual **PostgreSQL user password**.
-##### [Optional] Update bind ports
-If ThingsBoard Edge is going to be running on the same machine where ThingsBoard server (cloud) is running, you'll need to update configuration parameters to avoid port collision between ThingsBoard server and ThingsBoard Edge.
+##### [Optional] Update Bind Ports
+If ThingsBoard Edge runs on the same machine as the ThingsBoard Server, you need to update the port configuration to avoid conflicts between the two services.
Please execute the following command to update ThingsBoard Edge configuration file (**/etc/tb-edge/conf/tb-edge.conf**):
@@ -140,29 +127,26 @@ EOL'
{:copy-code}
```
-Make sure that ports above (18080, 11883, 15683) are not used by any other application.
+Make sure that ports **18080**, **11883**, and **15683–15688** are not being used by any other applications.
-#### Step 6. Run installation Script
+#### Step 5. Run Installation Script
-Once ThingsBoard Edge is installed and configured please execute the following install script:
+Once ThingsBoard Edge is installed and configured, please execute the following installation script:
```bash
sudo /usr/share/tb-edge/bin/install/install.sh
{:copy-code}
```
-#### Step 7. Restart ThingsBoard Edge Service
+#### Step 6. Start ThingsBoard Edge Service
```bash
-sudo service tb-edge restart
+sudo service tb-edge start
{:copy-code}
```
-#### Step 8. Open ThingsBoard Edge UI
-
-Once started, you will be able to open **ThingsBoard Edge UI** using the following link http://localhost:8080.
-
-###### NOTE: Edge HTTP bind port update
+#### Step 7. Open ThingsBoard Edge UI
-Use next **ThingsBoard Edge UI** link **http://localhost:18080** if you updated HTTP 8080 bind port to **18080**.
+Once the Edge service has started, open the Edge web interface at http://localhost:8080, or http://localhost:18080 if you modified the HTTP bind port configuration in the previous step.
+Log in using your **tenant credentials** from either your local ThingsBoard Server or the **ThingsBoard Live Demo**.
\ No newline at end of file
diff --git a/application/src/main/data/json/edge/instructions/upgrade/centos/instructions.md b/application/src/main/data/json/edge/instructions/upgrade/centos/instructions.md
index c0e721ace2..82774e89cb 100644
--- a/application/src/main/data/json/edge/instructions/upgrade/centos/instructions.md
+++ b/application/src/main/data/json/edge/instructions/upgrade/centos/instructions.md
@@ -1,15 +1,15 @@
#### Upgrading to ${TB_EDGE_VERSION}EDGE
-**ThingsBoard Edge package download:**
+ThingsBoard Edge package download:
```bash
wget https://github.com/thingsboard/thingsboard-edge/releases/download/v${TB_EDGE_TAG}/tb-edge-${TB_EDGE_TAG}.rpm
{:copy-code}
```
-##### ThingsBoard Edge service upgrade
Install package:
```bash
sudo rpm -Uvh tb-edge-${TB_EDGE_TAG}.rpm
{:copy-code}
```
+
${UPGRADE_DB}
diff --git a/application/src/main/data/json/edge/instructions/upgrade/docker/instructions.md b/application/src/main/data/json/edge/instructions/upgrade/docker/instructions.md
index f1b9931e8b..68c7597f84 100644
--- a/application/src/main/data/json/edge/instructions/upgrade/docker/instructions.md
+++ b/application/src/main/data/json/edge/instructions/upgrade/docker/instructions.md
@@ -1,10 +1,3 @@
#### Upgrading to ${TB_EDGE_VERSION}
-Execute the following command to pull **${TB_EDGE_VERSION}** image:
-
-```bash
-docker pull thingsboard/tb-edge:${TB_EDGE_VERSION}
-{:copy-code}
-```
-
${UPGRADE_DB}
diff --git a/application/src/main/data/json/edge/instructions/upgrade/docker/start_service.md b/application/src/main/data/json/edge/instructions/upgrade/docker/start_service.md
index a6c9c6731c..11dc99a637 100644
--- a/application/src/main/data/json/edge/instructions/upgrade/docker/start_service.md
+++ b/application/src/main/data/json/edge/instructions/upgrade/docker/start_service.md
@@ -1,11 +1,10 @@
-Modify ‘main’ docker compose (`docker-compose.yml`) file for ThingsBoard Edge and update version of the image:
+Modify ‘main’ docker compose (`docker-compose.yml`) a file for ThingsBoard Edge and update a version of the image:
```bash
nano docker-compose.yml
{:copy-code}
```
```text
-version: '3.8'
services:
mytbedge:
restart: always
@@ -13,7 +12,7 @@ services:
...
```
-Make sure your image is the set to **tb-edge-${TB_EDGE_VERSION}**.
+Make sure your image is set to **tb-edge:${TB_EDGE_VERSION}**.
Execute the following commands to up this docker compose directly:
```bash
diff --git a/application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_db.md b/application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_db.md
index 386c5f2743..523316cdae 100644
--- a/application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_db.md
+++ b/application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_db.md
@@ -1,4 +1,4 @@
-Create docker compose file for ThingsBoard Edge upgrade process:
+Create a docker compose file for ThingsBoard Edge upgrade process:
```bash
> docker-compose-upgrade.yml && nano docker-compose-upgrade.yml
@@ -8,7 +8,6 @@ Create docker compose file for ThingsBoard Edge upgrade process:
Add the following lines to the yml file:
```bash
-version: '3.8'
services:
mytbedge:
restart: on-failure
@@ -40,7 +39,7 @@ volumes:
{:copy-code}
```
-Execute the following command to start upgrade process:
+Execute the following command to start an upgrade process:
```bash
docker compose -f docker-compose-upgrade.yml up --abort-on-container-exit
diff --git a/application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_preparing.md b/application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_preparing.md
index f12eaa5fc0..d2a6f796a1 100644
--- a/application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_preparing.md
+++ b/application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_preparing.md
@@ -1,6 +1,6 @@
-Here is the list of commands, that can be used to quickly upgrade ThingsBoard Edge on Docker (Linux or MacOS).
+Here is the list of commands that can be used to quickly upgrade ThingsBoard Edge on Docker (Linux or macOS).
-#### Prepare for upgrading ThingsBoard Edge
+#### Prepare for Upgrading ThingsBoard Edge
Set the terminal in the directory which contains the `docker-compose.yml` file and execute the following command
to stop and remove currently running TB Edge container:
@@ -10,70 +10,6 @@ docker compose rm mytbedge
{:copy-code}
```
-**OPTIONAL:** If you still rely on Docker Compose as docker-compose (with a hyphen) here is the list of the above commands:
-```text
-docker-compose stop
-docker-compose rm mytbedge
-```
-
-##### Migrating Data from Docker Bind Mount Folders to Docker Volumes
-Starting with the **3.6.2** release, the ThingsBoard team has transitioned from using Docker bind mount folders to Docker volumes.
-This change aims to enhance security and efficiency in storing data for Docker containers and to mitigate permission issues across various environments.
-
-To migrate from Docker bind mounts to Docker volumes, please execute the following commands:
-
-```bash
-docker run --rm -v tb-edge-data:/volume -v ~/.mytb-edge-data:/backup busybox sh -c "cp -a /backup/. /volume"
-docker run --rm -v tb-edge-logs:/volume -v ~/.mytb-edge-logs:/backup busybox sh -c "cp -a /backup/. /volume"
-docker run --rm -v tb-edge-postgres-data:/volume -v ~/.mytb-edge-data/db:/backup busybox sh -c "cp -a /backup/. /volume"
-{:copy-code}
-```
-
-After completing the data migration to the newly created Docker volumes, you'll need to update the volume mounts in your Docker Compose configuration.
-Modify the `docker-compose.yml` file for ThingsBoard Edge to update the volume settings.
-
-Update volume mounts. Locate the following snippet:
-```text
- volumes:
- - ~/.mytb-edge-data:/data
- - ~/.mytb-edge-logs:/var/log/tb-edge
-...
-```
-
-And replace it with:
-```text
- volumes:
- - tb-edge-data:/data
- - tb-edge-logs:/var/log/tb-edge
-...
-```
-
-Apply a similar update for the PostgreSQL service. Find the section:
-```text
- volumes:
- - ~/.mytb-edge-data/db:/var/lib/postgresql/data
-...
-```
-
-And replace it with:
-```text
- volumes:
- - tb-edge-postgres-data:/var/lib/postgresql/data
-...
-```
-
-Finally, please add next volumes section at the end of the file:
-```text
-...
-volumes:
- tb-edge-data:
- name: tb-edge-data
- tb-edge-logs:
- name: tb-edge-logs
- tb-edge-postgres-data:
- name: tb-edge-postgres-data
-```
-
##### Backup Database
Make a copy of the database volume before upgrading:
diff --git a/application/src/main/data/json/edge/instructions/upgrade/start_service.md b/application/src/main/data/json/edge/instructions/upgrade/start_service.md
index c26c183154..109ccf5b4b 100644
--- a/application/src/main/data/json/edge/instructions/upgrade/start_service.md
+++ b/application/src/main/data/json/edge/instructions/upgrade/start_service.md
@@ -1,4 +1,4 @@
-Start the service
+#### Start the Service
```bash
sudo systemctl tb-edge start
diff --git a/application/src/main/data/json/edge/instructions/upgrade/ubuntu/instructions.md b/application/src/main/data/json/edge/instructions/upgrade/ubuntu/instructions.md
index 59d243c63c..8d4a8d6e26 100644
--- a/application/src/main/data/json/edge/instructions/upgrade/ubuntu/instructions.md
+++ b/application/src/main/data/json/edge/instructions/upgrade/ubuntu/instructions.md
@@ -1,15 +1,14 @@
#### Upgrading to ${TB_EDGE_VERSION}EDGE
-**ThingsBoard Edge package download:**
+ThingsBoard Edge package download:
```bash
wget https://github.com/thingsboard/thingsboard-edge/releases/download/v${TB_EDGE_TAG}/tb-edge-${TB_EDGE_TAG}.deb
{:copy-code}
```
-##### ThingsBoard Edge service upgrade
-
Install package:
```bash
sudo dpkg -i tb-edge-${TB_EDGE_TAG}.deb
{:copy-code}
```
+
${UPGRADE_DB}
diff --git a/application/src/main/data/json/edge/instructions/upgrade/upgrade_preparing.md b/application/src/main/data/json/edge/instructions/upgrade/upgrade_preparing.md
index bdd6c4ea18..3d3ac8c705 100644
--- a/application/src/main/data/json/edge/instructions/upgrade/upgrade_preparing.md
+++ b/application/src/main/data/json/edge/instructions/upgrade/upgrade_preparing.md
@@ -1,6 +1,6 @@
-Here is the list of commands, that can be used to quickly upgrade ThingsBoard Edge on ${OS}
+Here is the list of commands that can be used to quickly upgrade ThingsBoard Edge on ${OS}
-#### Prepare for upgrading ThingsBoard Edge
+#### Prepare for Upgrading ThingsBoard Edge
Stop ThingsBoard Edge service:
@@ -33,4 +33,4 @@ sudo -Hiu postgres pg_dump tb_edge > tb_edge.sql.bak
{:copy-code}
```
-Check backup file created successfully.
+Check the backup file created successfully.
diff --git a/application/src/main/data/json/system/widget_types/attributes_card.json b/application/src/main/data/json/system/widget_types/attributes_card.json
index 6246811769..2beedd2a76 100644
--- a/application/src/main/data/json/system/widget_types/attributes_card.json
+++ b/application/src/main/data/json/system/widget_types/attributes_card.json
@@ -11,7 +11,7 @@
"resources": [],
"templateHtml": "",
"templateCss": "#container {\n overflow: auto;\n}\n\n.tbDatasource-container {\n margin: 5px;\n padding: 8px;\n}\n\n.tbDatasource-title {\n font-size: 1.200rem;\n font-weight: 500;\n padding-bottom: 10px;\n}\n\n.tbDatasource-table {\n width: 100%;\n box-shadow: 0 0 10px #ccc;\n border-collapse: collapse;\n white-space: nowrap;\n font-size: 1.000rem;\n color: #757575;\n}\n\n.tbDatasource-table td {\n position: relative;\n border-top: 1px solid rgba(0, 0, 0, 0.12);\n border-bottom: 1px solid rgba(0, 0, 0, 0.12);\n padding: 0px 18px;\n box-sizing: border-box;\n}",
- "controllerScript": "self.onInit = function() {\n \n self.ctx.datasourceTitleCells = [];\n self.ctx.valueCells = [];\n self.ctx.labelCells = [];\n \n for (var i=0; i < self.ctx.datasources.length; i++) {\n var tbDatasource = self.ctx.datasources[i];\n\n var datasourceId = 'tbDatasource' + i;\n self.ctx.$container.append(\n \"\"\n );\n\n var datasourceContainer = $('#' + datasourceId,\n self.ctx.$container);\n\n datasourceContainer.append(\n \"\" +\n tbDatasource.name + \"
\"\n );\n \n var datasourceTitleCell = $('.tbDatasource-title', datasourceContainer);\n self.ctx.datasourceTitleCells.push(datasourceTitleCell);\n \n var tableId = 'table' + i;\n datasourceContainer.append(\n \"\"\n );\n var table = $('#' + tableId, self.ctx.$container);\n\n for (var a = 0; a < tbDatasource.dataKeys.length; a++) {\n var dataKey = tbDatasource.dataKeys[a];\n var labelCellId = 'labelCell' + a;\n var cellId = 'cell' + a;\n table.append(\"| \" + dataKey.label +\n \" | |
\");\n var labelCell = $('#' + labelCellId, table);\n self.ctx.labelCells.push(labelCell);\n var valueCell = $('#' + cellId, table);\n self.ctx.valueCells.push(valueCell);\n }\n } \n \n self.onResize();\n}\n\nself.onDataUpdated = function() {\n for (var i = 0; i < self.ctx.valueCells.length; i++) {\n var cellData = self.ctx.data[i];\n if (cellData && cellData.data && cellData.data.length > 0) {\n var tvPair = cellData.data[cellData.data.length -\n 1];\n var value = tvPair[1];\n var textValue;\n //toDo -> + IsNumber\n \n if (isNumber(value)) {\n var decimals = self.ctx.decimals;\n var units = self.ctx.units;\n if (cellData.dataKey.decimals || cellData.dataKey.decimals === 0) {\n decimals = cellData.dataKey.decimals;\n }\n if (cellData.dataKey.units) {\n units = cellData.dataKey.units;\n }\n txtValue = self.ctx.utils.formatValue(value, decimals, units, true);\n } else {\n txtValue = value;\n }\n self.ctx.valueCells[i].html(txtValue);\n }\n }\n \n function isNumber(n) {\n return !isNaN(parseFloat(n)) && isFinite(n);\n }\n}\n\nself.onResize = function() {\n var datasourceTitleFontSize = self.ctx.height/8;\n if (self.ctx.width/self.ctx.height <= 1.5) {\n datasourceTitleFontSize = self.ctx.width/12;\n }\n datasourceTitleFontSize = Math.min(datasourceTitleFontSize, 20);\n for (var i = 0; i < self.ctx.datasourceTitleCells.length; i++) {\n self.ctx.datasourceTitleCells[i].css('font-size', datasourceTitleFontSize+'px');\n }\n var valueFontSize = self.ctx.height/9;\n var labelFontSize = self.ctx.height/9;\n if (self.ctx.width/self.ctx.height <= 1.5) {\n valueFontSize = self.ctx.width/15;\n labelFontSize = self.ctx.width/15;\n }\n valueFontSize = Math.min(valueFontSize, 18);\n labelFontSize = Math.min(labelFontSize, 18);\n\n for (i = 0; i < self.ctx.valueCells; i++) {\n self.ctx.valueCells[i].css('font-size', valueFontSize+'px');\n self.ctx.valueCells[i].css('height', valueFontSize*2.5+'px');\n self.ctx.valueCells[i].css('padding', '0px ' + valueFontSize + 'px');\n self.ctx.labelCells[i].css('font-size', labelFontSize+'px');\n self.ctx.labelCells[i].css('height', labelFontSize*2.5+'px');\n self.ctx.labelCells[i].css('padding', '0px ' + labelFontSize + 'px');\n } \n}\n\nself.onDestroy = function() {\n}\n",
+ "controllerScript": "self.onInit = function() {\n\n self.ctx.datasourceTitleCells = [];\n self.ctx.valueCells = [];\n self.ctx.labelCells = [];\n\n for (var i = 0; i < self.ctx.datasources\n .length; i++) {\n var tbDatasource = self.ctx.datasources[i];\n\n var datasourceId = 'tbDatasource' + i;\n self.ctx.$container.append(\n \"\"\n );\n\n var datasourceContainer = $('#' + datasourceId,\n self.ctx.$container);\n\n datasourceContainer.append(\n \"\" +\n tbDatasource.name + \"
\"\n );\n\n var datasourceTitleCell = $(\n '.tbDatasource-title',\n datasourceContainer);\n self.ctx.datasourceTitleCells.push(\n datasourceTitleCell);\n\n var tableId = 'table' + i;\n datasourceContainer.append(\n \"\"\n );\n var table = $('#' + tableId, self.ctx\n .$container);\n\n for (var a = 0; a < tbDatasource.dataKeys\n .length; a++) {\n var dataKey = tbDatasource.dataKeys[a];\n var labelCellId = 'labelCell' + a;\n var cellId = 'cell' + a;\n table.append(\"| \" + dataKey.label +\n \" | |
\");\n var labelCell = $('#' + labelCellId, table);\n self.ctx.labelCells.push(labelCell);\n var valueCell = $('#' + cellId, table);\n self.ctx.valueCells.push(valueCell);\n }\n }\n\n self.onResize();\n}\n\nself.onDataUpdated = function() {\n for (var i = 0; i < self.ctx.valueCells\n .length; i++) {\n var cellData = self.ctx.data[i];\n if (cellData && cellData.data && cellData.data\n .length > 0) {\n var tvPair = cellData.data[cellData.data\n .length -\n 1];\n var value = tvPair[1];\n var textValue;\n //toDo -> + IsNumber\n\n if (isNumber(value)) {\n var decimals = self.ctx.decimals;\n var units = self.ctx.units;\n if (cellData.dataKey.decimals ||\n cellData.dataKey.decimals === 0) {\n decimals = cellData.dataKey\n .decimals;\n }\n if (cellData.dataKey.units) {\n units = cellData.dataKey.units;\n }\n txtValue = self.ctx.utils.formatValue(\n value, decimals, units, true);\n } else {\n txtValue = self.ctx.utilsService\n .customTranslation(value);\n }\n self.ctx.valueCells[i].html(txtValue);\n }\n }\n\n function isNumber(n) {\n return !isNaN(parseFloat(n)) && isFinite(n);\n }\n}\n\nself.onResize = function() {\n var datasourceTitleFontSize = self.ctx.height / 8;\n if (self.ctx.width / self.ctx.height <= 1.5) {\n datasourceTitleFontSize = self.ctx.width / 12;\n }\n datasourceTitleFontSize = Math.min(\n datasourceTitleFontSize, 20);\n for (var i = 0; i < self.ctx.datasourceTitleCells\n .length; i++) {\n self.ctx.datasourceTitleCells[i].css(\n 'font-size', datasourceTitleFontSize +\n 'px');\n }\n var valueFontSize = self.ctx.height / 9;\n var labelFontSize = self.ctx.height / 9;\n if (self.ctx.width / self.ctx.height <= 1.5) {\n valueFontSize = self.ctx.width / 15;\n labelFontSize = self.ctx.width / 15;\n }\n valueFontSize = Math.min(valueFontSize, 18);\n labelFontSize = Math.min(labelFontSize, 18);\n\n for (i = 0; i < self.ctx.valueCells; i++) {\n self.ctx.valueCells[i].css('font-size',\n valueFontSize + 'px');\n self.ctx.valueCells[i].css('height',\n valueFontSize * 2.5 + 'px');\n self.ctx.valueCells[i].css('padding', '0px ' +\n valueFontSize + 'px');\n self.ctx.labelCells[i].css('font-size',\n labelFontSize + 'px');\n self.ctx.labelCells[i].css('height',\n labelFontSize * 2.5 + 'px');\n self.ctx.labelCells[i].css('padding', '0px ' +\n labelFontSize + 'px');\n }\n}\n\nself.onDestroy = function() {}",
"settingsSchema": "{}",
"dataKeySettingsSchema": "{}\n",
"defaultConfig": "{\"datasources\":[{\"type\":\"function\",\"name\":\"function\",\"dataKeys\":[{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"Random\",\"color\":\"#2196f3\",\"settings\":{},\"_hash\":0.15479322438769105,\"funcBody\":\"var value = prevValue + Math.random() * 100 - 50;\\nvar multiplier = Math.pow(10, 2 || 0);\\nvar value = Math.round(value * multiplier) / multiplier;\\nif (value < -1000) {\\n\\tvalue = -1000;\\n} else if (value > 1000) {\\n\\tvalue = 1000;\\n}\\nreturn value;\"}]}],\"timewindow\":{\"realtime\":{\"timewindowMs\":60000}},\"showTitle\":true,\"backgroundColor\":\"#fff\",\"color\":\"rgba(0, 0, 0, 0.87)\",\"padding\":\"8px\",\"settings\":{},\"title\":\"Attributes card\",\"decimals\":null}"
@@ -29,4 +29,4 @@
"public": true
}
]
-}
\ No newline at end of file
+}
diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
index 35539834c3..257e9d8565 100644
--- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
@@ -36,6 +36,7 @@ import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto;
@@ -48,6 +49,7 @@ import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry;
+import org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry;
import java.util.ArrayList;
import java.util.Collection;
@@ -62,6 +64,8 @@ import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
+import static org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry.getValueForTsRecord;
+
/**
* @author Andrew Shvayka
@@ -112,6 +116,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
} else {
states.remove(cfId);
}
+ msg.getCallback().onSuccess();
}
public void process(EntityInitCalculatedFieldMsg msg) throws CalculatedFieldException {
@@ -346,21 +351,48 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
return mapToArguments(argNames, data);
}
- private Map mapToArguments(Map argNames, List data) {
- if (argNames.isEmpty()) {
+ private Map mapToArguments(Map> args, List data) {
+ if (args.isEmpty()) {
return Collections.emptyMap();
}
Map arguments = new HashMap<>();
for (TsKvProto item : data) {
ReferencedEntityKey key = new ReferencedEntityKey(item.getKv().getKey(), ArgumentType.TS_LATEST, null);
- String argName = argNames.get(key);
- if (argName != null) {
- arguments.put(argName, new SingleValueArgumentEntry(item));
+ Set argNames = args.get(key);
+ if (argNames != null) {
+ SingleValueArgumentEntry incoming = new SingleValueArgumentEntry(item);
+ argNames.forEach(argName -> arguments.compute(argName, (name, existing) -> {
+ if (existing == null) {
+ return incoming;
+ }
+ existing.updateEntry(incoming);
+ return existing;
+ }));
}
+
key = new ReferencedEntityKey(item.getKv().getKey(), ArgumentType.TS_ROLLING, null);
- argName = argNames.get(key);
- if (argName != null) {
- arguments.put(argName, new SingleValueArgumentEntry(item));
+ argNames = args.get(key);
+ if (argNames != null) {
+ Double recordValue = getValueForTsRecord(ProtoUtils.fromProto(item.getKv()));
+ argNames.forEach(argName -> arguments.compute(argName, (name, existing) -> {
+ if (existing instanceof TsRollingArgumentEntry rolling) {
+ if (recordValue != null) {
+ rolling.getTsRecords().put(item.getTs(), recordValue);
+ }
+ return rolling;
+ }
+ TsRollingArgumentEntry rolling = new TsRollingArgumentEntry();
+ if (recordValue != null) {
+ rolling.getTsRecords().put(item.getTs(), recordValue);
+ }
+ if (existing instanceof SingleValueArgumentEntry single) {
+ Double existingValue = getValueForTsRecord(single.getKvEntryValue());
+ if (existingValue != null) {
+ rolling.getTsRecords().put(single.getTs(), existingValue);
+ }
+ }
+ return rolling;
+ }));
}
}
return arguments;
@@ -378,13 +410,13 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
return mapToArguments(argNames, scope, attrDataList);
}
- private Map mapToArguments(Map argNames, AttributeScopeProto scope, List attrDataList) {
+ private Map mapToArguments(Map> args, AttributeScopeProto scope, List attrDataList) {
Map arguments = new HashMap<>();
for (AttributeValueProto item : attrDataList) {
ReferencedEntityKey key = new ReferencedEntityKey(item.getKey(), ArgumentType.ATTRIBUTE, AttributeScope.valueOf(scope.name()));
- String argName = argNames.get(key);
- if (argName != null) {
- arguments.put(argName, new SingleValueArgumentEntry(item));
+ Set argNames = args.get(key);
+ if (argNames != null) {
+ argNames.forEach(argName -> arguments.put(argName, new SingleValueArgumentEntry(item)));
}
}
return arguments;
@@ -402,18 +434,19 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
return mapToArgumentsWithDefaultValue(ctx.getMainEntityArguments(), ctx.getArguments(), scope, removedAttrKeys);
}
- private Map mapToArgumentsWithDefaultValue(Map argNames, Map configArguments, AttributeScopeProto scope, List removedAttrKeys) {
+ private Map mapToArgumentsWithDefaultValue(Map> args, Map configArguments, AttributeScopeProto scope, List removedAttrKeys) {
Map arguments = new HashMap<>();
for (String removedKey : removedAttrKeys) {
ReferencedEntityKey key = new ReferencedEntityKey(removedKey, ArgumentType.ATTRIBUTE, AttributeScope.valueOf(scope.name()));
- String argName = argNames.get(key);
- if (argName != null) {
- Argument argument = configArguments.get(argName);
- String defaultValue = (argument != null) ? argument.getDefaultValue() : null;
- arguments.put(argName, StringUtils.isNotEmpty(defaultValue)
- ? new SingleValueArgumentEntry(System.currentTimeMillis(), new StringDataEntry(removedKey, defaultValue), null)
- : new SingleValueArgumentEntry());
-
+ Set argNames = args.get(key);
+ if (argNames != null) {
+ argNames.forEach(argName -> {
+ Argument argument = configArguments.get(argName);
+ String defaultValue = (argument != null) ? argument.getDefaultValue() : null;
+ arguments.put(argName, StringUtils.isNotEmpty(defaultValue)
+ ? new SingleValueArgumentEntry(System.currentTimeMillis(), new StringDataEntry(removedKey, defaultValue), null)
+ : new SingleValueArgumentEntry());
+ });
}
}
return arguments;
diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
index 590e097928..b8bc74db7a 100644
--- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
@@ -125,10 +125,13 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
if (msg.getState() != null) {
msg.getState().setRequiredArguments(calculatedField.getArgNames());
}
- log.debug("Pushing CF state restore msg to specific actor [{}]", msg.getId().entityId());
+ log.debug("[{}] Pushing CF state restore msg to specific actor [{}]", tenantId, msg.getId().entityId());
getOrCreateActor(msg.getId().entityId()).tell(msg);
- } else {
+ } else if (msg.getState() != null) {
+ log.debug("[{}] Received CF state restore msg for non-existing CF [{}]. Removing state", tenantId, cfId);
cfStateService.removeState(msg.getId(), msg.getCallback());
+ } else {
+ msg.getCallback().onSuccess();
}
}
diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldStateRestoreMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldStateRestoreMsg.java
index 19be7c02fa..fbe666c662 100644
--- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldStateRestoreMsg.java
+++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldStateRestoreMsg.java
@@ -19,6 +19,7 @@ import lombok.Data;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.MsgType;
import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
+import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState;
@@ -27,6 +28,7 @@ public class CalculatedFieldStateRestoreMsg implements ToCalculatedFieldSystemMs
private final CalculatedFieldEntityCtxId id;
private final CalculatedFieldState state;
+ private final TbCallback callback;
@Override
public MsgType getMsgType() {
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 c143214e2b..b24b803950 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
@@ -115,14 +115,9 @@ import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;
-
-/**
- * @author Andrew Shvayka
- */
@Slf4j
public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
- static final String SESSION_TIMEOUT_MESSAGE = "session timeout!";
final TenantId tenantId;
final DeviceId deviceId;
final LinkedHashMapRemoveEldest sessions;
@@ -178,7 +173,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
private EdgeId findRelatedEdgeId() {
List result =
systemContext.getRelationService().findByToAndType(tenantId, deviceId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE);
- if (result != null && result.size() > 0) {
+ if (result != null && !result.isEmpty()) {
EntityRelation relationToEdge = result.get(0);
if (relationToEdge.getFrom() != null && relationToEdge.getFrom().getId() != null) {
log.trace("[{}][{}] found edge [{}] for device", tenantId, deviceId, relationToEdge.getFrom().getId());
@@ -501,7 +496,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
UUID sessionId = getSessionId(sessionInfo);
DeviceId deviceId = new DeviceId(new UUID(msg.getDeviceIdMSB(), msg.getDeviceIdLSB()));
ListenableFuture registrationFuture = systemContext.getClaimDevicesService()
- .registerClaimingInfo(tenantId, deviceId, msg.getSecretKey(), msg.getDurationMs());
+ .registerClaimingInfo(tenantId, deviceId, msg.getSecretKey(), msg.getDurationMs());
Futures.addCallback(registrationFuture, new FutureCallback<>() {
@Override
public void onSuccess(Void result) {
@@ -723,7 +718,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
toDeviceRpcPendingMap.remove(requestId);
status = RpcStatus.FAILED;
response = JacksonUtil.newObjectNode().put("error", "There was a Timeout and all retry " +
- "attempts have been exhausted. Retry attempts set: " + maxRpcRetries);
+ "attempts have been exhausted. Retry attempts set: " + maxRpcRetries);
}
} else {
md.setRetries(md.getRetries() + 1);
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 91d25a633b..fa17c0ca2c 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
@@ -43,7 +43,6 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.data.rule.RuleChain;
import org.thingsboard.server.common.data.rule.RuleChainType;
-import org.thingsboard.server.common.msg.MsgType;
import org.thingsboard.server.common.msg.TbActorMsg;
import org.thingsboard.server.common.msg.TbActorStopReason;
import org.thingsboard.server.common.msg.TbMsg;
@@ -138,13 +137,22 @@ public class TenantActor extends RuleChainManagerActor {
@Override
protected boolean doProcess(TbActorMsg msg) {
if (cantFindTenant) {
- log.info("[{}] Processing missing Tenant msg: {}", tenantId, msg);
- if (msg.getMsgType().equals(MsgType.QUEUE_TO_RULE_ENGINE_MSG)) {
- QueueToRuleEngineMsg queueMsg = (QueueToRuleEngineMsg) msg;
- queueMsg.getMsg().getCallback().onSuccess();
- } else if (msg.getMsgType().equals(MsgType.TRANSPORT_TO_DEVICE_ACTOR_MSG)) {
- TransportToDeviceActorMsgWrapper transportMsg = (TransportToDeviceActorMsgWrapper) msg;
- transportMsg.getCallback().onSuccess();
+ log.debug("[{}] Processing message for non-existing tenant: {}", tenantId, msg);
+ switch (msg.getMsgType()) {
+ case QUEUE_TO_RULE_ENGINE_MSG -> {
+ ((QueueToRuleEngineMsg) msg).getMsg().getCallback().onSuccess();
+ }
+ case TRANSPORT_TO_DEVICE_ACTOR_MSG -> {
+ ((TransportToDeviceActorMsgWrapper) msg).getCallback().onSuccess();
+ }
+ case CF_STATE_RESTORE_MSG -> {
+ ((CalculatedFieldStateRestoreMsg) msg).getCallback().onSuccess();
+ }
+ default -> {
+ if (!log.isDebugEnabled()) {
+ log.info("[{}] Processing message for non-existing tenant: {}", tenantId, msg);
+ }
+ }
}
return true;
}
@@ -390,6 +398,7 @@ public class TenantActor extends RuleChainManagerActor {
public TbActor createActor() {
return new TenantActor(context, tenantId);
}
+
}
}
diff --git a/application/src/main/java/org/thingsboard/server/controller/UserController.java b/application/src/main/java/org/thingsboard/server/controller/UserController.java
index b9cb0c88b3..a2a7c993e9 100644
--- a/application/src/main/java/org/thingsboard/server/controller/UserController.java
+++ b/application/src/main/java/org/thingsboard/server/controller/UserController.java
@@ -266,6 +266,9 @@ public class UserController extends BaseController {
if (user.getAuthority() == Authority.SYS_ADMIN && getCurrentUser().getId().equals(userId)) {
throw new ThingsboardException("Sysadmin is not allowed to delete himself", ThingsboardErrorCode.PERMISSION_DENIED);
}
+ if (user.getAuthority() == Authority.TENANT_ADMIN && userService.countTenantAdmins(user.getTenantId()) == 1) {
+ throw new ThingsboardException("At least one tenant administrator must remain!", ThingsboardErrorCode.BAD_REQUEST_PARAMS);
+ }
tbUserService.delete(getTenantId(), getCurrentUser().getCustomerId(), user, getCurrentUser());
}
diff --git a/application/src/main/java/org/thingsboard/server/service/ai/AiChatModelServiceImpl.java b/application/src/main/java/org/thingsboard/server/service/ai/AiChatModelServiceImpl.java
index c1829cbf84..639e2025fe 100644
--- a/application/src/main/java/org/thingsboard/server/service/ai/AiChatModelServiceImpl.java
+++ b/application/src/main/java/org/thingsboard/server/service/ai/AiChatModelServiceImpl.java
@@ -15,6 +15,7 @@
*/
package org.thingsboard.server.service.ai;
+import com.fasterxml.jackson.core.io.JsonStringEncoder;
import com.google.common.util.concurrent.FluentFuture;
import dev.langchain4j.data.message.ChatMessage;
import dev.langchain4j.data.message.Content;
@@ -32,8 +33,6 @@ import org.thingsboard.server.common.data.ai.model.chat.Langchain4jChatModelConf
import java.util.List;
import java.util.stream.Collectors;
-import static org.thingsboard.server.common.data.StringUtils.escapeControlChars;
-
@Service
@RequiredArgsConstructor
class AiChatModelServiceImpl implements AiChatModelService {
@@ -74,7 +73,7 @@ class AiChatModelServiceImpl implements AiChatModelService {
private Content prepareContent(Content content) {
if (content instanceof TextContent txt) {
- return new TextContent(escapeControlChars(txt.text()));
+ return new TextContent(new String(JsonStringEncoder.getInstance().quoteAsString(txt.text())));
}
return content;
}
diff --git a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java
index 70b41f069e..87bdda2b30 100644
--- a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java
@@ -62,14 +62,14 @@ public abstract class AbstractCalculatedFieldStateService implements CalculatedF
protected abstract void doRemove(CalculatedFieldEntityCtxId stateId, TbCallback callback);
- protected void processRestoredState(CalculatedFieldStateProto stateMsg) {
+ protected void processRestoredState(CalculatedFieldStateProto stateMsg, TbCallback callback) {
var id = fromProto(stateMsg.getId());
var state = fromProto(stateMsg);
- processRestoredState(id, state);
+ processRestoredState(id, state, callback);
}
- protected void processRestoredState(CalculatedFieldEntityCtxId id, CalculatedFieldState state) {
- actorSystemContext.tell(new CalculatedFieldStateRestoreMsg(id, state));
+ protected void processRestoredState(CalculatedFieldEntityCtxId id, CalculatedFieldState state, TbCallback callback) {
+ actorSystemContext.tell(new CalculatedFieldStateRestoreMsg(id, state, callback));
}
@Override
diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java
index f2a6916751..b6e1193cfb 100644
--- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java
@@ -23,7 +23,6 @@ import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.math.NumberUtils;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg;
@@ -31,21 +30,14 @@ import org.thingsboard.server.actors.calculatedField.MultipleTbCallback;
import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.EntityType;
-import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.cf.configuration.Argument;
import org.thingsboard.server.common.data.cf.configuration.OutputType;
import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.Aggregation;
-import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
-import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
-import org.thingsboard.server.common.data.kv.BooleanDataEntry;
-import org.thingsboard.server.common.data.kv.DoubleDataEntry;
-import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
-import org.thingsboard.server.common.data.kv.StringDataEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.msg.TbMsgType;
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
@@ -70,9 +62,6 @@ import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState;
-import org.thingsboard.server.service.cf.ctx.state.ScriptCalculatedFieldState;
-import org.thingsboard.server.service.cf.ctx.state.SimpleCalculatedFieldState;
-import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry;
import java.util.ArrayList;
@@ -80,12 +69,15 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
import static org.thingsboard.server.common.data.DataConstants.SCOPE;
+import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultAttributeEntry;
+import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultTsKvEntry;
+import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createStateByType;
+import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformSingleValueArgument;
import static org.thingsboard.server.utils.CalculatedFieldUtils.toProto;
@TbRuleEngineComponent
@@ -244,30 +236,17 @@ public class DefaultCalculatedFieldProcessingService implements CalculatedFieldP
private ListenableFuture fetchKvEntry(TenantId tenantId, EntityId entityId, Argument argument) {
return switch (argument.getRefEntityKey().getType()) {
case TS_ROLLING -> fetchTsRolling(tenantId, entityId, argument);
- case ATTRIBUTE -> transformSingleValueArgument(
- Futures.transform(
- attributesService.find(tenantId, entityId, argument.getRefEntityKey().getScope(), argument.getRefEntityKey().getKey()),
- result -> result.or(() -> Optional.of(new BaseAttributeKvEntry(createDefaultKvEntry(argument), System.currentTimeMillis(), 0L))),
- calculatedFieldCallbackExecutor)
- );
- case TS_LATEST -> transformSingleValueArgument(
- Futures.transform(
- timeseriesService.findLatest(tenantId, entityId, argument.getRefEntityKey().getKey()),
- result -> result.or(() -> Optional.of(new BasicTsKvEntry(System.currentTimeMillis(), createDefaultKvEntry(argument), 0L))),
- calculatedFieldCallbackExecutor));
+ case ATTRIBUTE -> Futures.transform(
+ attributesService.find(tenantId, entityId, argument.getRefEntityKey().getScope(), argument.getRefEntityKey().getKey()),
+ result -> transformSingleValueArgument(result.orElseGet(() -> createDefaultAttributeEntry(argument, System.currentTimeMillis()))),
+ calculatedFieldCallbackExecutor);
+ case TS_LATEST -> Futures.transform(
+ timeseriesService.findLatest(tenantId, entityId, argument.getRefEntityKey().getKey()),
+ result -> transformSingleValueArgument(result.orElseGet(() -> createDefaultTsKvEntry(argument, System.currentTimeMillis()))),
+ calculatedFieldCallbackExecutor);
};
}
- private ListenableFuture transformSingleValueArgument(ListenableFuture> kvEntryFuture) {
- return Futures.transform(kvEntryFuture, kvEntry -> {
- if (kvEntry.isPresent() && kvEntry.get().getValue() != null) {
- return ArgumentEntry.createSingleValueArgument(kvEntry.get());
- } else {
- return new SingleValueArgumentEntry();
- }
- }, calculatedFieldCallbackExecutor);
- }
-
private ListenableFuture fetchTsRolling(TenantId tenantId, EntityId entityId, Argument argument) {
long currentTime = System.currentTimeMillis();
long timeWindow = argument.getTimeWindow() == 0 ? System.currentTimeMillis() : argument.getTimeWindow();
@@ -282,28 +261,6 @@ public class DefaultCalculatedFieldProcessingService implements CalculatedFieldP
return Futures.transform(tsRollingFuture, tsRolling -> tsRolling == null ? new TsRollingArgumentEntry(limit, timeWindow) : ArgumentEntry.createTsRollingArgument(tsRolling, limit, timeWindow), calculatedFieldCallbackExecutor);
}
- private KvEntry createDefaultKvEntry(Argument argument) {
- String key = argument.getRefEntityKey().getKey();
- String defaultValue = argument.getDefaultValue();
- if (StringUtils.isBlank(defaultValue)) {
- return new StringDataEntry(key, null);
- }
- if (NumberUtils.isParsable(defaultValue)) {
- return new DoubleDataEntry(key, Double.parseDouble(defaultValue));
- }
- if ("true".equalsIgnoreCase(defaultValue) || "false".equalsIgnoreCase(defaultValue)) {
- return new BooleanDataEntry(key, Boolean.parseBoolean(defaultValue));
- }
- return new StringDataEntry(key, defaultValue);
- }
-
- private CalculatedFieldState createStateByType(CalculatedFieldCtx ctx) {
- return switch (ctx.getCfType()) {
- case SIMPLE -> new SimpleCalculatedFieldState(ctx.getArgNames());
- case SCRIPT -> new ScriptCalculatedFieldState(ctx.getArgNames());
- };
- }
-
private static class TbCallbackWrapper implements TbQueueCallback {
private final TbCallback callback;
diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
index e21d56b6d2..14370aa68c 100644
--- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
@@ -31,19 +31,19 @@ import static org.thingsboard.server.utils.CalculatedFieldUtils.toSingleValueArg
@AllArgsConstructor
public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
+ static final long DEFAULT_LAST_UPDATE_TS = -1L;
+
protected List requiredArguments;
protected Map arguments;
protected boolean sizeExceedsLimit;
- protected long latestTimestamp = -1;
-
public BaseCalculatedFieldState(List requiredArguments) {
this.requiredArguments = requiredArguments;
this.arguments = new HashMap<>();
}
public BaseCalculatedFieldState() {
- this(new ArrayList<>(), new HashMap<>(), false, -1);
+ this(new ArrayList<>(), new HashMap<>(), false);
}
@Override
@@ -73,7 +73,6 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
if (entryUpdated) {
stateUpdated = true;
- updateLastUpdateTimestamp(newEntry);
}
}
@@ -109,15 +108,29 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
protected abstract void validateNewEntry(ArgumentEntry newEntry);
- private void updateLastUpdateTimestamp(ArgumentEntry entry) {
- long newTs = this.latestTimestamp;
- if (entry instanceof SingleValueArgumentEntry singleValueArgumentEntry) {
- newTs = singleValueArgumentEntry.getTs();
- } else if (entry instanceof TsRollingArgumentEntry tsRollingArgumentEntry) {
- Map.Entry lastEntry = tsRollingArgumentEntry.getTsRecords().lastEntry();
- newTs = (lastEntry != null) ? lastEntry.getKey() : System.currentTimeMillis();
+ public long getLatestTimestamp() {
+ long latestTs = DEFAULT_LAST_UPDATE_TS;
+
+ boolean allDefault = arguments.values().stream().allMatch(entry -> {
+ if (entry instanceof SingleValueArgumentEntry single) {
+ return single.isDefaultValue();
+ }
+ return false;
+ });
+
+ for (ArgumentEntry entry : arguments.values()) {
+ if (entry instanceof SingleValueArgumentEntry single) {
+ if (allDefault) {
+ latestTs = Math.max(latestTs, single.getTs());
+ } else if (!single.isDefaultValue()) {
+ latestTs = Math.max(latestTs, single.getTs());
+ }
+ } else if (entry instanceof TsRollingArgumentEntry rolling) {
+ latestTs = Math.max(latestTs, rolling.getLatestTs());
+ }
}
- this.latestTimestamp = Math.max(this.latestTimestamp, newTs);
+
+ return latestTs;
}
}
diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
index c9eaaef19a..9ef5a8c2a9 100644
--- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
@@ -35,6 +35,7 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
+import org.thingsboard.server.common.data.util.CollectionsUtil;
import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.dao.usagerecord.ApiLimitService;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto;
@@ -44,6 +45,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import static org.thingsboard.common.util.ExpressionFunctionsUtil.userDefinedFunctions;
@@ -57,8 +59,8 @@ public class CalculatedFieldCtx {
private EntityId entityId;
private CalculatedFieldType cfType;
private final Map arguments;
- private final Map mainEntityArguments;
- private final Map> linkedEntityArguments;
+ private final Map> mainEntityArguments;
+ private final Map>> linkedEntityArguments;
private final List argNames;
private Output output;
private String expression;
@@ -88,9 +90,10 @@ public class CalculatedFieldCtx {
var refId = entry.getValue().getRefEntityId();
var refKey = entry.getValue().getRefEntityKey();
if (refId == null || refId.equals(calculatedField.getEntityId())) {
- mainEntityArguments.put(refKey, entry.getKey());
+ mainEntityArguments.compute(refKey, (key, existingNames) -> CollectionsUtil.addToSet(existingNames, entry.getKey()));
} else {
- linkedEntityArguments.computeIfAbsent(refId, key -> new HashMap<>()).put(refKey, entry.getKey());
+ linkedEntityArguments.computeIfAbsent(refId, key -> new HashMap<>())
+ .compute(refKey, (key, existingNames) -> CollectionsUtil.addToSet(existingNames, entry.getKey()));
}
}
this.argNames = new ArrayList<>(arguments.keySet());
@@ -182,7 +185,7 @@ public class CalculatedFieldCtx {
return map != null && matchesTimeSeries(map, values);
}
- private boolean matchesAttributes(Map argMap, List values, AttributeScope scope) {
+ private boolean matchesAttributes(Map> argMap, List values, AttributeScope scope) {
if (argMap.isEmpty() || values.isEmpty()) {
return false;
}
@@ -196,7 +199,7 @@ public class CalculatedFieldCtx {
return false;
}
- private boolean matchesTimeSeries(Map argMap, List values) {
+ private boolean matchesTimeSeries(Map> argMap, List values) {
if (argMap.isEmpty() || values.isEmpty()) {
return false;
}
@@ -225,7 +228,7 @@ public class CalculatedFieldCtx {
return matchesTimeSeriesKeys(mainEntityArguments, keys);
}
- private boolean matchesAttributesKeys(Map argMap, List keys, AttributeScope scope) {
+ private boolean matchesAttributesKeys(Map> argMap, List keys, AttributeScope scope) {
if (argMap.isEmpty() || keys.isEmpty()) {
return false;
}
@@ -240,7 +243,7 @@ public class CalculatedFieldCtx {
return false;
}
- private boolean matchesTimeSeriesKeys(Map argMap, List keys) {
+ private boolean matchesTimeSeriesKeys(Map> argMap, List keys) {
if (argMap.isEmpty() || keys.isEmpty()) {
return false;
}
diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java
index 2b52892744..2773e13fa3 100644
--- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java
@@ -43,6 +43,8 @@ import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory;
import org.thingsboard.server.service.cf.AbstractCalculatedFieldStateService;
import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static org.thingsboard.server.queue.common.AbstractTbQueueTemplate.bytesToString;
@@ -61,6 +63,8 @@ public class KafkaCalculatedFieldStateService extends AbstractCalculatedFieldSta
@Value("${queue.calculated_fields.poll_interval:25}")
private long pollInterval;
+ @Value("${queue.calculated_fields.pack_processing_timeout:60000}")
+ private long packProcessingTimeout;
private TbKafkaProducerTemplate> stateProducer;
@@ -74,21 +78,39 @@ public class KafkaCalculatedFieldStateService extends AbstractCalculatedFieldSta
.topic(partitionService.getTopic(queueKey))
.pollInterval(pollInterval)
.msgPackProcessor((msgs, consumer, consumerKey, config) -> {
+ CountDownLatch completionLatch = new CountDownLatch(msgs.size());
for (TbProtoQueueMsg msg : msgs) {
+ TbCallback callback = new TbCallback() {
+ @Override
+ public void onSuccess() {
+ int processedMsgCount = counter.incrementAndGet();
+ if (processedMsgCount % 10000 == 0) {
+ log.info("Processed {} CF state messages", processedMsgCount);
+ }
+ completionLatch.countDown();
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ log.error("Failed to process CF state message: {}", msg, t);
+ completionLatch.countDown();
+ }
+ };
+
try {
if (msg.getValue() != null) {
- processRestoredState(msg.getValue());
+ processRestoredState(msg.getValue(), callback);
} else {
- processRestoredState(getStateId(msg.getHeaders()), null);
+ processRestoredState(getStateId(msg.getHeaders()), null, callback);
}
} catch (Throwable t) {
- log.error("Failed to process state message: {}", msg, t);
+ callback.onFailure(t);
}
+ }
- int processedMsgCount = counter.incrementAndGet();
- if (processedMsgCount % 10000 == 0) {
- log.info("Processed {} calculated field state msgs", processedMsgCount);
- }
+ boolean success = completionLatch.await(packProcessingTimeout, TimeUnit.MILLISECONDS);
+ if (!success) {
+ log.error("Timeout to process CF state messages pack of size {}", msgs.size());
}
})
.consumerCreator((queueConfig, tpi) -> queueFactory.createCalculatedFieldStateConsumer())
diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java
index 9dc6139ca5..d17060d093 100644
--- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java
@@ -15,7 +15,6 @@
*/
package org.thingsboard.server.service.cf.ctx.state;
-import com.google.protobuf.InvalidProtocolBufferException;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
@@ -63,11 +62,22 @@ public class RocksDBCalculatedFieldStateService extends AbstractCalculatedFieldS
public void restore(QueueKey queueKey, Set partitions) {
if (stateService.getPartitions().isEmpty()) {
cfRocksDb.forEach((key, value) -> {
+ CalculatedFieldStateProto stateMsg;
try {
- processRestoredState(CalculatedFieldStateProto.parseFrom(value));
- } catch (InvalidProtocolBufferException e) {
- log.error("[{}] Failed to process restored state", key, e);
+ stateMsg = CalculatedFieldStateProto.parseFrom(value);
+ } catch (Exception e) {
+ log.error("Failed to parse CalculatedFieldStateProto for key {}", key, e);
+ return;
}
+ processRestoredState(stateMsg, new TbCallback() {
+ @Override
+ public void onSuccess() {}
+
+ @Override
+ public void onFailure(Throwable t) {
+ log.error("Failed to process CF state message: {}", stateMsg, t);
+ }
+ });
});
}
super.restore(queueKey, partitions);
diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java
index 577ff80219..80f5964582 100644
--- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldState.java
@@ -100,7 +100,7 @@ public class SimpleCalculatedFieldState extends BaseCalculatedFieldState {
}
long latestTs = getLatestTimestamp();
- if (useLatestTs && latestTs != -1) {
+ if (useLatestTs && latestTs != DEFAULT_LAST_UPDATE_TS) {
ObjectNode resultNode = JacksonUtil.newObjectNode();
resultNode.put("ts", latestTs);
resultNode.set("values", valuesNode);
diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
index 1ceea2c621..2f9a7de940 100644
--- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
@@ -19,7 +19,6 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.core.type.TypeReference;
import lombok.AllArgsConstructor;
import lombok.Data;
-import lombok.NoArgsConstructor;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.script.api.tbel.TbelCfArg;
import org.thingsboard.script.api.tbel.TbelCfSingleValueArg;
@@ -32,17 +31,25 @@ import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto;
import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
+import static org.thingsboard.server.service.cf.ctx.state.BaseCalculatedFieldState.DEFAULT_LAST_UPDATE_TS;
+
@Data
-@NoArgsConstructor
@AllArgsConstructor
public class SingleValueArgumentEntry implements ArgumentEntry {
+ public static final Long DEFAULT_VERSION = -1L;
+
private long ts;
private BasicKvEntry kvEntryValue;
private Long version;
private boolean forceResetPrevious;
+ public SingleValueArgumentEntry() {
+ this.ts = DEFAULT_LAST_UPDATE_TS;
+ this.version = DEFAULT_VERSION;
+ }
+
public SingleValueArgumentEntry(TsKvProto entry) {
this.ts = entry.getTs();
if (entry.hasVersion()) {
@@ -112,7 +119,7 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
@Override
public boolean updateEntry(ArgumentEntry entry) {
if (entry instanceof SingleValueArgumentEntry singleValueEntry) {
- if (singleValueEntry.getTs() <= this.ts) {
+ if (singleValueEntry.getTs() < this.ts) {
return false;
}
@@ -128,4 +135,9 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
}
return false;
}
+
+ public boolean isDefaultValue() {
+ return DEFAULT_VERSION.equals(this.version);
+ }
+
}
diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java
index b5a680a072..9f9ad5db73 100644
--- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java
@@ -31,6 +31,8 @@ import java.util.List;
import java.util.Map;
import java.util.TreeMap;
+import static org.thingsboard.server.service.cf.ctx.state.BaseCalculatedFieldState.DEFAULT_LAST_UPDATE_TS;
+
@Data
@NoArgsConstructor
@AllArgsConstructor
@@ -83,6 +85,11 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
return tsRecords;
}
+ public long getLatestTs() {
+ var lastEntry = tsRecords.lastEntry();
+ return (lastEntry != null) ? lastEntry.getKey() : DEFAULT_LAST_UPDATE_TS;
+ }
+
@Override
public TbelCfArg toTbelCfArg() {
List values = new ArrayList<>(tsRecords.size());
@@ -115,20 +122,11 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
}
private void addTsRecord(Long ts, KvEntry value) {
- try {
- switch (value.getDataType()) {
- case LONG -> value.getLongValue().ifPresent(aLong -> tsRecords.put(ts, aLong.doubleValue()));
- case DOUBLE -> value.getDoubleValue().ifPresent(aDouble -> tsRecords.put(ts, aDouble));
- case BOOLEAN -> value.getBooleanValue().ifPresent(aBoolean -> tsRecords.put(ts, aBoolean ? 1.0 : 0.0));
- case STRING -> value.getStrValue().ifPresent(aString -> tsRecords.put(ts, Double.parseDouble(aString)));
- case JSON -> value.getJsonValue().ifPresent(aString -> tsRecords.put(ts, Double.parseDouble(aString)));
- }
- } catch (Exception e) {
- tsRecords.put(ts, Double.NaN);
- log.debug("Invalid value '{}' for time series rolling arguments. Only numeric values are supported.", value.getValue());
- } finally {
- cleanupExpiredRecords();
+ Double recordValue = getValueForTsRecord(value);
+ if (recordValue != null) {
+ tsRecords.put(ts, recordValue);
}
+ cleanupExpiredRecords();
}
private void addTsRecord(Long ts, double value) {
@@ -143,4 +141,19 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
tsRecords.entrySet().removeIf(tsRecord -> tsRecord.getKey() < System.currentTimeMillis() - timeWindow);
}
+ public static Double getValueForTsRecord(KvEntry value) {
+ try {
+ return switch (value.getDataType()) {
+ case LONG -> value.getLongValue().map(Long::doubleValue).orElse(null);
+ case DOUBLE -> value.getDoubleValue().orElse(null);
+ case BOOLEAN -> value.getBooleanValue().map(b -> b ? 1.0 : 0.0).orElse(null);
+ case STRING -> value.getStrValue().map(Double::parseDouble).orElse(null);
+ case JSON -> value.getJsonValue().map(Double::parseDouble).orElse(null);
+ };
+ } catch (Exception e) {
+ log.debug("Invalid value '{}' for time series rolling arguments. Only numeric values are supported.", value.getValue());
+ return Double.NaN;
+ }
+ }
+
}
diff --git a/application/src/main/java/org/thingsboard/server/service/device/DeviceBulkImportService.java b/application/src/main/java/org/thingsboard/server/service/device/DeviceBulkImportService.java
index d042fb2657..c8925f9ee9 100644
--- a/application/src/main/java/org/thingsboard/server/service/device/DeviceBulkImportService.java
+++ b/application/src/main/java/org/thingsboard/server/service/device/DeviceBulkImportService.java
@@ -258,8 +258,7 @@ public class DeviceBulkImportService extends AbstractBulkImportService {
Lwm2mDeviceProfileTransportConfiguration transportConfiguration = new Lwm2mDeviceProfileTransportConfiguration();
transportConfiguration.setBootstrap(Collections.emptyList());
- transportConfiguration.setClientLwM2mSettings(new OtherConfiguration(false,1, 1, 1, PowerMode.DRX, null, null, null, null, null, V1_0.toString()));
- transportConfiguration.setObserveAttr(new TelemetryMappingConfiguration(Collections.emptyMap(), Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), Collections.emptyMap(), SINGLE));
+ transportConfiguration.setClientLwM2mSettings(new OtherConfiguration());
DeviceProfileData deviceProfileData = new DeviceProfileData();
DefaultDeviceProfileConfiguration configuration = new DefaultDeviceProfileConfiguration();
diff --git a/application/src/main/java/org/thingsboard/server/service/device/DeviceProvisionServiceImpl.java b/application/src/main/java/org/thingsboard/server/service/device/DeviceProvisionServiceImpl.java
index 0778d61ee7..ffd16c1287 100644
--- a/application/src/main/java/org/thingsboard/server/service/device/DeviceProvisionServiceImpl.java
+++ b/application/src/main/java/org/thingsboard/server/service/device/DeviceProvisionServiceImpl.java
@@ -186,9 +186,14 @@ public class DeviceProvisionServiceImpl implements DeviceProvisionService {
try {
Optional provisionState = attributesService.find(device.getTenantId(), device.getId(),
AttributeScope.SERVER_SCOPE, DEVICE_PROVISION_STATE).get();
- if (provisionState != null && provisionState.isPresent() && !provisionState.get().getValueAsString().equals(PROVISIONED_STATE)) {
- notify(device, provisionRequest, TbMsgType.PROVISION_FAILURE, false);
- throw new ProvisionFailedException(ProvisionResponseStatus.FAILURE.name());
+ if (provisionState != null && provisionState.isPresent()) {
+ if (provisionState.get().getValueAsString().equals(PROVISIONED_STATE)) {
+ notify(device, provisionRequest, TbMsgType.PROVISION_FAILURE, false);
+ throw new ProvisionFailedException(ProvisionResponseStatus.FAILURE.name());
+ } else {
+ log.error("[{}][{}] Unknown provision state: {}!", device.getName(), DEVICE_PROVISION_STATE, provisionState.get().getValueAsString());
+ throw new ProvisionFailedException(ProvisionResponseStatus.FAILURE.name());
+ }
} else {
saveProvisionStateAttribute(device).get();
notify(device, provisionRequest, TbMsgType.PROVISION_SUCCESS, true);
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/instructions/BaseEdgeInstallUpgradeInstructionsService.java b/application/src/main/java/org/thingsboard/server/service/edge/instructions/BaseEdgeInstallUpgradeInstructionsService.java
index e8910492a3..abce90c253 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/instructions/BaseEdgeInstallUpgradeInstructionsService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/instructions/BaseEdgeInstallUpgradeInstructionsService.java
@@ -18,7 +18,6 @@ package org.thingsboard.server.service.edge.instructions;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Value;
import org.thingsboard.server.service.install.InstallScripts;
import java.io.IOException;
@@ -26,6 +25,8 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
+import static org.thingsboard.edge.rpc.EdgeGrpcClient.getNewestEdgeVersion;
+
@Slf4j
@RequiredArgsConstructor
public abstract class BaseEdgeInstallUpgradeInstructionsService {
@@ -35,9 +36,12 @@ public abstract class BaseEdgeInstallUpgradeInstructionsService {
private final InstallScripts installScripts;
- @Value("${app.version:unknown}")
@Setter
- protected String appVersion;
+ protected String platformEdgeVersion = convertEdgeVersionToDocsFormat(getNewestEdgeVersion().name());
+
+ protected String convertEdgeVersionToDocsFormat(String edgeVersion) {
+ return edgeVersion.replace("_", ".").substring(2);
+ }
protected String readFile(Path file) {
try {
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/instructions/DefaultEdgeInstallInstructionsService.java b/application/src/main/java/org/thingsboard/server/service/edge/instructions/DefaultEdgeInstallInstructionsService.java
index e16a8be5b5..ce37bba290 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/instructions/DefaultEdgeInstallInstructionsService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/instructions/DefaultEdgeInstallInstructionsService.java
@@ -65,9 +65,7 @@ public class DefaultEdgeInstallInstructionsService extends BaseEdgeInstallUpgrad
dockerInstallInstructions = dockerInstallInstructions.replace("${EXTRA_HOSTS}", "");
dockerInstallInstructions = dockerInstallInstructions.replace("${BASE_URL}", baseUrl);
}
- String edgeVersion = appVersion + "EDGE";
- edgeVersion = edgeVersion.replace("-SNAPSHOT", "");
- dockerInstallInstructions = dockerInstallInstructions.replace("${TB_EDGE_VERSION}", edgeVersion);
+ dockerInstallInstructions = dockerInstallInstructions.replace("${TB_EDGE_VERSION}", platformEdgeVersion + "EDGE");
dockerInstallInstructions = replacePlaceholders(dockerInstallInstructions, edge);
return new EdgeInstructions(dockerInstallInstructions);
}
@@ -76,9 +74,8 @@ public class DefaultEdgeInstallInstructionsService extends BaseEdgeInstallUpgrad
String ubuntuInstallInstructions = readFile(resolveFile(os, "instructions.md"));
ubuntuInstallInstructions = replacePlaceholders(ubuntuInstallInstructions, edge);
ubuntuInstallInstructions = ubuntuInstallInstructions.replace("${BASE_URL}", request.getServerName());
- String edgeVersion = appVersion.replace("-SNAPSHOT", "");
- ubuntuInstallInstructions = ubuntuInstallInstructions.replace("${TB_EDGE_VERSION}", edgeVersion);
- ubuntuInstallInstructions = ubuntuInstallInstructions.replace("${TB_EDGE_TAG}", getTagVersion(edgeVersion));
+ ubuntuInstallInstructions = ubuntuInstallInstructions.replace("${TB_EDGE_VERSION}", platformEdgeVersion);
+ ubuntuInstallInstructions = ubuntuInstallInstructions.replace("${TB_EDGE_TAG}", getTagVersion(platformEdgeVersion));
return new EdgeInstructions(ubuntuInstallInstructions);
}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/instructions/DefaultEdgeUpgradeInstructionsService.java b/application/src/main/java/org/thingsboard/server/service/edge/instructions/DefaultEdgeUpgradeInstructionsService.java
index 2df75d603b..8855e78683 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/instructions/DefaultEdgeUpgradeInstructionsService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/instructions/DefaultEdgeUpgradeInstructionsService.java
@@ -52,12 +52,11 @@ public class DefaultEdgeUpgradeInstructionsService extends BaseEdgeInstallUpgrad
@Override
public EdgeInstructions getUpgradeInstructions(String edgeVersion, String upgradeMethod) {
- String tbVersion = appVersion.replace("-SNAPSHOT", "");
String currentEdgeVersion = convertEdgeVersionToDocsFormat(edgeVersion);
return switch (upgradeMethod.toLowerCase()) {
- case "docker" -> getDockerUpgradeInstructions(tbVersion, currentEdgeVersion);
+ case "docker" -> getDockerUpgradeInstructions(this.platformEdgeVersion, currentEdgeVersion);
case "ubuntu", "centos" ->
- getLinuxUpgradeInstructions(tbVersion, currentEdgeVersion, upgradeMethod.toLowerCase());
+ getLinuxUpgradeInstructions(this.platformEdgeVersion, currentEdgeVersion, upgradeMethod.toLowerCase());
default -> throw new IllegalArgumentException("Unsupported upgrade method for Edge: " + upgradeMethod);
};
}
@@ -74,8 +73,7 @@ public class DefaultEdgeUpgradeInstructionsService extends BaseEdgeInstallUpgrad
Optional attributeKvEntryOpt = attributesService.find(tenantId, edgeId, AttributeScope.SERVER_SCOPE, DataConstants.EDGE_VERSION_ATTR_KEY).get();
if (attributeKvEntryOpt.isPresent()) {
String edgeVersionFormatted = convertEdgeVersionToDocsFormat(attributeKvEntryOpt.get().getValueAsString());
- String appVersionFormatted = appVersion.replace("-SNAPSHOT", "");
- return isVersionGreaterOrEqualsThan(edgeVersionFormatted, "3.6.0") && !isVersionGreaterOrEqualsThan(edgeVersionFormatted, appVersionFormatted);
+ return isVersionGreaterOrEqualsThan(edgeVersionFormatted, "3.6.0") && !isVersionGreaterOrEqualsThan(edgeVersionFormatted, platformEdgeVersion);
}
return false;
}
@@ -98,13 +96,13 @@ public class DefaultEdgeUpgradeInstructionsService extends BaseEdgeInstallUpgrad
return true;
}
- private EdgeInstructions getDockerUpgradeInstructions(String tbVersion, String currentEdgeVersion) {
+ private EdgeInstructions getDockerUpgradeInstructions(String platformEdgeVersion, String currentEdgeVersion) {
EdgeUpgradeInfo edgeUpgradeInfo = upgradeVersionHashMap.get(currentEdgeVersion);
- if (edgeUpgradeInfo == null || edgeUpgradeInfo.getNextEdgeVersion() == null || tbVersion.equals(currentEdgeVersion)) {
+ if (edgeUpgradeInfo == null || edgeUpgradeInfo.getNextEdgeVersion() == null || platformEdgeVersion.equals(currentEdgeVersion)) {
return new EdgeInstructions("Edge upgrade instruction for " + currentEdgeVersion + "EDGE is not available.");
}
StringBuilder result = new StringBuilder(readFile(resolveFile("docker", "upgrade_preparing.md")));
- while (edgeUpgradeInfo.getNextEdgeVersion() != null && !tbVersion.equals(currentEdgeVersion)) {
+ while (edgeUpgradeInfo.getNextEdgeVersion() != null && !platformEdgeVersion.equals(currentEdgeVersion)) {
String edgeVersion = edgeUpgradeInfo.getNextEdgeVersion();
String dockerUpgradeInstructions = readFile(resolveFile("docker", "instructions.md"));
if (edgeUpgradeInfo.isRequiresUpdateDb()) {
@@ -125,15 +123,15 @@ public class DefaultEdgeUpgradeInstructionsService extends BaseEdgeInstallUpgrad
return new EdgeInstructions(result.toString());
}
- private EdgeInstructions getLinuxUpgradeInstructions(String tbVersion, String currentEdgeVersion, String os) {
+ private EdgeInstructions getLinuxUpgradeInstructions(String platformEdgeVersion, String currentEdgeVersion, String os) {
EdgeUpgradeInfo edgeUpgradeInfo = upgradeVersionHashMap.get(currentEdgeVersion);
- if (edgeUpgradeInfo == null || edgeUpgradeInfo.getNextEdgeVersion() == null || tbVersion.equals(currentEdgeVersion)) {
+ if (edgeUpgradeInfo == null || edgeUpgradeInfo.getNextEdgeVersion() == null || platformEdgeVersion.equals(currentEdgeVersion)) {
return new EdgeInstructions("Edge upgrade instruction for " + currentEdgeVersion + "EDGE is not available.");
}
String upgrade_preparing = readFile(resolveFile("upgrade_preparing.md"));
upgrade_preparing = upgrade_preparing.replace("${OS}", os.equals("centos") ? "RHEL/CentOS 7/8" : "Ubuntu");
StringBuilder result = new StringBuilder(upgrade_preparing);
- while (edgeUpgradeInfo.getNextEdgeVersion() != null && !tbVersion.equals(currentEdgeVersion)) {
+ while (edgeUpgradeInfo.getNextEdgeVersion() != null && !platformEdgeVersion.equals(currentEdgeVersion)) {
String edgeVersion = edgeUpgradeInfo.getNextEdgeVersion();
String linuxUpgradeInstructions = readFile(resolveFile(os, "instructions.md"));
if (edgeUpgradeInfo.isRequiresUpdateDb()) {
@@ -155,10 +153,6 @@ public class DefaultEdgeUpgradeInstructionsService extends BaseEdgeInstallUpgrad
return new EdgeInstructions(result.toString());
}
- private String convertEdgeVersionToDocsFormat(String edgeVersion) {
- return edgeVersion.replace("_", ".").substring(2);
- }
-
@Override
protected String getBaseDirName() {
return UPGRADE_DIR;
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/instructions/EdgeInstallInstructionsService.java b/application/src/main/java/org/thingsboard/server/service/edge/instructions/EdgeInstallInstructionsService.java
index d1719422ce..c2f2d51a2a 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/instructions/EdgeInstallInstructionsService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/instructions/EdgeInstallInstructionsService.java
@@ -23,6 +23,6 @@ public interface EdgeInstallInstructionsService {
EdgeInstructions getInstallInstructions(Edge edge, String installationMethod, HttpServletRequest request);
- void setAppVersion(String version);
+ void setPlatformEdgeVersion(String version);
}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/instructions/EdgeUpgradeInstructionsService.java b/application/src/main/java/org/thingsboard/server/service/edge/instructions/EdgeUpgradeInstructionsService.java
index 9ae68ab2d0..83a8da281d 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/instructions/EdgeUpgradeInstructionsService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/instructions/EdgeUpgradeInstructionsService.java
@@ -28,7 +28,7 @@ public interface EdgeUpgradeInstructionsService {
void updateInstructionMap(Map upgradeVersions);
- void setAppVersion(String version);
+ void setPlatformEdgeVersion(String version);
boolean isUpgradeAvailable(TenantId tenantId, EdgeId edgeId) throws Exception;
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
index ec3f839cb3..c8b2c564ff 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
@@ -69,6 +69,7 @@ import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -82,6 +83,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
+import java.util.function.Function;
import static org.thingsboard.server.service.state.DefaultDeviceStateService.ACTIVITY_STATE;
import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAST_CONNECT_TIME;
@@ -94,6 +96,7 @@ import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAS
public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase implements EdgeRpcService {
private final ConcurrentMap sessions = new ConcurrentHashMap<>();
+ private final ConcurrentMap sessionsById = new ConcurrentHashMap<>();
private final ConcurrentMap sessionNewEventsLocks = new ConcurrentHashMap<>();
private final Map sessionNewEvents = new HashMap<>();
private final ConcurrentMap> sessionEdgeEventChecks = new ConcurrentHashMap<>();
@@ -283,6 +286,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
destroySession(session);
session.cleanUp();
sessions.remove(edgeId);
+ sessionsById.remove(session.getSessionId());
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
newEventLock.lock();
try {
@@ -332,9 +336,15 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
TenantId tenantId = edge.getTenantId();
log.info("[{}][{}] edge [{}] connected successfully.", tenantId, edgeGrpcSession.getSessionId(), edgeId);
if (sessions.containsKey(edgeId)) {
- destroySession(sessions.get(edgeId));
+ EdgeGrpcSession existing = sessions.get(edgeId);
+ if (existing != null) {
+ log.info("[{}][{}] Replacing existing session [{}] for edge [{}]", tenantId, edgeGrpcSession.getSessionId(), existing.getSessionId(), edgeId);
+ destroySession(existing);
+ sessionsById.remove(existing.getSessionId());
+ }
}
sessions.put(edgeId, edgeGrpcSession);
+ sessionsById.put(edgeGrpcSession.getSessionId(), edgeGrpcSession);
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
newEventLock.lock();
try {
@@ -492,9 +502,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
private void onEdgeDisconnect(Edge edge, UUID sessionId) {
EdgeId edgeId = edge.getId();
log.info("[{}][{}] edge disconnected!", edgeId, sessionId);
- EdgeGrpcSession toRemove = sessions.get(edgeId);
- if (toRemove.getSessionId().equals(sessionId)) {
- toRemove = sessions.remove(edgeId);
+ EdgeGrpcSession current = sessions.get(edgeId);
+ if (current != null && current.getSessionId().equals(sessionId)) {
+ EdgeGrpcSession toRemove = sessions.remove(edgeId);
final Lock newEventLock = sessionNewEventsLocks.computeIfAbsent(edgeId, id -> new ReentrantLock());
newEventLock.lock();
try {
@@ -503,6 +513,7 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
newEventLock.unlock();
}
destroySession(toRemove);
+ sessionsById.remove(sessionId);
TenantId tenantId = toRemove.getEdge().getTenantId();
save(tenantId, edgeId, ACTIVITY_STATE, false);
long lastDisconnectTs = System.currentTimeMillis();
@@ -510,7 +521,18 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
pushRuleEngineMessage(toRemove.getEdge().getTenantId(), edge, lastDisconnectTs, TbMsgType.DISCONNECT_EVENT);
cancelScheduleEdgeEventsCheck(edgeId);
} else {
- log.debug("[{}] edge session [{}] is not available anymore, nothing to remove. most probably this session is already outdated!", edgeId, sessionId);
+ log.info("[{}] edge session [{}] is not current anymore. Attempting to destroy it by sessionId.", edgeId, sessionId);
+ EdgeGrpcSession stale = sessionsById.remove(sessionId);
+ if (stale != null) {
+ try {
+ destroySession(stale);
+ log.info("[{}][{}] Successfully destroyed stale session for edge [{}]", stale.getTenantId(), sessionId, edgeId);
+ } catch (Exception e) {
+ log.warn("[{}][{}] Failed to destroy stale session for edge [{}]", stale.getTenantId(), sessionId, edgeId, e);
+ }
+ } else {
+ log.debug("[{}] No session found by sessionId [{}] to destroy", edgeId, sessionId);
+ }
}
edgeIdServiceIdCache.evict(edgeId);
}
@@ -522,6 +544,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
session.getTenantId(), session.getEdge().getId(), session.getEdge().getName(), session.getSessionId());
zombieSessions.add(session);
}
+ } catch (Exception e) {
+ log.warn("[{}][{}] Exception during session destroy for edge [{}] with session id [{}]",
+ session.getTenantId(), session.getEdge().getId(), session.getEdge().getName(), session.getSessionId(), e);
}
}
@@ -631,25 +656,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
private void cleanupZombieSessions() {
try {
- List toRemove = new ArrayList<>();
- for (EdgeGrpcSession session : sessions.values()) {
- if (session instanceof KafkaEdgeGrpcSession kafkaSession &&
- !kafkaSession.isConnected() &&
- kafkaSession.getConsumer() != null &&
- kafkaSession.getConsumer().getConsumer() != null &&
- !kafkaSession.getConsumer().getConsumer().isStopped()) {
- toRemove.add(kafkaSession.getEdge().getId());
- }
- }
- for (EdgeId edgeId : toRemove) {
- log.info("[{}] Destroying session for edge because edge is not connected", edgeId);
- EdgeGrpcSession removed = sessions.get(edgeId);
- if (removed instanceof KafkaEdgeGrpcSession kafkaSession) {
- if (kafkaSession.destroy()) {
- sessions.remove(edgeId);
- }
- }
- }
+ tryToDestroyZombieSessions(getZombieSessions(sessions.values()), s -> sessions.remove(s.getEdge().getId()));
+ tryToDestroyZombieSessions(getZombieSessions(sessionsById.values()), s -> sessionsById.remove(s.getSessionId()));
+
zombieSessions.removeIf(zombie -> {
if (zombie.destroy()) {
log.info("[{}][{}] Successfully cleaned up zombie session [{}] for edge [{}].",
@@ -666,4 +675,38 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
}
}
+ private List getZombieSessions(Collection sessions) {
+ List result = new ArrayList<>();
+ for (EdgeGrpcSession session : sessions) {
+ if (isKafkaSessionAndZombie(session)) {
+ result.add(session);
+ }
+ }
+ return result;
+ }
+
+ private void tryToDestroyZombieSessions(List sessionsToRemove, Function removeFunc) {
+ for (EdgeGrpcSession toRemove : sessionsToRemove) {
+ log.info("[{}] Destroying session for edge because edge is not connected", toRemove.getEdge().getId());
+ if (toRemove.destroy()) {
+ removeFunc.apply(toRemove);
+ }
+ }
+ }
+
+ private boolean isKafkaSessionAndZombie(EdgeGrpcSession session) {
+ if (session instanceof KafkaEdgeGrpcSession kafkaSession) {
+ log.debug("[{}] kafkaSession.isConnected() = {}, kafkaSession.getConsumer().getConsumer().isStopped() = {}",
+ kafkaSession.getEdge().getId(),
+ kafkaSession.isConnected(),
+ kafkaSession.getConsumer() != null ? kafkaSession.getConsumer().getConsumer() != null ? kafkaSession.getConsumer().getConsumer().isStopped() : null : null);
+ return !kafkaSession.isConnected() &&
+ kafkaSession.getConsumer() != null &&
+ kafkaSession.getConsumer().getConsumer() != null &&
+ !kafkaSession.getConsumer().getConsumer().isStopped();
+ }
+ return false;
+
+ }
+
}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
index 521730741f..c7c27ae4d0 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
@@ -113,7 +113,7 @@ public abstract class EdgeGrpcSession implements Closeable {
private static final int MAX_DOWNLINK_ATTEMPTS = 3;
private static final String RATE_LIMIT_REACHED = "Rate limit reached";
- protected static final ConcurrentLinkedQueue highPriorityQueue = new ConcurrentLinkedQueue<>();
+ protected final ConcurrentLinkedQueue highPriorityQueue = new ConcurrentLinkedQueue<>();
protected UUID sessionId;
private BiConsumer sessionOpenListener;
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java
index d165be33d4..e6eaec23e4 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java
@@ -82,16 +82,18 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
edgeEvents.add(edgeEvent);
}
List downlinkMsgsPack = convertToDownlinkMsgsPack(edgeEvents);
+ boolean isInterrupted = true;
try {
- boolean isInterrupted = sendDownlinkMsgsPack(downlinkMsgsPack).get();
+ isInterrupted = sendDownlinkMsgsPack(downlinkMsgsPack).get();
if (isInterrupted) {
log.debug("[{}][{}] Send downlink messages task was interrupted", tenantId, edge.getId());
- } else {
- consumer.commit();
}
} catch (Exception e) {
log.error("[{}][{}] Failed to process downlink messages", tenantId, edge.getId(), e);
}
+ if (!isInterrupted) {
+ consumer.commit();
+ }
}
@Override
@@ -101,8 +103,21 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
@Override
public ListenableFuture processEdgeEvents() {
+ if (!isConnected() || isSyncInProgress() || isHighPriorityProcessing) {
+ log.warn("[{}][{}] Session is not ready (connected={}, syncInProgress={}, highPriority={}), skip starting edge event consumer",
+ tenantId, edge != null ? edge.getId() : null, isConnected(), isSyncInProgress(), isHighPriorityProcessing);
+ return Futures.immediateFuture(Boolean.FALSE);
+ }
if (consumer == null || (consumer.getConsumer() != null && consumer.getConsumer().isStopped())) {
try {
+ if (consumerExecutor != null && !consumerExecutor.isShutdown()) {
+ try {
+ consumerExecutor.shutdown();
+ awaitConsumerTermination();
+ } catch (Exception e) {
+ log.warn("[{}][{}] Failed to shutdown previous consumer executor", tenantId, edge.getId(), e);
+ }
+ }
this.consumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("edge-event-consumer"));
this.consumer = QueueConsumerManager.>builder()
.name("TB Edge events [" + edge.getId() + "]")
@@ -133,6 +148,7 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
public boolean destroy() {
try {
if (consumer != null) {
+ log.info("[{}][{}] Stopping edge event consumer...", tenantId, edge != null ? edge.getId() : null);
consumer.stop();
}
} catch (Exception e) {
@@ -141,16 +157,25 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession {
}
consumer = null;
try {
- if (consumerExecutor != null) {
+ if (consumerExecutor != null && !consumerExecutor.isShutdown()) {
consumerExecutor.shutdown();
+ awaitConsumerTermination();
}
} catch (Exception e) {
- log.warn("[{}][{}] Failed to shutdown consumer executor", tenantId, edge.getId(), e);
+ log.warn("[{}][{}] Failed to shutdown edge event consumer executor", tenantId, edge.getId(), e);
return false;
}
return true;
}
+ private void awaitConsumerTermination() {
+ try {
+ consumerExecutor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS);
+ } catch (InterruptedException ie) {
+ log.warn("[{}][{}] Interrupted while awaiting consumer executor termination", tenantId, edge.getId());
+ }
+ }
+
@Override
public void cleanUp() {
String topic = topicService.buildEdgeEventNotificationsTopicPartitionInfo(tenantId, edge.getId()).getTopic();
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/BaseDashboardProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/BaseDashboardProcessor.java
index 56efb9157e..a1ac38138b 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/BaseDashboardProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/BaseDashboardProcessor.java
@@ -63,12 +63,12 @@ public abstract class BaseDashboardProcessor extends BaseEdgeProcessor {
Dashboard savedDashboard = edgeCtx.getDashboardService().saveDashboard(dashboard, false);
- updateDashboardAssignments(tenantId, dashboardById, savedDashboard, newAssignedCustomers);
+ updateDashboardAssignments(tenantId, customerId, dashboardById, savedDashboard, newAssignedCustomers);
return created;
}
- private void updateDashboardAssignments(TenantId tenantId, Dashboard dashboardById, Dashboard savedDashboard, Set newAssignedCustomers) {
+ private void updateDashboardAssignments(TenantId tenantId, CustomerId edgeCustomerId, Dashboard dashboardById, Dashboard savedDashboard, Set newAssignedCustomers) {
Set currentAssignedCustomers = new HashSet<>();
if (dashboardById != null) {
if (dashboardById.getAssignedCustomers() != null) {
@@ -76,7 +76,7 @@ public abstract class BaseDashboardProcessor extends BaseEdgeProcessor {
}
}
- newAssignedCustomers = filterNonExistingCustomers(tenantId, currentAssignedCustomers, newAssignedCustomers);
+ newAssignedCustomers = filterNonExistingCustomers(tenantId, edgeCustomerId, currentAssignedCustomers, newAssignedCustomers);
Set addedCustomerIds = new HashSet<>();
Set removedCustomerIds = new HashSet<>();
@@ -100,6 +100,6 @@ public abstract class BaseDashboardProcessor extends BaseEdgeProcessor {
}
}
- protected abstract Set filterNonExistingCustomers(TenantId tenantId, Set currentAssignedCustomers, Set newAssignedCustomers);
+ protected abstract Set filterNonExistingCustomers(TenantId tenantId, CustomerId customerId, Set currentAssignedCustomers, Set newAssignedCustomers);
}
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/DashboardEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/DashboardEdgeProcessor.java
index e1259a7e0e..d517b53303 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/DashboardEdgeProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/DashboardEdgeProcessor.java
@@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.ShortCustomerInfo;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.edge.EdgeEvent;
import org.thingsboard.server.common.data.edge.EdgeEventType;
+import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DashboardId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.msg.TbMsgType;
@@ -38,8 +39,10 @@ import org.thingsboard.server.gen.edge.v1.UpdateMsgType;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.service.edge.EdgeMsgConstructorUtils;
+import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
+import java.util.stream.Collectors;
@Slf4j
@Component
@@ -127,14 +130,24 @@ public class DashboardEdgeProcessor extends BaseDashboardProcessor implements Da
}
@Override
- protected Set filterNonExistingCustomers(TenantId tenantId, Set currentAssignedCustomers, Set newAssignedCustomers) {
- newAssignedCustomers.addAll(currentAssignedCustomers);
- return newAssignedCustomers;
+ protected Set filterNonExistingCustomers(TenantId tenantId, CustomerId edgeCustomerId, Set currentAssignedCustomers, Set newAssignedCustomers) {
+ boolean edgeCustomerPresentInNewAssignments = newAssignedCustomers.stream()
+ .map(ShortCustomerInfo::getCustomerId)
+ .anyMatch(edgeCustomerId::equals);
+
+ if (edgeCustomerPresentInNewAssignments) {
+ Set result = new HashSet<>(newAssignedCustomers);
+ result.addAll(currentAssignedCustomers);
+ return result;
+ } else {
+ return currentAssignedCustomers.stream()
+ .filter(info -> !edgeCustomerId.equals(info.getCustomerId()))
+ .collect(Collectors.toSet());
+ }
}
@Override
public EdgeEventType getEdgeEventType() {
return EdgeEventType.DASHBOARD;
}
-
}
diff --git a/application/src/main/java/org/thingsboard/server/service/install/DefaultDatabaseSchemaSettingsService.java b/application/src/main/java/org/thingsboard/server/service/install/DefaultDatabaseSchemaSettingsService.java
index f41a530630..44aa14b782 100644
--- a/application/src/main/java/org/thingsboard/server/service/install/DefaultDatabaseSchemaSettingsService.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/DefaultDatabaseSchemaSettingsService.java
@@ -17,7 +17,6 @@ package org.thingsboard.server.service.install;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
-import org.springframework.context.annotation.Profile;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.thingsboard.server.service.install.update.DefaultDataUpdateService;
@@ -25,14 +24,13 @@ import org.thingsboard.server.service.install.update.DefaultDataUpdateService;
import java.util.List;
@Service
-@Profile("install")
@Slf4j
@RequiredArgsConstructor
public class DefaultDatabaseSchemaSettingsService implements DatabaseSchemaSettingsService {
- // This list should include all versions which are compatible for the upgrade.
- // The compatibility cycle usually breaks when we have some scripts written in Java that may not work after new release.
- private static final List SUPPORTED_VERSIONS_FOR_UPGRADE = List.of("4.2.0");
+ // This list should include all versions that are compatible for the upgrade in 4 digits format (like 4.2.0.0, etc.).
+ // The compatibility cycle usually breaks when we have some scripts written in Java that may not work after a new release.
+ private static final List SUPPORTED_VERSIONS_FOR_UPGRADE = List.of("4.2.0.0", "4.2.1.0");
private final ProjectInfo projectInfo;
private final JdbcTemplate jdbcTemplate;
@@ -80,7 +78,7 @@ public class DefaultDatabaseSchemaSettingsService implements DatabaseSchemaSetti
@Override
public String getPackageSchemaVersion() {
if (packageSchemaVersion == null) {
- packageSchemaVersion = projectInfo.getProjectVersion();
+ packageSchemaVersion = normalizeVersion(projectInfo.getProjectVersion());
}
return packageSchemaVersion;
}
@@ -88,17 +86,28 @@ public class DefaultDatabaseSchemaSettingsService implements DatabaseSchemaSetti
@Override
public String getDbSchemaVersion() {
if (schemaVersionFromDb == null) {
- Long version = getSchemaVersionFromDb();
- if (version == null) {
+ Long dbVersion = getSchemaVersionFromDb();
+ if (dbVersion == null) {
onSchemaSettingsError("Upgrade failed: the database schema version is missing.");
}
@SuppressWarnings("DataFlowIssue")
- long major = version / 1000000;
- long minor = (version % 1000000) / 1000;
- long patch = version % 1000;
-
- schemaVersionFromDb = major + "." + minor + "." + patch;
+ long version = dbVersion;
+
+ if (version < 1_000_000_000) {
+ // Old format: MMM mmm ppp (e.g., 4002001 = 4.2.1)
+ long major = version / 1_000_000;
+ long minor = (version % 1_000_000) / 1000;
+ long maintenance = version % 1000;
+ schemaVersionFromDb = major + "." + minor + "." + maintenance + ".0";
+ } else {
+ // New format: MMM mmm mmm ppp (e.g., 4002001001 = 4.2.1.1)
+ long major = version / 1_000_000_000;
+ long minor = (version % 1_000_000_000) / 1_000_000;
+ long maintenance = (version % 1_000_000) / 1000;
+ long patch = version % 1000;
+ schemaVersionFromDb = major + "." + minor + "." + maintenance + "." + patch;
+ }
}
return schemaVersionFromDb;
}
@@ -116,13 +125,26 @@ public class DefaultDatabaseSchemaSettingsService implements DatabaseSchemaSetti
long major = Integer.parseInt(versionParts[0]);
long minor = Integer.parseInt(versionParts[1]);
- long patch = versionParts.length > 2 ? Integer.parseInt(versionParts[2]) : 0;
+ long maintenance = Integer.parseInt(versionParts[2]);
+ long patch = Integer.parseInt(versionParts[3]);
- return major * 1000000 + minor * 1000 + patch;
+ return major * 1_000_000_000L + minor * 1_000_000L + maintenance * 1000L + patch;
}
private void onSchemaSettingsError(String message) {
Runtime.getRuntime().addShutdownHook(new Thread(() -> log.error(message)));
throw new RuntimeException(message);
}
+
+ private String normalizeVersion(String version) {
+ String[] parts = version.split("\\.");
+
+ int major = Integer.parseInt(parts[0]);
+ int minor = parts.length > 1 ? Integer.parseInt(parts[1]) : 0;
+ int maintenance = parts.length > 2 ? Integer.parseInt(parts[2]) : 0;
+ int patch = parts.length > 3 ? Integer.parseInt(parts[3]) : 0;
+
+ return major + "." + minor + "." + maintenance + "." + patch;
+ }
+
}
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
index 4e44517ca0..f9b5c282a8 100644
--- a/application/src/main/java/org/thingsboard/server/service/install/InstallScripts.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/InstallScripts.java
@@ -65,9 +65,6 @@ import java.util.stream.Stream;
import static org.thingsboard.server.utils.LwM2mObjectModelUtils.toLwm2mResource;
-/**
- * Created by ashvayka on 18.04.18.
- */
@Component
@Slf4j
public class InstallScripts {
@@ -134,6 +131,10 @@ public class InstallScripts {
return Paths.get(getDataDir(), JSON_DIR, EDGE_DIR, RULE_CHAINS_DIR);
}
+ public Path getWidgetTypesDir() {
+ return Paths.get(getDataDir(), JSON_DIR, SYSTEM_DIR, WIDGET_TYPES_DIR);
+ }
+
public String getDataDir() {
if (!StringUtils.isEmpty(dataDir)) {
if (!Paths.get(this.dataDir).toFile().isDirectory()) {
@@ -237,7 +238,7 @@ public class InstallScripts {
}
);
}
- Path widgetTypesDir = Paths.get(getDataDir(), JSON_DIR, SYSTEM_DIR, WIDGET_TYPES_DIR);
+ Path widgetTypesDir = getWidgetTypesDir();
if (Files.exists(widgetTypesDir)) {
try (Stream dirStream = listDir(widgetTypesDir).filter(path -> path.toString().endsWith(JSON_EXT))) {
dirStream.forEach(
diff --git a/application/src/main/java/org/thingsboard/server/service/install/SqlEntityDatabaseSchemaService.java b/application/src/main/java/org/thingsboard/server/service/install/SqlEntityDatabaseSchemaService.java
index aa43d77439..a773d0ab23 100644
--- a/application/src/main/java/org/thingsboard/server/service/install/SqlEntityDatabaseSchemaService.java
+++ b/application/src/main/java/org/thingsboard/server/service/install/SqlEntityDatabaseSchemaService.java
@@ -22,12 +22,13 @@ import org.springframework.stereotype.Service;
@Service
@Profile("install")
@Slf4j
-public class SqlEntityDatabaseSchemaService extends SqlAbstractDatabaseSchemaService
- implements EntityDatabaseSchemaService {
+public class SqlEntityDatabaseSchemaService extends SqlAbstractDatabaseSchemaService implements EntityDatabaseSchemaService {
+
public static final String SCHEMA_ENTITIES_SQL = "schema-entities.sql";
public static final String SCHEMA_ENTITIES_IDX_SQL = "schema-entities-idx.sql";
public static final String SCHEMA_ENTITIES_IDX_PSQL_ADDON_SQL = "schema-entities-idx-psql-addon.sql";
- public static final String SCHEMA_VIEWS_AND_FUNCTIONS_SQL = "schema-views-and-functions.sql";
+ public static final String SCHEMA_VIEWS_SQL = "schema-views.sql";
+ public static final String SCHEMA_FUNCTIONS_SQL = "schema-functions.sql";
public SqlEntityDatabaseSchemaService() {
super(SCHEMA_ENTITIES_SQL, SCHEMA_ENTITIES_IDX_SQL);
@@ -49,8 +50,10 @@ public class SqlEntityDatabaseSchemaService extends SqlAbstractDatabaseSchemaSer
@Override
public void createOrUpdateViewsAndFunctions() throws Exception {
- log.info("Installing SQL DataBase schema views and functions: " + SCHEMA_VIEWS_AND_FUNCTIONS_SQL);
- executeQueryFromFile(SCHEMA_VIEWS_AND_FUNCTIONS_SQL);
+ log.info("Installing SQL DataBase schema views: " + SCHEMA_VIEWS_SQL);
+ executeQueryFromFile(SCHEMA_VIEWS_SQL);
+ log.info("Installing SQL DataBase schema functions: " + SCHEMA_FUNCTIONS_SQL);
+ executeQueryFromFile(SCHEMA_FUNCTIONS_SQL);
}
}
diff --git a/application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java b/application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java
index 4d9e145711..0b4d3bec6f 100644
--- a/application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java
+++ b/application/src/main/java/org/thingsboard/server/service/ota/DefaultOtaPackageStateService.java
@@ -328,7 +328,7 @@ public class DefaultOtaPackageStateService implements OtaPackageStateService {
attributes.add(new BaseAttributeKvEntry(ts, new LongDataEntry(getAttributeKey(otaPackageType, SIZE), otaPackage.getDataSize())));
}
- if (otaPackage.getChecksumAlgorithm() != null) {
+ if (otaPackage.getChecksumAlgorithm() == null) {
attrToRemove.add(getAttributeKey(otaPackageType, CHECKSUM_ALGORITHM));
} else {
attributes.add(new BaseAttributeKvEntry(ts, new StringDataEntry(getAttributeKey(otaPackageType, CHECKSUM_ALGORITHM), otaPackage.getChecksumAlgorithm().name())));
diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmStatusSubCtx.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmStatusSubCtx.java
index b3cd136726..6b3d38e8eb 100644
--- a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmStatusSubCtx.java
+++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmStatusSubCtx.java
@@ -95,6 +95,9 @@ public class TbAlarmStatusSubCtx extends TbAbstractSubCtx {
private void handleAlarmStatusSubscriptionUpdate(TbSubscription sub, AlarmSubscriptionUpdate subscriptionUpdate) {
try {
AlarmInfo alarm = subscriptionUpdate.getAlarm();
+ if (!alarm.getOriginator().equals(subscription.getEntityId())) {
+ return;
+ }
Set alarmsIds = subscription.getAlarmIds();
if (alarmsIds.contains(alarm.getId().getId())) {
if (!subscription.matches(alarm) || subscriptionUpdate.isAlarmDeleted()) {
diff --git a/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java b/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java
index 9850e2d1a1..63a4f100e1 100644
--- a/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java
+++ b/application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java
@@ -186,7 +186,7 @@ public abstract class AbstractBulkImportService kvs.add(dataEntry.getKey().getKey(), dataEntry.getValue().toJsonPrimitive()));
return Map.entry(kvType, kvs);
})
- .filter(kvsEntry -> kvsEntry.getValue().entrySet().size() > 0)
+ .filter(kvsEntry -> !kvsEntry.getValue().entrySet().isEmpty())
.forEach(kvsEntry -> {
BulkImportColumnType kvType = kvsEntry.getKey();
if (kvType == BulkImportColumnType.SHARED_ATTRIBUTE || kvType == BulkImportColumnType.SERVER_ATTRIBUTE) {
diff --git a/application/src/main/java/org/thingsboard/server/service/system/SystemInfoService.java b/application/src/main/java/org/thingsboard/server/service/system/SystemInfoService.java
index faae80a942..e2edd8a9cc 100644
--- a/application/src/main/java/org/thingsboard/server/service/system/SystemInfoService.java
+++ b/application/src/main/java/org/thingsboard/server/service/system/SystemInfoService.java
@@ -19,7 +19,9 @@ import org.thingsboard.server.common.data.FeaturesInfo;
import org.thingsboard.server.common.data.SystemInfo;
public interface SystemInfoService {
+
SystemInfo getSystemInfo();
FeaturesInfo getFeaturesInfo();
+
}
diff --git a/application/src/main/java/org/thingsboard/server/service/system/SystemPatchApplier.java b/application/src/main/java/org/thingsboard/server/service/system/SystemPatchApplier.java
new file mode 100644
index 0000000000..76af940685
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/service/system/SystemPatchApplier.java
@@ -0,0 +1,290 @@
+/**
+ * Copyright © 2016-2025 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.service.system;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.base.Charsets;
+import com.google.common.hash.Hashing;
+import com.google.common.io.Resources;
+import jakarta.annotation.PostConstruct;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.widget.WidgetTypeDetails;
+import org.thingsboard.server.dao.widget.WidgetTypeService;
+import org.thingsboard.server.queue.util.TbCoreComponent;
+import org.thingsboard.server.service.install.DatabaseSchemaSettingsService;
+import org.thingsboard.server.service.install.InstallScripts;
+import org.thingsboard.server.service.install.update.DefaultDataUpdateService;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+
+/**
+ * Runs at application startup and applies no-downtime data updates
+ * when the package PATCH version increases (e.g., 4.2.1.0 -> 4.2.1.1).
+ */
+@Slf4j
+@Component
+@TbCoreComponent
+@RequiredArgsConstructor
+public class SystemPatchApplier {
+
+ private static final String SCHEMA_VIEWS_SQL = "sql/schema-views.sql";
+
+ private static final long ADVISORY_LOCK_ID = 7536891047216478431L;
+
+ private final JdbcTemplate jdbcTemplate;
+ private final InstallScripts installScripts;
+ private final DatabaseSchemaSettingsService schemaSettingsService;
+ private final WidgetTypeService widgetTypeService;
+
+ @PostConstruct
+ private void init() {
+ ExecutorService executor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("system-patch-applier"));
+ executor.submit(() -> {
+ try {
+ applyPatchIfNeeded();
+ } catch (Exception e) {
+ log.error("Failed to apply system data patch updates", e);
+ } finally {
+ executor.shutdown();
+ }
+ });
+ }
+
+ private void applyPatchIfNeeded() {
+ boolean skipVersionCheck = DefaultDataUpdateService.getEnv("SKIP_PATCH_VERSION_CHECK", false);
+ if (!skipVersionCheck && !isVersionChanged()) {
+ return;
+ }
+
+ if (!acquireAdvisoryLock()) {
+ log.trace("Could not acquire advisory lock. Another node is processing patch updates.");
+ return;
+ }
+
+ try {
+ updateSqlViews();
+ log.info("Updated sql database views");
+
+ int updated = updateWidgetTypes();
+ log.info("Updated {} widget types", updated);
+
+ schemaSettingsService.updateSchemaVersion();
+ log.info("System data patch update completed successfully");
+
+ } finally {
+ releaseAdvisoryLock();
+ }
+ }
+
+ private boolean isVersionChanged() {
+ String packageVersion = schemaSettingsService.getPackageSchemaVersion();
+ String dbVersion = schemaSettingsService.getDbSchemaVersion();
+
+ log.trace("Package version: {}, DB schema version: {}", packageVersion, dbVersion);
+
+ VersionInfo packageVersionInfo = parseVersion(packageVersion);
+ VersionInfo dbVersionInfo = parseVersion(dbVersion);
+
+ if (packageVersionInfo == null || dbVersionInfo == null) {
+ log.warn("Unable to parse versions. Package: {}, DB: {}", packageVersion, dbVersion);
+ return false;
+ }
+
+ if (!isPatchVersionChanged(packageVersionInfo, dbVersionInfo)) {
+ return false;
+ }
+
+ log.info("Patch version increased from {} to {}. Starting system data update.", dbVersion, packageVersion);
+ return true;
+ }
+
+ private boolean isPatchVersionChanged(VersionInfo packageVersion, VersionInfo dbVersion) {
+ return packageVersion.major == dbVersion.major && packageVersion.minor == dbVersion.minor
+ && packageVersion.maintenance == dbVersion.maintenance && packageVersion.patch > dbVersion.patch;
+ }
+
+ private void updateSqlViews() {
+ try {
+ URL schemaViewsUrl = Resources.getResource(SCHEMA_VIEWS_SQL);
+ String sql = Resources.toString(schemaViewsUrl, Charsets.UTF_8);
+ jdbcTemplate.execute(sql);
+ } catch (IOException e) {
+ throw new RuntimeException("Unable to update database views from schema-views.sql", e);
+ }
+ }
+
+ private int updateWidgetTypes() {
+ AtomicInteger updated = new AtomicInteger();
+ Path widgetTypesDir = installScripts.getWidgetTypesDir();
+
+ if (!Files.exists(widgetTypesDir)) {
+ log.trace("Widget types directory does not exist: {}", widgetTypesDir);
+ return 0;
+ }
+
+ try (Stream dirStream = listDir(widgetTypesDir).filter(path -> path.toString().endsWith(InstallScripts.JSON_EXT))) {
+ dirStream.forEach(
+ path -> {
+ try {
+ if (updateWidgetTypeFromFile(path)) {
+ updated.incrementAndGet();
+ }
+ } catch (Exception e) {
+ log.error("Unable to update widget type from json: [{}]", path.toString());
+ throw new RuntimeException("Unable to update widget type from json", e);
+ }
+ }
+ );
+ }
+
+ return updated.get();
+ }
+
+ private boolean updateWidgetTypeFromFile(Path filePath) {
+ JsonNode json = JacksonUtil.toJsonNode(filePath.toFile());
+ WidgetTypeDetails fileWidgetType = JacksonUtil.treeToValue(json, WidgetTypeDetails.class);
+ String fqn = fileWidgetType.getFqn();
+
+ WidgetTypeDetails existingWidgetType = widgetTypeService.findWidgetTypeDetailsByTenantIdAndFqn(TenantId.SYS_TENANT_ID, fqn);
+ if (existingWidgetType == null) {
+ // We expect only update here, so it's probably never happening, but for test purpose leave it like this:
+ throw new RuntimeException("Widget type not found: " + fqn);
+ }
+ if (isWidgetTypeChanged(existingWidgetType, fileWidgetType)) {
+ existingWidgetType.setDescription(fileWidgetType.getDescription());
+ existingWidgetType.setName(fileWidgetType.getName());
+ existingWidgetType.setDescriptor(fileWidgetType.getDescriptor());
+ widgetTypeService.saveWidgetType(existingWidgetType);
+ log.trace("Updated widget type: {}", fqn);
+ return true;
+ }
+
+ log.trace("Widget type unchanged: {}", fqn);
+ return false;
+ }
+
+ private boolean isWidgetTypeChanged(WidgetTypeDetails existing, WidgetTypeDetails file) {
+ if (!isDescriptorEqual(existing.getDescriptor(), file.getDescriptor())) {
+ return true;
+ }
+
+ if (!Objects.equals(existing.getName(), file.getName())) {
+ return true;
+ }
+
+ return !Objects.equals(existing.getDescription(), file.getDescription());
+ }
+
+ private boolean isDescriptorEqual(JsonNode desc1, JsonNode desc2) {
+ if (desc1 == null && desc2 == null) {
+ return true;
+ }
+ if (desc1 == null || desc2 == null) {
+ return false;
+ }
+
+ try {
+ String hash1 = computeChecksum(desc1);
+ String hash2 = computeChecksum(desc2);
+ return Objects.equals(hash1, hash2);
+ } catch (Exception e) {
+ log.warn("Failed to compare descriptors using checksum, falling back to equals", e);
+ return desc1.equals(desc2);
+ }
+ }
+
+ private String computeChecksum(JsonNode node) {
+ String canonicalString = JacksonUtil.toCanonicalString(node);
+ if (canonicalString == null) {
+ return null;
+ }
+ return Hashing.sha256().hashBytes(canonicalString.getBytes()).toString();
+ }
+
+ private boolean acquireAdvisoryLock() {
+ try {
+ Boolean acquired = jdbcTemplate.queryForObject(
+ "SELECT pg_try_advisory_lock(?)",
+ Boolean.class,
+ ADVISORY_LOCK_ID
+ );
+ if (Boolean.TRUE.equals(acquired)) {
+ log.trace("Acquired advisory lock");
+ return true;
+ }
+ return false;
+ } catch (Exception e) {
+ log.error("Failed to acquire advisory lock", e);
+ return false;
+ }
+ }
+
+ private void releaseAdvisoryLock() {
+ try {
+ jdbcTemplate.queryForObject(
+ "SELECT pg_advisory_unlock(?)",
+ Boolean.class,
+ ADVISORY_LOCK_ID
+ );
+ log.debug("Released advisory lock");
+ } catch (Exception e) {
+ log.error("Failed to release advisory lock", e);
+ }
+ }
+
+ private VersionInfo parseVersion(String version) {
+ try {
+ String[] parts = version.split("\\.");
+ int major = Integer.parseInt(parts[0]);
+ int minor = parts.length > 1 ? Integer.parseInt(parts[1]) : 0;
+ int maintenance = parts.length > 2 ? Integer.parseInt(parts[2]) : 0;
+ int patch = parts.length > 3 ? Integer.parseInt(parts[3]) : 0;
+ return new VersionInfo(major, minor, maintenance, patch);
+ } catch (Exception e) {
+ log.error("Failed to parse version: {}", version, e);
+ return null;
+ }
+ }
+
+ private Stream listDir(Path dir) {
+ try {
+ return Files.list(dir);
+ } catch (NoSuchFileException e) {
+ return Stream.empty();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ public record VersionInfo(int major, int minor, int maintenance, int patch) {}
+
+}
diff --git a/application/src/main/java/org/thingsboard/server/service/update/DefaultUpdateService.java b/application/src/main/java/org/thingsboard/server/service/update/DefaultUpdateService.java
index 3c5ab63cd2..cb82e9b133 100644
--- a/application/src/main/java/org/thingsboard/server/service/update/DefaultUpdateService.java
+++ b/application/src/main/java/org/thingsboard/server/service/update/DefaultUpdateService.java
@@ -150,10 +150,10 @@ public class DefaultUpdateService implements UpdateService {
.build());
}
ObjectNode edgeRequest = JacksonUtil.newObjectNode().put(VERSION_PARAM, version);
- String edgeInstallVersion = restClient.postForObject(UPDATE_SERVER_BASE_URL + "/api/v1/edge/installMapping", new HttpEntity<>(edgeRequest.toString(), headers), String.class);
- if (edgeInstallVersion != null) {
- edgeInstallInstructionsService.setAppVersion(edgeInstallVersion);
- edgeUpgradeInstructionsService.setAppVersion(edgeInstallVersion);
+ String edgePlatformVersion = restClient.postForObject(UPDATE_SERVER_BASE_URL + "/api/v1/edge/installMapping", new HttpEntity<>(edgeRequest.toString(), headers), String.class);
+ if (edgePlatformVersion != null) {
+ edgeInstallInstructionsService.setPlatformEdgeVersion(edgePlatformVersion);
+ edgeUpgradeInstructionsService.setPlatformEdgeVersion(edgePlatformVersion);
}
EdgeUpgradeMessage edgeUpgradeMessage = restClient.postForObject(UPDATE_SERVER_BASE_URL + "/api/v1/edge/upgradeMapping", new HttpEntity<>(edgeRequest.toString(), headers), EdgeUpgradeMessage.class);
if (edgeUpgradeMessage != null) {
diff --git a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java
new file mode 100644
index 0000000000..1296df6e87
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java
@@ -0,0 +1,75 @@
+/**
+ * Copyright © 2016-2025 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.utils;
+
+import lombok.NonNull;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.thingsboard.server.common.data.StringUtils;
+import org.thingsboard.server.common.data.cf.configuration.Argument;
+import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
+import org.thingsboard.server.common.data.kv.BooleanDataEntry;
+import org.thingsboard.server.common.data.kv.DoubleDataEntry;
+import org.thingsboard.server.common.data.kv.KvEntry;
+import org.thingsboard.server.common.data.kv.StringDataEntry;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
+import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
+import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState;
+import org.thingsboard.server.service.cf.ctx.state.ScriptCalculatedFieldState;
+import org.thingsboard.server.service.cf.ctx.state.SimpleCalculatedFieldState;
+import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry;
+
+import static org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry.DEFAULT_VERSION;
+
+public class CalculatedFieldArgumentUtils {
+
+ public static ArgumentEntry transformSingleValueArgument(@NonNull KvEntry kvEntry) {
+ return kvEntry.getValue() != null ? ArgumentEntry.createSingleValueArgument(kvEntry) : new SingleValueArgumentEntry();
+ }
+
+ public static TsKvEntry createDefaultTsKvEntry(Argument argument, long ts) {
+ return new BasicTsKvEntry(ts, createDefaultKvEntry(argument), DEFAULT_VERSION);
+ }
+
+ public static AttributeKvEntry createDefaultAttributeEntry(Argument argument, long ts) {
+ return new BaseAttributeKvEntry(createDefaultKvEntry(argument), ts, DEFAULT_VERSION);
+ }
+
+ private static KvEntry createDefaultKvEntry(Argument argument) {
+ String key = argument.getRefEntityKey().getKey();
+ String defaultValue = argument.getDefaultValue();
+ if (StringUtils.isBlank(defaultValue)) {
+ return new StringDataEntry(key, null);
+ }
+ if (NumberUtils.isParsable(defaultValue)) {
+ return new DoubleDataEntry(key, Double.parseDouble(defaultValue));
+ }
+ if ("true".equalsIgnoreCase(defaultValue) || "false".equalsIgnoreCase(defaultValue)) {
+ return new BooleanDataEntry(key, Boolean.parseBoolean(defaultValue));
+ }
+ return new StringDataEntry(key, defaultValue);
+ }
+
+ public static CalculatedFieldState createStateByType(CalculatedFieldCtx ctx) {
+ return switch (ctx.getCfType()) {
+ case SIMPLE -> new SimpleCalculatedFieldState(ctx.getArgNames());
+ case SCRIPT -> new ScriptCalculatedFieldState(ctx.getArgNames());
+ };
+ }
+
+}
diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml
index 7da26fdc81..cda7f5782a 100644
--- a/application/src/main/resources/thingsboard.yml
+++ b/application/src/main/resources/thingsboard.yml
@@ -1295,6 +1295,8 @@ transport:
ignore_type_cast_errors: "${SNMP_RESPONSE_IGNORE_TYPE_CAST_ERRORS:false}"
# Thread pool size for scheduler that executes device querying tasks
scheduler_thread_pool_size: "${SNMP_SCHEDULER_THREAD_POOL_SIZE:4}"
+ # Maximum number of retry attempts for a single SNMP devices batch during bootstrap.
+ batch_retries: "${SNMP_BOOTSTRAP_RETRIES:8}"
stats:
# Enable/Disable the collection of transport statistics
enabled: "${TB_TRANSPORT_STATS_ENABLED:true}"
@@ -1718,6 +1720,8 @@ queue:
print-interval-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_MIN_PRINT_INTERVAL_MS:60000}"
# Time to wait for the stats-loading requests to Kafka to finish
kafka-response-timeout-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_RESPONSE_TIMEOUT_MS:1000}"
+ # Topics cache TTL in milliseconds. 5 minutes by default
+ topics_cache_ttl_ms: "${TB_QUEUE_KAFKA_TOPICS_CACHE_TTL_MS:300000}"
partitions:
hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256
transport_api:
diff --git a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java
index da4b5758cc..ffa39d8894 100644
--- a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java
@@ -45,6 +45,7 @@ import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
@DaoSqlTest
public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTest {
@@ -570,6 +571,7 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
@Test
public void testScriptCalculatedFieldWhenUsedLatestTsInScript() throws Exception {
Device testDevice = createDevice("Test device", "1234567890");
+
long ts = System.currentTimeMillis() - 300000L;
doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"temperature\":30}}", ts)));
@@ -606,6 +608,91 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
});
}
+ @Test
+ public void testSimpleCalculatedFieldWhenUseLatestTsIsTrueAndDefaultArguments() throws Exception {
+ Device testDevice = createDevice("Test device", "1234567890");
+
+ CalculatedField calculatedField = new CalculatedField();
+ calculatedField.setEntityId(testDevice.getId());
+ calculatedField.setType(CalculatedFieldType.SIMPLE);
+ calculatedField.setName("a + b + c");
+ calculatedField.setDebugSettings(DebugSettings.all());
+ calculatedField.setConfigurationVersion(1);
+
+ SimpleCalculatedFieldConfiguration config = new SimpleCalculatedFieldConfiguration();
+
+ Argument argument1 = new Argument();
+ ReferencedEntityKey refEntityKey1 = new ReferencedEntityKey("a", ArgumentType.TS_LATEST, null);
+ argument1.setRefEntityKey(refEntityKey1);
+ argument1.setDefaultValue("100");
+ Argument argument2 = new Argument();
+ ReferencedEntityKey refEntityKey2 = new ReferencedEntityKey("b", ArgumentType.TS_LATEST, null);
+ argument2.setRefEntityKey(refEntityKey2);
+ argument2.setDefaultValue("200");
+ Argument argument3 = new Argument();
+ ReferencedEntityKey refEntityKey3 = new ReferencedEntityKey("c", ArgumentType.TS_LATEST, null);
+ argument3.setRefEntityKey(refEntityKey3);
+ argument3.setDefaultValue("300");
+ config.setArguments(Map.of("a", argument1, "b", argument2, "c", argument3));
+ config.setExpression("a + b + c");
+
+ Output output = new Output();
+ output.setName("d");
+ output.setType(OutputType.TIME_SERIES);
+ output.setDecimalsByDefault(0);
+ config.setOutput(output);
+
+ config.setUseLatestTs(true);
+
+ calculatedField.setConfiguration(config);
+
+ CalculatedField savedCalculatedField = doPost("/api/calculatedField", calculatedField, CalculatedField.class);
+
+ await().alias("create CF -> perform initial calculation with default arguments").atMost(TIMEOUT, TimeUnit.SECONDS)
+ .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ ObjectNode d = getLatestTelemetry(testDevice.getId(), "d");
+ assertThat(d).isNotNull();
+ assertThat(d.get("d").get(0).get("value").asText()).isEqualTo("600");
+ });
+
+ doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"a\":10}"));
+
+ await().alias("update telemetry -> save result with ts of 'a' argument").atMost(TIMEOUT, TimeUnit.SECONDS)
+ .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ ObjectNode keys = getLatestTelemetry(testDevice.getId(), "d", "a");
+ assertThat(keys).isNotNull();
+ String aTs = keys.get("a").get(0).get("ts").asText();
+ assertThat(keys.get("d").get(0).get("ts").asText()).isEqualTo(aTs);
+ assertThat(keys.get("d").get(0).get("value").asText()).isEqualTo("510");
+ });
+
+ doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"b\":20}"));
+ doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"c\":30}"));
+
+ await().alias("update telemetry -> save result with latest ts of updated arguments").atMost(TIMEOUT, TimeUnit.SECONDS)
+ .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ ObjectNode keys = getLatestTelemetry(testDevice.getId(), "d");
+ assertThat(keys).isNotNull();
+ assertThat(keys.get("d").get(0).get("value").asText()).isEqualTo("60");
+ });
+
+ String latestTs = getLatestTelemetry(testDevice.getId(), "d").get("d").get(0).get("ts").asText();
+
+ doDelete("/api/plugins/telemetry/DEVICE/" + testDevice.getId() + "/timeseries/delete?keys=b&deleteAllDataForKeys=true").andExpect(status().isOk());
+
+ await().alias("delete telemetry -> save result with previous latest ts and default argument").atMost(TIMEOUT, TimeUnit.SECONDS)
+ .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ ObjectNode keys = getLatestTelemetry(testDevice.getId(), "d");
+ assertThat(keys).isNotNull();
+ assertThat(keys.get("d").get(0).get("ts").asText()).isEqualTo(latestTs);
+ assertThat(keys.get("d").get(0).get("value").asText()).isEqualTo("240");
+ });
+ }
+
@Test
public void testSimpleCalculatedFieldWhenCtxBecameUninitialized() throws Exception {
Device testDevice = createDevice("Test device", "1234567890");
@@ -659,6 +746,135 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
});
}
+ @Test
+ public void testCalculatedFieldWhenTheSameTelemetryKeysUsed() throws Exception {
+ Device testDevice = createDevice("Test device", "1234567890");
+ doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"a\":5}"));
+
+ CalculatedField calculatedField = new CalculatedField();
+ calculatedField.setEntityId(testDevice.getId());
+ calculatedField.setType(CalculatedFieldType.SIMPLE);
+ calculatedField.setName("a + b");
+ calculatedField.setDebugSettings(DebugSettings.all());
+
+ SimpleCalculatedFieldConfiguration config = new SimpleCalculatedFieldConfiguration();
+
+ ReferencedEntityKey refEntityKey = new ReferencedEntityKey("a", ArgumentType.TS_LATEST, null);
+ Argument argumentA = new Argument();
+ argumentA.setRefEntityKey(refEntityKey);
+ Argument argumentB = new Argument();
+ argumentB.setRefEntityKey(refEntityKey);
+ config.setArguments(Map.of("a", argumentA, "b", argumentB));
+ config.setExpression("a + b");
+
+ Output output = new Output();
+ output.setName("c");
+ output.setType(OutputType.TIME_SERIES);
+ output.setDecimalsByDefault(0);
+ config.setOutput(output);
+
+ calculatedField.setConfiguration(config);
+
+ doPost("/api/calculatedField", calculatedField, CalculatedField.class);
+
+ await().alias("create CF -> perform initial calculation").atMost(TIMEOUT, TimeUnit.SECONDS)
+ .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ ObjectNode c = getLatestTelemetry(testDevice.getId(), "c");
+ assertThat(c).isNotNull();
+ assertThat(c.get("c").get(0).get("value").asText()).isEqualTo("10");
+ });
+
+ doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"a\":10}"));
+
+ await().alias("update telemetry -> recalculate state").atMost(TIMEOUT, TimeUnit.SECONDS)
+ .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ ObjectNode c = getLatestTelemetry(testDevice.getId(), "c");
+ assertThat(c).isNotNull();
+ assertThat(c.get("c").get(0).get("value").asText()).isEqualTo("20");
+ });
+ }
+
+ @Test
+ public void testCalculatedFieldWhenBatchOfTelemetrySent() throws Exception {
+ Device testDevice = createDevice("Test device", "1234567890");
+ long now = System.currentTimeMillis();
+ doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"a\":5, \"b\":10}}", now - TimeUnit.MINUTES.toMillis(3))));
+ doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"b\":20}}", now - TimeUnit.MINUTES.toMillis(1))));
+
+ CalculatedField calculatedField = new CalculatedField();
+ calculatedField.setEntityId(testDevice.getId());
+ calculatedField.setType(CalculatedFieldType.SCRIPT);
+ calculatedField.setName("Script CF");
+ calculatedField.setDebugSettings(DebugSettings.all());
+
+ ScriptCalculatedFieldConfiguration config = new ScriptCalculatedFieldConfiguration();
+
+ ReferencedEntityKey refEntityKeyA = new ReferencedEntityKey("a", ArgumentType.TS_LATEST, null);
+ Argument argumentA = new Argument();
+ argumentA.setRefEntityKey(refEntityKeyA);
+ Argument argumentB = new Argument();
+ ReferencedEntityKey refEntityKeyB = new ReferencedEntityKey("b", ArgumentType.TS_ROLLING, null);
+ argumentB.setTimeWindow(TimeUnit.MINUTES.toMillis(10));
+ argumentB.setLimit(1000);
+ argumentB.setRefEntityKey(refEntityKeyB);
+ config.setArguments(Map.of("a", argumentA, "b", argumentB));
+ config.setExpression("""
+ return {
+ "latestA": a,
+ "avgB": b.avg
+ };
+ """);
+
+ Output output = new Output();
+ output.setType(OutputType.TIME_SERIES);
+ config.setOutput(output);
+
+ calculatedField.setConfiguration(config);
+
+ doPost("/api/calculatedField", calculatedField, CalculatedField.class);
+
+ await().alias("create CF -> perform initial calculation").atMost(TIMEOUT, TimeUnit.SECONDS)
+ .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ ObjectNode result = getLatestTelemetry(testDevice.getId(), "latestA", "avgB");
+ assertThat(result).isNotNull();
+ assertThat(result.get("latestA").get(0).get("value").asText()).isEqualTo("5");
+ assertThat(result.get("avgB").get(0).get("value").asText()).isEqualTo("15.0");
+ });
+
+ doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("""
+ [{
+ "ts": %s,
+ "values": {
+ "a": 6,
+ "b": 100
+ }
+ }, {
+ "ts": %s,
+ "values": {
+ "a": 7,
+ "b": 200
+ }
+ }, {
+ "ts": %s,
+ "values": {
+ "a": 8,
+ "b": 300
+ }
+ }]""", now - TimeUnit.MINUTES.toMillis(2), now, now - TimeUnit.MINUTES.toMillis(5))));
+
+ await().alias("update telemetry -> recalculate state").atMost(TIMEOUT, TimeUnit.SECONDS)
+ .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+ .untilAsserted(() -> {
+ ObjectNode result = getLatestTelemetry(testDevice.getId(), "latestA", "avgB");
+ assertThat(result).isNotNull();
+ assertThat(result.get("latestA").get(0).get("value").asText()).isEqualTo("7");
+ assertThat(result.get("avgB").get(0).get("value").asText()).isEqualTo("126.0");
+ });
+ }
+
private ObjectNode getLatestTelemetry(EntityId entityId, String... keys) throws Exception {
return doGetAsync("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() + "/values/timeseries?keys=" + String.join(",", keys), ObjectNode.class);
}
diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractNotifyEntityTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractNotifyEntityTest.java
index 01c24ad493..d4eec0d648 100644
--- a/application/src/test/java/org/thingsboard/server/controller/AbstractNotifyEntityTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/AbstractNotifyEntityTest.java
@@ -128,11 +128,17 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
protected void testNotifyEntityAllOneTimeLogEntityActionEntityEqClass(HasName entity, EntityId entityId, EntityId originatorId,
TenantId tenantId, CustomerId customerId, UserId userId, String userName,
ActionType actionType, ActionType actionTypeEdge, Object... additionalInfo) {
+ testNotifyEntityAllOneTimeLogEntityActionEntityEqClass(tenantId, entity, entityId, originatorId, tenantId, customerId, userId, userName, actionType, actionTypeEdge, additionalInfo);
+ }
+
+ protected void testNotifyEntityAllOneTimeLogEntityActionEntityEqClass(TenantId entityTenantId, HasName entity, EntityId entityId, EntityId originatorId,
+ TenantId authTenantId, CustomerId customerId, UserId userId, String userName,
+ ActionType actionType, ActionType actionTypeEdge, Object... additionalInfo) {
int cntTime = 1;
- testNotificationMsgToEdgeServiceTime(entityId, tenantId, actionTypeEdge, cntTime);
- testLogEntityActionEntityEqClass(entity, originatorId, tenantId, customerId, userId, userName, actionType, cntTime, additionalInfo);
+ testNotificationMsgToEdgeServiceTime(entityId, entityTenantId, actionTypeEdge, cntTime);
+ testLogEntityActionEntityEqClass(entity, originatorId, authTenantId, customerId, userId, userName, actionType, cntTime, additionalInfo);
ArgumentMatcher matcherOriginatorId = argument -> argument.equals(originatorId);
- testPushMsgToRuleEngineTime(matcherOriginatorId, tenantId, entity, cntTime);
+ testPushMsgToRuleEngineTime(matcherOriginatorId, authTenantId, entity, cntTime);
Mockito.reset(tbClusterService, auditLogService);
}
@@ -159,17 +165,26 @@ public abstract class AbstractNotifyEntityTest extends AbstractWebTest {
TenantId tenantId, CustomerId customerId, UserId userId, String userName,
ActionType actionType,
int cntTime, int cntTimeEdge, int cntTimeRuleEngine, Object... additionalInfo) {
+ testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAny(tenantId, entity, originator, tenantId, customerId, userId, userName, actionType,
+ cntTime, cntTimeEdge, cntTimeRuleEngine, additionalInfo);
+ }
+
+ protected void testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAny(TenantId entityTenantId, HasName entity, HasName originator,
+ TenantId authTenantId, CustomerId customerId, UserId userId, String userName,
+ ActionType actionType,
+ int cntTime, int cntTimeEdge, int cntTimeRuleEngine, Object... additionalInfo) {
EntityId originatorId = createEntityId_NULL_UUID(originator);
- testSendNotificationMsgToEdgeServiceTimeEntityEqAny(tenantId, actionType, cntTimeEdge);
+ testSendNotificationMsgToEdgeServiceTimeEntityEqAny(entityTenantId, actionType, cntTimeEdge);
ArgumentMatcher matcherEntityClassEquals = argument -> argument.getClass().equals(entity.getClass());
ArgumentMatcher matcherOriginatorId = argument -> argument.getClass().equals(originatorId.getClass());
ArgumentMatcher matcherCustomerId = customerId == null ?
argument -> argument.getClass().equals(CustomerId.class) : argument -> argument.equals(customerId);
ArgumentMatcher matcherUserId = userId == null ?
argument -> argument.getClass().equals(UserId.class) : argument -> argument.equals(userId);
- testLogEntityActionAdditionalInfo(matcherEntityClassEquals, matcherOriginatorId, tenantId, matcherCustomerId, matcherUserId, userName, actionType, cntTime,
+ testLogEntityActionAdditionalInfo(matcherEntityClassEquals, matcherOriginatorId, authTenantId, matcherCustomerId, matcherUserId, userName, actionType, cntTime,
extractMatcherAdditionalInfoClass(additionalInfo));
- testPushMsgToRuleEngineTime(matcherOriginatorId, tenantId, entity, cntTimeRuleEngine);
+ testPushMsgToRuleEngineTime(matcherOriginatorId, authTenantId, entity, cntTimeRuleEngine);
+
}
protected void testNotifyManyEntityManyTimeMsgToEdgeServiceEntityEqAnyAdditionalInfoAny(HasName entity, HasName originator,
diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
index b93ff0f623..54b2bbf6cf 100644
--- a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
@@ -38,8 +38,6 @@ import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.test.mock.mockito.MockBean;
-import org.springframework.boot.test.mock.mockito.SpyBean;
import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager;
import org.springframework.http.HttpHeaders;
@@ -53,6 +51,8 @@ import org.springframework.mock.http.MockHttpInputMessage;
import org.springframework.mock.http.MockHttpOutputMessage;
import org.springframework.mock.web.MockMultipartFile;
import org.springframework.mock.web.MockPart;
+import org.springframework.test.context.bean.override.mockito.MockitoBean;
+import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
import org.springframework.test.util.ReflectionTestUtils;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.MvcResult;
@@ -154,6 +154,7 @@ import org.thingsboard.server.service.entitiy.tenant.profile.TbTenantProfileServ
import org.thingsboard.server.service.security.auth.jwt.RefreshTokenRequest;
import org.thingsboard.server.service.security.auth.rest.LoginRequest;
import org.thingsboard.server.service.security.model.token.JwtTokenFactory;
+import org.thingsboard.server.service.system.SystemPatchApplier;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
@@ -278,18 +279,21 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
@Autowired
private JwtTokenFactory jwtTokenFactory;
- @SpyBean
- protected MailService mailService;
-
@Autowired
protected InMemoryStorage storage;
@Autowired
protected JdbcTemplate jdbcTemplate;
- @MockBean
+ @MockitoSpyBean
+ protected MailService mailService;
+
+ @MockitoBean
protected CfRocksDb cfRocksDb;
+ @MockitoBean
+ protected SystemPatchApplier systemPatchApplier;
+
@Rule
public TestRule watcher = new TestWatcher() {
protected void starting(Description description) {
@@ -1274,7 +1278,7 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
protected List findJobs(List types, List entities) throws Exception {
return doGetTypedWithPageLink("/api/jobs?types=" + types.stream().map(Enum::name).collect(Collectors.joining(",")) +
- "&entities=" + entities.stream().map(UUID::toString).collect(Collectors.joining(",")) + "&",
+ "&entities=" + entities.stream().map(UUID::toString).collect(Collectors.joining(",")) + "&",
new TypeReference>() {}, new PageLink(100, 0, null, new SortOrder("createdTime", SortOrder.Direction.DESC))).getData();
}
diff --git a/application/src/test/java/org/thingsboard/server/controller/DeviceControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/DeviceControllerTest.java
index 36cede9e84..4551e530d2 100644
--- a/application/src/test/java/org/thingsboard/server/controller/DeviceControllerTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/DeviceControllerTest.java
@@ -81,6 +81,7 @@ import org.thingsboard.server.service.state.DeviceStateService;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
@@ -387,6 +388,48 @@ public class DeviceControllerTest extends AbstractControllerTest {
.andExpect(statusReason(containsString("Device can`t be referencing to device profile from different tenant!")));
}
+ @Test
+ public void testSaveDeviceWithFirmware() throws Exception {
+ loginTenantAdmin();
+ DeviceProfile profile = createDeviceProfile("Profile to test ota updates");
+ profile = doPost("/api/deviceProfile", profile, DeviceProfile.class);
+
+ SaveOtaPackageInfoRequest firmwareInfo = new SaveOtaPackageInfoRequest();
+ firmwareInfo.setDeviceProfileId(profile.getId());
+ firmwareInfo.setType(FIRMWARE);
+ String title = "title";
+ firmwareInfo.setTitle(title);
+ String fwVersion = "1.0";
+ firmwareInfo.setVersion(fwVersion);
+ String url = "test.url";
+ firmwareInfo.setUrl(url);
+ firmwareInfo.setUsesUrl(true);
+ OtaPackageInfo savedFw = doPost("/api/otaPackage", firmwareInfo, OtaPackageInfo.class);
+
+ Device device = new Device();
+ device.setName("My ota device");
+ device.setDeviceProfileId(profile.getId());
+ device.setFirmwareId(savedFw.getId());
+ device = doPost("/api/device", device, Device.class);
+
+ //check shared attributes
+ Device finalDevice = device;
+ await().atMost(TIMEOUT, TimeUnit.SECONDS).until(() -> {
+ List