From ae33b5bbfeb130f7b5b559f32a3e11935d5d12e5 Mon Sep 17 00:00:00 2001 From: nickAS21 Date: Fri, 12 Dec 2025 18:23:57 +0200 Subject: [PATCH 1/8] lwm2m: fix bug? delete fluky tests 19_65534/65534 --- .../ota/AbstractOtaLwM2MIntegrationTest.java | 23 --------- .../ota/sql/Ota5LwM2MIntegrationTest.java | 47 ------------------- .../ota/sql/Ota9LwM2MIntegrationTest.java | 46 +----------------- 3 files changed, 2 insertions(+), 114 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/AbstractOtaLwM2MIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/AbstractOtaLwM2MIntegrationTest.java index cff56694e5..d5af0a1857 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/AbstractOtaLwM2MIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/AbstractOtaLwM2MIntegrationTest.java @@ -247,27 +247,4 @@ public abstract class AbstractOtaLwM2MIntegrationTest extends AbstractLwM2MInteg log.warn("{}", statuses); return statuses.containsAll(expectedStatuses); } - - protected void resultReadOtaParams_19(String resourceIdVer, OtaPackageInfo otaPackageInfo) throws Exception { - String actualResult = sendRPCById(resourceIdVer); - ObjectNode rpcActualResult = JacksonUtil.fromString(actualResult, ObjectNode.class); - assertEquals(ResponseCode.CONTENT.getName(), rpcActualResult.get("result").asText()); - String valStr = rpcActualResult.get("value").asText(); - String start = "{ id=0 value="; - String valHexDec = valStr.substring(valStr.indexOf(start) + start.length(), (valStr.indexOf("}"))); - String valNode = new String(Hex.decodeHex((valHexDec).toCharArray())); - ObjectNode actualResultVal = JacksonUtil.fromString(valNode, ObjectNode.class); - assert actualResultVal != null; - assertEquals(otaPackageInfo.getTitle(), actualResultVal.get(OTA_INFO_19_TITLE).asText()); - assertEquals(otaPackageInfo.getVersion(), actualResultVal.get(OTA_INFO_19_VERSION).asText()); - assertEquals(otaPackageInfo.getChecksum(), actualResultVal.get(OTA_INFO_19_FILE_CHECKSUM256).asText()); - assertEquals(otaPackageInfo.getFileName(), actualResultVal.get(OTA_INFO_19_FILE_NAME).asText()); - assertEquals(Optional.of(otaPackageInfo.getDataSize()), Optional.of((long) actualResultVal.get(OTA_INFO_19_FILE_SIZE).asInt())); - } - - private String sendRPCById(String path) throws Exception { - String setRpcRequest = "{\"method\": \"Read\", \"params\": {\"id\": \"" + path + "\"}}"; - return doPostAsync("/api/plugins/rpc/twoway/" + lwM2MTestClient.getDeviceIdStr(), setRpcRequest, String.class, status().isOk()); - } - } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota5LwM2MIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota5LwM2MIntegrationTest.java index e4db6f70bf..6e272a4f64 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota5LwM2MIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota5LwM2MIntegrationTest.java @@ -111,51 +111,4 @@ public class Ota5LwM2MIntegrationTest extends AbstractOtaLwM2MIntegrationTest { .until(() -> getFwSwStateTelemetryFromAPI(device.getId().getId(), "fw_state"), this::predicateForStatuses); log.warn("Object5: Got the ts: {}", ts); } - - /** - * ObjectId = 19/65533/0 - * { - * "title" : "My firmware", - * "version" : "fw.v.1.5.0-update", - * "checksum" : "4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a", - * "fileSize" : 1, - * "fileName" : "filename.txt" - * } - * to base64 - * /5/0/5 -> Update Result (Res); 5/0/3 -> State; - * => ((Res>=0 && Res<=9) && State=0) - * => Write to Package/Write to Package URI -> DOWNLOADING ((Res>=0 && Res<=9) && State=1) - * => Download Finished -> DOWNLOADED ((Res==0 || Res=8) && State=2) - * => Executable resource Update is triggered / Initiate Firmware Update -> UPDATING (Res=0 && State=3) - * => Update Successful [Res==1] - * => Start / Res=0 -> "IDLE" .... - * @throws Exception - */ - @Test - public void testFirmwareUpdateByObject5WithObject19_Ok() throws Exception { - Lwm2mDeviceProfileTransportConfiguration transportConfiguration = getTransportConfiguration19(OBSERVE_ATTRIBUTES_WITH_PARAMS_OTA5_19, getBootstrapServerCredentialsNoSec(NONE)); - DeviceProfile deviceProfile = createLwm2mDeviceProfile("profileFor" + this.CLIENT_ENDPOINT_OTA5 + "19_Ok", transportConfiguration); - String endpoint = this.CLIENT_ENDPOINT_OTA5 + "19_Ok"; - LwM2MDeviceCredentials deviceCredentials = getDeviceCredentialsNoSec(createNoSecClientCredentials(endpoint)); - final Device device = createLwm2mDevice(deviceCredentials, endpoint, deviceProfile.getId()); - createNewClient(SECURITY_NO_SEC, null, false, endpoint, device.getId().getId().toString()); - awaitObserveReadAll(6, device.getId().getId().toString()); - - OtaPackageInfo otaPackageInfo = createFirmware(TARGET_FW_VERSION, deviceProfile.getId()); - device.setFirmwareId(otaPackageInfo.getId()); - final Device savedDevice = doPost("/api/device", device, Device.class); - - assertThat(savedDevice).as("saved device").isNotNull(); - assertThat(getDeviceFromAPI(device.getId().getId())).as("fetched device").isEqualTo(savedDevice); - - expectedStatuses = Arrays.asList(QUEUED, INITIATED, DOWNLOADING, DOWNLOADED, UPDATING, UPDATED); - List ts = await("await on timeseries for FW") - .atMost(TIMEOUT, TimeUnit.SECONDS) - .until(() -> getFwSwStateTelemetryFromAPI(device.getId().getId(), "fw_state"), this::predicateForStatuses); - - String ver_Id_19 = lwM2MTestClient.getLeshanClient().getObjectTree().getModel().getObjectModel(BINARY_APP_DATA_CONTAINER).version; - String resourceIdVer = "/" + BINARY_APP_DATA_CONTAINER + "_" + ver_Id_19 + "/" + FW_INSTANCE_ID + "/" + RESOURCE_ID_0; - resultReadOtaParams_19(resourceIdVer, otaPackageInfo); - log.warn("Object5 with Object19: Got the ts: {}", ts); - } } diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota9LwM2MIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota9LwM2MIntegrationTest.java index b2617dd7ba..a091821444 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota9LwM2MIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota9LwM2MIntegrationTest.java @@ -51,7 +51,8 @@ public class Ota9LwM2MIntegrationTest extends AbstractOtaLwM2MIntegrationTest { * => PKG integrity verified -> DELIVERED (Res=3 (Successfully Downloaded and package integrity verified) && State=3) -> INSTALLED; * => Install -> INSTALLED (Res=2 SW successfully installed) && State=4) -> Start * - * */ + * + */ @Test public void testSoftwareUpdateByObject9() throws Exception { String clientEndpoint = this.CLIENT_ENDPOINT_OTA9; @@ -75,47 +76,4 @@ public class Ota9LwM2MIntegrationTest extends AbstractOtaLwM2MIntegrationTest { .until(() -> getFwSwStateTelemetryFromAPI(device.getId().getId(), "sw_state"), this::predicateForStatuses); log.warn("Object9: Got the ts: {}", ts); } - /** - * ObjectId = 19/65534/0 - * { - * "title" : "My sw", - * "version" : "v1.0.19", - * "checksum" : "4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a", - * "fileSize" : 1, - * "fileName" : "filename.txt" - * } - * => Start -> INITIAL (State=0) -> DOWNLOAD STARTED; - * => PKG / URI Write -> DOWNLOAD STARTED (Res=1 (Downloading) && State=1) -> DOWNLOADED - * => PKG Written -> DOWNLOADED (Res=1 Initial && State=2) -> DELIVERED; - * => PKG integrity verified -> DELIVERED (Res=3 (Successfully Downloaded and package integrity verified) && State=3) -> INSTALLED; - * => Install -> INSTALLED (Res=2 SW successfully installed) && State=4) -> Start - * - * */ - @Test - public void testSoftwareUpdateByObject9WithObject19_Ok() throws Exception { - String clientEndpoint = this.CLIENT_ENDPOINT_OTA9_19; - Lwm2mDeviceProfileTransportConfiguration transportConfiguration = getTransportConfiguration19(OBSERVE_ATTRIBUTES_WITH_PARAMS_OTA9_19, getBootstrapServerCredentialsNoSec(NONE)); - DeviceProfile deviceProfile = createLwm2mDeviceProfile("profileFor" + clientEndpoint, transportConfiguration); - LwM2MDeviceCredentials deviceCredentials = getDeviceCredentialsNoSec(createNoSecClientCredentials(clientEndpoint)); - final Device device = createLwm2mDevice(deviceCredentials, clientEndpoint, deviceProfile.getId()); - createNewClient(SECURITY_NO_SEC, null, false, clientEndpoint, device.getId().getId().toString()); - awaitObserveReadAll(5, device.getId().getId().toString()); - OtaPackageInfo otaPackageInfo = createSoftware(deviceProfile.getId(), "v1.0.19"); - device.setSoftwareId(otaPackageInfo.getId()); - final Device savedDevice = doPost("/api/device", device, Device.class); //sync call - - assertThat(savedDevice).as("saved device").isNotNull(); - assertThat(getDeviceFromAPI(device.getId().getId())).as("fetched device").isEqualTo(savedDevice); - - expectedStatuses = List.of( - QUEUED, INITIATED, DOWNLOADING, DOWNLOADING, DOWNLOADING, DOWNLOADED, VERIFIED, UPDATED); - List ts = await("await on timeseries") - .atMost(TIMEOUT, TimeUnit.SECONDS) - .until(() -> getFwSwStateTelemetryFromAPI(device.getId().getId(), "sw_state"), this::predicateForStatuses); - - String ver_Id_19 = lwM2MTestClient.getLeshanClient().getObjectTree().getModel().getObjectModel(BINARY_APP_DATA_CONTAINER).version; - String resourceIdVer = "/" + BINARY_APP_DATA_CONTAINER + "_" + ver_Id_19 + "/" + SW_INSTANCE_ID + "/" + RESOURCE_ID_0; - resultReadOtaParams_19(resourceIdVer, otaPackageInfo); - log.warn("Object9: Got the ts: {}", ts); - } } From f0f15dbfe292cd03e937f84b69d57ffc8dbc9776 Mon Sep 17 00:00:00 2001 From: nickAS21 Date: Mon, 15 Dec 2025 11:10:48 +0200 Subject: [PATCH 2/8] lwm2m: fix bug? delete fluky tests 19_65534/65534 - refactoring --- .../ota/AbstractOtaLwM2MIntegrationTest.java | 43 ------------------- .../ota/sql/Ota5LwM2MIntegrationTest.java | 4 -- .../ota/sql/Ota9LwM2MIntegrationTest.java | 4 -- 3 files changed, 51 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/AbstractOtaLwM2MIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/AbstractOtaLwM2MIntegrationTest.java index d5af0a1857..0681f00d27 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/AbstractOtaLwM2MIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/AbstractOtaLwM2MIntegrationTest.java @@ -16,14 +16,10 @@ package org.thingsboard.server.transport.lwm2m.ota; import com.fasterxml.jackson.core.type.TypeReference; -import com.fasterxml.jackson.databind.node.ObjectNode; -import com.github.dockerjava.zerodep.shaded.org.apache.commons.codec.binary.Hex; import lombok.extern.slf4j.Slf4j; -import org.eclipse.leshan.core.ResponseCode; import org.springframework.mock.web.MockMultipartFile; import org.springframework.test.web.servlet.request.MockMultipartHttpServletRequestBuilder; import org.springframework.test.web.servlet.request.MockMvcRequestBuilders; -import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.OtaPackageInfo; import org.thingsboard.server.common.data.id.DeviceProfileId; @@ -35,20 +31,13 @@ import org.thingsboard.server.transport.lwm2m.AbstractLwM2MIntegrationTest; import java.util.Comparator; import java.util.List; -import java.util.Optional; import java.util.UUID; import java.util.stream.Collectors; -import static org.junit.Assert.assertEquals; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; import static org.thingsboard.rest.client.utils.RestJsonConverter.toTimeseries; import static org.thingsboard.server.common.data.ota.OtaPackageType.FIRMWARE; import static org.thingsboard.server.common.data.ota.OtaPackageType.SOFTWARE; -import static org.thingsboard.server.transport.lwm2m.server.ota.DefaultLwM2MOtaUpdateService.OTA_INFO_19_FILE_CHECKSUM256; -import static org.thingsboard.server.transport.lwm2m.server.ota.DefaultLwM2MOtaUpdateService.OTA_INFO_19_FILE_NAME; -import static org.thingsboard.server.transport.lwm2m.server.ota.DefaultLwM2MOtaUpdateService.OTA_INFO_19_FILE_SIZE; -import static org.thingsboard.server.transport.lwm2m.server.ota.DefaultLwM2MOtaUpdateService.OTA_INFO_19_TITLE; -import static org.thingsboard.server.transport.lwm2m.server.ota.DefaultLwM2MOtaUpdateService.OTA_INFO_19_VERSION; @Slf4j @DaoSqlTest @@ -57,7 +46,6 @@ public abstract class AbstractOtaLwM2MIntegrationTest extends AbstractLwM2MInteg protected static final String CLIENT_ENDPOINT_WITHOUT_FW_INFO = "WithoutFirmwareInfoDevice"; protected static final String CLIENT_ENDPOINT_OTA5 = "Ota5_Device"; protected static final String CLIENT_ENDPOINT_OTA9 = "Ota9_Device"; - protected static final String CLIENT_ENDPOINT_OTA9_19 = "Ota9_Device_19"; protected List expectedStatuses; protected final String OBSERVE_ATTRIBUTES_WITH_PARAMS_OTA5 = @@ -88,37 +76,6 @@ public abstract class AbstractOtaLwM2MIntegrationTest extends AbstractLwM2MInteg " \"attributeLwm2m\": {}\n" + " }"; - protected final String OBSERVE_ATTRIBUTES_WITH_PARAMS_OTA5_19 = - - " {\n" + - " \"keyName\": {\n" + - " \"/5_1.2/0/3\": \"state\",\n" + - " \"/5_1.2/0/5\": \"updateResult\",\n" + - " \"/5_1.2/0/6\": \"pkgname\",\n" + - " \"/5_1.2/0/7\": \"pkgversion\",\n" + - " \"/5_1.2/0/9\": \"firmwareUpdateDeliveryMethod\",\n" + - " \"/19_1.1/0/0\": \"dataRead\"\n" + - " },\n" + - " \"observe\": [\n" + - " \"/5_1.2/0/3\",\n" + - " \"/5_1.2/0/5\",\n" + - " \"/5_1.2/0/6\",\n" + - " \"/5_1.2/0/7\",\n" + - " \"/5_1.2/0/9\",\n" + - " \"/19_1.1/0/0\"\n" + - " ],\n" + - " \"attribute\": [],\n" + - " \"telemetry\": [\n" + - " \"/5_1.2/0/3\",\n" + - " \"/5_1.2/0/5\",\n" + - " \"/5_1.2/0/6\",\n" + - " \"/5_1.2/0/7\",\n" + - " \"/5_1.2/0/9\",\n" + - " \"/19_1.1/0/0\"\n" + - " ],\n" + - " \"attributeLwm2m\": {}\n" + - " }"; - public static final String CLIENT_LWM2M_SETTINGS_19 = " {\n" + " \"useObject19ForOtaInfo\": true,\n" + diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota5LwM2MIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota5LwM2MIntegrationTest.java index 6e272a4f64..0943491045 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota5LwM2MIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota5LwM2MIntegrationTest.java @@ -21,7 +21,6 @@ import org.junit.Assert; import org.junit.Test; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; -import org.thingsboard.server.common.data.OtaPackageInfo; import org.thingsboard.server.common.data.device.credentials.lwm2m.LwM2MDeviceCredentials; import org.thingsboard.server.common.data.device.profile.Lwm2mDeviceProfileTransportConfiguration; import org.thingsboard.server.common.data.kv.KvEntry; @@ -46,10 +45,7 @@ import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.QUEU import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.UPDATED; import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.UPDATING; import static org.thingsboard.server.dao.service.OtaPackageServiceTest.TARGET_FW_VERSION; -import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.BINARY_APP_DATA_CONTAINER; import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.LwM2MProfileBootstrapConfigType.NONE; -import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.RESOURCE_ID_0; -import static org.thingsboard.server.transport.lwm2m.server.ota.DefaultLwM2MOtaUpdateService.FW_INSTANCE_ID; @Slf4j public class Ota5LwM2MIntegrationTest extends AbstractOtaLwM2MIntegrationTest { diff --git a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota9LwM2MIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota9LwM2MIntegrationTest.java index a091821444..5371a6a3c0 100644 --- a/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota9LwM2MIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/lwm2m/ota/sql/Ota9LwM2MIntegrationTest.java @@ -19,7 +19,6 @@ import lombok.extern.slf4j.Slf4j; import org.junit.Test; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; -import org.thingsboard.server.common.data.OtaPackageInfo; import org.thingsboard.server.common.data.device.credentials.lwm2m.LwM2MDeviceCredentials; import org.thingsboard.server.common.data.device.profile.Lwm2mDeviceProfileTransportConfiguration; import org.thingsboard.server.common.data.kv.TsKvEntry; @@ -36,10 +35,7 @@ import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.INIT import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.QUEUED; import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.UPDATED; import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.VERIFIED; -import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.BINARY_APP_DATA_CONTAINER; import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.LwM2MProfileBootstrapConfigType.NONE; -import static org.thingsboard.server.transport.lwm2m.Lwm2mTestHelper.RESOURCE_ID_0; -import static org.thingsboard.server.transport.lwm2m.server.ota.DefaultLwM2MOtaUpdateService.SW_INSTANCE_ID; @Slf4j public class Ota9LwM2MIntegrationTest extends AbstractOtaLwM2MIntegrationTest { From 12fbfc30f35e2da6c1ce9ff5af23d0ed4c892cdc Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Mon, 15 Dec 2025 16:13:13 +0200 Subject: [PATCH 3/8] Update sql views in system patch applier --- .../SqlEntityDatabaseSchemaService.java | 13 ++- .../service/system/SystemPatchApplier.java | 18 +++ ...and-functions.sql => schema-functions.sql} | 93 +-------------- dao/src/main/resources/sql/schema-views.sql | 106 ++++++++++++++++++ .../server/dao/PostgreSqlInitializer.java | 3 +- .../server/dao/TimescaleSqlInitializer.java | 4 +- .../install/sql/EntitiesSchemaSqlTest.java | 6 +- 7 files changed, 143 insertions(+), 100 deletions(-) rename dao/src/main/resources/sql/{schema-views-and-functions.sql => schema-functions.sql} (73%) create mode 100644 dao/src/main/resources/sql/schema-views.sql diff --git a/application/src/main/java/org/thingsboard/server/service/install/SqlEntityDatabaseSchemaService.java b/application/src/main/java/org/thingsboard/server/service/install/SqlEntityDatabaseSchemaService.java index aa43d77439..a773d0ab23 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/SqlEntityDatabaseSchemaService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/SqlEntityDatabaseSchemaService.java @@ -22,12 +22,13 @@ import org.springframework.stereotype.Service; @Service @Profile("install") @Slf4j -public class SqlEntityDatabaseSchemaService extends SqlAbstractDatabaseSchemaService - implements EntityDatabaseSchemaService { +public class SqlEntityDatabaseSchemaService extends SqlAbstractDatabaseSchemaService implements EntityDatabaseSchemaService { + public static final String SCHEMA_ENTITIES_SQL = "schema-entities.sql"; public static final String SCHEMA_ENTITIES_IDX_SQL = "schema-entities-idx.sql"; public static final String SCHEMA_ENTITIES_IDX_PSQL_ADDON_SQL = "schema-entities-idx-psql-addon.sql"; - public static final String SCHEMA_VIEWS_AND_FUNCTIONS_SQL = "schema-views-and-functions.sql"; + public static final String SCHEMA_VIEWS_SQL = "schema-views.sql"; + public static final String SCHEMA_FUNCTIONS_SQL = "schema-functions.sql"; public SqlEntityDatabaseSchemaService() { super(SCHEMA_ENTITIES_SQL, SCHEMA_ENTITIES_IDX_SQL); @@ -49,8 +50,10 @@ public class SqlEntityDatabaseSchemaService extends SqlAbstractDatabaseSchemaSer @Override public void createOrUpdateViewsAndFunctions() throws Exception { - log.info("Installing SQL DataBase schema views and functions: " + SCHEMA_VIEWS_AND_FUNCTIONS_SQL); - executeQueryFromFile(SCHEMA_VIEWS_AND_FUNCTIONS_SQL); + log.info("Installing SQL DataBase schema views: " + SCHEMA_VIEWS_SQL); + executeQueryFromFile(SCHEMA_VIEWS_SQL); + log.info("Installing SQL DataBase schema functions: " + SCHEMA_FUNCTIONS_SQL); + executeQueryFromFile(SCHEMA_FUNCTIONS_SQL); } } diff --git a/application/src/main/java/org/thingsboard/server/service/system/SystemPatchApplier.java b/application/src/main/java/org/thingsboard/server/service/system/SystemPatchApplier.java index 204d2b5d6c..76af940685 100644 --- a/application/src/main/java/org/thingsboard/server/service/system/SystemPatchApplier.java +++ b/application/src/main/java/org/thingsboard/server/service/system/SystemPatchApplier.java @@ -16,7 +16,9 @@ package org.thingsboard.server.service.system; import com.fasterxml.jackson.databind.JsonNode; +import com.google.common.base.Charsets; import com.google.common.hash.Hashing; +import com.google.common.io.Resources; import jakarta.annotation.PostConstruct; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -34,6 +36,7 @@ import org.thingsboard.server.service.install.update.DefaultDataUpdateService; import java.io.IOException; import java.io.UncheckedIOException; +import java.net.URL; import java.nio.file.Files; import java.nio.file.NoSuchFileException; import java.nio.file.Path; @@ -53,6 +56,8 @@ import java.util.stream.Stream; @RequiredArgsConstructor public class SystemPatchApplier { + private static final String SCHEMA_VIEWS_SQL = "sql/schema-views.sql"; + private static final long ADVISORY_LOCK_ID = 7536891047216478431L; private final JdbcTemplate jdbcTemplate; @@ -86,6 +91,9 @@ public class SystemPatchApplier { } try { + updateSqlViews(); + log.info("Updated sql database views"); + int updated = updateWidgetTypes(); log.info("Updated {} widget types", updated); @@ -124,6 +132,16 @@ public class SystemPatchApplier { && packageVersion.maintenance == dbVersion.maintenance && packageVersion.patch > dbVersion.patch; } + private void updateSqlViews() { + try { + URL schemaViewsUrl = Resources.getResource(SCHEMA_VIEWS_SQL); + String sql = Resources.toString(schemaViewsUrl, Charsets.UTF_8); + jdbcTemplate.execute(sql); + } catch (IOException e) { + throw new RuntimeException("Unable to update database views from schema-views.sql", e); + } + } + private int updateWidgetTypes() { AtomicInteger updated = new AtomicInteger(); Path widgetTypesDir = installScripts.getWidgetTypesDir(); diff --git a/dao/src/main/resources/sql/schema-views-and-functions.sql b/dao/src/main/resources/sql/schema-functions.sql similarity index 73% rename from dao/src/main/resources/sql/schema-views-and-functions.sql rename to dao/src/main/resources/sql/schema-functions.sql index a6a7ae43cb..e763210a3d 100644 --- a/dao/src/main/resources/sql/schema-views-and-functions.sql +++ b/dao/src/main/resources/sql/schema-functions.sql @@ -14,84 +14,6 @@ -- limitations under the License. -- -DROP VIEW IF EXISTS device_info_active_attribute_view CASCADE; -CREATE OR REPLACE VIEW device_info_active_attribute_view AS -SELECT d.* - , c.title as customer_title - , COALESCE((c.additional_info::json->>'isPublic')::bool, FALSE) as customer_is_public - , d.type as device_profile_name - , COALESCE(da.bool_v, FALSE) as active -FROM device d - LEFT JOIN customer c ON c.id = d.customer_id - LEFT JOIN attribute_kv da ON da.entity_id = d.id AND da.attribute_type = 2 AND da.attribute_key = (select key_id from key_dictionary where key = 'active'); - -DROP VIEW IF EXISTS device_info_active_ts_view CASCADE; -CREATE OR REPLACE VIEW device_info_active_ts_view AS -SELECT d.* - , c.title as customer_title - , COALESCE((c.additional_info::json->>'isPublic')::bool, FALSE) as customer_is_public - , d.type as device_profile_name - , COALESCE(dt.bool_v, FALSE) as active -FROM device d - LEFT JOIN customer c ON c.id = d.customer_id - LEFT JOIN ts_kv_latest dt ON dt.entity_id = d.id and dt.key = (select key_id from key_dictionary where key = 'active'); - -DROP VIEW IF EXISTS device_info_view CASCADE; -CREATE OR REPLACE VIEW device_info_view AS SELECT * FROM device_info_active_attribute_view; - -DROP VIEW IF EXISTS alarm_info CASCADE; -CREATE VIEW alarm_info AS -SELECT a.*, -(CASE WHEN a.acknowledged AND a.cleared THEN 'CLEARED_ACK' - WHEN NOT a.acknowledged AND a.cleared THEN 'CLEARED_UNACK' - WHEN a.acknowledged AND NOT a.cleared THEN 'ACTIVE_ACK' - WHEN NOT a.acknowledged AND NOT a.cleared THEN 'ACTIVE_UNACK' END) as status, -COALESCE(CASE WHEN a.originator_type = 0 THEN (select title from tenant where id = a.originator_id) - WHEN a.originator_type = 1 THEN (select title from customer where id = a.originator_id) - WHEN a.originator_type = 2 THEN (select email from tb_user where id = a.originator_id) - WHEN a.originator_type = 3 THEN (select title from dashboard where id = a.originator_id) - WHEN a.originator_type = 4 THEN (select name from asset where id = a.originator_id) - WHEN a.originator_type = 5 THEN (select name from device where id = a.originator_id) - WHEN a.originator_type = 9 THEN (select name from entity_view where id = a.originator_id) - WHEN a.originator_type = 13 THEN (select name from device_profile where id = a.originator_id) - WHEN a.originator_type = 14 THEN (select name from asset_profile where id = a.originator_id) - WHEN a.originator_type = 18 THEN (select name from edge where id = a.originator_id) END - , 'Deleted') originator_name, -COALESCE(CASE WHEN a.originator_type = 0 THEN (select title from tenant where id = a.originator_id) - WHEN a.originator_type = 1 THEN (select COALESCE(NULLIF(title, ''), email) from customer where id = a.originator_id) - WHEN a.originator_type = 2 THEN (select email from tb_user where id = a.originator_id) - WHEN a.originator_type = 3 THEN (select title from dashboard where id = a.originator_id) - WHEN a.originator_type = 4 THEN (select COALESCE(NULLIF(label, ''), name) from asset where id = a.originator_id) - WHEN a.originator_type = 5 THEN (select COALESCE(NULLIF(label, ''), name) from device where id = a.originator_id) - WHEN a.originator_type = 9 THEN (select name from entity_view where id = a.originator_id) - WHEN a.originator_type = 13 THEN (select name from device_profile where id = a.originator_id) - WHEN a.originator_type = 14 THEN (select name from asset_profile where id = a.originator_id) - WHEN a.originator_type = 18 THEN (select COALESCE(NULLIF(label, ''), name) from edge where id = a.originator_id) END - , 'Deleted') as originator_label, -u.first_name as assignee_first_name, u.last_name as assignee_last_name, u.email as assignee_email -FROM alarm a -LEFT JOIN tb_user u ON u.id = a.assignee_id; - -DROP VIEW IF EXISTS edge_active_attribute_view CASCADE; -CREATE OR REPLACE VIEW edge_active_attribute_view AS -SELECT ee.id - , ee.created_time - , ee.additional_info - , ee.customer_id - , ee.root_rule_chain_id - , ee.type - , ee.name - , ee.label - , ee.routing_key - , ee.secret - , ee.tenant_id - , ee.version -FROM edge ee - JOIN attribute_kv ON ee.id = attribute_kv.entity_id - JOIN key_dictionary ON attribute_kv.attribute_key = key_dictionary.key_id -WHERE attribute_kv.bool_v = true AND key_dictionary.key = 'active' -ORDER BY ee.id; - CREATE OR REPLACE FUNCTION create_or_update_active_alarm( t_id uuid, c_id uuid, a_id uuid, a_created_ts bigint, a_o_id uuid, a_o_type integer, a_type varchar, @@ -306,19 +228,6 @@ BEGIN END $$; -DROP VIEW IF EXISTS widget_type_info_view CASCADE; -CREATE OR REPLACE VIEW widget_type_info_view AS -SELECT t.*, - COALESCE((t.descriptor::json->>'type')::text, '') as widget_type, - array_to_json(ARRAY( - SELECT json_build_object('id', wb.widgets_bundle_id, 'name', b.title) - FROM widgets_bundle_widget wb - JOIN widgets_bundle b ON wb.widgets_bundle_id = b.id - WHERE wb.widget_type_id = t.id - ORDER BY b.title - )) AS bundles -FROM widget_type t; - CREATE OR REPLACE PROCEDURE cleanup_timeseries_by_ttl(IN null_uuid uuid, IN system_ttl bigint, INOUT deleted bigint) LANGUAGE plpgsql AS @@ -389,4 +298,4 @@ BEGIN FETCH tenant_cursor INTO tenant_id_record; END LOOP; END -$$; \ No newline at end of file +$$; diff --git a/dao/src/main/resources/sql/schema-views.sql b/dao/src/main/resources/sql/schema-views.sql new file mode 100644 index 0000000000..c8f7ea6483 --- /dev/null +++ b/dao/src/main/resources/sql/schema-views.sql @@ -0,0 +1,106 @@ +-- +-- 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. +-- + +DROP VIEW IF EXISTS device_info_active_attribute_view CASCADE; +CREATE OR REPLACE VIEW device_info_active_attribute_view AS +SELECT d.* + , c.title as customer_title + , COALESCE((c.additional_info::json->>'isPublic')::bool, FALSE) as customer_is_public + , d.type as device_profile_name + , COALESCE(da.bool_v, FALSE) as active +FROM device d + LEFT JOIN customer c ON c.id = d.customer_id + LEFT JOIN attribute_kv da ON da.entity_id = d.id AND da.attribute_type = 2 AND da.attribute_key = (select key_id from key_dictionary where key = 'active'); + +DROP VIEW IF EXISTS device_info_active_ts_view CASCADE; +CREATE OR REPLACE VIEW device_info_active_ts_view AS +SELECT d.* + , c.title as customer_title + , COALESCE((c.additional_info::json->>'isPublic')::bool, FALSE) as customer_is_public + , d.type as device_profile_name + , COALESCE(dt.bool_v, FALSE) as active +FROM device d + LEFT JOIN customer c ON c.id = d.customer_id + LEFT JOIN ts_kv_latest dt ON dt.entity_id = d.id and dt.key = (select key_id from key_dictionary where key = 'active'); + +DROP VIEW IF EXISTS device_info_view CASCADE; +CREATE OR REPLACE VIEW device_info_view AS SELECT * FROM device_info_active_attribute_view; + +DROP VIEW IF EXISTS alarm_info CASCADE; +CREATE VIEW alarm_info AS +SELECT a.*, +(CASE WHEN a.acknowledged AND a.cleared THEN 'CLEARED_ACK' + WHEN NOT a.acknowledged AND a.cleared THEN 'CLEARED_UNACK' + WHEN a.acknowledged AND NOT a.cleared THEN 'ACTIVE_ACK' + WHEN NOT a.acknowledged AND NOT a.cleared THEN 'ACTIVE_UNACK' END) as status, +COALESCE(CASE WHEN a.originator_type = 0 THEN (select title from tenant where id = a.originator_id) + WHEN a.originator_type = 1 THEN (select title from customer where id = a.originator_id) + WHEN a.originator_type = 2 THEN (select email from tb_user where id = a.originator_id) + WHEN a.originator_type = 3 THEN (select title from dashboard where id = a.originator_id) + WHEN a.originator_type = 4 THEN (select name from asset where id = a.originator_id) + WHEN a.originator_type = 5 THEN (select name from device where id = a.originator_id) + WHEN a.originator_type = 9 THEN (select name from entity_view where id = a.originator_id) + WHEN a.originator_type = 13 THEN (select name from device_profile where id = a.originator_id) + WHEN a.originator_type = 14 THEN (select name from asset_profile where id = a.originator_id) + WHEN a.originator_type = 18 THEN (select name from edge where id = a.originator_id) END + , 'Deleted') originator_name, +COALESCE(CASE WHEN a.originator_type = 0 THEN (select title from tenant where id = a.originator_id) + WHEN a.originator_type = 1 THEN (select COALESCE(NULLIF(title, ''), email) from customer where id = a.originator_id) + WHEN a.originator_type = 2 THEN (select email from tb_user where id = a.originator_id) + WHEN a.originator_type = 3 THEN (select title from dashboard where id = a.originator_id) + WHEN a.originator_type = 4 THEN (select COALESCE(NULLIF(label, ''), name) from asset where id = a.originator_id) + WHEN a.originator_type = 5 THEN (select COALESCE(NULLIF(label, ''), name) from device where id = a.originator_id) + WHEN a.originator_type = 9 THEN (select name from entity_view where id = a.originator_id) + WHEN a.originator_type = 13 THEN (select name from device_profile where id = a.originator_id) + WHEN a.originator_type = 14 THEN (select name from asset_profile where id = a.originator_id) + WHEN a.originator_type = 18 THEN (select COALESCE(NULLIF(label, ''), name) from edge where id = a.originator_id) END + , 'Deleted') as originator_label, +u.first_name as assignee_first_name, u.last_name as assignee_last_name, u.email as assignee_email +FROM alarm a +LEFT JOIN tb_user u ON u.id = a.assignee_id; + +DROP VIEW IF EXISTS edge_active_attribute_view CASCADE; +CREATE OR REPLACE VIEW edge_active_attribute_view AS +SELECT ee.id + , ee.created_time + , ee.additional_info + , ee.customer_id + , ee.root_rule_chain_id + , ee.type + , ee.name + , ee.label + , ee.routing_key + , ee.secret + , ee.tenant_id + , ee.version +FROM edge ee + JOIN attribute_kv ON ee.id = attribute_kv.entity_id + JOIN key_dictionary ON attribute_kv.attribute_key = key_dictionary.key_id +WHERE attribute_kv.bool_v = true AND key_dictionary.key = 'active' +ORDER BY ee.id; + +DROP VIEW IF EXISTS widget_type_info_view CASCADE; +CREATE OR REPLACE VIEW widget_type_info_view AS +SELECT t.*, + COALESCE((t.descriptor::json->>'type')::text, '') as widget_type, + array_to_json(ARRAY( + SELECT json_build_object('id', wb.widgets_bundle_id, 'name', b.title) + FROM widgets_bundle_widget wb + JOIN widgets_bundle b ON wb.widgets_bundle_id = b.id + WHERE wb.widget_type_id = t.id + ORDER BY b.title + )) AS bundles +FROM widget_type t; diff --git a/dao/src/test/java/org/thingsboard/server/dao/PostgreSqlInitializer.java b/dao/src/test/java/org/thingsboard/server/dao/PostgreSqlInitializer.java index b2bb965380..49956bd6d4 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/PostgreSqlInitializer.java +++ b/dao/src/test/java/org/thingsboard/server/dao/PostgreSqlInitializer.java @@ -33,7 +33,8 @@ public class PostgreSqlInitializer { "sql/schema-entities.sql", "sql/schema-entities-idx.sql", "sql/schema-entities-idx-psql-addon.sql", - "sql/schema-views-and-functions.sql", + "sql/schema-views.sql", + "sql/schema-functions.sql", "sql/system-data.sql", "sql/system-test-psql.sql"); private static final String dropAllTablesSqlFile = "sql/psql/drop-all-tables.sql"; diff --git a/dao/src/test/java/org/thingsboard/server/dao/TimescaleSqlInitializer.java b/dao/src/test/java/org/thingsboard/server/dao/TimescaleSqlInitializer.java index c9d13b6926..511afd1eca 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/TimescaleSqlInitializer.java +++ b/dao/src/test/java/org/thingsboard/server/dao/TimescaleSqlInitializer.java @@ -33,7 +33,8 @@ public class TimescaleSqlInitializer { "sql/schema-entities.sql", "sql/schema-entities-idx.sql", "sql/schema-entities-idx-psql-addon.sql", - "sql/schema-views-and-functions.sql", + "sql/schema-views.sql", + "sql/schema-functions.sql", "sql/system-data.sql", "sql/system-test-psql.sql"); private static final String dropAllTablesSqlFile = "sql/psql/drop-all-tables.sql"; @@ -63,4 +64,5 @@ public class TimescaleSqlInitializer { throw new RuntimeException("Unable to clean up the Timescale database. Reason: " + e.getMessage(), e); } } + } diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/install/sql/EntitiesSchemaSqlTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/install/sql/EntitiesSchemaSqlTest.java index 9f10ecfb7b..b28b410a12 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/install/sql/EntitiesSchemaSqlTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/install/sql/EntitiesSchemaSqlTest.java @@ -35,8 +35,10 @@ public class EntitiesSchemaSqlTest extends AbstractServiceTest { @Value("${classpath:sql/schema-entities.sql}") private Path installEntitiesPath; - @Value("${classpath:sql/schema-views-and-functions.sql}") + @Value("${classpath:sql/schema-views.sql}") private Path installViewsPath; + @Value("${classpath:sql/schema-functions.sql}") + private Path installFunctionsPath; @Autowired private JdbcTemplate jdbcTemplate; @@ -45,10 +47,12 @@ public class EntitiesSchemaSqlTest extends AbstractServiceTest { public void testRepeatedInstall() throws IOException { String entitiesScript = Files.readString(installEntitiesPath); String viewsScript = Files.readString(installViewsPath); + String functionsScript = Files.readString(installFunctionsPath); try { for (int i = 1; i <= 2; i++) { jdbcTemplate.execute(entitiesScript); jdbcTemplate.execute(viewsScript); + jdbcTemplate.execute(functionsScript); } } catch (Exception e) { Assertions.fail("Failed to execute reinstall", e); From eb86dc91514d5effc67a076b590eb19a7b986a9d Mon Sep 17 00:00:00 2001 From: Nikita Mazurenko Date: Tue, 16 Dec 2025 15:51:36 +0200 Subject: [PATCH 4/8] Fix events from different edges being mixed together in one queue --- .../thingsboard/server/service/edge/rpc/EdgeGrpcSession.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java index 521730741f..c7c27ae4d0 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java @@ -113,7 +113,7 @@ public abstract class EdgeGrpcSession implements Closeable { private static final int MAX_DOWNLINK_ATTEMPTS = 3; private static final String RATE_LIMIT_REACHED = "Rate limit reached"; - protected static final ConcurrentLinkedQueue highPriorityQueue = new ConcurrentLinkedQueue<>(); + protected final ConcurrentLinkedQueue highPriorityQueue = new ConcurrentLinkedQueue<>(); protected UUID sessionId; private BiConsumer sessionOpenListener; From 6627bd9c0630e4bf75bf85539854de56f6a89275 Mon Sep 17 00:00:00 2001 From: Maksym Tsymbarov Date: Tue, 16 Dec 2025 16:51:57 +0200 Subject: [PATCH 5/8] Fixed mutation in ignore Array Merge Function --- ui-ngx/src/app/core/utils.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ui-ngx/src/app/core/utils.ts b/ui-ngx/src/app/core/utils.ts index e0fadd768a..5bdb6a48dd 100644 --- a/ui-ngx/src/app/core/utils.ts +++ b/ui-ngx/src/app/core/utils.ts @@ -403,7 +403,7 @@ export function mergeDeep(target: T, ...sources: T[]): T { function ignoreArrayMergeFunc(target: any, sources: any) { if (_.isArray(target)) { - return sources; + return deepClone(sources); } } From ca71ced6833231d564b3d2fd2c07e31a35442e42 Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Tue, 16 Dec 2025 18:04:14 +0200 Subject: [PATCH 6/8] KafkaEdgeGrpcSession - move commit outside try/catch to propagate expection in case consumer already closed --- .../server/service/edge/rpc/KafkaEdgeGrpcSession.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java index 63669a8e3d..e6eaec23e4 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/KafkaEdgeGrpcSession.java @@ -82,16 +82,18 @@ public class KafkaEdgeGrpcSession extends EdgeGrpcSession { edgeEvents.add(edgeEvent); } List downlinkMsgsPack = convertToDownlinkMsgsPack(edgeEvents); + boolean isInterrupted = true; try { - boolean isInterrupted = sendDownlinkMsgsPack(downlinkMsgsPack).get(); + isInterrupted = sendDownlinkMsgsPack(downlinkMsgsPack).get(); if (isInterrupted) { log.debug("[{}][{}] Send downlink messages task was interrupted", tenantId, edge.getId()); - } else { - consumer.commit(); } } catch (Exception e) { log.error("[{}][{}] Failed to process downlink messages", tenantId, edge.getId(), e); } + if (!isInterrupted) { + consumer.commit(); + } } @Override From e0357aff503e71a0fa0e32f2d59d626da70a0000 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Wed, 17 Dec 2025 12:44:59 +0200 Subject: [PATCH 7/8] fixed processing telemetry batch --- ...CalculatedFieldEntityMessageProcessor.java | 34 +++++++- .../cf/ctx/state/TsRollingArgumentEntry.java | 32 +++++--- .../cf/CalculatedFieldIntegrationTest.java | 80 +++++++++++++++++++ 3 files changed, 131 insertions(+), 15 deletions(-) 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 49d1af2b0f..257e9d8565 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java @@ -36,6 +36,7 @@ import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto; import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; @@ -48,6 +49,7 @@ import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; +import org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry; import java.util.ArrayList; import java.util.Collection; @@ -62,6 +64,8 @@ import java.util.UUID; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; +import static org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry.getValueForTsRecord; + /** * @author Andrew Shvayka @@ -356,13 +360,39 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM ReferencedEntityKey key = new ReferencedEntityKey(item.getKv().getKey(), ArgumentType.TS_LATEST, null); Set argNames = args.get(key); if (argNames != null) { - argNames.forEach(argName -> arguments.put(argName, new SingleValueArgumentEntry(item))); + SingleValueArgumentEntry incoming = new SingleValueArgumentEntry(item); + argNames.forEach(argName -> arguments.compute(argName, (name, existing) -> { + if (existing == null) { + return incoming; + } + existing.updateEntry(incoming); + return existing; + })); } key = new ReferencedEntityKey(item.getKv().getKey(), ArgumentType.TS_ROLLING, null); argNames = args.get(key); if (argNames != null) { - argNames.forEach(argName -> arguments.put(argName, new SingleValueArgumentEntry(item))); + Double recordValue = getValueForTsRecord(ProtoUtils.fromProto(item.getKv())); + argNames.forEach(argName -> arguments.compute(argName, (name, existing) -> { + if (existing instanceof TsRollingArgumentEntry rolling) { + if (recordValue != null) { + rolling.getTsRecords().put(item.getTs(), recordValue); + } + return rolling; + } + TsRollingArgumentEntry rolling = new TsRollingArgumentEntry(); + if (recordValue != null) { + rolling.getTsRecords().put(item.getTs(), recordValue); + } + if (existing instanceof SingleValueArgumentEntry single) { + Double existingValue = getValueForTsRecord(single.getKvEntryValue()); + if (existingValue != null) { + rolling.getTsRecords().put(single.getTs(), existingValue); + } + } + return rolling; + })); } } return arguments; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java index e01d8b7369..9f9ad5db73 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java @@ -122,20 +122,11 @@ public class TsRollingArgumentEntry implements ArgumentEntry { } private void addTsRecord(Long ts, KvEntry value) { - try { - switch (value.getDataType()) { - case LONG -> value.getLongValue().ifPresent(aLong -> tsRecords.put(ts, aLong.doubleValue())); - case DOUBLE -> value.getDoubleValue().ifPresent(aDouble -> tsRecords.put(ts, aDouble)); - case BOOLEAN -> value.getBooleanValue().ifPresent(aBoolean -> tsRecords.put(ts, aBoolean ? 1.0 : 0.0)); - case STRING -> value.getStrValue().ifPresent(aString -> tsRecords.put(ts, Double.parseDouble(aString))); - case JSON -> value.getJsonValue().ifPresent(aString -> tsRecords.put(ts, Double.parseDouble(aString))); - } - } catch (Exception e) { - tsRecords.put(ts, Double.NaN); - log.debug("Invalid value '{}' for time series rolling arguments. Only numeric values are supported.", value.getValue()); - } finally { - cleanupExpiredRecords(); + Double recordValue = getValueForTsRecord(value); + if (recordValue != null) { + tsRecords.put(ts, recordValue); } + cleanupExpiredRecords(); } private void addTsRecord(Long ts, double value) { @@ -150,4 +141,19 @@ public class TsRollingArgumentEntry implements ArgumentEntry { tsRecords.entrySet().removeIf(tsRecord -> tsRecord.getKey() < System.currentTimeMillis() - timeWindow); } + public static Double getValueForTsRecord(KvEntry value) { + try { + return switch (value.getDataType()) { + case LONG -> value.getLongValue().map(Long::doubleValue).orElse(null); + case DOUBLE -> value.getDoubleValue().orElse(null); + case BOOLEAN -> value.getBooleanValue().map(b -> b ? 1.0 : 0.0).orElse(null); + case STRING -> value.getStrValue().map(Double::parseDouble).orElse(null); + case JSON -> value.getJsonValue().map(Double::parseDouble).orElse(null); + }; + } catch (Exception e) { + log.debug("Invalid value '{}' for time series rolling arguments. Only numeric values are supported.", value.getValue()); + return Double.NaN; + } + } + } diff --git a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java index c83e8c5ee3..1da3673695 100644 --- a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java @@ -796,6 +796,86 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes }); } + @Test + public void testCalculatedFieldWhenBatchOfTelemetrySent() throws Exception { + Device testDevice = createDevice("Test device", "1234567890"); + doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"a\":5}")); + doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"b\":10}")); + doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"b\":20}")); + + CalculatedField calculatedField = new CalculatedField(); + calculatedField.setEntityId(testDevice.getId()); + calculatedField.setType(CalculatedFieldType.SCRIPT); + calculatedField.setName("Script CF"); + calculatedField.setDebugSettings(DebugSettings.all()); + + SimpleCalculatedFieldConfiguration config = new SimpleCalculatedFieldConfiguration(); + + ReferencedEntityKey refEntityKeyA = new ReferencedEntityKey("a", ArgumentType.TS_LATEST, null); + Argument argumentA = new Argument(); + argumentA.setRefEntityKey(refEntityKeyA); + Argument argumentB = new Argument(); + ReferencedEntityKey refEntityKeyB = new ReferencedEntityKey("b", ArgumentType.TS_ROLLING, null); + argumentB.setTimeWindow(TimeUnit.MINUTES.toMillis(10)); + argumentB.setLimit(1000); + argumentB.setRefEntityKey(refEntityKeyB); + config.setArguments(Map.of("a", argumentA, "b", argumentB)); + config.setExpression(""" + return { + "latestA": a, + "avgB": b.avg + }; + """); + + Output output = new Output(); + output.setType(OutputType.TIME_SERIES); + config.setOutput(output); + + calculatedField.setConfiguration(config); + + doPost("/api/calculatedField", calculatedField, CalculatedField.class); + + await().alias("create CF -> perform initial calculation").atMost(TIMEOUT, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + ObjectNode result = getLatestTelemetry(testDevice.getId(), "latestA", "avgB"); + assertThat(result).isNotNull(); + assertThat(result.get("latestA").get(0).get("value").asText()).isEqualTo("5"); + assertThat(result.get("avgB").get(0).get("value").asText()).isEqualTo("15.0"); + }); + + long now = System.currentTimeMillis(); + doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format(""" + [{ + "ts": %s, + "values": { + "a": 6, + "b": 100 + } + }, { + "ts": %s, + "values": { + "a": 7, + "b": 200 + } + }, { + "ts": %s, + "values": { + "a": 8, + "b": 300 + } + }]""", now - TimeUnit.MINUTES.toMillis(2), now, now - TimeUnit.MINUTES.toMillis(5)))); + + await().alias("update telemetry -> recalculate state").atMost(TIMEOUT, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + ObjectNode result = getLatestTelemetry(testDevice.getId(), "latestA", "avgB"); + assertThat(result).isNotNull(); + assertThat(result.get("latestA").get(0).get("value").asText()).isEqualTo("7"); + assertThat(result.get("avgB").get(0).get("value").asText()).isEqualTo("126.0"); + }); + } + private ObjectNode getLatestTelemetry(EntityId entityId, String... keys) throws Exception { return doGetAsync("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() + "/values/timeseries?keys=" + String.join(",", keys), ObjectNode.class); } From 3f0388b7c0cf7737ebf0b5dc9c73f8833003f113 Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Wed, 17 Dec 2025 12:15:23 +0100 Subject: [PATCH 8/8] TbMqttSslTransportComponent for MqttSslHandlerProvider --- .../mqtt/MqttSslHandlerProvider.java | 3 +- .../mqtt/TbMqttSslTransportComponent.java | 31 +++++++++++++++++++ .../mqtt/TbMqttTransportComponent.java | 6 +++- 3 files changed, 37 insertions(+), 3 deletions(-) create mode 100644 common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttSslTransportComponent.java diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttSslHandlerProvider.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttSslHandlerProvider.java index 867b39cd23..77396c69fa 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttSslHandlerProvider.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttSslHandlerProvider.java @@ -20,7 +20,6 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; @@ -48,7 +47,7 @@ import java.util.concurrent.TimeUnit; @Slf4j @Component("MqttSslHandlerProvider") -@ConditionalOnProperty(prefix = "transport.mqtt.ssl", value = "enabled", havingValue = "true", matchIfMissing = false) +@TbMqttSslTransportComponent public class MqttSslHandlerProvider { @Value("${transport.mqtt.ssl.protocol}") diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttSslTransportComponent.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttSslTransportComponent.java new file mode 100644 index 0000000000..2a0e8ef42f --- /dev/null +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttSslTransportComponent.java @@ -0,0 +1,31 @@ +/** + * 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.transport.mqtt; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; + +import java.lang.annotation.Inherited; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; + +/** + * Same as @TbMqttTransportComponent with additional condition by `transport.mqtt.ssl.enabled == true` + */ + +@Inherited +@Retention(RetentionPolicy.RUNTIME) +@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.mqtt.enabled:true}'=='true' && '${transport.mqtt.ssl.enabled:false}'=='true')") +public @interface TbMqttSslTransportComponent {} diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttTransportComponent.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttTransportComponent.java index b517cda103..0d034aebb2 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttTransportComponent.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttTransportComponent.java @@ -21,7 +21,11 @@ import java.lang.annotation.Inherited; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; +/** + * See also @TbMqttSslTransportComponent + */ + @Inherited @Retention(RetentionPolicy.RUNTIME) -@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.mqtt.enabled}'=='true')") +@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.mqtt.enabled:true}'=='true')") public @interface TbMqttTransportComponent {}