diff --git a/application/src/main/data/json/edge/rule_chains/edge_root_rule_chain.json b/application/src/main/data/json/edge/rule_chains/edge_root_rule_chain.json
index 81f9e6a14d..e614c9b54c 100644
--- a/application/src/main/data/json/edge/rule_chains/edge_root_rule_chain.json
+++ b/application/src/main/data/json/edge/rule_chains/edge_root_rule_chain.json
@@ -10,27 +10,9 @@
"externalId": null
},
"metadata": {
- "firstNodeIndex": 0,
+ "firstNodeIndex": 2,
"nodes": [
{
- "additionalInfo": {
- "description": "Process incoming messages from devices with the alarm rules defined in the device profile. Dispatch all incoming messages with \"Success\" relation type.",
- "layoutX": 187,
- "layoutY": 468
- },
- "type": "org.thingsboard.rule.engine.profile.TbDeviceProfileNode",
- "name": "Device Profile Node",
- "configuration": {
- "persistAlarmRulesState": false,
- "fetchAlarmRulesStateOnStart": false
- },
- "externalId": null
- },
- {
- "additionalInfo": {
- "layoutX": 823,
- "layoutY": 157
- },
"type": "org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNode",
"name": "Save Timeseries",
"configurationVersion": 1,
@@ -41,13 +23,12 @@
"type": "ON_EVERY_MESSAGE"
}
},
- "externalId": null
+ "additionalInfo": {
+ "layoutX": 823,
+ "layoutY": 157
+ }
},
{
- "additionalInfo": {
- "layoutX": 824,
- "layoutY": 52
- },
"type": "org.thingsboard.rule.engine.telemetry.TbMsgAttributesNode",
"name": "Save Client Attributes",
"configurationVersion": 3,
@@ -60,25 +41,23 @@
"sendAttributesUpdatedNotification": false,
"updateAttributesOnlyOnValueChange": true
},
- "externalId": null
+ "additionalInfo": {
+ "layoutX": 824,
+ "layoutY": 52
+ }
},
{
- "additionalInfo": {
- "layoutX": 347,
- "layoutY": 149
- },
"type": "org.thingsboard.rule.engine.filter.TbMsgTypeSwitchNode",
"name": "Message Type Switch",
"configuration": {
"version": 0
},
- "externalId": null
+ "additionalInfo": {
+ "layoutX": 347,
+ "layoutY": 149
+ }
},
{
- "additionalInfo": {
- "layoutX": 825,
- "layoutY": 266
- },
"type": "org.thingsboard.rule.engine.action.TbLogNode",
"name": "Log RPC from Device",
"configuration": {
@@ -86,13 +65,12 @@
"jsScript": "return '\\nIncoming message:\\n' + JSON.stringify(msg) + '\\nIncoming metadata:\\n' + JSON.stringify(metadata);",
"tbelScript": "return '\\nIncoming message:\\n' + JSON.stringify(msg) + '\\nIncoming metadata:\\n' + JSON.stringify(metadata);"
},
- "externalId": null
+ "additionalInfo": {
+ "layoutX": 825,
+ "layoutY": 266
+ }
},
{
- "additionalInfo": {
- "layoutX": 824,
- "layoutY": 378
- },
"type": "org.thingsboard.rule.engine.action.TbLogNode",
"name": "Log Other",
"configuration": {
@@ -100,97 +78,92 @@
"jsScript": "return '\\nIncoming message:\\n' + JSON.stringify(msg) + '\\nIncoming metadata:\\n' + JSON.stringify(metadata);",
"tbelScript": "return '\\nIncoming message:\\n' + JSON.stringify(msg) + '\\nIncoming metadata:\\n' + JSON.stringify(metadata);"
},
- "externalId": null
- },
- {
"additionalInfo": {
"layoutX": 824,
- "layoutY": 466
- },
+ "layoutY": 378
+ }
+ },
+ {
"type": "org.thingsboard.rule.engine.rpc.TbSendRPCRequestNode",
"name": "RPC Call Request",
"configuration": {
"timeoutInSeconds": 60
},
- "externalId": null
+ "additionalInfo": {
+ "layoutX": 824,
+ "layoutY": 466
+ }
},
{
- "additionalInfo": {
- "layoutX": 1126,
- "layoutY": 104
- },
"type": "org.thingsboard.rule.engine.edge.TbMsgPushToCloudNode",
"name": "Push to cloud",
"configuration": {
"scope": "CLIENT_SCOPE"
},
- "externalId": null
+ "additionalInfo": {
+ "layoutX": 1126,
+ "layoutY": 104
+ }
},
{
- "additionalInfo": {
- "layoutX": 826,
- "layoutY": 601
- },
"type": "org.thingsboard.rule.engine.edge.TbMsgPushToCloudNode",
"name": "Push to cloud",
"configuration": {
"scope": "SERVER_SCOPE"
},
- "externalId": null
+ "additionalInfo": {
+ "layoutX": 826,
+ "layoutY": 601
+ }
}
],
"connections": [
{
"fromIndex": 0,
- "toIndex": 3,
+ "toIndex": 6,
"type": "Success"
},
{
"fromIndex": 1,
- "toIndex": 7,
+ "toIndex": 6,
"type": "Success"
},
{
"fromIndex": 2,
- "toIndex": 7,
- "type": "Success"
- },
- {
- "fromIndex": 3,
- "toIndex": 1,
+ "toIndex": 0,
"type": "Post telemetry"
},
{
- "fromIndex": 3,
- "toIndex": 2,
+ "fromIndex": 2,
+ "toIndex": 1,
"type": "Post attributes"
},
{
- "fromIndex": 3,
- "toIndex": 4,
+ "fromIndex": 2,
+ "toIndex": 3,
"type": "RPC Request from Device"
},
{
- "fromIndex": 3,
- "toIndex": 5,
+ "fromIndex": 2,
+ "toIndex": 4,
"type": "Other"
},
{
- "fromIndex": 3,
- "toIndex": 6,
+ "fromIndex": 2,
+ "toIndex": 5,
"type": "RPC Request to Device"
},
{
- "fromIndex": 3,
- "toIndex": 8,
+ "fromIndex": 2,
+ "toIndex": 7,
"type": "Attributes Deleted"
},
{
- "fromIndex": 3,
- "toIndex": 8,
+ "fromIndex": 2,
+ "toIndex": 7,
"type": "Attributes Updated"
}
],
"ruleChainConnections": null
}
-}
+}
\ No newline at end of file
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/data/json/system/widget_types/photo_camera_input.json b/application/src/main/data/json/system/widget_types/photo_camera_input.json
index 69327395cc..fc676380ab 100644
--- a/application/src/main/data/json/system/widget_types/photo_camera_input.json
+++ b/application/src/main/data/json/system/widget_types/photo_camera_input.json
@@ -15,7 +15,7 @@
"settingsSchema": "",
"dataKeySettingsSchema": "{}\n",
"settingsDirective": "tb-photo-camera-input-widget-settings",
- "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\":\"Photo camera input\",\"showTitleIcon\":false,\"titleIcon\":\"more_horiz\",\"iconColor\":\"rgba(0, 0, 0, 0.87)\",\"iconSize\":\"24px\",\"titleTooltip\":\"\",\"dropShadow\":true,\"enableFullscreen\":false,\"widgetStyle\":{},\"titleStyle\":{\"fontSize\":\"16px\",\"fontWeight\":400},\"useDashboardTimewindow\":true,\"displayTimewindow\":true,\"showLegend\":false,\"actions\":{}}"
+ "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;\"}]}],\"showTitle\":true,\"backgroundColor\":\"#fff\",\"color\":\"rgba(0, 0, 0, 0.87)\",\"padding\":\"8px\",\"settings\":{\"widgetTitle\":\"\",\"saveToGallery\":true,\"usePublicGalleryLink\":false,\"imageFormat\":\"image/png\",\"imageQuality\":0.92,\"maxWidth\":640,\"maxHeight\":480},\"title\":\"Photo camera input\",\"showTitleIcon\":false,\"titleIcon\":\"more_horiz\",\"iconColor\":\"rgba(0, 0, 0, 0.87)\",\"iconSize\":\"24px\",\"titleTooltip\":\"\",\"dropShadow\":true,\"enableFullscreen\":false,\"widgetStyle\":{},\"titleStyle\":{\"fontSize\":\"16px\",\"fontWeight\":400},\"showLegend\":false,\"actions\":{}}"
},
"resources": [
{
diff --git a/application/src/main/data/json/system/widget_types/timeseries_table.json b/application/src/main/data/json/system/widget_types/timeseries_table.json
index 8343eb918c..15779eb902 100644
--- a/application/src/main/data/json/system/widget_types/timeseries_table.json
+++ b/application/src/main/data/json/system/widget_types/timeseries_table.json
@@ -17,7 +17,7 @@
"latestDataKeySettingsDirective": "tb-timeseries-table-latest-key-settings",
"hasBasicMode": true,
"basicModeDirective": "tb-timeseries-table-basic-config",
- "defaultConfig": "{\"datasources\":[{\"type\":\"function\",\"name\":\"function\",\"entityAliasId\":null,\"filterId\":null,\"dataKeys\":[{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"Temperature °C\",\"color\":\"#2196f3\",\"settings\":{\"useCellStyleFunction\":true,\"cellStyleFunction\":\"if (value) {\\n var percent = (value + 60)/120 * 100;\\n var color = tinycolor.mix('blue', 'red', percent);\\n color.setAlpha(.5);\\n return {\\n paddingLeft: '20px',\\n color: '#ffffff',\\n background: color.toRgbString(),\\n fontSize: '18px'\\n };\\n} else {\\n return {};\\n}\",\"useCellContentFunction\":false},\"_hash\":0.8587686344902596,\"funcBody\":\"var value = prevValue + Math.random() * 40 - 20;\\nvar multiplier = Math.pow(10, 1 || 0);\\nvar value = Math.round(value * multiplier) / multiplier;\\nif (value < -60) {\\n\\tvalue = -60;\\n} else if (value > 60) {\\n\\tvalue = 60;\\n}\\nreturn value;\",\"units\":null,\"decimals\":null,\"usePostProcessing\":null,\"postFuncBody\":null},{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"Humidity, %\",\"color\":\"#ffc107\",\"settings\":{\"useCellStyleFunction\":true,\"cellStyleFunction\":\"if (value) {\\n var percent = value;\\n var backgroundColor = tinycolor('blue');\\n backgroundColor.setAlpha(value/100);\\n var color = 'blue';\\n if (value > 50) {\\n color = 'white';\\n }\\n \\n return {\\n paddingLeft: '20px',\\n color: color,\\n background: backgroundColor.toRgbString(),\\n fontSize: '18px'\\n };\\n} else {\\n return {};\\n}\",\"useCellContentFunction\":false},\"_hash\":0.12775350966079668,\"funcBody\":\"var value = prevValue + Math.random() * 20 - 10;\\nvar multiplier = Math.pow(10, 1 || 0);\\nvar value = Math.round(value * multiplier) / multiplier;\\nif (value < 5) {\\n\\tvalue = 5;\\n} else if (value > 100) {\\n\\tvalue = 100;\\n}\\nreturn value;\"}],\"latestDataKeys\":null}],\"timewindow\":{\"realtime\":{\"interval\":1000,\"timewindowMs\":60000},\"aggregation\":{\"type\":\"NONE\",\"limit\":200}},\"showTitle\":true,\"backgroundColor\":\"rgb(255, 255, 255)\",\"color\":\"rgba(0, 0, 0, 0.87)\",\"padding\":\"8px\",\"settings\":{\"showTimestamp\":true,\"displayPagination\":true,\"defaultPageSize\":10},\"title\":\"Timeseries table\",\"dropShadow\":true,\"enableFullscreen\":true,\"titleStyle\":{\"fontSize\":\"16px\",\"fontWeight\":400,\"padding\":\"5px 10px 5px 10px\"},\"useDashboardTimewindow\":false,\"showLegend\":false,\"widgetStyle\":{},\"actions\":{},\"showTitleIcon\":false,\"iconColor\":\"rgba(0, 0, 0, 0.87)\",\"iconSize\":\"24px\",\"displayTimewindow\":true,\"configMode\":\"basic\"}"
+ "defaultConfig": "{\"datasources\":[{\"type\":\"function\",\"name\":\"function\",\"entityAliasId\":null,\"filterId\":null,\"dataKeys\":[{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"Temperature °C\",\"color\":\"#2196f3\",\"settings\":{\"useCellStyleFunction\":true,\"cellStyleFunction\":\"if (value) {\\n var percent = (value + 60)/120 * 100;\\n var color = tinycolor.mix('blue', 'red', percent);\\n color.setAlpha(.5);\\n return {\\n paddingLeft: '20px',\\n color: '#ffffff',\\n background: color.toRgbString(),\\n fontSize: '18px'\\n };\\n} else {\\n return {};\\n}\",\"useCellContentFunction\":false},\"_hash\":0.8587686344902596,\"funcBody\":\"var value = prevValue + Math.random() * 40 - 20;\\nvar multiplier = Math.pow(10, 1 || 0);\\nvar value = Math.round(value * multiplier) / multiplier;\\nif (value < -60) {\\n\\tvalue = -60;\\n} else if (value > 60) {\\n\\tvalue = 60;\\n}\\nreturn value;\",\"units\":null,\"decimals\":null,\"usePostProcessing\":null,\"postFuncBody\":null},{\"name\":\"f(x)\",\"type\":\"function\",\"label\":\"Humidity, %\",\"color\":\"#ffc107\",\"settings\":{\"useCellStyleFunction\":true,\"cellStyleFunction\":\"if (value) {\\n var percent = value;\\n var backgroundColor = tinycolor('blue');\\n backgroundColor.setAlpha(value/100);\\n var color = 'blue';\\n if (value > 50) {\\n color = 'white';\\n }\\n \\n return {\\n paddingLeft: '20px',\\n color: color,\\n background: backgroundColor.toRgbString(),\\n fontSize: '18px'\\n };\\n} else {\\n return {};\\n}\",\"useCellContentFunction\":false},\"_hash\":0.12775350966079668,\"funcBody\":\"var value = prevValue + Math.random() * 20 - 10;\\nvar multiplier = Math.pow(10, 1 || 0);\\nvar value = Math.round(value * multiplier) / multiplier;\\nif (value < 5) {\\n\\tvalue = 5;\\n} else if (value > 100) {\\n\\tvalue = 100;\\n}\\nreturn value;\"}],\"latestDataKeys\":null}],\"timewindow\":{\"realtime\":{\"interval\":1000,\"timewindowMs\":60000},\"aggregation\":{\"type\":\"NONE\",\"limit\":200}},\"showTitle\":true,\"backgroundColor\":\"rgb(255, 255, 255)\",\"color\":\"rgba(0, 0, 0, 0.87)\",\"padding\":\"8px\",\"settings\":{\"enableSearch\":true,\"enableSelectColumnDisplay\":true,\"enableStickyHeader\":true,\"enableStickyAction\":true,\"showCellActionsMenu\":true,\"reserveSpaceForHiddenAction\":\"true\",\"showTimestamp\":true,\"dateFormat\":{\"format\":\"yyyy-MM-dd HH:mm:ss\"},\"displayPagination\":true,\"useEntityLabel\":false,\"defaultPageSize\":10,\"pageStepCount\":3,\"pageStepIncrement\":10,\"hideEmptyLines\":false,\"disableStickyHeader\":false,\"useRowStyleFunction\":false,\"rowStyleFunction\":\"\",\"tabSortKey\":\"timestamp\"},\"title\":\"Timeseries table\",\"dropShadow\":true,\"enableFullscreen\":true,\"titleStyle\":{\"fontSize\":\"16px\",\"fontWeight\":400,\"padding\":\"5px 10px 5px 10px\"},\"useDashboardTimewindow\":false,\"showLegend\":false,\"widgetStyle\":{},\"actions\":{},\"showTitleIcon\":false,\"iconColor\":\"rgba(0, 0, 0, 0.87)\",\"iconSize\":\"24px\",\"displayTimewindow\":true,\"configMode\":\"basic\"}"
},
"resources": [
{
@@ -32,4 +32,4 @@
"public": true
}
]
-}
\ No newline at end of file
+}
diff --git a/application/src/main/data/json/tenant/device_profile/rule_chain_template.json b/application/src/main/data/json/tenant/device_profile/rule_chain_template.json
index 305dc04961..8773a2d6aa 100644
--- a/application/src/main/data/json/tenant/device_profile/rule_chain_template.json
+++ b/application/src/main/data/json/tenant/device_profile/rule_chain_template.json
@@ -10,12 +10,12 @@
"configuration": null
},
"metadata": {
- "firstNodeIndex": 6,
+ "firstNodeIndex": 2,
"nodes": [
{
"additionalInfo": {
- "layoutX": 822,
- "layoutY": 294
+ "layoutX": 824,
+ "layoutY": 156
},
"type": "org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNode",
"name": "Save Timeseries",
@@ -30,8 +30,8 @@
},
{
"additionalInfo": {
- "layoutX": 824,
- "layoutY": 221
+ "layoutX": 825,
+ "layoutY": 52
},
"type": "org.thingsboard.rule.engine.telemetry.TbMsgAttributesNode",
"name": "Save Client Attributes",
@@ -48,8 +48,8 @@
},
{
"additionalInfo": {
- "layoutX": 494,
- "layoutY": 309
+ "layoutX": 347,
+ "layoutY": 149
},
"type": "org.thingsboard.rule.engine.filter.TbMsgTypeSwitchNode",
"name": "Message Type Switch",
@@ -59,8 +59,8 @@
},
{
"additionalInfo": {
- "layoutX": 824,
- "layoutY": 383
+ "layoutX": 825,
+ "layoutY": 266
},
"type": "org.thingsboard.rule.engine.action.TbLogNode",
"name": "Log RPC from Device",
@@ -72,8 +72,8 @@
},
{
"additionalInfo": {
- "layoutX": 823,
- "layoutY": 444
+ "layoutX": 825,
+ "layoutY": 379
},
"type": "org.thingsboard.rule.engine.action.TbLogNode",
"name": "Log Other",
@@ -85,27 +85,14 @@
},
{
"additionalInfo": {
- "layoutX": 822,
- "layoutY": 507
+ "layoutX": 825,
+ "layoutY": 468
},
"type": "org.thingsboard.rule.engine.rpc.TbSendRPCRequestNode",
"name": "RPC Call Request",
"configuration": {
"timeoutInSeconds": 60
}
- },
- {
- "additionalInfo": {
- "description": "",
- "layoutX": 209,
- "layoutY": 307
- },
- "type": "org.thingsboard.rule.engine.profile.TbDeviceProfileNode",
- "name": "Device Profile Node",
- "configuration": {
- "persistAlarmRulesState": false,
- "fetchAlarmRulesStateOnStart": false
- }
}
],
"connections": [
@@ -133,11 +120,6 @@
"fromIndex": 2,
"toIndex": 5,
"type": "RPC Request to Device"
- },
- {
- "fromIndex": 6,
- "toIndex": 2,
- "type": "Success"
}
],
"ruleChainConnections": null
diff --git a/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json b/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json
index a988c9d5eb..c48dab1964 100644
--- a/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json
+++ b/application/src/main/data/json/tenant/rule_chains/root_rule_chain.json
@@ -9,7 +9,7 @@
"configuration": null
},
"metadata": {
- "firstNodeIndex": 6,
+ "firstNodeIndex": 2,
"nodes": [
{
"additionalInfo": {
@@ -92,27 +92,9 @@
"configuration": {
"timeoutInSeconds": 60
}
- },
- {
- "additionalInfo": {
- "description": "Process incoming messages from devices with the alarm rules defined in the device profile. Dispatch all incoming messages with \"Success\" relation type.",
- "layoutX": 204,
- "layoutY": 240
- },
- "type": "org.thingsboard.rule.engine.profile.TbDeviceProfileNode",
- "name": "Device Profile Node",
- "configuration": {
- "persistAlarmRulesState": false,
- "fetchAlarmRulesStateOnStart": false
- }
}
],
"connections": [
- {
- "fromIndex": 6,
- "toIndex": 2,
- "type": "Success"
- },
{
"fromIndex": 2,
"toIndex": 4,
diff --git a/application/src/main/data/upgrade/basic/schema_update.sql b/application/src/main/data/upgrade/basic/schema_update.sql
index 0add4c0545..e5fdba571b 100644
--- a/application/src/main/data/upgrade/basic/schema_update.sql
+++ b/application/src/main/data/upgrade/basic/schema_update.sql
@@ -34,6 +34,18 @@ SET profile_data = jsonb_set(
WHEN (profile_data -> 'configuration') ? 'maxRelationLevelPerCfArgument'
THEN NULL
ELSE to_jsonb(10)
+ END,
+ 'maxRelatedEntitiesToReturnPerCfArgument',
+ CASE
+ WHEN (profile_data -> 'configuration') ? 'maxRelatedEntitiesToReturnPerCfArgument'
+ THEN NULL
+ ELSE to_jsonb(100)
+ END,
+ 'minAllowedDeduplicationIntervalInSecForCF',
+ CASE
+ WHEN (profile_data -> 'configuration') ? 'minAllowedDeduplicationIntervalInSecForCF'
+ THEN NULL
+ ELSE to_jsonb(60)
END
)
),
@@ -43,6 +55,17 @@ WHERE NOT (
(profile_data -> 'configuration') ? 'minAllowedScheduledUpdateIntervalInSecForCF'
AND
(profile_data -> 'configuration') ? 'maxRelationLevelPerCfArgument'
+ AND
+ (profile_data -> 'configuration') ? 'maxRelatedEntitiesToReturnPerCfArgument'
+ AND
+ (profile_data -> 'configuration') ? 'minAllowedDeduplicationIntervalInSecForCF'
);
-- UPDATE TENANT PROFILE CONFIGURATION END
+
+-- CALCULATED FIELD UNIQUE CONSTRAINT UPDATE START
+
+ALTER TABLE calculated_field DROP CONSTRAINT IF EXISTS calculated_field_unq_key;
+ALTER TABLE calculated_field ADD CONSTRAINT calculated_field_unq_key UNIQUE (entity_id, type, name);
+
+-- CALCULATED FIELD UNIQUE CONSTRAINT UPDATE END
diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
index b23015a1fe..35cf9cb467 100644
--- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
@@ -117,6 +117,7 @@ import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
import org.thingsboard.server.service.cf.CalculatedFieldProcessingService;
import org.thingsboard.server.service.cf.CalculatedFieldQueueService;
import org.thingsboard.server.service.cf.CalculatedFieldStateService;
+import org.thingsboard.server.service.cf.OwnerService;
import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
import org.thingsboard.server.service.component.ComponentDiscoveryService;
import org.thingsboard.server.service.edge.rpc.EdgeRpcService;
@@ -571,6 +572,10 @@ public class ActorSystemContext {
@Getter
private JobManager jobManager;
+ @Autowired
+ @Getter
+ private OwnerService ownerService;
+
@Value("${actors.session.max_concurrent_sessions_per_device:1}")
@Getter
private int maxConcurrentSessionsPerDevice;
@@ -659,6 +664,10 @@ public class ActorSystemContext {
@Getter
private long cfCalculationResultTimeout;
+ @Value("${actors.alarms.reevaluation_interval:120}")
+ @Getter
+ private long alarmRulesReevaluationInterval;
+
@Autowired
@Getter
private MqttClientSettings mqttClientSettings;
@@ -851,8 +860,9 @@ public class ActorSystemContext {
if (errorMessage != null) {
eventBuilder.error(errorMessage);
}
-
- ListenableFuture future = eventService.saveAsync(eventBuilder.build());
+ CalculatedFieldDebugEvent event = eventBuilder.build();
+ log.debug("Persisting calculated field debug event: {}", event);
+ ListenableFuture future = eventService.saveAsync(event);
Futures.addCallback(future, CALCULATED_FIELD_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor());
} catch (IllegalArgumentException ex) {
log.warn("Failed to persist calculated field debug message", ex);
@@ -862,7 +872,7 @@ public class ActorSystemContext {
private boolean checkLimits(TenantId tenantId) {
if (debugModeRateLimitsConfig.isCalculatedFieldDebugPerTenantLimitsEnabled() &&
- !rateLimitService.checkRateLimit(LimitedApi.CALCULATED_FIELD_DEBUG_EVENTS, (Object) tenantId, debugModeRateLimitsConfig.getCalculatedFieldDebugPerTenantLimitsConfiguration())) {
+ !rateLimitService.checkRateLimit(LimitedApi.CALCULATED_FIELD_DEBUG_EVENTS, (Object) tenantId, debugModeRateLimitsConfig.getCalculatedFieldDebugPerTenantLimitsConfiguration())) {
log.trace("[{}] Calculated field debug event limits exceeded!", tenantId);
return false;
}
@@ -886,12 +896,13 @@ public class ActorSystemContext {
return getScheduler().scheduleWithFixedDelay(() -> ctx.tell(msg), delayInMs, periodInMs, TimeUnit.MILLISECONDS);
}
- public void scheduleMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs) {
+ public ScheduledFuture> scheduleMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs) {
log.debug("Scheduling msg {} with delay {} ms", msg, delayInMs);
if (delayInMs > 0) {
- getScheduler().schedule(() -> ctx.tell(msg), delayInMs, TimeUnit.MILLISECONDS);
+ return getScheduler().schedule(() -> ctx.tell(msg), delayInMs, TimeUnit.MILLISECONDS);
} else {
ctx.tell(msg);
+ return null;
}
}
diff --git a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
index 27bdec2422..20cacda26a 100644
--- a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
@@ -43,7 +43,6 @@ import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
import org.thingsboard.server.common.msg.queue.RuleEngineException;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.dao.tenant.TenantService;
-import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper;
import java.util.HashSet;
import java.util.Optional;
@@ -89,16 +88,20 @@ public class AppActor extends ContextAwareActor {
break;
case PARTITION_CHANGE_MSG:
case CF_PARTITIONS_CHANGE_MSG:
+ case CF_STATE_PARTITION_RESTORE_MSG:
ctx.broadcastToChildren(msg, true);
break;
case COMPONENT_LIFE_CYCLE_MSG:
onComponentLifecycleMsg((ComponentLifecycleMsg) msg);
break;
+ case CF_ENTITY_ACTION_EVENT_MSG:
+ forwardToTenantActor((TenantAwareMsg) msg, true);
+ break;
case QUEUE_TO_RULE_ENGINE_MSG:
onQueueToRuleEngineMsg((QueueToRuleEngineMsg) msg);
break;
case TRANSPORT_TO_DEVICE_ACTOR_MSG:
- onToDeviceActorMsg((TenantAwareMsg) msg, false);
+ forwardToTenantActor((TenantAwareMsg) msg, false);
break;
case DEVICE_ATTRIBUTES_UPDATE_TO_DEVICE_ACTOR_MSG:
case DEVICE_CREDENTIALS_UPDATE_TO_DEVICE_ACTOR_MSG:
@@ -108,7 +111,7 @@ public class AppActor extends ContextAwareActor {
case DEVICE_RPC_RESPONSE_TO_DEVICE_ACTOR_MSG:
case SERVER_RPC_RESPONSE_TO_DEVICE_ACTOR_MSG:
case REMOVE_RPC_TO_DEVICE_ACTOR_MSG:
- onToDeviceActorMsg((TenantAwareMsg) msg, true);
+ forwardToTenantActor((TenantAwareMsg) msg, true);
break;
case SESSION_TIMEOUT_MSG:
ctx.broadcastToChildrenByType(msg, EntityType.TENANT);
@@ -117,11 +120,11 @@ public class AppActor extends ContextAwareActor {
case CF_STATE_RESTORE_MSG:
//TODO: use priority from the message body. For example, messages about CF lifecycle are important and Device lifecycle are not.
// same for the Linked telemetry.
- onToCalculatedFieldSystemActorMsg((ToCalculatedFieldSystemMsg) msg, true);
+ forwardToTenantActor((ToCalculatedFieldSystemMsg) msg, true);
break;
case CF_TELEMETRY_MSG:
case CF_LINKED_TELEMETRY_MSG:
- onToCalculatedFieldSystemActorMsg((ToCalculatedFieldSystemMsg) msg, false);
+ forwardToTenantActor((ToCalculatedFieldSystemMsg) msg, false);
break;
default:
return false;
@@ -162,7 +165,7 @@ public class AppActor extends ContextAwareActor {
private void onComponentLifecycleMsg(ComponentLifecycleMsg msg) {
TbActorRef target = null;
if (TenantId.SYS_TENANT_ID.equals(msg.getTenantId())) {
- if (!EntityType.TENANT_PROFILE.equals(msg.getEntityId().getEntityType())) {
+ if (!msg.getEntityId().getEntityType().isOneOf(EntityType.TENANT_PROFILE, EntityType.TB_RESOURCE)) {
log.warn("Message has system tenant id: {}", msg);
}
} else {
@@ -187,7 +190,7 @@ public class AppActor extends ContextAwareActor {
}
}
- private void onToCalculatedFieldSystemActorMsg(ToCalculatedFieldSystemMsg msg, boolean priority) {
+ private void forwardToTenantActor(TenantAwareMsg msg, boolean priority) {
getOrCreateTenantActor(msg.getTenantId()).ifPresentOrElse(tenantActor -> {
if (priority) {
tenantActor.tellWithHighPriority(msg);
@@ -199,21 +202,6 @@ public class AppActor extends ContextAwareActor {
});
}
-
- private void onToDeviceActorMsg(TenantAwareMsg msg, boolean priority) {
- getOrCreateTenantActor(msg.getTenantId()).ifPresentOrElse(tenantActor -> {
- if (priority) {
- tenantActor.tellWithHighPriority(msg);
- } else {
- tenantActor.tell(msg);
- }
- }, () -> {
- if (msg instanceof TransportToDeviceActorMsgWrapper) {
- ((TransportToDeviceActorMsgWrapper) msg).getCallback().onSuccess();
- }
- });
- }
-
private Optional getOrCreateTenantActor(TenantId tenantId) {
if (deletedTenants.contains(tenantId)) {
return Optional.empty();
@@ -245,6 +233,7 @@ public class AppActor extends ContextAwareActor {
public TbActor createActor() {
return new AppActor(context);
}
+
}
}
diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldAlarmActionMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldAlarmActionMsg.java
new file mode 100644
index 0000000000..3202296345
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldAlarmActionMsg.java
@@ -0,0 +1,41 @@
+/**
+ * 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.actors.calculatedField;
+
+import lombok.Builder;
+import lombok.Data;
+import org.thingsboard.server.common.data.alarm.Alarm;
+import org.thingsboard.server.common.data.audit.ActionType;
+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;
+
+@Data
+@Builder
+public class CalculatedFieldAlarmActionMsg implements ToCalculatedFieldSystemMsg {
+
+ private final TenantId tenantId;
+ private final Alarm alarm;
+ private final ActionType action;
+ private final TbCallback callback;
+
+ @Override
+ public MsgType getMsgType() {
+ return MsgType.CF_ALARM_ACTION_MSG;
+ }
+
+}
diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldArgumentResetMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldArgumentResetMsg.java
new file mode 100644
index 0000000000..8b5927827e
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldArgumentResetMsg.java
@@ -0,0 +1,37 @@
+/**
+ * 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.actors.calculatedField;
+
+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.state.CalculatedFieldCtx;
+
+@Data
+public class CalculatedFieldArgumentResetMsg implements ToCalculatedFieldSystemMsg {
+
+ private final TenantId tenantId;
+ private final CalculatedFieldCtx ctx;
+ private final TbCallback callback;
+
+ @Override
+ public MsgType getMsgType() {
+ return MsgType.CF_ARGUMENT_RESET_MSG;
+ }
+
+}
diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActionEventMsg.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActionEventMsg.java
new file mode 100644
index 0000000000..6fc191e3db
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActionEventMsg.java
@@ -0,0 +1,57 @@
+/**
+ * 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.actors.calculatedField;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import lombok.Builder;
+import lombok.Data;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.server.common.data.audit.ActionType;
+import org.thingsboard.server.common.data.id.EntityId;
+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.common.util.ProtoUtils;
+import org.thingsboard.server.gen.transport.TransportProtos.EntityActionEventProto;
+
+@Data
+@Builder
+public class CalculatedFieldEntityActionEventMsg implements ToCalculatedFieldSystemMsg {
+
+ private final TenantId tenantId;
+ private final EntityId entityId;
+ private final JsonNode entity;
+ private final ActionType action;
+ private final TbCallback callback;
+
+ public static CalculatedFieldEntityActionEventMsg fromProto(EntityActionEventProto proto,
+ TbCallback callback) {
+ return CalculatedFieldEntityActionEventMsg.builder()
+ .tenantId((TenantId) ProtoUtils.fromProto(proto.getTenantId()))
+ .entityId(ProtoUtils.fromProto(proto.getEntityId()))
+ .entity(JacksonUtil.toJsonNode(proto.getEntity()))
+ .action(ActionType.valueOf(proto.getAction()))
+ .callback(callback)
+ .build();
+ }
+
+ @Override
+ public MsgType getMsgType() {
+ return MsgType.CF_ENTITY_ACTION_EVENT_MSG;
+ }
+
+}
diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java
index c57984ef3d..160cd995d5 100644
--- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java
@@ -21,6 +21,7 @@ import org.thingsboard.server.actors.TbActorCtx;
import org.thingsboard.server.actors.TbActorException;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.msg.CalculatedFieldStatePartitionRestoreMsg;
import org.thingsboard.server.common.msg.TbActorStopReason;
import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
@@ -63,18 +64,33 @@ public class CalculatedFieldEntityActor extends AbstractCalculatedFieldActor {
case CF_STATE_RESTORE_MSG:
processor.process((CalculatedFieldStateRestoreMsg) msg);
break;
+ case CF_STATE_PARTITION_RESTORE_MSG:
+ processor.process((CalculatedFieldStatePartitionRestoreMsg) msg);
+ break;
case CF_ENTITY_INIT_CF_MSG:
processor.process((EntityInitCalculatedFieldMsg) msg);
break;
case CF_ENTITY_DELETE_MSG:
processor.process((CalculatedFieldEntityDeleteMsg) msg);
break;
+ case CF_RELATION_ACTION_MSG:
+ processor.process((CalculatedFieldRelationActionMsg) msg);
+ break;
case CF_ENTITY_TELEMETRY_MSG:
processor.process((EntityCalculatedFieldTelemetryMsg) msg);
break;
case CF_LINKED_TELEMETRY_MSG:
processor.process((EntityCalculatedFieldLinkedTelemetryMsg) msg);
break;
+ case CF_REEVALUATE_MSG:
+ processor.process((CalculatedFieldReevaluateMsg) msg);
+ break;
+ case CF_ALARM_ACTION_MSG:
+ processor.process((CalculatedFieldAlarmActionMsg) msg);
+ break;
+ case CF_ARGUMENT_RESET_MSG:
+ processor.process((CalculatedFieldArgumentResetMsg) msg);
+ break;
default:
return false;
}
@@ -85,4 +101,5 @@ public class CalculatedFieldEntityActor extends AbstractCalculatedFieldActor {
void logProcessingException(Exception e) {
log.warn("[{}][{}] Processing failure", tenantId, processor.entityId, e);
}
+
}
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 0927bf6267..673db74863 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
@@ -21,10 +21,13 @@ import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.DebugModeUtil;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.actors.TbActorCtx;
+import org.thingsboard.server.actors.calculatedField.EntityInitCalculatedFieldMsg.StateAction;
import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.StringUtils;
+import org.thingsboard.server.common.data.alarm.Alarm;
+import org.thingsboard.server.common.data.cf.CalculatedFieldType;
import org.thingsboard.server.common.data.cf.configuration.Argument;
import org.thingsboard.server.common.data.cf.configuration.ArgumentType;
import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey;
@@ -33,6 +36,7 @@ import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.StringDataEntry;
import org.thingsboard.server.common.data.msg.TbMsgType;
+import org.thingsboard.server.common.msg.CalculatedFieldStatePartitionRestoreMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TbCallback;
@@ -48,6 +52,8 @@ 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.aggregation.RelatedEntitiesAggregationCalculatedFieldState;
+import org.thingsboard.server.service.cf.ctx.state.alarm.AlarmCalculatedFieldState;
import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingArgumentEntry;
import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingCalculatedFieldState;
@@ -64,6 +70,7 @@ import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
+import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createStateByType;
/**
* @author Andrew Shvayka
@@ -78,7 +85,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
final CalculatedFieldProcessingService cfService;
final CalculatedFieldStateService cfStateService;
- TbActorCtx ctx;
+ TbActorCtx actorCtx;
Map states = new HashMap<>();
CalculatedFieldEntityMessageProcessor(ActorSystemContext systemContext, TenantId tenantId, EntityId entityId) {
@@ -90,7 +97,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
}
void init(TbActorCtx ctx) {
- this.ctx = ctx;
+ this.actorCtx = ctx;
}
public void stop(boolean partitionChanged) {
@@ -98,8 +105,9 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
"[{}][{}] Stopping entity actor due to change partition event." :
"[{}][{}] Stopping entity actor.",
tenantId, entityId);
+ states.values().forEach(this::closeState);
states.clear();
- ctx.stop(ctx.getSelf());
+ actorCtx.stop(actorCtx.getSelf());
}
public void process(CalculatedFieldPartitionChangeMsg msg) {
@@ -111,28 +119,55 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
public void process(CalculatedFieldStateRestoreMsg msg) {
CalculatedFieldId cfId = msg.getId().cfId();
log.debug("[{}] [{}] Processing CF state restore msg.", msg.getId().entityId(), cfId);
- if (msg.getState() != null) {
- states.put(cfId, msg.getState());
+ CalculatedFieldState state = msg.getState();
+ if (state != null) {
+ state.setCtx(msg.getCtx(), actorCtx);
+ state.setPartition(msg.getPartition());
+ if (state instanceof RelatedEntitiesAggregationCalculatedFieldState relatedEntitiesAggState) {
+ relatedEntitiesAggState.scheduleReevaluation();
+ }
+ states.put(cfId, state);
} else {
- states.remove(cfId);
+ removeState(cfId);
+ }
+ }
+
+ public void process(CalculatedFieldStatePartitionRestoreMsg msg) {
+ log.debug("Processing CF state partition restore msg: {}", msg);
+ for (CalculatedFieldState state : states.values()) {
+ if (msg.getPartition().equals(state.getPartition())) {
+ state.init();
+ }
}
}
public void process(EntityInitCalculatedFieldMsg msg) throws CalculatedFieldException {
- log.debug("[{}] Processing entity init CF msg.", msg.getCtx().getCfId());
+ log.debug("[{}] Processing entity init CF msg: {}", msg.getCtx().getCfId(), msg);
var ctx = msg.getCtx();
- if (msg.isForceReinit()) {
- log.debug("Force reinitialization of CF: [{}].", ctx.getCfId());
- states.remove(ctx.getCfId());
+ CalculatedFieldState state;
+ if (msg.getStateAction() == StateAction.RECREATE) {
+ removeState(ctx.getCfId());
+ state = null;
+ } else {
+ state = states.get(ctx.getCfId());
}
try {
- var state = getOrInitState(ctx);
+ if (state == null) {
+ state = createState(ctx);
+ } else if (msg.getStateAction() == StateAction.REINIT) {
+ log.debug("Force reinitialization of CF: [{}].", ctx.getCfId());
+ state.reset();
+ initState(state, ctx);
+ } else {
+ state.setCtx(ctx, actorCtx);
+ }
if (state.isSizeOk()) {
- processStateIfReady(ctx, Collections.singletonList(ctx.getCfId()), state, null, null, msg.getCallback());
+ processStateIfReady(state, Collections.emptyMap(), ctx, Collections.singletonList(ctx.getCfId()), null, null, msg.getCallback());
} else {
throw new RuntimeException(ctx.getSizeExceedsLimitMessage());
}
} catch (Exception e) {
+ log.debug("[{}][{}] Failed to initialize CF state", entityId, ctx.getCfId(), e);
if (e instanceof CalculatedFieldException cfe) {
throw cfe;
}
@@ -140,31 +175,110 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
}
}
- public void process(CalculatedFieldEntityDeleteMsg msg) {
+ public void process(CalculatedFieldArgumentResetMsg msg) throws CalculatedFieldException {
+ log.debug("[{}] Processing CF argument reset msg.", entityId);
+ var ctx = msg.getCtx();
+ try {
+ Map dynamicSourceArgs = ctx.getArguments().entrySet().stream()
+ .filter(entry -> entry.getValue().hasOwnerSource())
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ Map fetchedArgs = cfService.fetchArgsFromDb(tenantId, entityId, dynamicSourceArgs);
+ fetchedArgs.values().forEach(arg -> arg.setForceResetPrevious(true));
+
+ processArgumentValuesUpdate(ctx, Collections.singletonList(ctx.getCfId()), msg.getCallback(), fetchedArgs, null, null);
+ } catch (Exception e) {
+ throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
+ }
+ }
+
+ public void process(CalculatedFieldEntityDeleteMsg msg) throws CalculatedFieldException {
log.debug("[{}] Processing CF entity delete msg.", msg.getEntityId());
if (this.entityId.equals(msg.getEntityId())) {
if (states.isEmpty()) {
msg.getCallback().onSuccess();
} else {
MultipleTbCallback multipleTbCallback = new MultipleTbCallback(states.size(), msg.getCallback());
- states.forEach((cfId, state) -> cfStateService.removeState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), multipleTbCallback));
- ctx.stop(ctx.getSelf());
+ states.forEach((cfId, state) -> cfStateService.deleteState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), multipleTbCallback));
+ actorCtx.stop(actorCtx.getSelf());
}
} else {
var cfId = new CalculatedFieldId(msg.getEntityId().getId());
- var state = states.remove(cfId);
+ var state = removeState(cfId);
if (state != null) {
- cfStateService.removeState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), msg.getCallback());
+ cfStateService.deleteState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), msg.getCallback());
} else {
msg.getCallback().onSuccess();
}
}
}
+ public void process(CalculatedFieldRelationActionMsg msg) throws CalculatedFieldException {
+ log.debug("[{}] Processing CF {} related entity msg.", msg.getRelatedEntityId(), msg.getAction());
+ switch (msg.getAction()) {
+ case UPDATED -> handleRelationUpdate(msg);
+ case DELETED -> handleRelationDelete(msg);
+ default -> msg.getCallback().onSuccess();
+ }
+ }
+
+ private void handleRelationUpdate(CalculatedFieldRelationActionMsg msg) throws CalculatedFieldException {
+ CalculatedFieldCtx ctx = msg.getCalculatedField();
+ var callback = new MultipleTbCallback(CALLBACKS_PER_CF, msg.getCallback());
+ var state = states.get(ctx.getCfId());
+ try {
+ Map updatedArgs = new HashMap<>();
+ if (state == null) {
+ state = createState(ctx);
+ } else {
+ if (state instanceof RelatedEntitiesAggregationCalculatedFieldState relatedEntitiesAggState) {
+ Map fetchedArgs = cfService.fetchArgsFromDb(tenantId, msg.getRelatedEntityId(), ctx.getArguments());
+ updatedArgs = relatedEntitiesAggState.updateEntityData(setEntityIdToSingleEntityArguments(msg.getRelatedEntityId(), fetchedArgs));
+ }
+
+ state.checkStateSize(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), ctx.getMaxStateSize());
+ }
+ if (state.isSizeOk()) {
+ processStateIfReady(state, updatedArgs, ctx, Collections.singletonList(ctx.getCfId()), null, null, callback);
+ } else {
+ throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build();
+ }
+ } catch (Exception e) {
+ log.debug("[{}][{}] Failed to initialize CF state", entityId, ctx.getCfId(), e);
+ if (e instanceof CalculatedFieldException cfe) {
+ throw cfe;
+ }
+ throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
+ }
+ }
+
+ private void handleRelationDelete(CalculatedFieldRelationActionMsg msg) throws CalculatedFieldException {
+ CalculatedFieldCtx ctx = msg.getCalculatedField();
+ CalculatedFieldId cfId = ctx.getCfId();
+ CalculatedFieldState state = states.get(cfId);
+ if (state == null) {
+ msg.getCallback().onSuccess();
+ return;
+ }
+ if (state instanceof RelatedEntitiesAggregationCalculatedFieldState aggState) {
+ aggState.cleanupEntityData(msg.getRelatedEntityId());
+
+ state.checkStateSize(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), ctx.getMaxStateSize());
+
+ if (state.isSizeOk()) {
+ processStateIfReady(state, Collections.emptyMap(), ctx, Collections.singletonList(ctx.getCfId()), null, null, msg.getCallback());
+ } else {
+ throw new RuntimeException(ctx.getSizeExceedsLimitMessage());
+ }
+ } else {
+ msg.getCallback().onSuccess();
+ }
+ }
+
public void process(EntityCalculatedFieldTelemetryMsg msg) throws CalculatedFieldException {
- log.debug("[{}] Processing CF telemetry msg.", msg.getEntityId());
+ log.trace("[{}] Processing CF telemetry msg: {}", msg.getEntityId(), msg);
var proto = msg.getProto();
- var numberOfCallbacks = CALLBACKS_PER_CF * (msg.getEntityIdFields().size() + msg.getProfileIdFields().size());
+ var numberOfCallbacks = msg.getEntityIdFields().size() + msg.getProfileIdFields().size();
MultipleTbCallback callback = new MultipleTbCallback(numberOfCallbacks, msg.getCallback());
List cfIdList = getCalculatedFieldIds(proto);
Set cfIdSet = new HashSet<>(cfIdList);
@@ -177,36 +291,37 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
}
public void process(EntityCalculatedFieldLinkedTelemetryMsg msg) throws CalculatedFieldException {
- log.debug("[{}] Processing CF link telemetry msg.", msg.getEntityId());
+ log.trace("[{}] Processing CF link telemetry msg: {}", msg.getEntityId(), msg);
var proto = msg.getProto();
var ctx = msg.getCtx();
- var callback = new MultipleTbCallback(CALLBACKS_PER_CF, msg.getCallback());
+ var callback = msg.getCallback();
try {
List cfIds = getCalculatedFieldIds(proto);
if (cfIds.contains(ctx.getCfId())) {
- callback.onSuccess(CALLBACKS_PER_CF);
+ callback.onSuccess();
} else {
if (proto.getTsDataCount() > 0) {
processArgumentValuesUpdate(ctx, cfIds, callback, mapToArguments(ctx, msg.getEntityId(), proto.getTsDataList()), toTbMsgId(proto), toTbMsgType(proto));
} else if (proto.getAttrDataCount() > 0) {
processArgumentValuesUpdate(ctx, cfIds, callback, mapToArguments(ctx, msg.getEntityId(), proto.getScope(), proto.getAttrDataList()), toTbMsgId(proto), toTbMsgType(proto));
} else if (proto.getRemovedTsKeysCount() > 0) {
- processArgumentValuesUpdate(ctx, cfIds, callback, mapToArgumentsWithFetchedValue(ctx, proto.getRemovedTsKeysList()), toTbMsgId(proto), toTbMsgType(proto));
+ processArgumentValuesUpdate(ctx, cfIds, callback, mapToArgumentsWithFetchedValue(ctx, msg.getEntityId(), proto.getRemovedTsKeysList()), toTbMsgId(proto), toTbMsgType(proto));
} else if (proto.getRemovedAttrKeysCount() > 0) {
processArgumentValuesUpdate(ctx, cfIds, callback, mapToArgumentsWithDefaultValue(ctx, msg.getEntityId(), proto.getScope(), proto.getRemovedAttrKeysList()), toTbMsgId(proto), toTbMsgType(proto));
} else {
- callback.onSuccess(CALLBACKS_PER_CF);
+ callback.onSuccess();
}
}
} catch (Exception e) {
+ log.debug("[{}][{}] Failed to process linked CF telemetry msg: {}", entityId, ctx.getCfId(), msg, e);
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
}
}
- private void process(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, Collection cfIds, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException {
+ private void process(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, Collection cfIds, List cfIdList, TbCallback callback) throws CalculatedFieldException {
try {
if (cfIds.contains(ctx.getCfId())) {
- callback.onSuccess(CALLBACKS_PER_CF);
+ callback.onSuccess();
} else {
if (proto.getTsDataCount() > 0) {
processTelemetry(ctx, proto, cfIdList, callback);
@@ -217,10 +332,11 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
} else if (proto.getRemovedAttrKeysCount() > 0) {
processRemovedAttributes(ctx, proto, cfIdList, callback);
} else {
- callback.onSuccess(CALLBACKS_PER_CF);
+ callback.onSuccess();
}
}
} catch (Exception e) {
+ log.debug("[{}][{}] Failed to process CF telemetry msg: {}", entityId, ctx.getCfId(), proto, e);
if (e instanceof CalculatedFieldException cfe) {
throw cfe;
}
@@ -228,81 +344,147 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
}
}
- private void processTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException {
+ public void process(CalculatedFieldReevaluateMsg msg) throws CalculatedFieldException {
+ CalculatedFieldId cfId = msg.getCtx().getCfId();
+ CalculatedFieldState state = states.get(cfId);
+ if (state == null) {
+ log.debug("[{}][{}] Failed to find CF state for entity to handle {}", entityId, cfId, msg);
+ } else {
+ if (state.isSizeOk()) {
+ log.debug("[{}][{}] Reevaluating CF state", entityId, cfId);
+ processStateIfReady(state, null, msg.getCtx(), Collections.singletonList(cfId), null, null, msg.getCallback());
+ } else {
+ throw new RuntimeException(msg.getCtx().getSizeExceedsLimitMessage());
+ }
+ }
+ }
+
+ public void process(CalculatedFieldAlarmActionMsg msg) {
+ log.debug("[{}] Processing alarm action event msg: {}", entityId, msg);
+ for (CalculatedFieldState state : states.values()) {
+ if (state instanceof AlarmCalculatedFieldState alarmCfState) {
+ Alarm stateAlarm = alarmCfState.getCurrentAlarm();
+ if (stateAlarm != null && stateAlarm.getId().equals(msg.getAlarm().getId())) {
+ alarmCfState.processAlarmAction(msg.getAlarm(), msg.getAction());
+ }
+ }
+ }
+ msg.getCallback().onSuccess();
+ }
+
+ private void processTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, TbCallback callback) throws CalculatedFieldException {
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getTsDataList()), toTbMsgId(proto), toTbMsgType(proto));
}
- private void processAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException {
+ private void processAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, TbCallback callback) throws CalculatedFieldException {
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getScope(), proto.getAttrDataList()), toTbMsgId(proto), toTbMsgType(proto));
}
- private void processRemovedTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException {
- processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArgumentsWithFetchedValue(ctx, proto.getRemovedTsKeysList()), toTbMsgId(proto), toTbMsgType(proto));
+ private void processRemovedTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, TbCallback callback) throws CalculatedFieldException {
+ processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArgumentsWithFetchedValue(ctx, entityId, proto.getRemovedTsKeysList()), toTbMsgId(proto), toTbMsgType(proto));
}
- private void processRemovedAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, MultipleTbCallback callback) throws CalculatedFieldException {
+ private void processRemovedAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List cfIdList, TbCallback callback) throws CalculatedFieldException {
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArgumentsWithDefaultValue(ctx, proto.getScope(), proto.getRemovedAttrKeysList()), toTbMsgId(proto), toTbMsgType(proto));
}
- private void processArgumentValuesUpdate(CalculatedFieldCtx ctx, List cfIdList, MultipleTbCallback callback,
+ private void processArgumentValuesUpdate(CalculatedFieldCtx ctx, List cfIdList, TbCallback callback,
Map newArgValues, UUID tbMsgId, TbMsgType tbMsgType) throws CalculatedFieldException {
if (newArgValues.isEmpty()) {
log.debug("[{}] No new argument values to process for CF.", ctx.getCfId());
- callback.onSuccess(CALLBACKS_PER_CF);
+ callback.onSuccess();
}
CalculatedFieldState state = states.get(ctx.getCfId());
boolean justRestored = false;
if (state == null) {
- state = getOrInitState(ctx);
+ state = createState(ctx);
justRestored = true;
- } else if (ctx.shouldFetchDynamicArgumentsFromDb(state)) {
+ } else if (ctx.shouldFetchRelationQueryDynamicArgumentsFromDb(state)) {
log.debug("[{}][{}] Going to update dynamic arguments for CF.", entityId, ctx.getCfId());
try {
Map dynamicArgsFromDb = cfService.fetchDynamicArgsFromDb(ctx, entityId);
dynamicArgsFromDb.forEach(newArgValues::putIfAbsent);
- var geofencingState = (GeofencingCalculatedFieldState) state;
- geofencingState.setLastDynamicArgumentsRefreshTs(System.currentTimeMillis());
+ if (ctx.getCfType() == CalculatedFieldType.GEOFENCING) {
+ var geofencingState = (GeofencingCalculatedFieldState) state;
+ geofencingState.updateLastDynamicArgumentsRefreshTs();
+ }
+ } catch (Exception e) {
+ throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
+ }
+ } else if (ctx.shouldFetchEntityRelations(state)) {
+ log.debug("[{}][{}] Going to update related entities for CF.", entityId, ctx.getCfId());
+ try {
+ if (state instanceof RelatedEntitiesAggregationCalculatedFieldState relatedEntitiesState) {
+ List relatedEntities = cfService.fetchRelatedEntities(ctx, entityId);
+ List missingEntities = relatedEntitiesState.checkRelatedEntities(relatedEntities);
+ if (!missingEntities.isEmpty()) {
+ missingEntities.forEach(missingEntityId -> {
+ Map fetchedArgs = cfService.fetchArgsFromDb(tenantId, missingEntityId, ctx.getArguments());
+ relatedEntitiesState.updateEntityData(setEntityIdToSingleEntityArguments(missingEntityId, fetchedArgs));
+ });
+ justRestored = true;
+ }
+ }
} catch (Exception e) {
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
}
}
if (state.isSizeOk()) {
- if (state.updateState(ctx, newArgValues) || justRestored) {
+ Map updatedArgs = state.update(newArgValues, ctx);
+ if (!updatedArgs.isEmpty() || justRestored) {
cfIdList = new ArrayList<>(cfIdList);
cfIdList.add(ctx.getCfId());
- processStateIfReady(ctx, cfIdList, state, tbMsgId, tbMsgType, callback);
+ processStateIfReady(state, updatedArgs, ctx, cfIdList, tbMsgId, tbMsgType, callback);
} else {
- callback.onSuccess(CALLBACKS_PER_CF);
+ callback.onSuccess();
}
} else {
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build();
}
}
- @SneakyThrows
- private CalculatedFieldState getOrInitState(CalculatedFieldCtx ctx) {
- CalculatedFieldState state = states.get(ctx.getCfId());
- if (state != null) {
- return state;
- } else {
- ListenableFuture stateFuture = cfService.fetchStateFromDb(ctx, entityId);
- // Ugly but necessary. We do not expect to often fetch data from DB. Only once per pair lifetime.
- // This call happens while processing the CF pack from the queue consumer. So the timeout should be relatively low.
- // Alternatively, we can fetch the state outside the actor system and push separate command to create this actor,
- // but this will significantly complicate the code.
- state = stateFuture.get(1, TimeUnit.MINUTES);
- state.checkStateSize(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), ctx.getMaxStateSize());
- states.put(ctx.getCfId(), state);
- }
+ private CalculatedFieldState createState(CalculatedFieldCtx ctx) {
+ CalculatedFieldState state = createStateByType(ctx, entityId);
+ initState(state, ctx);
return state;
}
- private void processStateIfReady(CalculatedFieldCtx ctx, List cfIdList, CalculatedFieldState state, UUID tbMsgId, TbMsgType tbMsgType, TbCallback callback) throws CalculatedFieldException {
+ private void initState(CalculatedFieldState state, CalculatedFieldCtx ctx) {
+ state.setCtx(ctx, actorCtx);
+ state.init();
+
+ if (ctx.getCfType() == CalculatedFieldType.GEOFENCING && ctx.isRelationQueryDynamicArguments()) {
+ GeofencingCalculatedFieldState geofencingState = (GeofencingCalculatedFieldState) state;
+ geofencingState.updateLastDynamicArgumentsRefreshTs();
+ }
+
+ Map arguments = fetchArguments(ctx);
+ state.update(arguments, ctx);
+
+ state.checkStateSize(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), ctx.getMaxStateSize());
+ states.put(ctx.getCfId(), state);
+ }
+
+ @SneakyThrows
+ private Map fetchArguments(CalculatedFieldCtx ctx) {
+ ListenableFuture