Andrii Landiak
2 years ago
7107 changed files with 39012 additions and 15437 deletions
|
Before Width: | Height: | Size: 74 KiB After Width: | Height: | Size: 73 KiB |
|
Before Width: | Height: | Size: 11 KiB After Width: | Height: | Size: 11 KiB |
|
Before Width: | Height: | Size: 30 KiB After Width: | Height: | Size: 30 KiB |
|
After Width: | Height: | Size: 14 KiB |
|
After Width: | Height: | Size: 13 KiB |
|
After Width: | Height: | Size: 24 KiB |
|
After Width: | Height: | Size: 13 KiB |
|
Before Width: | Height: | Size: 5.2 KiB After Width: | Height: | Size: 5.3 KiB |
|
Before Width: | Height: | Size: 77 KiB After Width: | Height: | Size: 76 KiB |
|
After Width: | Height: | Size: 13 KiB |
|
After Width: | Height: | Size: 13 KiB |
|
After Width: | Height: | Size: 14 KiB |
|
After Width: | Height: | Size: 13 KiB |
|
Before Width: | Height: | Size: 6.0 KiB After Width: | Height: | Size: 6.0 KiB |
|
Before Width: | Height: | Size: 57 KiB After Width: | Height: | Size: 57 KiB |
|
Before Width: | Height: | Size: 19 KiB After Width: | Height: | Size: 19 KiB |
|
Before Width: | Height: | Size: 16 KiB After Width: | Height: | Size: 16 KiB |
|
Before Width: | Height: | Size: 74 KiB After Width: | Height: | Size: 72 KiB |
|
Before Width: | Height: | Size: 6.9 KiB After Width: | Height: | Size: 6.9 KiB |
|
Before Width: | Height: | Size: 5.2 KiB After Width: | Height: | Size: 5.3 KiB |
|
Before Width: | Height: | Size: 6.9 KiB After Width: | Height: | Size: 7.0 KiB |
|
Before Width: | Height: | Size: 5.2 KiB After Width: | Height: | Size: 5.3 KiB |
|
After Width: | Height: | Size: 16 KiB |
|
Before Width: | Height: | Size: 57 KiB After Width: | Height: | Size: 57 KiB |
|
Before Width: | Height: | Size: 20 KiB After Width: | Height: | Size: 20 KiB |
|
Before Width: | Height: | Size: 16 KiB After Width: | Height: | Size: 16 KiB |
|
Before Width: | Height: | Size: 74 KiB After Width: | Height: | Size: 72 KiB |
|
Before Width: | Height: | Size: 7.0 KiB After Width: | Height: | Size: 7.0 KiB |
|
Before Width: | Height: | Size: 14 KiB After Width: | Height: | Size: 14 KiB |
|
Before Width: | Height: | Size: 14 KiB After Width: | Height: | Size: 14 KiB |
|
Before Width: | Height: | Size: 7.0 KiB After Width: | Height: | Size: 7.1 KiB |
|
Before Width: | Height: | Size: 27 KiB After Width: | Height: | Size: 27 KiB |
|
After Width: | Height: | Size: 13 KiB |
|
After Width: | Height: | Size: 13 KiB |
|
After Width: | Height: | Size: 13 KiB |
|
After Width: | Height: | Size: 13 KiB |
|
Before Width: | Height: | Size: 73 KiB After Width: | Height: | Size: 72 KiB |
|
After Width: | Height: | Size: 15 KiB |
|
Before Width: | Height: | Size: 5.2 KiB After Width: | Height: | Size: 5.3 KiB |
|
Before Width: | Height: | Size: 77 KiB After Width: | Height: | Size: 76 KiB |
File diff suppressed because one or more lines are too long
@ -0,0 +1,51 @@ |
|||
{ |
|||
"fqn": "value_stepper", |
|||
"name": "Value stepper", |
|||
"deprecated": false, |
|||
"image": "tb-image;/api/images/system/value-stepper-widget.svg", |
|||
"description": "Allows users to click the buttons to send commands to devices or update attributes/time series data. Configurable settings let users define how to retrieve the initial state and specify actions for each button.", |
|||
"descriptor": { |
|||
"type": "rpc", |
|||
"sizeX": 3.5, |
|||
"sizeY": 2, |
|||
"resources": [], |
|||
"templateHtml": "<tb-value-stepper-widget\n [ctx]='ctx'\n [widgetTitlePanel]=\"widgetTitlePanel\">\n</tb-value-stepper-widget>", |
|||
"templateCss": "", |
|||
"controllerScript": "self.onInit = function() {\n self.ctx.$scope.actionWidget.onInit();\n}\n\nself.typeParameters = function() {\n return {\n previewWidth: '230px',\n previewHeight: '110px',\n embedTitlePanel: true,\n displayRpcMessageToast: false\n };\n};\n\nself.onDestroy = function() {\n}\n", |
|||
"dataKeySettingsForm": [], |
|||
"settingsDirective": "tb-value-stepper-widget-settings", |
|||
"hasBasicMode": true, |
|||
"basicModeDirective": "tb-value-stepper-basic-config", |
|||
"defaultConfig": "{\"showTitle\":true,\"backgroundColor\":\"#ffffff\",\"color\":\"rgba(0, 0, 0, 0.87)\",\"padding\":\"0px\",\"settings\":{\"initialState\":{\"action\":\"EXECUTE_RPC\",\"defaultValue\":0,\"executeRpc\":{\"method\":\"getState\",\"requestTimeout\":5000,\"requestPersistent\":false,\"persistentPollingInterval\":1000},\"getAttribute\":{\"key\":\"state\",\"scope\":null},\"getTimeSeries\":{\"key\":\"state\"},\"getAlarmStatus\":{\"severityList\":null,\"typeList\":null},\"dataToValue\":{\"type\":\"NONE\",\"compareToValue\":true,\"dataToValueFunction\":\"/* Should return integer value */\\nreturn data;\"}},\"disabledState\":{\"action\":\"DO_NOTHING\",\"defaultValue\":false,\"getAttribute\":{\"key\":\"state\",\"scope\":null},\"getTimeSeries\":{\"key\":\"state\"},\"getAlarmStatus\":{\"severityList\":null,\"typeList\":null},\"dataToValue\":{\"type\":\"NONE\",\"compareToValue\":true,\"dataToValueFunction\":\"/* Should return boolean value */\\nreturn data;\"}},\"leftButtonClick\":{\"action\":\"EXECUTE_RPC\",\"executeRpc\":{\"method\":\"setState\",\"requestTimeout\":5000,\"requestPersistent\":false,\"persistentPollingInterval\":1000},\"setAttribute\":{\"key\":\"state\",\"scope\":\"SERVER_SCOPE\"},\"putTimeSeries\":{\"key\":\"state\"},\"valueToData\":{\"type\":\"VALUE\",\"constantValue\":0,\"valueToDataFunction\":\"/* Convert input integer value to RPC parameters or attribute/time-series value */\\nreturn value;\"}},\"rightButtonClick\":{\"action\":\"EXECUTE_RPC\",\"executeRpc\":{\"method\":\"setState\",\"requestTimeout\":5000,\"requestPersistent\":false,\"persistentPollingInterval\":1000},\"setAttribute\":{\"key\":\"state\",\"scope\":\"SERVER_SCOPE\"},\"putTimeSeries\":{\"key\":\"state\"},\"valueToData\":{\"type\":\"VALUE\",\"constantValue\":0,\"valueToDataFunction\":\"/* Convert input integer value to RPC parameters or attribute/time-series value */\\nreturn value;\"}},\"appearance\":{\"type\":\"simplified\",\"autoScale\":true,\"minValueRange\":-100,\"maxValueRange\":100,\"valueStep\":0.5,\"showValueBox\":true,\"valueUnits\":\"\",\"valueDecimals\":1,\"valueFont\":{\"family\":\"Roboto\",\"weight\":\"500\",\"style\":\"normal\",\"size\":16,\"sizeUnit\":\"px\",\"lineHeight\":\"24px\"},\"valueColor\":\"#000\",\"valueBoxBackground\":\"rgba(0, 0, 0, 0.04)\",\"showBorder\":true,\"borderWidth\":1,\"borderColor\":\"#305680\"},\"buttonAppearance\":{\"leftButton\":{\"showButton\":true,\"icon\":\"arrow_back_ios_new\",\"iconSize\":24,\"iconSizeUnit\":\"px\",\"mainColorOn\":\"#3F52DD\",\"backgroundColorOn\":\"#FFFFFF\",\"mainColorDisabled\":\"rgba(0,0,0,0.12)\",\"backgroundColorDisabled\":\"#FFFFFF\",\"customStyle\":{\"enabled\":null,\"hovered\":null,\"pressed\":null,\"activated\":null,\"disabled\":null}},\"rightButton\":{\"showButton\":true,\"icon\":\"arrow_forward_ios\",\"iconSize\":24,\"iconSizeUnit\":\"px\",\"mainColorOn\":\"#3F52DD\",\"backgroundColorOn\":\"#FFFFFF\",\"mainColorDisabled\":\"rgba(0,0,0,0.12)\",\"backgroundColorDisabled\":\"#FFFFFF\",\"customStyle\":{\"enabled\":null,\"hovered\":null,\"pressed\":null,\"activated\":null,\"disabled\":null}}},\"background\":{\"type\":\"color\",\"color\":\"#fff\",\"overlay\":{\"enabled\":false,\"color\":\"rgba(255,255,255,0.72)\",\"blur\":3}},\"padding\":\"12px\"},\"title\":\"Value stepper\",\"dropShadow\":true,\"enableFullscreen\":false,\"widgetStyle\":{},\"actions\":{},\"widgetCss\":\"\",\"noDataDisplayMessage\":\"\",\"titleFont\":{\"size\":16,\"sizeUnit\":\"px\",\"family\":null,\"weight\":\"500\",\"style\":null,\"lineHeight\":\"24px\"},\"showTitleIcon\":false,\"titleTooltip\":\"\",\"titleStyle\":{\"fontSize\":\"16px\",\"fontWeight\":400},\"pageSize\":1024,\"titleIcon\":\"\",\"iconColor\":\"rgba(0, 0, 0, 0.87)\",\"iconSize\":\"14px\",\"configMode\":\"basic\",\"titleColor\":\"rgba(0, 0, 0, 0.87)\",\"datasources\":null,\"borderRadius\":null}" |
|||
}, |
|||
"resources": [ |
|||
{ |
|||
"link": "/api/images/system/value-stepper-widget.svg", |
|||
"title": "\"Value stepper\" system widget image", |
|||
"type": "IMAGE", |
|||
"subType": "IMAGE", |
|||
"fileName": "value-stepper-widget.svg", |
|||
"publicResourceKey": "s0UKoqbiMCcKVn0pD55XZzPUR89XlXAO", |
|||
"mediaType": "image/svg+xml", |
|||
"data": "PHN2ZyB3aWR0aD0iMjE0IiBoZWlnaHQ9Ijc2IiBmaWxsPSJub25lIiB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciPjxnIGZpbHRlcj0idXJsKCNhKSI+PHJlY3QgeD0iOC41IiB5PSI0LjUiIHdpZHRoPSIxOTciIGhlaWdodD0iNTkiIHJ4PSI0IiBmaWxsPSIjZmZmIiBzaGFwZS1yZW5kZXJpbmc9ImNyaXNwRWRnZXMiLz48cmVjdCB4PSI5IiB5PSI1IiB3aWR0aD0iMTk2IiBoZWlnaHQ9IjU4IiByeD0iMy41IiBzdHJva2U9IiMzMDU2ODAiIHNoYXBlLXJlbmRlcmluZz0iY3Jpc3BFZGdlcyIvPjxyZWN0IHg9IjIwLjUiIHk9IjE4IiB3aWR0aD0iMzIiIGhlaWdodD0iMzIiIHJ4PSIxNiIgZmlsbD0iIzMwNTY4MCIvPjxwYXRoIGQ9Im0zOC41IDQwIDEuNDEtMS40MUwzNS4zMyAzNGw0LjU4LTQuNTlMMzguNSAyOGwtNiA2IDYgNloiIGZpbGw9IiNmZmYiLz48cmVjdCB4PSI2NSIgeT0iMTguNSIgd2lkdGg9Ijg0IiBoZWlnaHQ9IjMxIiByeD0iMy41IiBmaWxsPSIjMzA1NjgwIiBmaWxsLW9wYWNpdHk9Ii4wNCIvPjxyZWN0IHg9IjY1IiB5PSIxOC41IiB3aWR0aD0iODQiIGhlaWdodD0iMzEiIHJ4PSIzLjUiIHN0cm9rZT0iIzMwNTY4MCIvPjxwYXRoIGQ9Ik04OC41OSAzNy41VjM5aC03LjYzdi0xLjI5bDMuNy00LjA0Yy40MS0uNDYuNzMtLjg1Ljk3LTEuMTkuMjMtLjMzLjQtLjYzLjQ5LS45YTIuMyAyLjMgMCAwIDAtLjA2LTEuNzNjLS4xMy0uMjctLjMyLS41LS41OC0uNjVhMS43IDEuNyAwIDAgMC0uOTMtLjI0Yy0uNDIgMC0uNzcuMS0xLjA2LjI3LS4yOC4xOS0uNS40NC0uNjUuNzZhMi42IDIuNiAwIDAgMC0uMjIgMS4xaC0xLjg4YzAtLjY3LjE1LTEuMjcuNDYtMS44Mi4zLS41NS43My0uOTkgMS4zLTEuM2E0LjEgNC4xIDAgMCAxIDIuMDgtLjVjLjc2IDAgMS40LjEzIDEuOTQuMzguNTMuMjYuOTMuNjIgMS4yIDEuMDlhMy4zNiAzLjM2IDAgMCAxIC4yNSAyLjcyIDUgNSAwIDAgMS0uNDkgMS4wNCA5IDkgMCAwIDEtLjc0IDEuMDRjLS4yOC4zNS0uNi43LS45NCAxLjA1bC0yLjQ2IDIuNzFoNS4yNVptOS4yNy05Ljg4djEuMDRMOTMuMyAzOWgtMS45OGw0LjU0LTkuODhoLTUuOXYtMS41aDcuODlabTEuOTggMTAuNDRjMC0uMjkuMS0uNTMuMy0uNzMuMi0uMi40Ni0uMy44LS4zcy42LjEuOC4zYy4yLjIuMy40NC4zLjczYTEgMSAwIDAgMS0uMy43NGMtLjIuMi0uNDYuMy0uOC4zcy0uNi0uMS0uOC0uM2ExIDEgMCAwIDEtLjMtLjc0Wm02LjQ5LTQuMzUtMS41LS4zNy42MS01LjcyaDYuMTR2MS42SDEwN2wtLjMyIDIuNzlhMy42NyAzLjY3IDAgMCAxIDEuODEtLjQ2Yy41NCAwIDEuMDIuMDkgMS40NC4yNi40Mi4xNy43OS40MyAxLjA4Ljc2LjMuMzMuNTMuNzMuNjggMS4yYTUgNSAwIDAgMSAwIDMuMDcgMy4xNyAzLjE3IDAgMCAxLTEuODUgMi4wM2MtLjQ2LjE5LTEuMDEuMjktMS42NS4yOWE0LjYgNC42IDAgMCAxLTEuMzYtLjIgMy43MyAzLjczIDAgMCAxLTEuMTctLjYyIDMuMTQgMy4xNCAwIDAgMS0xLjE5LTIuNDJoMS44NWMuMDUuMzcuMTUuNjkuMy45NS4xNi4yNS4zOC40NS42NC41OGEyIDIgMCAwIDAgLjkyLjJjLjMyIDAgLjYtLjA1LjgzLS4xNi4yMy0uMTEuNDItLjI3LjU3LS40OC4xNS0uMjIuMjctLjQ3LjM0LS43NWEzLjYyIDMuNjIgMCAwIDAtLjAyLTEuODcgMS45OCAxLjk4IDAgMCAwLS4zOC0uNzJjLS4xNy0uMi0uMzgtLjM2LS42My0uNDdhMi4xMyAyLjEzIDAgMCAwLS44OC0uMTdjLS40NSAwLS44LjA3LTEuMDQuMi0uMjMuMTMtLjQ1LjI5LS42NS40OFptMTEuNzUtNC4xNmMwLS4zOC4xLS43Mi4yOC0xLjA0LjE5LS4zMi40NC0uNTcuNzUtLjc2YTEuOTYgMS45NiAwIDAgMSAyLjA1IDBjLjMxLjE5LjU2LjQ0Ljc0Ljc2LjE5LjMyLjI4LjY2LjI4IDEuMDRzLS4xLjczLS4yOCAxLjA1Yy0uMTguMzEtLjQzLjU2LS43NC43NGEyLjA0IDIuMDQgMCAwIDEtMi44LS43NCAyLjAzIDIuMDMgMCAwIDEtLjI4LTEuMDVabTEuMDUgMGExIDEgMCAwIDAgMSAxIC45Ny45NyAwIDAgMCAuOTgtMSAxIDEgMCAwIDAtLjI3LS43Mi45My45MyAwIDAgMC0uNy0uM2MtLjI4IDAtLjUxLjEtLjcxLjMtLjIuMi0uMy40My0uMy43MlptMTIuMTkgNS43NWgxLjk1YTQuNSA0LjUgMCAwIDEtLjYyIDEuOTkgMy43MiAzLjcyIDAgMCAxLTEuNSAxLjM3IDUgNSAwIDAgMS0yLjMzLjUgNC4xNSA0LjE1IDAgMCAxLTMuMzQtMS40NWMtLjQtLjQ4LS43MS0xLjA0LS45Mi0xLjctLjIxLS42Ni0uMzItMS40LS4zMi0yLjIydi0uOTVjMC0uODEuMS0xLjU1LjMyLTIuMjIuMjItLjY2LjUzLTEuMjIuOTQtMS42OS40LS40Ny45LS44NCAxLjQ2LTEuMDlhNC43OCA0Ljc4IDAgMCAxIDEuOTMtLjM3Yy45IDAgMS42Ny4xNyAyLjMuNS42Mi4zMyAxLjEuOCAxLjQ1IDEuMzguMzUuNTkuNTYgMS4yNi42NCAyLjAyaC0xLjk1Yy0uMDUtLjQ4LS4xNy0uOS0uMzUtMS4yNWExLjc3IDEuNzcgMCAwIDAtLjc2LS44IDIuNzMgMi43MyAwIDAgMC0xLjMzLS4yOGMtLjQ1IDAtLjg0LjA4LTEuMTcuMjVhMi4yIDIuMiAwIDAgMC0uODQuNzNjLS4yMi4zMy0uMzkuNzItLjUgMS4yLS4xMS40Ny0uMTcgMS0uMTcgMS42di45N2MwIC41Ny4wNSAxLjEuMTUgMS41Ni4xLjQ3LjI2Ljg2LjQ3IDEuMi4yMS4zMy40OC41OS44MS43Ny4zMy4xOC43Mi4yNyAxLjE4LjI3LjU2IDAgMS0uMDggMS4zNS0uMjYuMzUtLjE4LjYxLS40NC44LS43OC4xNy0uMzQuMy0uNzYuMzUtMS4yNVoiIGZpbGw9IiMwMDAiIGZpbGwtb3BhY2l0eT0iLjg3Ii8+PHJlY3QgeD0iMTYxLjUiIHk9IjE4IiB3aWR0aD0iMzIiIGhlaWdodD0iMzIiIHJ4PSIxNiIgZmlsbD0iIzMwNTY4MCIvPjxwYXRoIGQ9Im0xNzUuNSAyOC0xLjQxIDEuNDEgNC41OCA0LjU5LTQuNTggNC41OUwxNzUuNSA0MGw2LTYtNi02WiIgZmlsbD0iI2ZmZiIvPjwvZz48ZGVmcz48ZmlsdGVyIGlkPSJhIiB4PSIuNSIgeT0iLjUiIHdpZHRoPSIyMTMiIGhlaWdodD0iNzUiIGZpbHRlclVuaXRzPSJ1c2VyU3BhY2VPblVzZSIgY29sb3ItaW50ZXJwb2xhdGlvbi1maWx0ZXJzPSJzUkdCIj48ZmVGbG9vZCBmbG9vZC1vcGFjaXR5PSIwIiByZXN1bHQ9IkJhY2tncm91bmRJbWFnZUZpeCIvPjxmZUNvbG9yTWF0cml4IGluPSJTb3VyY2VBbHBoYSIgdmFsdWVzPSIwIDAgMCAwIDAgMCAwIDAgMCAwIDAgMCAwIDAgMCAwIDAgMCAxMjcgMCIgcmVzdWx0PSJoYXJkQWxwaGEiLz48ZmVPZmZzZXQgZHk9IjQiLz48ZmVHYXVzc2lhbkJsdXIgc3RkRGV2aWF0aW9uPSI0Ii8+PGZlQ29tcG9zaXRlIGluMj0iaGFyZEFscGhhIiBvcGVyYXRvcj0ib3V0Ii8+PGZlQ29sb3JNYXRyaXggdmFsdWVzPSIwIDAgMCAwIDAgMCAwIDAgMCAwIDAgMCAwIDAgMCAwIDAgMCAwLjA0IDAiLz48ZmVCbGVuZCBpbjI9IkJhY2tncm91bmRJbWFnZUZpeCIgcmVzdWx0PSJlZmZlY3QxX2Ryb3BTaGFkb3dfNTY2OV8xNjA3MDUiLz48ZmVCbGVuZCBpbj0iU291cmNlR3JhcGhpYyIgaW4yPSJlZmZlY3QxX2Ryb3BTaGFkb3dfNTY2OV8xNjA3MDUiIHJlc3VsdD0ic2hhcGUiLz48L2ZpbHRlcj48L2RlZnM+PC9zdmc+", |
|||
"public": true |
|||
} |
|||
], |
|||
"scada": false, |
|||
"tags": [ |
|||
"command", |
|||
"downlink", |
|||
"device configuration", |
|||
"device control", |
|||
"invocation", |
|||
"remote method", |
|||
"remote function", |
|||
"interface", |
|||
"subroutine call", |
|||
"inter-process communication", |
|||
"server request", |
|||
"update attribute", |
|||
"set attribute", |
|||
"add time-series" |
|||
] |
|||
} |
|||
@ -0,0 +1,70 @@ |
|||
/** |
|||
* 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.extern.slf4j.Slf4j; |
|||
import org.thingsboard.common.util.DebugModeUtil; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.service.ContextAwareActor; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.TbActorMsg; |
|||
import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; |
|||
|
|||
@Slf4j |
|||
public abstract class AbstractCalculatedFieldActor extends ContextAwareActor { |
|||
|
|||
protected final TenantId tenantId; |
|||
|
|||
public AbstractCalculatedFieldActor(ActorSystemContext systemContext, TenantId tenantId) { |
|||
super(systemContext); |
|||
this.tenantId = tenantId; |
|||
} |
|||
|
|||
@Override |
|||
protected boolean doProcess(TbActorMsg msg) { |
|||
if (msg instanceof ToCalculatedFieldSystemMsg cfm) { |
|||
Exception cause; |
|||
try { |
|||
return doProcessCfMsg(cfm); |
|||
} catch (CalculatedFieldException cfe) { |
|||
if (DebugModeUtil.isDebugFailuresAvailable(cfe.getCtx().getCalculatedField())) { |
|||
String message; |
|||
if (cfe.getErrorMessage() != null) { |
|||
message = cfe.getErrorMessage(); |
|||
} else if (cfe.getCause() != null) { |
|||
message = cfe.getCause().getMessage(); |
|||
} else { |
|||
message = "N/A"; |
|||
} |
|||
systemContext.persistCalculatedFieldDebugEvent(tenantId, cfe.getCtx().getCfId(), cfe.getEventEntity(), cfe.getArguments(), cfe.getMsgId(), cfe.getMsgType(), null, message); |
|||
} |
|||
cause = cfe.getCause(); |
|||
} catch (Exception e) { |
|||
logProcessingException(e); |
|||
cause = e; |
|||
} |
|||
cfm.getCallback().onFailure(cause); |
|||
return true; |
|||
} else { |
|||
return false; |
|||
} |
|||
} |
|||
|
|||
abstract void logProcessingException(Exception e); |
|||
|
|||
abstract boolean doProcessCfMsg(ToCalculatedFieldSystemMsg msg) throws CalculatedFieldException; |
|||
|
|||
} |
|||
@ -0,0 +1,81 @@ |
|||
/** |
|||
* 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.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
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.ToCalculatedFieldSystemMsg; |
|||
import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg; |
|||
|
|||
@Slf4j |
|||
public class CalculatedFieldEntityActor extends AbstractCalculatedFieldActor { |
|||
|
|||
private final CalculatedFieldEntityMessageProcessor processor; |
|||
|
|||
CalculatedFieldEntityActor(ActorSystemContext systemContext, TenantId tenantId, EntityId entityId) { |
|||
super(systemContext, tenantId); |
|||
this.processor = new CalculatedFieldEntityMessageProcessor(systemContext, tenantId, entityId); |
|||
} |
|||
|
|||
@Override |
|||
public void init(TbActorCtx ctx) throws TbActorException { |
|||
super.init(ctx); |
|||
log.debug("[{}][{}] Starting CF entity actor.", processor.tenantId, processor.entityId); |
|||
try { |
|||
processor.init(ctx); |
|||
log.debug("[{}][{}] CF entity actor started.", processor.tenantId, processor.entityId); |
|||
} catch (Exception e) { |
|||
log.warn("[{}][{}] Unknown failure", processor.tenantId, processor.entityId, e); |
|||
throw new TbActorException("Failed to initialize CF entity actor", e); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
protected boolean doProcessCfMsg(ToCalculatedFieldSystemMsg msg) throws CalculatedFieldException { |
|||
switch (msg.getMsgType()) { |
|||
case CF_PARTITIONS_CHANGE_MSG: |
|||
processor.process((CalculatedFieldPartitionChangeMsg) msg); |
|||
break; |
|||
case CF_STATE_RESTORE_MSG: |
|||
processor.process((CalculatedFieldStateRestoreMsg) 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_ENTITY_TELEMETRY_MSG: |
|||
processor.process((EntityCalculatedFieldTelemetryMsg) msg); |
|||
break; |
|||
case CF_LINKED_TELEMETRY_MSG: |
|||
processor.process((EntityCalculatedFieldLinkedTelemetryMsg) msg); |
|||
break; |
|||
default: |
|||
return false; |
|||
} |
|||
return true; |
|||
} |
|||
|
|||
@Override |
|||
void logProcessingException(Exception e) { |
|||
log.warn("[{}][{}] Processing failure", tenantId, processor.entityId, e); |
|||
} |
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
/** |
|||
* 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 org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.TbActor; |
|||
import org.thingsboard.server.actors.TbActorId; |
|||
import org.thingsboard.server.actors.TbCalculatedFieldEntityActorId; |
|||
import org.thingsboard.server.actors.TbEntityActorId; |
|||
import org.thingsboard.server.actors.device.DeviceActor; |
|||
import org.thingsboard.server.actors.service.ContextBasedCreator; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
public class CalculatedFieldEntityActorCreator extends ContextBasedCreator { |
|||
|
|||
private final TenantId tenantId; |
|||
private final EntityId entityId; |
|||
|
|||
public CalculatedFieldEntityActorCreator(ActorSystemContext context, TenantId tenantId, EntityId entityId) { |
|||
super(context); |
|||
this.tenantId = tenantId; |
|||
this.entityId = entityId; |
|||
} |
|||
|
|||
@Override |
|||
public TbActorId createActorId() { |
|||
return new TbCalculatedFieldEntityActorId(entityId); |
|||
} |
|||
|
|||
@Override |
|||
public TbActor createActor() { |
|||
return new CalculatedFieldEntityActor(context, tenantId, entityId); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
/** |
|||
* 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.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; |
|||
|
|||
@Data |
|||
public class CalculatedFieldEntityDeleteMsg implements ToCalculatedFieldSystemMsg { |
|||
|
|||
private final TenantId tenantId; |
|||
private final EntityId entityId; |
|||
private final TbCallback callback; |
|||
|
|||
public CalculatedFieldEntityDeleteMsg(TenantId tenantId, |
|||
EntityId entityId, |
|||
TbCallback callback) { |
|||
this.tenantId = tenantId; |
|||
this.entityId = entityId; |
|||
this.callback = callback; |
|||
} |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.CF_ENTITY_DELETE_MSG; |
|||
} |
|||
} |
|||
@ -0,0 +1,439 @@ |
|||
/** |
|||
* 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.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.SneakyThrows; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.common.util.DebugModeUtil; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.TbActorCtx; |
|||
import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; |
|||
import org.thingsboard.server.common.data.AttributeScope; |
|||
import org.thingsboard.server.common.data.StringUtils; |
|||
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; |
|||
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.StringDataEntry; |
|||
import org.thingsboard.server.common.data.msg.TbMsgType; |
|||
import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg; |
|||
import org.thingsboard.server.common.msg.queue.TbCallback; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; |
|||
import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; |
|||
import org.thingsboard.server.service.cf.CalculatedFieldResult; |
|||
import org.thingsboard.server.service.cf.CalculatedFieldStateService; |
|||
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.SingleValueArgumentEntry; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Collection; |
|||
import java.util.Collections; |
|||
import java.util.HashMap; |
|||
import java.util.HashSet; |
|||
import java.util.LinkedList; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Set; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.stream.Collectors; |
|||
|
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Slf4j |
|||
public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareMsgProcessor { |
|||
// (1 for result persistence + 1 for the state persistence )
|
|||
public static final int CALLBACKS_PER_CF = 2; |
|||
|
|||
final TenantId tenantId; |
|||
final EntityId entityId; |
|||
final CalculatedFieldProcessingService cfService; |
|||
final CalculatedFieldStateService cfStateService; |
|||
final int partition; |
|||
|
|||
TbActorCtx ctx; |
|||
Map<CalculatedFieldId, CalculatedFieldState> states = new HashMap<>(); |
|||
|
|||
CalculatedFieldEntityMessageProcessor(ActorSystemContext systemContext, TenantId tenantId, EntityId entityId) { |
|||
super(systemContext); |
|||
this.tenantId = tenantId; |
|||
this.entityId = entityId; |
|||
this.cfService = systemContext.getCalculatedFieldProcessingService(); |
|||
this.cfStateService = systemContext.getCalculatedFieldStateService(); |
|||
this.partition = systemContext.getCalculatedFieldEntityProfileCache().getEntityIdPartition(tenantId, entityId); |
|||
} |
|||
|
|||
void init(TbActorCtx ctx) { |
|||
this.ctx = ctx; |
|||
} |
|||
|
|||
public void process(CalculatedFieldPartitionChangeMsg msg) { |
|||
if (!msg.getPartitions()[partition]) { |
|||
log.info("[{}][{}] Stopping entity actor due to change partition event.", partition, entityId); |
|||
ctx.stop(ctx.getSelf()); |
|||
} |
|||
} |
|||
|
|||
public void process(CalculatedFieldStateRestoreMsg msg) { |
|||
CalculatedFieldId cfId = msg.getId().cfId(); |
|||
log.info("[{}] [{}] Processing CF state restore msg.", msg.getId().entityId(), cfId); |
|||
if (msg.getState() != null) { |
|||
states.put(cfId, msg.getState()); |
|||
} else { |
|||
states.remove(cfId); |
|||
} |
|||
} |
|||
|
|||
public void process(EntityInitCalculatedFieldMsg msg) throws CalculatedFieldException { |
|||
log.info("[{}] Processing entity init CF msg.", msg.getCtx().getCfId()); |
|||
var ctx = msg.getCtx(); |
|||
if (msg.isForceReinit()) { |
|||
log.info("Force reinitialization of CF: [{}].", ctx.getCfId()); |
|||
states.remove(ctx.getCfId()); |
|||
} |
|||
try { |
|||
var state = getOrInitState(ctx); |
|||
if (state.isSizeOk()) { |
|||
processStateIfReady(ctx, Collections.singletonList(ctx.getCfId()), state, null, null, msg.getCallback()); |
|||
} else { |
|||
throw new RuntimeException(ctx.getSizeExceedsLimitMessage()); |
|||
} |
|||
} catch (Exception e) { |
|||
if (e instanceof CalculatedFieldException cfe) { |
|||
throw cfe; |
|||
} |
|||
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build(); |
|||
} |
|||
} |
|||
|
|||
public void process(CalculatedFieldEntityDeleteMsg msg) { |
|||
log.info("[{}] Processing CF entity delete msg.", msg.getEntityId()); |
|||
if (this.entityId.equals(msg.getEntityId())) { |
|||
MultipleTbCallback multipleTbCallback = new MultipleTbCallback(states.size(), msg.getCallback()); |
|||
states.forEach((cfId, state) -> cfStateService.removeState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), multipleTbCallback)); |
|||
ctx.stop(ctx.getSelf()); |
|||
} else { |
|||
var cfId = new CalculatedFieldId(msg.getEntityId().getId()); |
|||
var state = states.remove(cfId); |
|||
if (state != null) { |
|||
cfStateService.removeState(new CalculatedFieldEntityCtxId(tenantId, cfId, entityId), msg.getCallback()); |
|||
} |
|||
} |
|||
} |
|||
|
|||
public void process(EntityCalculatedFieldTelemetryMsg msg) throws CalculatedFieldException { |
|||
log.info("[{}] Processing CF telemetry msg.", msg.getEntityId()); |
|||
var proto = msg.getProto(); |
|||
var numberOfCallbacks = CALLBACKS_PER_CF * (msg.getEntityIdFields().size() + msg.getProfileIdFields().size()); |
|||
MultipleTbCallback callback = new MultipleTbCallback(numberOfCallbacks, msg.getCallback()); |
|||
List<CalculatedFieldId> cfIdList = getCalculatedFieldIds(proto); |
|||
Set<CalculatedFieldId> cfIdSet = new HashSet<>(cfIdList); |
|||
for (var ctx : msg.getEntityIdFields()) { |
|||
process(ctx, proto, cfIdSet, cfIdList, callback); |
|||
} |
|||
for (var ctx : msg.getProfileIdFields()) { |
|||
process(ctx, proto, cfIdSet, cfIdList, callback); |
|||
} |
|||
} |
|||
|
|||
public void process(EntityCalculatedFieldLinkedTelemetryMsg msg) throws CalculatedFieldException { |
|||
log.info("[{}] Processing CF link telemetry msg.", msg.getEntityId()); |
|||
var proto = msg.getProto(); |
|||
var ctx = msg.getCtx(); |
|||
var callback = new MultipleTbCallback(CALLBACKS_PER_CF, msg.getCallback()); |
|||
try { |
|||
List<CalculatedFieldId> cfIds = getCalculatedFieldIds(proto); |
|||
if (cfIds.contains(ctx.getCfId())) { |
|||
callback.onSuccess(CALLBACKS_PER_CF); |
|||
} 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)); |
|||
} 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); |
|||
} |
|||
} |
|||
} catch (Exception e) { |
|||
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build(); |
|||
} |
|||
} |
|||
|
|||
private void process(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, Collection<CalculatedFieldId> cfIds, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { |
|||
try { |
|||
if (cfIds.contains(ctx.getCfId())) { |
|||
callback.onSuccess(CALLBACKS_PER_CF); |
|||
} else { |
|||
if (proto.getTsDataCount() > 0) { |
|||
processTelemetry(ctx, proto, cfIdList, callback); |
|||
} else if (proto.getAttrDataCount() > 0) { |
|||
processAttributes(ctx, proto, cfIdList, callback); |
|||
} else if (proto.getRemovedTsKeysCount() > 0) { |
|||
processRemovedTelemetry(ctx, proto, cfIdList, callback); |
|||
} else if (proto.getRemovedAttrKeysCount() > 0) { |
|||
processRemovedAttributes(ctx, proto, cfIdList, callback); |
|||
} else { |
|||
callback.onSuccess(CALLBACKS_PER_CF); |
|||
} |
|||
} |
|||
} catch (Exception e) { |
|||
if (e instanceof CalculatedFieldException cfe) { |
|||
throw cfe; |
|||
} |
|||
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build(); |
|||
} |
|||
} |
|||
|
|||
private void processTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { |
|||
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getTsDataList()), toTbMsgId(proto), toTbMsgType(proto)); |
|||
} |
|||
|
|||
private void processAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, MultipleTbCallback 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<CalculatedFieldId> cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { |
|||
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArgumentsWithFetchedValue(ctx, proto.getRemovedTsKeysList()), toTbMsgId(proto), toTbMsgType(proto)); |
|||
} |
|||
|
|||
private void processRemovedAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback) throws CalculatedFieldException { |
|||
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArgumentsWithDefaultValue(ctx, proto.getScope(), proto.getRemovedAttrKeysList()), toTbMsgId(proto), toTbMsgType(proto)); |
|||
} |
|||
|
|||
private void processArgumentValuesUpdate(CalculatedFieldCtx ctx, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback, |
|||
Map<String, ArgumentEntry> newArgValues, UUID tbMsgId, TbMsgType tbMsgType) throws CalculatedFieldException { |
|||
if (newArgValues.isEmpty()) { |
|||
log.info("[{}] No new argument values to process for CF.", ctx.getCfId()); |
|||
callback.onSuccess(CALLBACKS_PER_CF); |
|||
} |
|||
CalculatedFieldState state = states.get(ctx.getCfId()); |
|||
boolean justRestored = false; |
|||
if (state == null) { |
|||
state = getOrInitState(ctx); |
|||
justRestored = true; |
|||
} |
|||
if (state.isSizeOk()) { |
|||
if (state.updateState(newArgValues) || justRestored) { |
|||
cfIdList = new ArrayList<>(cfIdList); |
|||
cfIdList.add(ctx.getCfId()); |
|||
processStateIfReady(ctx, cfIdList, state, tbMsgId, tbMsgType, callback); |
|||
} else { |
|||
callback.onSuccess(CALLBACKS_PER_CF); |
|||
} |
|||
} 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<CalculatedFieldState> stateFuture = systemContext.getCalculatedFieldProcessingService().fetchStateFromDb(ctx, entityId); |
|||
// Ugly but necessary. We do not expect to often fetch data from DB. Only once per <Entity, CalculatedField> 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); |
|||
} |
|||
return state; |
|||
} |
|||
|
|||
private void processStateIfReady(CalculatedFieldCtx ctx, List<CalculatedFieldId> cfIdList, CalculatedFieldState state, UUID tbMsgId, TbMsgType tbMsgType, TbCallback callback) throws CalculatedFieldException { |
|||
CalculatedFieldEntityCtxId ctxId = new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId); |
|||
boolean stateSizeOk; |
|||
if (ctx.isInitialized() && state.isReady()) { |
|||
try { |
|||
CalculatedFieldResult calculationResult = state.performCalculation(ctx).get(5, TimeUnit.SECONDS); |
|||
state.checkStateSize(ctxId, ctx.getMaxStateSize()); |
|||
stateSizeOk = state.isSizeOk(); |
|||
if (stateSizeOk) { |
|||
cfService.pushMsgToRuleEngine(tenantId, entityId, calculationResult, cfIdList, callback); |
|||
if (DebugModeUtil.isDebugAllAvailable(ctx.getCalculatedField())) { |
|||
systemContext.persistCalculatedFieldDebugEvent(tenantId, ctx.getCfId(), entityId, state.getArguments(), tbMsgId, tbMsgType, JacksonUtil.writeValueAsString(calculationResult.getResult()), null); |
|||
} |
|||
} |
|||
} catch (Exception e) { |
|||
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).msgId(tbMsgId).msgType(tbMsgType).arguments(state.getArguments()).cause(e).build(); |
|||
} |
|||
} else { |
|||
state.checkStateSize(ctxId, ctx.getMaxStateSize()); |
|||
stateSizeOk = state.isSizeOk(); |
|||
if (stateSizeOk) { |
|||
callback.onSuccess(); // State was updated but no calculation performed;
|
|||
} |
|||
} |
|||
if (stateSizeOk) { |
|||
cfStateService.persistState(ctxId, state, callback); |
|||
} else { |
|||
removeStateAndRaiseSizeException(ctxId, CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build(), callback); |
|||
} |
|||
} |
|||
|
|||
private void removeStateAndRaiseSizeException(CalculatedFieldEntityCtxId ctxId, CalculatedFieldException ex, TbCallback callback) { |
|||
// We remove the state, but remember that it is over-sized in a local map.
|
|||
cfStateService.removeState(ctxId, new TbCallback() { |
|||
@Override |
|||
public void onSuccess() { |
|||
callback.onFailure(ex); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
callback.onFailure(ex); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private Map<String, ArgumentEntry> mapToArguments(CalculatedFieldCtx ctx, List<TsKvProto> data) { |
|||
return mapToArguments(ctx.getMainEntityArguments(), data); |
|||
} |
|||
|
|||
private Map<String, ArgumentEntry> mapToArguments(CalculatedFieldCtx ctx, EntityId entityId, List<TsKvProto> data) { |
|||
var argNames = ctx.getLinkedEntityArguments().get(entityId); |
|||
if (argNames.isEmpty()) { |
|||
return Collections.emptyMap(); |
|||
} |
|||
return mapToArguments(argNames, data); |
|||
} |
|||
|
|||
private Map<String, ArgumentEntry> mapToArguments(Map<ReferencedEntityKey, String> argNames, List<TsKvProto> data) { |
|||
if (argNames.isEmpty()) { |
|||
return Collections.emptyMap(); |
|||
} |
|||
Map<String, ArgumentEntry> 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)); |
|||
} |
|||
key = new ReferencedEntityKey(item.getKv().getKey(), ArgumentType.TS_ROLLING, null); |
|||
argName = argNames.get(key); |
|||
if (argName != null) { |
|||
arguments.put(argName, new SingleValueArgumentEntry(item)); |
|||
} |
|||
} |
|||
return arguments; |
|||
} |
|||
|
|||
private Map<String, ArgumentEntry> mapToArguments(CalculatedFieldCtx ctx, AttributeScopeProto scope, List<AttributeValueProto> attrDataList) { |
|||
return mapToArguments(ctx.getMainEntityArguments(), scope, attrDataList); |
|||
} |
|||
|
|||
private Map<String, ArgumentEntry> mapToArguments(CalculatedFieldCtx ctx, EntityId entityId, AttributeScopeProto scope, List<AttributeValueProto> attrDataList) { |
|||
var argNames = ctx.getLinkedEntityArguments().get(entityId); |
|||
if (argNames.isEmpty()) { |
|||
return Collections.emptyMap(); |
|||
} |
|||
return mapToArguments(argNames, scope, attrDataList); |
|||
} |
|||
|
|||
private Map<String, ArgumentEntry> mapToArguments(Map<ReferencedEntityKey, String> argNames, AttributeScopeProto scope, List<AttributeValueProto> attrDataList) { |
|||
Map<String, ArgumentEntry> 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)); |
|||
} |
|||
} |
|||
return arguments; |
|||
} |
|||
|
|||
private Map<String, ArgumentEntry> mapToArgumentsWithDefaultValue(CalculatedFieldCtx ctx, EntityId entityId, AttributeScopeProto scope, List<String> removedAttrKeys) { |
|||
var argNames = ctx.getLinkedEntityArguments().get(entityId); |
|||
if (argNames.isEmpty()) { |
|||
return Collections.emptyMap(); |
|||
} |
|||
return mapToArgumentsWithDefaultValue(argNames, ctx.getArguments(), scope, removedAttrKeys); |
|||
} |
|||
|
|||
private Map<String, ArgumentEntry> mapToArgumentsWithDefaultValue(CalculatedFieldCtx ctx, AttributeScopeProto scope, List<String> removedAttrKeys) { |
|||
return mapToArgumentsWithDefaultValue(ctx.getMainEntityArguments(), ctx.getArguments(), scope, removedAttrKeys); |
|||
} |
|||
|
|||
private Map<String, ArgumentEntry> mapToArgumentsWithDefaultValue(Map<ReferencedEntityKey, String> argNames, Map<String, Argument> configArguments, AttributeScopeProto scope, List<String> removedAttrKeys) { |
|||
Map<String, ArgumentEntry> 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()); |
|||
|
|||
} |
|||
} |
|||
return arguments; |
|||
} |
|||
|
|||
private Map<String, ArgumentEntry> mapToArgumentsWithFetchedValue(CalculatedFieldCtx ctx, List<String> removedTelemetryKeys) { |
|||
Map<String, Argument> deletedArguments = ctx.getArguments().entrySet().stream() |
|||
.filter(entry -> removedTelemetryKeys.contains(entry.getKey())) |
|||
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); |
|||
|
|||
Map<String, ArgumentEntry> fetchedArgs = cfService.fetchArgsFromDb(tenantId, entityId, deletedArguments); |
|||
|
|||
fetchedArgs.values().forEach(arg -> arg.setForceResetPrevious(true)); |
|||
return fetchedArgs; |
|||
} |
|||
|
|||
private static List<CalculatedFieldId> getCalculatedFieldIds(CalculatedFieldTelemetryMsgProto proto) { |
|||
List<CalculatedFieldId> cfIds = new LinkedList<>(); |
|||
for (var cfId : proto.getPreviousCalculatedFieldsList()) { |
|||
cfIds.add(new CalculatedFieldId(new UUID(cfId.getCalculatedFieldIdMSB(), cfId.getCalculatedFieldIdLSB()))); |
|||
} |
|||
return cfIds; |
|||
} |
|||
|
|||
private UUID toTbMsgId(CalculatedFieldTelemetryMsgProto proto) { |
|||
if (proto.getTbMsgIdMSB() != 0 && proto.getTbMsgIdLSB() != 0) { |
|||
return new UUID(proto.getTbMsgIdMSB(), proto.getTbMsgIdLSB()); |
|||
} |
|||
return null; |
|||
} |
|||
|
|||
private TbMsgType toTbMsgType(CalculatedFieldTelemetryMsgProto proto) { |
|||
if (!proto.getTbMsgType().isEmpty()) { |
|||
return TbMsgType.valueOf(proto.getTbMsgType()); |
|||
} |
|||
return null; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,40 @@ |
|||
/** |
|||
* 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.Getter; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.msg.TbMsgType; |
|||
import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; |
|||
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; |
|||
|
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
|
|||
@Getter |
|||
@Builder |
|||
public class CalculatedFieldException extends Exception { |
|||
|
|||
private final CalculatedFieldCtx ctx; |
|||
private final EntityId eventEntity; |
|||
private final UUID msgId; |
|||
private final TbMsgType msgType; |
|||
private Map<String, ArgumentEntry> arguments; |
|||
private String errorMessage; |
|||
private Exception cause; |
|||
|
|||
} |
|||
@ -0,0 +1,40 @@ |
|||
/** |
|||
* 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.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.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; |
|||
|
|||
@Data |
|||
public class CalculatedFieldLinkedTelemetryMsg implements ToCalculatedFieldSystemMsg { |
|||
|
|||
private final TenantId tenantId; |
|||
private final EntityId entityId; |
|||
private final CalculatedFieldLinkedTelemetryMsgProto proto; |
|||
private final TbCallback callback; |
|||
|
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.CF_LINKED_TELEMETRY_MSG; |
|||
} |
|||
} |
|||
@ -0,0 +1,90 @@ |
|||
/** |
|||
* 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.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.TbActorCtx; |
|||
import org.thingsboard.server.actors.TbActorException; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg; |
|||
import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg; |
|||
import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg; |
|||
import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg; |
|||
import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg; |
|||
|
|||
/** |
|||
* Created by ashvayka on 15.03.18. |
|||
*/ |
|||
@Slf4j |
|||
public class CalculatedFieldManagerActor extends AbstractCalculatedFieldActor { |
|||
|
|||
private final CalculatedFieldManagerMessageProcessor processor; |
|||
|
|||
public CalculatedFieldManagerActor(ActorSystemContext systemContext, TenantId tenantId) { |
|||
super(systemContext, tenantId); |
|||
this.processor = new CalculatedFieldManagerMessageProcessor(systemContext, tenantId); |
|||
} |
|||
|
|||
@Override |
|||
public void init(TbActorCtx ctx) throws TbActorException { |
|||
super.init(ctx); |
|||
log.debug("[{}] Starting CF manager actor.", processor.tenantId); |
|||
try { |
|||
processor.init(ctx); |
|||
log.debug("[{}] CF manager actor started.", processor.tenantId); |
|||
} catch (Exception e) { |
|||
log.warn("[{}] Unknown failure", processor.tenantId, e); |
|||
throw new TbActorException("Failed to initialize manager actor", e); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
protected boolean doProcessCfMsg(ToCalculatedFieldSystemMsg msg) throws CalculatedFieldException { |
|||
switch (msg.getMsgType()) { |
|||
case CF_PARTITIONS_CHANGE_MSG: |
|||
processor.onPartitionChange((CalculatedFieldPartitionChangeMsg) msg); |
|||
break; |
|||
case CF_INIT_MSG: |
|||
processor.onFieldInitMsg((CalculatedFieldInitMsg) msg); |
|||
break; |
|||
case CF_LINK_INIT_MSG: |
|||
processor.onLinkInitMsg((CalculatedFieldLinkInitMsg) msg); |
|||
break; |
|||
case CF_STATE_RESTORE_MSG: |
|||
processor.onStateRestoreMsg((CalculatedFieldStateRestoreMsg) msg); |
|||
break; |
|||
case CF_ENTITY_LIFECYCLE_MSG: |
|||
processor.onEntityLifecycleMsg((CalculatedFieldEntityLifecycleMsg) msg); |
|||
break; |
|||
case CF_TELEMETRY_MSG: |
|||
processor.onTelemetryMsg((CalculatedFieldTelemetryMsg) msg); |
|||
break; |
|||
case CF_LINKED_TELEMETRY_MSG: |
|||
processor.onLinkedTelemetryMsg((CalculatedFieldLinkedTelemetryMsg) msg); |
|||
break; |
|||
default: |
|||
return false; |
|||
} |
|||
return true; |
|||
} |
|||
|
|||
@Override |
|||
void logProcessingException(Exception e) { |
|||
log.warn("[{}] Processing failure", tenantId, e); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,46 @@ |
|||
/** |
|||
* 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 org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.actors.TbActor; |
|||
import org.thingsboard.server.actors.TbActorId; |
|||
import org.thingsboard.server.actors.TbEntityActorId; |
|||
import org.thingsboard.server.actors.TbStringActorId; |
|||
import org.thingsboard.server.actors.service.ContextBasedCreator; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
public class CalculatedFieldManagerActorCreator extends ContextBasedCreator { |
|||
|
|||
private final TenantId tenantId; |
|||
|
|||
public CalculatedFieldManagerActorCreator(ActorSystemContext context, TenantId tenantId) { |
|||
super(context); |
|||
this.tenantId = tenantId; |
|||
} |
|||
|
|||
@Override |
|||
public TbActorId createActorId() { |
|||
return new TbStringActorId("CFM|" + tenantId); |
|||
} |
|||
|
|||
@Override |
|||
public TbActor createActor() { |
|||
return new CalculatedFieldManagerActor(context, tenantId); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,471 @@ |
|||
/** |
|||
* 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.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.TbActorRef; |
|||
import org.thingsboard.server.actors.TbCalculatedFieldEntityActorId; |
|||
import org.thingsboard.server.actors.service.DefaultActorService; |
|||
import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.cf.CalculatedField; |
|||
import org.thingsboard.server.common.data.cf.CalculatedFieldLink; |
|||
import org.thingsboard.server.common.data.id.AssetId; |
|||
import org.thingsboard.server.common.data.id.CalculatedFieldId; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg; |
|||
import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg; |
|||
import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg; |
|||
import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg; |
|||
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; |
|||
import org.thingsboard.server.common.msg.queue.TbCallback; |
|||
import org.thingsboard.server.dao.cf.CalculatedFieldService; |
|||
import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; |
|||
import org.thingsboard.server.service.cf.CalculatedFieldStateService; |
|||
import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; |
|||
import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; |
|||
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; |
|||
import org.thingsboard.server.service.profile.TbAssetProfileCache; |
|||
import org.thingsboard.server.service.profile.TbDeviceProfileCache; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Collections; |
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.CopyOnWriteArrayList; |
|||
|
|||
import static org.thingsboard.server.utils.CalculatedFieldUtils.fromProto; |
|||
|
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Slf4j |
|||
public class CalculatedFieldManagerMessageProcessor extends AbstractContextAwareMsgProcessor { |
|||
|
|||
private final Map<CalculatedFieldId, CalculatedFieldCtx> calculatedFields = new HashMap<>(); |
|||
private final Map<EntityId, List<CalculatedFieldCtx>> entityIdCalculatedFields = new HashMap<>(); |
|||
private final Map<EntityId, List<CalculatedFieldLink>> entityIdCalculatedFieldLinks = new HashMap<>(); |
|||
|
|||
private final CalculatedFieldProcessingService cfExecService; |
|||
private final CalculatedFieldStateService cfStateService; |
|||
private final CalculatedFieldEntityProfileCache cfEntityCache; |
|||
private final CalculatedFieldService cfDaoService; |
|||
private final TbAssetProfileCache assetProfileCache; |
|||
private final TbDeviceProfileCache deviceProfileCache; |
|||
protected final TenantId tenantId; |
|||
|
|||
protected TbActorCtx ctx; |
|||
|
|||
CalculatedFieldManagerMessageProcessor(ActorSystemContext systemContext, TenantId tenantId) { |
|||
super(systemContext); |
|||
this.cfEntityCache = systemContext.getCalculatedFieldEntityProfileCache(); |
|||
this.cfExecService = systemContext.getCalculatedFieldProcessingService(); |
|||
this.cfStateService = systemContext.getCalculatedFieldStateService(); |
|||
this.cfDaoService = systemContext.getCalculatedFieldService(); |
|||
this.assetProfileCache = systemContext.getAssetProfileCache(); |
|||
this.deviceProfileCache = systemContext.getDeviceProfileCache(); |
|||
this.tenantId = tenantId; |
|||
} |
|||
|
|||
void init(TbActorCtx ctx) { |
|||
this.ctx = ctx; |
|||
} |
|||
|
|||
public void onFieldInitMsg(CalculatedFieldInitMsg msg) throws CalculatedFieldException { |
|||
log.info("[{}] Processing CF init message.", msg.getCf().getId()); |
|||
var cf = msg.getCf(); |
|||
var cfCtx = new CalculatedFieldCtx(cf, systemContext.getTbelInvokeService(), systemContext.getApiLimitService()); |
|||
try { |
|||
cfCtx.init(); |
|||
} catch (Exception e) { |
|||
throw CalculatedFieldException.builder().ctx(cfCtx).eventEntity(cf.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build(); |
|||
} |
|||
calculatedFields.put(cf.getId(), cfCtx); |
|||
// We use copy on write lists to safely pass the reference to another actor for the iteration.
|
|||
// Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead)
|
|||
entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cfCtx); |
|||
msg.getCallback().onSuccess(); |
|||
} |
|||
|
|||
public void onLinkInitMsg(CalculatedFieldLinkInitMsg msg) { |
|||
log.info("[{}] Processing CF link init message for entity [{}].", msg.getLink().getCalculatedFieldId(), msg.getLink().getEntityId()); |
|||
var link = msg.getLink(); |
|||
// We use copy on write lists to safely pass the reference to another actor for the iteration.
|
|||
// Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead)
|
|||
entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(link); |
|||
msg.getCallback().onSuccess(); |
|||
} |
|||
|
|||
public void onStateRestoreMsg(CalculatedFieldStateRestoreMsg msg) { |
|||
var cfId = msg.getId().cfId(); |
|||
var calculatedField = calculatedFields.get(cfId); |
|||
|
|||
if (calculatedField != null) { |
|||
msg.getState().setRequiredArguments(calculatedField.getArgNames()); |
|||
log.info("Pushing CF state restore msg to specific actor [{}]", msg.getId().entityId()); |
|||
getOrCreateActor(msg.getId().entityId()).tell(msg); |
|||
} else { |
|||
cfStateService.removeState(msg.getId(), msg.getCallback()); |
|||
} |
|||
} |
|||
|
|||
public void onEntityLifecycleMsg(CalculatedFieldEntityLifecycleMsg msg) throws CalculatedFieldException { |
|||
log.info("Processing entity lifecycle event: [{}] for entity: [{}]", msg.getData().getEvent(), msg.getData().getEntityId()); |
|||
var entityType = msg.getData().getEntityId().getEntityType(); |
|||
var event = msg.getData().getEvent(); |
|||
switch (entityType) { |
|||
case CALCULATED_FIELD: { |
|||
switch (event) { |
|||
case CREATED: |
|||
onCfCreated(msg.getData(), msg.getCallback()); |
|||
break; |
|||
case UPDATED: |
|||
onCfUpdated(msg.getData(), msg.getCallback()); |
|||
break; |
|||
case DELETED: |
|||
onCfDeleted(msg.getData(), msg.getCallback()); |
|||
break; |
|||
default: |
|||
msg.getCallback().onSuccess(); |
|||
break; |
|||
} |
|||
break; |
|||
} |
|||
case DEVICE: |
|||
case ASSET: { |
|||
switch (event) { |
|||
case CREATED: |
|||
onEntityCreated(msg.getData(), msg.getCallback()); |
|||
break; |
|||
case UPDATED: |
|||
onEntityUpdated(msg.getData(), msg.getCallback()); |
|||
break; |
|||
case DELETED: |
|||
onEntityDeleted(msg.getData(), msg.getCallback()); |
|||
break; |
|||
default: |
|||
msg.getCallback().onSuccess(); |
|||
break; |
|||
} |
|||
break; |
|||
} |
|||
default: { |
|||
msg.getCallback().onSuccess(); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void onEntityCreated(ComponentLifecycleMsg msg, TbCallback callback) { |
|||
EntityId entityId = msg.getEntityId(); |
|||
EntityId profileId = getProfileId(tenantId, entityId); |
|||
cfEntityCache.add(tenantId, profileId, entityId); |
|||
var entityIdFields = getCalculatedFieldsByEntityId(entityId); |
|||
var profileIdFields = getCalculatedFieldsByEntityId(profileId); |
|||
var fieldsCount = entityIdFields.size() + profileIdFields.size(); |
|||
if (fieldsCount > 0) { |
|||
MultipleTbCallback multiCallback = new MultipleTbCallback(fieldsCount, callback); |
|||
entityIdFields.forEach(ctx -> initCfForEntity(entityId, ctx, true, multiCallback)); |
|||
profileIdFields.forEach(ctx -> initCfForEntity(entityId, ctx, true, multiCallback)); |
|||
} else { |
|||
callback.onSuccess(); |
|||
} |
|||
} |
|||
|
|||
private void onEntityUpdated(ComponentLifecycleMsg msg, TbCallback callback) { |
|||
if (msg.getOldProfileId() != null && msg.getOldProfileId() != msg.getProfileId()) { |
|||
cfEntityCache.update(tenantId, msg.getOldProfileId(), msg.getProfileId(), msg.getEntityId()); |
|||
var oldProfileCfs = getCalculatedFieldsByEntityId(msg.getOldProfileId()); |
|||
var newProfileCfs = getCalculatedFieldsByEntityId(msg.getProfileId()); |
|||
var fieldsCount = oldProfileCfs.size() + newProfileCfs.size(); |
|||
if (fieldsCount > 0) { |
|||
MultipleTbCallback multiCallback = new MultipleTbCallback(fieldsCount, callback); |
|||
var entityId = msg.getEntityId(); |
|||
oldProfileCfs.forEach(ctx -> deleteCfForEntity(entityId, ctx.getCfId(), callback)); |
|||
newProfileCfs.forEach(ctx -> initCfForEntity(entityId, ctx, true, multiCallback)); |
|||
} else { |
|||
callback.onSuccess(); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) { |
|||
cfEntityCache.evict(tenantId, msg.getEntityId()); |
|||
log.info("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId()); |
|||
getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); |
|||
} |
|||
|
|||
private void onCfCreated(ComponentLifecycleMsg msg, TbCallback callback) throws CalculatedFieldException { |
|||
var cfId = new CalculatedFieldId(msg.getEntityId().getId()); |
|||
if (calculatedFields.containsKey(cfId)) { |
|||
log.warn("[{}] CF was already initialized [{}]", tenantId, cfId); |
|||
callback.onSuccess(); |
|||
} else { |
|||
var cf = cfDaoService.findById(msg.getTenantId(), cfId); |
|||
if (cf == null) { |
|||
log.warn("[{}] Failed to lookup CF by id [{}]", tenantId, cfId); |
|||
callback.onSuccess(); |
|||
} else { |
|||
var cfCtx = new CalculatedFieldCtx(cf, systemContext.getTbelInvokeService(), systemContext.getApiLimitService()); |
|||
try { |
|||
cfCtx.init(); |
|||
} catch (Exception e) { |
|||
throw CalculatedFieldException.builder().ctx(cfCtx).eventEntity(cf.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build(); |
|||
} |
|||
calculatedFields.put(cf.getId(), cfCtx); |
|||
// We use copy on write lists to safely pass the reference to another actor for the iteration.
|
|||
// Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead)
|
|||
entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cfCtx); |
|||
addLinks(cf); |
|||
initCf(cfCtx, callback, false); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void onCfUpdated(ComponentLifecycleMsg msg, TbCallback callback) throws CalculatedFieldException { |
|||
var cfId = new CalculatedFieldId(msg.getEntityId().getId()); |
|||
var oldCfCtx = calculatedFields.get(cfId); |
|||
if (oldCfCtx == null) { |
|||
onCfCreated(msg, callback); |
|||
} else { |
|||
var newCf = cfDaoService.findById(msg.getTenantId(), cfId); |
|||
if (newCf == null) { |
|||
log.warn("[{}] Failed to lookup CF by id [{}]", tenantId, cfId); |
|||
callback.onSuccess(); |
|||
} else { |
|||
var newCfCtx = new CalculatedFieldCtx(newCf, systemContext.getTbelInvokeService(), systemContext.getApiLimitService()); |
|||
try { |
|||
newCfCtx.init(); |
|||
} catch (Exception e) { |
|||
throw CalculatedFieldException.builder().ctx(newCfCtx).eventEntity(newCfCtx.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build(); |
|||
} |
|||
calculatedFields.put(newCf.getId(), newCfCtx); |
|||
List<CalculatedFieldCtx> oldCfList = entityIdCalculatedFields.get(newCf.getEntityId()); |
|||
List<CalculatedFieldCtx> newCfList = new CopyOnWriteArrayList<>(); |
|||
boolean found = false; |
|||
for (CalculatedFieldCtx oldCtx : oldCfList) { |
|||
if (oldCtx.getCfId().equals(newCf.getId())) { |
|||
newCfList.add(newCfCtx); |
|||
found = true; |
|||
} else { |
|||
newCfList.add(oldCtx); |
|||
} |
|||
} |
|||
if (!found) { |
|||
newCfList.add(newCfCtx); |
|||
} |
|||
entityIdCalculatedFields.put(newCf.getEntityId(), newCfList); |
|||
|
|||
deleteLinks(oldCfCtx); |
|||
addLinks(newCf); |
|||
|
|||
// We use copy on write lists to safely pass the reference to another actor for the iteration.
|
|||
// Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead)
|
|||
var stateChanges = newCfCtx.hasStateChanges(oldCfCtx); |
|||
if (stateChanges || newCfCtx.hasOtherSignificantChanges(oldCfCtx)) { |
|||
initCf(newCfCtx, callback, stateChanges); |
|||
} else { |
|||
callback.onSuccess(); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void onCfDeleted(ComponentLifecycleMsg msg, TbCallback callback) { |
|||
var cfId = new CalculatedFieldId(msg.getEntityId().getId()); |
|||
var cfCtx = calculatedFields.remove(cfId); |
|||
if (cfCtx == null) { |
|||
log.warn("[{}] CF was already deleted [{}]", tenantId, cfId); |
|||
callback.onSuccess(); |
|||
} else { |
|||
entityIdCalculatedFields.get(cfCtx.getEntityId()).remove(cfCtx); |
|||
deleteLinks(cfCtx); |
|||
|
|||
EntityId entityId = cfCtx.getEntityId(); |
|||
EntityType entityType = cfCtx.getEntityId().getEntityType(); |
|||
if (isProfileEntity(entityType)) { |
|||
var entityIds = cfEntityCache.getMyEntityIdsByProfileId(tenantId, entityId); |
|||
if (!entityIds.isEmpty()) { |
|||
//TODO: no need to do this if we cache all created actors and know which one belong to us;
|
|||
var multiCallback = new MultipleTbCallback(entityIds.size(), callback); |
|||
entityIds.forEach(id -> deleteCfForEntity(entityId, cfId, multiCallback)); |
|||
} else { |
|||
callback.onSuccess(); |
|||
} |
|||
} else { |
|||
deleteCfForEntity(entityId, cfId, callback); |
|||
} |
|||
} |
|||
} |
|||
|
|||
public void onTelemetryMsg(CalculatedFieldTelemetryMsg msg) { |
|||
EntityId entityId = msg.getEntityId(); |
|||
log.info("Received telemetry msg from entity [{}]", entityId); |
|||
// 2 = 1 for CF processing + 1 for links processing
|
|||
MultipleTbCallback callback = new MultipleTbCallback(2, msg.getCallback()); |
|||
// process all cfs related to entity, or it's profile;
|
|||
var entityIdFields = getCalculatedFieldsByEntityId(entityId); |
|||
var profileIdFields = getCalculatedFieldsByEntityId(getProfileId(tenantId, entityId)); |
|||
if (!entityIdFields.isEmpty() || !profileIdFields.isEmpty()) { |
|||
log.info("Pushing telemetry msg to specific actor [{}]", entityId); |
|||
getOrCreateActor(entityId).tell(new EntityCalculatedFieldTelemetryMsg(msg, entityIdFields, profileIdFields, callback)); |
|||
} else { |
|||
callback.onSuccess(); |
|||
} |
|||
// process all links (if any);
|
|||
List<CalculatedFieldEntityCtxId> linkedCalculatedFields = filterCalculatedFieldLinks(msg); |
|||
var linksSize = linkedCalculatedFields.size(); |
|||
if (linksSize > 0) { |
|||
cfExecService.pushMsgToLinks(msg, linkedCalculatedFields, callback); |
|||
} else { |
|||
callback.onSuccess(); |
|||
} |
|||
} |
|||
|
|||
public void onLinkedTelemetryMsg(CalculatedFieldLinkedTelemetryMsg msg) { |
|||
EntityId sourceEntityId = msg.getEntityId(); |
|||
log.info("Received linked telemetry msg from entity [{}]", sourceEntityId); |
|||
var proto = msg.getProto(); |
|||
var linksList = proto.getLinksList(); |
|||
for (var linkProto : linksList) { |
|||
var link = fromProto(linkProto); |
|||
var targetEntityId = link.entityId(); |
|||
var targetEntityType = targetEntityId.getEntityType(); |
|||
var cf = calculatedFields.get(link.cfId()); |
|||
if (EntityType.DEVICE_PROFILE.equals(targetEntityType) || EntityType.ASSET_PROFILE.equals(targetEntityType)) { |
|||
// iterate over all entities that belong to profile and push the message for corresponding CF
|
|||
var entityIds = cfEntityCache.getMyEntityIdsByProfileId(tenantId, targetEntityId); |
|||
if (!entityIds.isEmpty()) { |
|||
MultipleTbCallback callback = new MultipleTbCallback(entityIds.size(), msg.getCallback()); |
|||
var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, callback); |
|||
entityIds.forEach(entityId -> { |
|||
log.info("Pushing linked telemetry msg to specific actor [{}]", entityId); |
|||
getOrCreateActor(entityId).tell(newMsg); |
|||
}); |
|||
} else { |
|||
msg.getCallback().onSuccess(); |
|||
} |
|||
} else { |
|||
log.info("Pushing linked telemetry msg to specific actor [{}]", targetEntityId); |
|||
var newMsg = new EntityCalculatedFieldLinkedTelemetryMsg(tenantId, sourceEntityId, proto.getMsg(), cf, msg.getCallback()); |
|||
getOrCreateActor(targetEntityId).tell(newMsg); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private List<CalculatedFieldEntityCtxId> filterCalculatedFieldLinks(CalculatedFieldTelemetryMsg msg) { |
|||
EntityId entityId = msg.getEntityId(); |
|||
var proto = msg.getProto(); |
|||
List<CalculatedFieldEntityCtxId> result = new ArrayList<>(); |
|||
for (var link : getCalculatedFieldLinksByEntityId(entityId)) { |
|||
CalculatedFieldCtx ctx = calculatedFields.get(link.getCalculatedFieldId()); |
|||
if (ctx.linkMatches(entityId, proto)) { |
|||
result.add(ctx.toCalculatedFieldEntityCtxId()); |
|||
} |
|||
} |
|||
return result; |
|||
} |
|||
|
|||
private List<CalculatedFieldCtx> getCalculatedFieldsByEntityId(EntityId entityId) { |
|||
if (entityId == null) { |
|||
return Collections.emptyList(); |
|||
} |
|||
var result = entityIdCalculatedFields.get(entityId); |
|||
if (result == null) { |
|||
result = Collections.emptyList(); |
|||
} |
|||
return result; |
|||
} |
|||
|
|||
private List<CalculatedFieldLink> getCalculatedFieldLinksByEntityId(EntityId entityId) { |
|||
if (entityId == null) { |
|||
return Collections.emptyList(); |
|||
} |
|||
var result = entityIdCalculatedFieldLinks.get(entityId); |
|||
if (result == null) { |
|||
result = Collections.emptyList(); |
|||
} |
|||
return result; |
|||
} |
|||
|
|||
private void initCf(CalculatedFieldCtx cfCtx, TbCallback callback, boolean forceStateReinit) { |
|||
EntityId entityId = cfCtx.getEntityId(); |
|||
EntityType entityType = cfCtx.getEntityId().getEntityType(); |
|||
if (isProfileEntity(entityType)) { |
|||
var entityIds = cfEntityCache.getMyEntityIdsByProfileId(tenantId, entityId); |
|||
if (!entityIds.isEmpty()) { |
|||
var multiCallback = new MultipleTbCallback(entityIds.size(), callback); |
|||
entityIds.forEach(id -> initCfForEntity(id, cfCtx, forceStateReinit, multiCallback)); |
|||
} else { |
|||
callback.onSuccess(); |
|||
} |
|||
} else { |
|||
initCfForEntity(entityId, cfCtx, forceStateReinit, callback); |
|||
} |
|||
} |
|||
|
|||
private void deleteCfForEntity(EntityId entityId, CalculatedFieldId cfId, TbCallback callback) { |
|||
log.info("Pushing delete CF msg to specific actor [{}]", entityId); |
|||
getOrCreateActor(entityId).tell(new CalculatedFieldEntityDeleteMsg(tenantId, cfId, callback)); |
|||
} |
|||
|
|||
private void initCfForEntity(EntityId entityId, CalculatedFieldCtx cfCtx, boolean forceStateReinit, TbCallback callback) { |
|||
log.info("Pushing entity init CF msg to specific actor [{}]", entityId); |
|||
getOrCreateActor(entityId).tell(new EntityInitCalculatedFieldMsg(tenantId, cfCtx, callback, forceStateReinit)); |
|||
} |
|||
|
|||
private static boolean isProfileEntity(EntityType entityType) { |
|||
return EntityType.DEVICE_PROFILE.equals(entityType) || EntityType.ASSET_PROFILE.equals(entityType); |
|||
} |
|||
|
|||
private EntityId getProfileId(TenantId tenantId, EntityId entityId) { |
|||
return switch (entityId.getEntityType()) { |
|||
case ASSET -> assetProfileCache.get(tenantId, (AssetId) entityId).getId(); |
|||
case DEVICE -> deviceProfileCache.get(tenantId, (DeviceId) entityId).getId(); |
|||
default -> null; |
|||
}; |
|||
} |
|||
|
|||
private TbActorRef getOrCreateActor(EntityId entityId) { |
|||
return ctx.getOrCreateChildActor(new TbCalculatedFieldEntityActorId(entityId), |
|||
() -> DefaultActorService.CF_ENTITY_DISPATCHER_NAME, |
|||
() -> new CalculatedFieldEntityActorCreator(systemContext, tenantId, entityId), |
|||
() -> true); |
|||
} |
|||
|
|||
private void addLinks(CalculatedField newCf) { |
|||
var newLinks = newCf.getConfiguration().buildCalculatedFieldLinks(tenantId, newCf.getEntityId(), newCf.getId()); |
|||
newLinks.forEach(link -> entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(link)); |
|||
} |
|||
|
|||
private void deleteLinks(CalculatedFieldCtx cfCtx) { |
|||
var oldCf = cfCtx.getCalculatedField(); |
|||
var oldLinks = oldCf.getConfiguration().buildCalculatedFieldLinks(tenantId, oldCf.getEntityId(), oldCf.getId()); |
|||
oldLinks.forEach(link -> entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new CopyOnWriteArrayList<>()).remove(link)); |
|||
} |
|||
|
|||
public void onPartitionChange(CalculatedFieldPartitionChangeMsg msg) { |
|||
ctx.broadcastToChildren(msg, true); |
|||
} |
|||
} |
|||
@ -0,0 +1,40 @@ |
|||
/** |
|||
* 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.service.cf.ctx.CalculatedFieldEntityCtxId; |
|||
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; |
|||
|
|||
@Data |
|||
public class CalculatedFieldStateRestoreMsg implements ToCalculatedFieldSystemMsg { |
|||
|
|||
private final CalculatedFieldEntityCtxId id; |
|||
private final CalculatedFieldState state; |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.CF_STATE_RESTORE_MSG; |
|||
} |
|||
|
|||
@Override |
|||
public TenantId getTenantId() { |
|||
return id.tenantId(); |
|||
} |
|||
} |
|||
@ -0,0 +1,39 @@ |
|||
/** |
|||
* 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.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.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; |
|||
|
|||
@Data |
|||
public class CalculatedFieldTelemetryMsg implements ToCalculatedFieldSystemMsg { |
|||
|
|||
private final TenantId tenantId; |
|||
private final EntityId entityId; |
|||
private final CalculatedFieldTelemetryMsgProto proto; |
|||
private final TbCallback callback; |
|||
|
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.CF_TELEMETRY_MSG; |
|||
} |
|||
} |
|||
@ -0,0 +1,42 @@ |
|||
/** |
|||
* 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.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.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; |
|||
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Data |
|||
public class EntityCalculatedFieldLinkedTelemetryMsg implements ToCalculatedFieldSystemMsg { |
|||
|
|||
private final TenantId tenantId; |
|||
private final EntityId entityId; |
|||
private final CalculatedFieldTelemetryMsgProto proto; |
|||
private final CalculatedFieldCtx ctx; |
|||
private final TbCallback callback; |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.CF_LINKED_TELEMETRY_MSG; |
|||
} |
|||
} |
|||
@ -0,0 +1,56 @@ |
|||
/** |
|||
* 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.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.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; |
|||
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Data |
|||
public class EntityCalculatedFieldTelemetryMsg implements ToCalculatedFieldSystemMsg { |
|||
|
|||
private final TenantId tenantId; |
|||
private final EntityId entityId; |
|||
private final CalculatedFieldTelemetryMsgProto proto; |
|||
// Both lists are effectively immutable in CalculatedFieldManagerMessageProcessor and must stay so.
|
|||
private final List<CalculatedFieldCtx> entityIdFields; |
|||
private final List<CalculatedFieldCtx> profileIdFields; |
|||
private final TbCallback callback; |
|||
|
|||
public EntityCalculatedFieldTelemetryMsg(CalculatedFieldTelemetryMsg msg, |
|||
List<CalculatedFieldCtx> entityIdFields, |
|||
List<CalculatedFieldCtx> profileIdFields, |
|||
TbCallback callback) { |
|||
this.tenantId = msg.getTenantId(); |
|||
this.entityId = msg.getEntityId(); |
|||
this.proto = msg.getProto(); |
|||
this.entityIdFields = entityIdFields; |
|||
this.profileIdFields = profileIdFields; |
|||
this.callback = callback; |
|||
} |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.CF_ENTITY_TELEMETRY_MSG; |
|||
} |
|||
} |
|||
@ -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.Data; |
|||
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.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; |
|||
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Data |
|||
public class EntityInitCalculatedFieldMsg implements ToCalculatedFieldSystemMsg { |
|||
|
|||
private final TenantId tenantId; |
|||
private final CalculatedFieldCtx ctx; |
|||
private final TbCallback callback; |
|||
private final boolean forceReinit; |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.CF_ENTITY_INIT_CF_MSG; |
|||
} |
|||
} |
|||
@ -0,0 +1,56 @@ |
|||
/** |
|||
* 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.Getter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.common.msg.queue.TbCallback; |
|||
|
|||
import java.util.UUID; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
@Slf4j |
|||
public class MultipleTbCallback implements TbCallback { |
|||
@Getter |
|||
private final UUID id; |
|||
private final AtomicInteger counter; |
|||
private final TbCallback callback; |
|||
|
|||
public MultipleTbCallback(int count, TbCallback callback) { |
|||
id = UUID.randomUUID(); |
|||
this.counter = new AtomicInteger(count); |
|||
this.callback = callback; |
|||
} |
|||
|
|||
@Override |
|||
public void onSuccess() { |
|||
onSuccess(1); |
|||
} |
|||
|
|||
public void onSuccess(int number) { |
|||
log.trace("[{}][{}] onSuccess({})", id, callback.getId(), number); |
|||
if (counter.addAndGet(-number) <= 0) { |
|||
log.trace("[{}][{}] Done.", id, callback.getId()); |
|||
callback.onSuccess(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
log.warn("[{}][{}] onFailure.", id, callback.getId()); |
|||
callback.onFailure(t); |
|||
} |
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue