diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java index e7b0113dbe..75499b9a8d 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java @@ -43,6 +43,7 @@ import org.thingsboard.server.gen.transport.TransportProtos; @Slf4j public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor { + private static final String UNKNOWN_NAME = "Unknown"; private final String ruleChainName; private final TbApiUsageReportClient apiUsageClient; private final DefaultTbContext defaultCtx; @@ -57,7 +58,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor highPriorityQueue = new ConcurrentLinkedQueue<>(); - private static final int MAX_DOWNLINK_ATTEMPTS = 10; // max number of attemps to send downlink message if edge connected + private static final int MAX_DOWNLINK_ATTEMPTS = 10; // max number of attempts to send downlink message if edge connected private static final String QUEUE_START_TS_ATTR_KEY = "queueStartTs"; private static final String QUEUE_START_SEQ_ID_ATTR_KEY = "queueStartSeqId"; diff --git a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java index 2d5ac5a99b..b3d20ff5e9 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java @@ -122,7 +122,21 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService updateSchema("3.6.4", 3006004, "3.7.0", 3007000, null); break; case "3.7.0": - updateSchema("3.7.0", 3007000, "3.7.1", 3007001, null); + updateSchema("3.7.0", 3007000, "3.7.1", 3007001, connection -> { + try { + connection.createStatement().execute("UPDATE rule_node SET " + + "configuration = CASE " + + " WHEN (configuration::jsonb ->> 'persistAlarmRulesState') = 'false'" + + " THEN (configuration::jsonb || '{\"fetchAlarmRulesStateOnStart\": \"false\"}'::jsonb)::varchar " + + " ELSE configuration " + + "END, " + + "configuration_version = 1 " + + "WHERE type = 'org.thingsboard.rule.engine.profile.TbDeviceProfileNode' " + + "AND configuration_version < 1;"); + } catch (Exception e) { + log.warn("Failed to execute update script for device profile rule nodes due to: ", e); + } + }); break; default: throw new RuntimeException("Unable to upgrade SQL database, unsupported fromVersion: " + fromVersion); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java index c713bece6b..4043b519da 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java @@ -334,7 +334,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer } private void onTimeSeriesUpdate(UUID entityId, List data, TbCallback callback) { - entityUpdates.get(entityId).timeSeriesUpdateTs = System.currentTimeMillis(); + getEntityUpdatesInfo(entityId).timeSeriesUpdateTs = System.currentTimeMillis(); processSubscriptionData(entityId, sub -> TbSubscriptionType.TIMESERIES.equals(sub.getType()), s -> { @@ -371,7 +371,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer } private void onAttributesUpdate(UUID entityId, String scope, List data, TbCallback callback) { - entityUpdates.get(entityId).attributesUpdateTs = System.currentTimeMillis(); + getEntityUpdatesInfo(entityId).attributesUpdateTs = System.currentTimeMillis(); processSubscriptionData(entityId, sub -> TbSubscriptionType.ATTRIBUTES.equals(sub.getType()), s -> { @@ -639,4 +639,8 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer throw new TbRateLimitsException(message); } + private TbEntityUpdatesInfo getEntityUpdatesInfo(UUID entityId) { + return entityUpdates.computeIfAbsent(entityId, id -> new TbEntityUpdatesInfo(0)); + } + } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 440d8c6ba8..536c35e816 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1020,6 +1020,8 @@ transport: # MQTT disconnect timeout in milliseconds. The time to wait for the client to disconnect after the server sends a disconnect message. disconnect_timeout: "${MQTT_DISCONNECT_TIMEOUT:1000}" msg_queue_size_per_device_limit: "${MQTT_MSG_QUEUE_SIZE_PER_DEVICE_LIMIT:100}" # messages await in the queue before the device connected state. This limit works on the low level before TenantProfileLimits mechanism + # Interval of periodic report of the gateway metrics + gateway_metrics_report_interval_sec: "${MQTT_GATEWAY_METRICS_REPORT_INTERVAL_SEC:60}" netty: # Netty leak detector level leak_detector_level: "${NETTY_LEAK_DETECTOR_LVL:DISABLED}" diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java index 7c8cdb5d6e..5fd68417dd 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java @@ -20,22 +20,34 @@ import io.netty.handler.codec.mqtt.MqttQoS; import lombok.extern.slf4j.Slf4j; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.boot.test.mock.mockito.SpyBean; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; +import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService; +import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; +import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsState; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestCallback; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; +import java.util.ArrayList; import java.util.Arrays; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Random; import java.util.Set; import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.verify; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_ATTRIBUTES_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_TELEMETRY_SHORT_JSON_TOPIC; @@ -43,6 +55,7 @@ import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVIC import static org.thingsboard.server.common.data.device.profile.MqttTopics.DEVICE_TELEMETRY_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_CONNECT_TOPIC; import static org.thingsboard.server.common.data.device.profile.MqttTopics.GATEWAY_TELEMETRY_TOPIC; +import static org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService.GATEWAY_METRICS; @Slf4j public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqttIntegrationTest { @@ -53,6 +66,9 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt protected static final String MALFORMED_JSON_PAYLOAD = "{\"key1\":, \"key2\":true, \"key3\": 3.0, \"key4\": 4," + " \"key5\": {\"someNumber\": 42, \"someArray\": [1,2,3], \"someNestedObject\": {\"key\": \"value\"}}}"; + @SpyBean + GatewayMetricsService gatewayMetricsService; + @Before public void beforeTest() throws Exception { MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() @@ -106,13 +122,98 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt String deviceName = "Device A"; Device device = doExecuteWithRetriesAndInterval(() -> doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class), - 20, - 100); + 20, + 100); assertNotNull(device); client.disconnect(); } + @Test + public void testPushMetricsGateway() throws Exception { + MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() + .gatewayName("Test metrics gateway") + .build(); + processBeforeTest(configProperties); + + Map> gwLatencies = new HashMap<>(); + Map> transportLatencies = new HashMap<>(); + + publishLatency(gwLatencies, transportLatencies, 5); + + gatewayMetricsService.reportMetrics(); + + List actualKeys = getActualKeysList(savedGateway.getId(), List.of(GATEWAY_METRICS)); + assertEquals(GATEWAY_METRICS, actualKeys.get(0)); + + String telemetryUrl = String.format("/api/plugins/telemetry/DEVICE/%s/values/timeseries?startTs=%d&endTs=%d&keys=%s", savedGateway.getId(), 0, System.currentTimeMillis(), GATEWAY_METRICS); + + Map>> gatewayTelemetry = doGetAsyncTyped(telemetryUrl, new TypeReference<>() {}); + Map latencyCheckTelemetry = gatewayTelemetry.get(GATEWAY_METRICS).get(0); + Map latencyCheckValue = JacksonUtil.fromString((String) latencyCheckTelemetry.get("value"), new TypeReference<>() {}); + assertNotNull(latencyCheckValue); + + gwLatencies.forEach((connectorName, gwLatencyList) -> { + long avgGwLatency = (long) gwLatencyList.stream().mapToLong(Long::longValue).average().getAsDouble(); + long minGwLatency = gwLatencyList.stream().mapToLong(Long::longValue).min().getAsLong(); + long maxGwLatency = gwLatencyList.stream().mapToLong(Long::longValue).max().getAsLong(); + + List transportLatencyList = transportLatencies.get(connectorName); + assertNotNull(transportLatencyList); + + long avgTransportLatency = (long) transportLatencyList.stream().mapToLong(Long::longValue).average().getAsDouble(); + long minTransportLatency = transportLatencyList.stream().mapToLong(Long::longValue).min().getAsLong(); + long maxTransportLatency = transportLatencyList.stream().mapToLong(Long::longValue).max().getAsLong(); + + GatewayMetricsState.ConnectorMetricsResult connectorLatencyResult = latencyCheckValue.get(connectorName); + assertNotNull(connectorLatencyResult); + checkConnectorLatencyResult(connectorLatencyResult, avgGwLatency, minGwLatency, maxGwLatency, avgTransportLatency, minTransportLatency, maxTransportLatency); + }); + } + + private void publishLatency(Map> gwLatencies, Map> transportLatencies, int n) throws Exception { + Random random = new Random(); + for (int i = 0; i < n; i++) { + long publishedTs = System.currentTimeMillis() - 10; + long gatewayLatencyA = random.nextLong(100, 500); + var firstData = new GatewayMetadata("connectorA", publishedTs - gatewayLatencyA, publishedTs); + gwLatencies.computeIfAbsent("connectorA", key -> new ArrayList<>()).add(gatewayLatencyA); + long gatewayLatencyB = random.nextLong(120, 450); + var secondData = new GatewayMetadata("connectorB", publishedTs - gatewayLatencyB, publishedTs); + gwLatencies.computeIfAbsent("connectorB", key -> new ArrayList<>()).add(gatewayLatencyB); + + List expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5"); + String deviceName1 = "Device A"; + String deviceName2 = "Device B"; + String firstMetadata = JacksonUtil.writeValueAsString(firstData); + String secondMetadata = JacksonUtil.writeValueAsString(secondData); + String payload = getGatewayTelemetryJsonPayloadWithMetadata(deviceName1, deviceName2, "10000", "20000", firstMetadata, secondMetadata); + processGatewayTelemetryTest(GATEWAY_TELEMETRY_TOPIC, expectedKeys, payload.getBytes(), deviceName1, deviceName2); + + ArgumentCaptor transportReceiveTsCaptorA = ArgumentCaptor.forClass(Long.class); + ArgumentCaptor transportReceiveTsCaptorB = ArgumentCaptor.forClass(Long.class); + verify(gatewayMetricsService).process(any(), eq(savedGateway.getId()), eq(List.of(firstData)), transportReceiveTsCaptorA.capture()); + verify(gatewayMetricsService).process(any(), eq(savedGateway.getId()), eq(List.of(secondData)), transportReceiveTsCaptorB.capture()); + Long transportReceiveTsA = transportReceiveTsCaptorA.getValue(); + Long transportReceiveTsB = transportReceiveTsCaptorB.getValue(); + Long transportLatencyA = transportReceiveTsA - publishedTs; + Long transportLatencyB = transportReceiveTsB - publishedTs; + transportLatencies.computeIfAbsent("connectorA", key -> new ArrayList<>()).add(transportLatencyA); + transportLatencies.computeIfAbsent("connectorB", key -> new ArrayList<>()).add(transportLatencyB); + } + } + + private void checkConnectorLatencyResult(GatewayMetricsState.ConnectorMetricsResult result, long avgGwLatency, long minGwLatency, long maxGwLatency, + long avgTransportLatency, long minTransportLatency, long maxTransportLatency) { + assertNotNull(result); + assertEquals(avgGwLatency, result.avgGwLatency()); + assertEquals(minGwLatency, result.minGwLatency()); + assertEquals(maxGwLatency, result.maxGwLatency()); + assertEquals(avgTransportLatency, result.avgTransportLatency()); + assertEquals(minTransportLatency, result.minTransportLatency()); + assertEquals(maxTransportLatency, result.maxTransportLatency()); + } + protected void processJsonPayloadTelemetryTest(String topic, List expectedKeys, byte[] payload, boolean withTs) throws Exception { processTelemetryTest(topic, expectedKeys, payload, withTs, false); } @@ -143,7 +244,8 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt long end = System.currentTimeMillis() + 5000; Map>> values = null; while (start <= end) { - values = doGetAsyncTyped(getTelemetryValuesUrl, new TypeReference<>() {}); + values = doGetAsyncTyped(getTelemetryValuesUrl, new TypeReference<>() { + }); boolean valid = values.size() == expectedKeys.size(); if (valid) { for (String key : expectedKeys) { @@ -182,7 +284,7 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt MqttTestClient client = new MqttTestClient(); client.connectAndWait(gatewayAccessToken); client.publishAndWait(topic, payload); - client.disconnect(); + client.disconnectAndWait(); Device firstDevice = doExecuteWithRetriesAndInterval(() -> doGet("/api/tenant/devices?deviceName=" + firstDeviceName, Device.class), 20, @@ -239,6 +341,16 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt return "{\"" + deviceA + "\": " + payload + ", \"" + deviceB + "\": " + payload + "}"; } + protected String getGatewayTelemetryJsonPayloadWithMetadata(String deviceA, String deviceB, String firstTsValue, String secondTsValue, String firstMetadata, String secondMetadata) { + String payloadA = "[{\"ts\": " + firstTsValue + ", \"values\": " + PAYLOAD_VALUES_STR + ", \"metadata\":" + firstMetadata + "}, " + + "{\"ts\": " + secondTsValue + ", \"values\": " + PAYLOAD_VALUES_STR + "}]"; + + String payloadB = "[{\"ts\": " + firstTsValue + ", \"values\": " + PAYLOAD_VALUES_STR + ", \"metadata\":" + secondMetadata + "}, " + + "{\"ts\": " + secondTsValue + ", \"values\": " + PAYLOAD_VALUES_STR + "}]"; + + return "{\"" + deviceA + "\": " + payloadA + ", \"" + deviceB + "\": " + payloadB + "}"; + } + private String getTelemetryValuesUrl(DeviceId deviceId, Set actualKeySet) { return "/api/plugins/telemetry/DEVICE/" + deviceId + "/values/timeseries?startTs=0&endTs=25000&keys=" + String.join(",", actualKeySet); } diff --git a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java index 776d5657ce..ee4af20639 100644 --- a/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java +++ b/common/actor/src/main/java/org/thingsboard/server/actors/TbActorMailbox.java @@ -240,7 +240,7 @@ public final class TbActorMailbox implements TbActorCtx { highPriorityMsgs.forEach(msg -> msg.onTbActorStopped(stopReason)); normalPriorityMsgs.forEach(msg -> msg.onTbActorStopped(stopReason)); } catch (Throwable t) { - log.warn("[{}] Failed to destroy actor: {}", selfId, t); + log.warn("[{}] Failed to destroy actor: ", selfId, t); } }); } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/gateway/metrics/GatewayMetadata.java b/common/message/src/main/java/org/thingsboard/server/common/msg/gateway/metrics/GatewayMetadata.java new file mode 100644 index 0000000000..cc905f9383 --- /dev/null +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/gateway/metrics/GatewayMetadata.java @@ -0,0 +1,19 @@ +/** + * Copyright © 2016-2024 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.common.msg.gateway.metrics; + +public record GatewayMetadata(String connector, long receivedTs, long publishedTs) { +} diff --git a/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java b/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java index d1283d3ba0..f4abf5f3a0 100644 --- a/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java +++ b/common/proto/src/main/java/org/thingsboard/server/common/adaptor/JsonConverter.java @@ -34,6 +34,8 @@ import org.thingsboard.server.common.data.kv.JsonDataEntry; import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.LongDataEntry; import org.thingsboard.server.common.data.kv.StringDataEntry; +import org.thingsboard.server.common.data.util.TbPair; +import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg; @@ -83,6 +85,49 @@ public class JsonConverter { return convertToTelemetryProto(jsonElement, System.currentTimeMillis()); } + public static TbPair> convertToGatewayTelemetry(JsonElement jsonElement, long systemTs) { + List metadataResult = null; + PostTelemetryMsg.Builder builder = PostTelemetryMsg.newBuilder(); + if (jsonElement.isJsonArray()) { + var ja = jsonElement.getAsJsonArray(); + for (int i = 0; i < ja.size(); i++) { + var je = ja.get(i); + if (je.isJsonObject()) { + JsonObject jo = je.getAsJsonObject(); + JsonElement metadataElem = jo.remove("metadata"); + if (metadataElem != null) { + if (metadataResult == null) { + metadataResult = new ArrayList<>(); + } + if (metadataElem.isJsonObject()) { + JsonObject metadataObj = metadataElem.getAsJsonObject(); + var connector = getAndValidateMetadataElement(metadataObj, "connector").getAsString(); + var receivedTs = getAndValidateMetadataElement(metadataObj, "receivedTs").getAsLong(); + var publishedTs = getAndValidateMetadataElement(metadataObj, "publishedTs").getAsLong(); + metadataResult.add(new GatewayMetadata(connector, receivedTs, publishedTs)); + } else { + throw new JsonSyntaxException("Can't parse gateway metadata: " + metadataElem); + } + } + parseObject(systemTs, null, builder, jo); + } else { + throw new JsonSyntaxException(CAN_T_PARSE_VALUE + je); + } + } + } else { + throw new JsonSyntaxException(CAN_T_PARSE_VALUE + jsonElement); + } + return TbPair.of(builder.build(), metadataResult); + } + + private static JsonElement getAndValidateMetadataElement(JsonObject metadata, String elementName) { + var element = metadata.get(elementName); + if (element == null || element.isJsonNull()) { + throw new JsonSyntaxException(String.format("Can't parse gateway element in metadata: [%s][%s]", metadata, elementName)); + } + return element; + } + private static void convertToTelemetry(JsonElement jsonElement, long systemTs, Map> result, PostTelemetryMsg.Builder builder) { if (jsonElement.isJsonObject()) { parseObject(systemTs, result, builder, jsonElement.getAsJsonObject()); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java index b8fd79be4b..d78a4afbc2 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java @@ -22,12 +22,12 @@ import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Component; import org.thingsboard.server.common.transport.TransportContext; import org.thingsboard.server.common.transport.TransportTenantProfileCache; import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor; +import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService; import java.net.InetSocketAddress; import java.util.concurrent.atomic.AtomicInteger; @@ -37,7 +37,7 @@ import java.util.concurrent.atomic.AtomicInteger; */ @Slf4j @Component -@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.mqtt.enabled}'=='true')") +@TbMqttTransportComponent public class MqttTransportContext extends TransportContext { @Getter @@ -56,6 +56,10 @@ public class MqttTransportContext extends TransportContext { @Autowired private TransportTenantProfileCache tenantProfileCache; + @Getter + @Autowired + private GatewayMetricsService gatewayMetricsService; + @Getter @Value("${transport.mqtt.netty.max_payload_size}") private Integer maxPayloadSize; diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index f839ed4b18..d7f64e58ee 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -1464,7 +1464,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional deviceProfileOpt) { deviceSessionCtx.onDeviceUpdate(sessionInfo, device, deviceProfileOpt); if (gatewaySessionHandler != null) { - gatewaySessionHandler.onDeviceUpdate(sessionInfo, device, deviceProfileOpt); + gatewaySessionHandler.onGatewayUpdate(sessionInfo, device, deviceProfileOpt); } } @@ -1473,6 +1473,9 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement context.onAuthFailure(address); ChannelHandlerContext ctx = deviceSessionCtx.getChannel(); closeCtx(ctx, MqttReasonCodes.Disconnect.ADMINISTRATIVE_ACTION); + if (gatewaySessionHandler != null) { + gatewaySessionHandler.onGatewayDelete(deviceId); + } } public void sendErrorRpcResponse(TransportProtos.SessionInfoProto sessionInfo, int requestId, ThingsboardErrorCode result, String errorMsg) { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java index ffc7e9e561..b8d1dbe582 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java @@ -39,7 +39,7 @@ import java.net.InetSocketAddress; * @author Andrew Shvayka */ @Service("MqttTransportService") -@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.mqtt.enabled}'=='true')") +@TbMqttTransportComponent @Slf4j public class MqttTransportService implements TbTransportService { 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 new file mode 100644 index 0000000000..da0047cc7e --- /dev/null +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/TbMqttTransportComponent.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2024 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.Retention; +import java.lang.annotation.RetentionPolicy; + +@Retention(RetentionPolicy.RUNTIME) +@ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.mqtt.enabled}'=='true')") +public @interface TbMqttTransportComponent { +} diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java new file mode 100644 index 0000000000..4f67866d06 --- /dev/null +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/GatewayMetricsService.java @@ -0,0 +1,113 @@ +/** + * Copyright © 2016-2024 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.gateway; + +import jakarta.annotation.PostConstruct; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.transport.TransportService; +import org.thingsboard.server.common.transport.TransportServiceCallback; +import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.queue.scheduler.SchedulerComponent; +import org.thingsboard.server.transport.mqtt.TbMqttTransportComponent; +import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; +import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsState; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; + +@Slf4j +@Service +@TbMqttTransportComponent +public class GatewayMetricsService { + + public static final String GATEWAY_METRICS = "gatewayMetrics"; + + @Value("${transport.mqtt.gateway_metrics_report_interval_sec:60}") + private int metricsReportIntervalSec; + + @Autowired + private SchedulerComponent scheduler; + + @Autowired + private TransportService transportService; + + private Map states = new ConcurrentHashMap<>(); + + @PostConstruct + private void init() { + scheduler.scheduleAtFixedRate(this::reportMetrics, metricsReportIntervalSec, metricsReportIntervalSec, TimeUnit.SECONDS); + } + + public void process(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId, List data, long serverReceiveTs) { + states.computeIfAbsent(gatewayId, k -> new GatewayMetricsState(sessionInfo)).update(data, serverReceiveTs); + } + + public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceId gatewayId) { + var state = states.get(gatewayId); + if (state != null) { + state.updateSessionInfo(sessionInfo); + } + } + + public void onDeviceDelete(DeviceId deviceId) { + states.remove(deviceId); + } + + public void reportMetrics() { + if (states.isEmpty()) { + return; + } + Map statesToReport = states; + states = new ConcurrentHashMap<>(); + + long ts = System.currentTimeMillis(); + + statesToReport.forEach((gatewayId, state) -> { + reportMetrics(state, ts); + }); + } + + private void reportMetrics(GatewayMetricsState state, long ts) { + if (state.isEmpty()) { + return; + } + var result = state.getStateResult(); + var kvProto = TransportProtos.KeyValueProto.newBuilder() + .setKey(GATEWAY_METRICS) + .setType(TransportProtos.KeyValueType.JSON_V) + .setJsonV(JacksonUtil.toString(result)) + .build(); + + TransportProtos.TsKvListProto tsKvList = TransportProtos.TsKvListProto.newBuilder() + .setTs(ts) + .addKv(kvProto) + .build(); + + TransportProtos.PostTelemetryMsg telemetryMsg = TransportProtos.PostTelemetryMsg.newBuilder() + .addTsKvList(tsKvList) + .build(); + + transportService.process(state.getSessionInfo(), telemetryMsg, TransportServiceCallback.EMPTY); + } + +} diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java new file mode 100644 index 0000000000..a800c78e28 --- /dev/null +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/gateway/metrics/GatewayMetricsState.java @@ -0,0 +1,123 @@ +/** + * Copyright © 2016-2024 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.gateway.metrics; + +import lombok.Getter; +import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; +import org.thingsboard.server.gen.transport.TransportProtos; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +public class GatewayMetricsState { + + private final Map connectors; + private final Lock updateLock; + + @Getter + private volatile TransportProtos.SessionInfoProto sessionInfo; + + public GatewayMetricsState(TransportProtos.SessionInfoProto sessionInfo) { + this.connectors = new HashMap<>(); + this.updateLock = new ReentrantLock(); + this.sessionInfo = sessionInfo; + } + + public void updateSessionInfo(TransportProtos.SessionInfoProto sessionInfo) { + this.sessionInfo = sessionInfo; + } + + public void update(List metricsData, long serverReceiveTs) { + updateLock.lock(); + try { + metricsData.forEach(data -> { + connectors.computeIfAbsent(data.connector(), k -> new ConnectorMetricsState()).update(data, serverReceiveTs); + }); + } finally { + updateLock.unlock(); + } + } + + public Map getStateResult() { + Map result = new HashMap<>(); + updateLock.lock(); + try { + connectors.forEach((name, state) -> result.put(name, state.getResult())); + connectors.clear(); + } finally { + updateLock.unlock(); + } + + return result; + } + + public boolean isEmpty() { + return connectors.isEmpty(); + } + + private static class ConnectorMetricsState { + private final AtomicInteger count; + private final AtomicLong gwLatencySum; + private final AtomicLong transportLatencySum; + private volatile long minGwLatency; + private volatile long maxGwLatency; + private volatile long minTransportLatency; + private volatile long maxTransportLatency; + + private ConnectorMetricsState() { + this.count = new AtomicInteger(0); + this.gwLatencySum = new AtomicLong(0); + this.transportLatencySum = new AtomicLong(0); + } + + private void update(GatewayMetadata metricsData, long serverReceiveTs) { + long gwLatency = metricsData.publishedTs() - metricsData.receivedTs(); + long transportLatency = serverReceiveTs - metricsData.publishedTs(); + count.incrementAndGet(); + gwLatencySum.addAndGet(gwLatency); + transportLatencySum.addAndGet(transportLatency); + if (minGwLatency == 0 || minGwLatency > gwLatency) { + minGwLatency = gwLatency; + } + if (maxGwLatency < gwLatency) { + maxGwLatency = gwLatency; + } + if (minTransportLatency == 0 || minTransportLatency > transportLatency) { + minTransportLatency = transportLatency; + } + if (maxTransportLatency < transportLatency) { + maxTransportLatency = transportLatency; + } + } + + private ConnectorMetricsResult getResult() { + long count = this.count.get(); + long avgGwLatency = gwLatencySum.get() / count; + long avgTransportLatency = transportLatencySum.get() / count; + return new ConnectorMetricsResult(avgGwLatency, minGwLatency, maxGwLatency, avgTransportLatency, minTransportLatency, maxTransportLatency); + } + } + + public record ConnectorMetricsResult(long avgGwLatency, long minGwLatency, long maxGwLatency, + long avgTransportLatency, long minTransportLatency, long maxTransportLatency) { + } + +} diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index 3b3aaef983..2e205ff840 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -48,6 +48,8 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.util.TbPair; +import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; import org.thingsboard.server.common.msg.tools.TbRateLimitsException; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; @@ -62,6 +64,7 @@ import org.thingsboard.server.transport.mqtt.MqttTransportHandler; import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor; import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor; +import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService; import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState; import java.util.ArrayList; @@ -114,6 +117,7 @@ public abstract class AbstractGatewaySessionHandler mqttQoSMap; protected final ChannelHandlerContext channel; protected final DeviceSessionCtx deviceSessionCtx; + protected final GatewayMetricsService gatewayMetricsService; @Getter @Setter @@ -131,6 +135,7 @@ public abstract class AbstractGatewaySessionHandler createWeakMap() { @@ -380,7 +385,13 @@ public abstract class AbstractGatewaySessionHandler> gatewayPayloadPair = JsonConverter.convertToGatewayTelemetry(msg.getAsJsonArray(), systemTs); + TransportProtos.PostTelemetryMsg postTelemetryMsg = gatewayPayloadPair.getFirst(); + List metadata = gatewayPayloadPair.getSecond(); + if (!CollectionUtils.isEmpty(metadata)) { + gatewayMetricsService.process(deviceSessionCtx.getSessionInfo(), gateway.getDeviceId(), metadata, systemTs); + } transportService.process(deviceCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(channel, deviceName, msgId, postTelemetryMsg)); } catch (Throwable e) { log.warn("[{}][{}][{}] Failed to convert telemetry: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, msg, e); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java index 9aa228afff..5dce8e8893 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java @@ -17,14 +17,21 @@ package org.thingsboard.server.transport.mqtt.session; import io.netty.buffer.ByteBuf; import io.netty.handler.codec.mqtt.MqttPublishMessage; +import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.adaptor.AdaptorException; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.DeviceProfile; +import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse; +import org.thingsboard.server.gen.transport.TransportProtos; +import java.util.Optional; import java.util.UUID; /** * Created by nickAS21 on 26.12.22 */ +@Slf4j public class GatewaySessionHandler extends AbstractGatewaySessionHandler { public GatewaySessionHandler(DeviceSessionCtx deviceSessionCtx, UUID sessionId, boolean overwriteDevicesActivity) { @@ -51,7 +58,16 @@ public class GatewaySessionHandler extends AbstractGatewaySessionHandler deviceProfileOpt) { + this.onDeviceUpdate(sessionInfo, device, deviceProfileOpt); + gatewayMetricsService.onDeviceUpdate(sessionInfo, gateway.getDeviceId()); + } + + public void onGatewayDelete(DeviceId deviceId) { + gatewayMetricsService.onDeviceDelete(deviceId); } } diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index 140839ccc3..c9fd10a99d 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -146,6 +146,8 @@ transport: # MQTT disconnect timeout in milliseconds. The time to wait for the client to disconnect after the server sends a disconnect message. disconnect_timeout: "${MQTT_DISCONNECT_TIMEOUT:1000}" msg_queue_size_per_device_limit: "${MQTT_MSG_QUEUE_SIZE_PER_DEVICE_LIMIT:100}" # messages await in the queue before device connected state. This limit works on low level before TenantProfileLimits mechanism + # Interval of periodic report of the gateway metrics + gateway_metrics_report_interval_sec: "${MQTT_GATEWAY_METRICS_REPORT_INTERVAL_SEC:60}" netty: # Netty leak detector level leak_detector_level: "${NETTY_LEAK_DETECTOR_LVL:DISABLED}" diff --git a/ui-ngx/src/app/modules/home/components/widget/lib/scada/scada-symbol.models.ts b/ui-ngx/src/app/modules/home/components/widget/lib/scada/scada-symbol.models.ts index 837e3a44ee..0418665d69 100644 --- a/ui-ngx/src/app/modules/home/components/widget/lib/scada/scada-symbol.models.ts +++ b/ui-ngx/src/app/modules/home/components/widget/lib/scada/scada-symbol.models.ts @@ -15,7 +15,7 @@ /// import { ValueType } from '@shared/models/constants'; -import { Box, Element, Runner, SVG, Svg, Text, Timeline } from '@svgdotjs/svg.js'; +import { Box, Element, Runner, Style, SVG, Svg, Text, Timeline } from '@svgdotjs/svg.js'; import '@svgdotjs/svg.panzoom.js'; import { DataToValueType, @@ -58,6 +58,7 @@ export interface ScadaSymbolApi { enable: (element: Element | Element[]) => void; callAction: (event: Event, behaviorId: string, value?: any, observer?: Partial>) => void; setValue: (valueId: string, value: any) => void; + cssAnimate: CssScadaSymbolAnimations; } export interface ScadaSymbolContext { @@ -621,7 +622,8 @@ export class ScadaSymbolObject { disable: this.disableElement.bind(this), enable: this.enableElement.bind(this), callAction: this.callAction.bind(this), - setValue: this.setValue.bind(this) + setValue: this.setValue.bind(this), + cssAnimate: new CssScadaSymbolAnimations(this.svgShape) }, tags: {}, properties: {}, @@ -1008,3 +1010,301 @@ export class ScadaSymbolObject { } } } + +const scadaSymbolAnimationId = 'scadaSymbolAnimation'; + +class CssScadaSymbolAnimations { + constructor(private svgShape: Svg) {} + + public rotate(element: Element, rotation: number, duration = 1000, loop = false): CssScadaSymbolAnimation { + this.checkOldAnimation(element); + return this.setupAnimation(element, + new RotateCssScadaSymbolAnimation(this.svgShape, element, loop, rotation || 0)).duration(duration); + } + + public move(element: Element, deltaX: number, deltaY: number, duration = 1000, loop = false): CssScadaSymbolAnimation { + this.checkOldAnimation(element); + return this.setupAnimation(element, + new MoveCssScadaSymbolAnimation(this.svgShape, element, loop, deltaX, deltaY).duration(duration)); + } + + public attr(element: Element, attrName: string, value: any, duration = 1000, loop = false): CssScadaSymbolAnimation { + this.checkOldAnimation(element); + return this.setupAnimation(element, + new AttrsCssScadaSymbolAnimation(this.svgShape, element, loop, {[attrName]: value}).duration(duration)); + } + + public attrs(element: Element, attr: any, duration = 1000, loop = false): CssScadaSymbolAnimation { + this.checkOldAnimation(element); + return this.setupAnimation(element, + new AttrsCssScadaSymbolAnimation(this.svgShape, element, loop, attr).duration(duration)); + } + + public animation(element: Element): CssScadaSymbolAnimation | undefined { + return element.remember(scadaSymbolAnimationId); + } + + private checkOldAnimation(element: Element) { + const previousAnimation: CssScadaSymbolAnimation = element.remember(scadaSymbolAnimationId); + if (previousAnimation) { + previousAnimation.destroy(); + } + } + + private setupAnimation(element: Element, animation: CssScadaSymbolAnimation): CssScadaSymbolAnimation { + animation.init(); + element.remember(scadaSymbolAnimationId, animation); + return animation; + } +} + +interface ScadaSymbolAnimationKeyframe { + stop: string; + style: any; +} + +abstract class CssScadaSymbolAnimation { + + private _animationName: string; + private _animationStyle: Style; + + private _running = true; + private _speed = 1; + private _duration = 1000; + private _easing = 'linear'; + + protected constructor(protected svgShape: Svg, + protected element: Element, + protected loop: boolean) { + } + + public init() { + this.prepareAnimation(); + } + + public start() { + if (!this._running) { + this.updateAnimationStyle('animation-play-state', 'running'); + this._running = true; + } + } + + public pause() { + if (this._running) { + this.updateAnimationStyle('animation-play-state', 'paused'); + this._running = false; + } + } + + public running(): boolean { + return this._running; + } + + public destroy() { + if (this._animationStyle) { + this._animationStyle.remove(); + this.element.removeClass(this._animationName); + this._animationStyle = null; + this._animationName = null; + } + } + + public duration(duration: number): CssScadaSymbolAnimation { + this._duration = duration; + this.updateAnimationStyle(this.loop ? 'animation-duration' : 'transition-duration', + Math.round(this._duration / this._speed) + 'ms'); + return this; + } + + public speed(speed: number): CssScadaSymbolAnimation { + this._speed = speed; + this.updateAnimationStyle(this.loop ? 'animation-duration' : 'transition-duration', + Math.round(this._duration / this._speed) + 'ms'); + return this; + } + + public easing(easing: string): CssScadaSymbolAnimation { + this._easing = easing; + this.updateAnimationStyle(this.loop ? 'animation-timing-function' : 'transition-timing-function', this._easing); + return this; + } + + private prepareAnimation() { + this._animationName = 'animation_' + generateElementId(); + this._animationStyle = this.svgShape.style(); + let styles: any; + if (this.loop) { + styles = { + 'animation-name': this._animationName, + 'animation-duration': this._duration + 'ms', + 'animation-timing-function': this._easing, + 'animation-iteration-count': 'infinite', + 'animation-fill-mode': 'forwards', + 'animation-play-state': 'running', + ...this.animationStyles() + }; + } else { + styles = { + 'transition-property': this.transitionProperties(), + 'transition-duration': this._duration + 'ms', + 'transition-timing-function': this._easing, + ...this.animationStyles() + }; + } + this._animationStyle.rule('.' + this._animationName, styles); + + if (this.loop) { + const keyframes = this.animationKeyframes(); + let keyframesCss = `\n@keyframes ${this._animationName} {\n`; + for (const keyframe of keyframes) { + let keyframeCss = ` ${keyframe.stop} {\n`; + for (const i of Object.keys(keyframe.style)) { + keyframeCss += ' ' + i + ':' + keyframe.style[i] + ';\n'; + } + keyframeCss += ' }\n'; + keyframesCss += keyframeCss; + } + keyframesCss += '}'; + this._animationStyle.addText(keyframesCss); + } + this.element.addClass(this._animationName); + if (!this.loop) { + this.doTransform(); + } + } + + private updateAnimationStyle(attrName: string, value: any) { + if (this._animationStyle) { + const styleText = this._animationStyle.node.innerHTML; + const attrValueRegex = new RegExp(`${attrName}:([^;]+);`); + this._animationStyle.node.innerHTML = styleText.replace(attrValueRegex, `${attrName}:${value};`); + } + } + + protected animationStyles(): any { + return {}; + } + + protected abstract animationKeyframes(): ScadaSymbolAnimationKeyframe[]; + + protected abstract transitionProperties(): string; + + protected doTransform() { + } + +} + +class RotateCssScadaSymbolAnimation extends CssScadaSymbolAnimation { + + constructor(protected svgShape: Svg, + protected element: Element, + protected loop: boolean, + private rotation: number) { + super(svgShape, element, loop); + } + + protected animationStyles(): any { + return { + 'transform-origin': `${this.element.cx()}px ${this.element.cy()}px` + }; + } + + protected animationKeyframes(): ScadaSymbolAnimationKeyframe[] { + const transform = this.element.transform(); + return [ + { + stop: '0%', + style: { + transform: `translate(${transform.translateX}px, ${transform.translateY}px) rotate(${transform.rotate}deg)` + } + }, + { + stop: '100%', + style: { + transform: `translate(${transform.translateX}px, ${transform.translateY}px) rotate(${this.rotation}deg)` + } + } + ]; + } + + protected transitionProperties(): string { + return 'transform'; + } + + protected doTransform() { + const transform = this.element.transform(); + this.element.attr({transform: `translate(${transform.translateX} ${transform.translateY}) rotate(${this.rotation})`}); + } + +} + +class MoveCssScadaSymbolAnimation extends CssScadaSymbolAnimation { + + constructor(protected svgShape: Svg, + protected element: Element, + protected loop: boolean, + private deltaX: number, + private deltaY: number) { + super(svgShape, element, loop); + } + + protected animationStyles(): any { + return { + 'transform-origin': `${this.element.cx()}px ${this.element.cy()}px` + }; + } + + protected animationKeyframes(): ScadaSymbolAnimationKeyframe[] { + const transform = this.element.transform(); + return [ + { + stop: '0%', + style: { + transform: `translate(${transform.translateX}px, ${transform.translateY}px)` + } + }, + { + stop: '100%', + style: { + transform: `translate(${transform.translateX+this.deltaX}px, ${transform.translateY+this.deltaY}px)` + } + } + ]; + } + + protected transitionProperties(): string { + return 'transform'; + } + + protected doTransform() { + this.element.relative(this.deltaX, this.deltaY); + } + +} + +class AttrsCssScadaSymbolAnimation extends CssScadaSymbolAnimation { + + constructor(protected svgShape: Svg, + protected element: Element, + protected loop: boolean, + private attr: any) { + super(svgShape, element, loop); + } + + protected animationStyles(): any { + return {}; + } + + protected animationKeyframes(): ScadaSymbolAnimationKeyframe[] { + return []; + } + + protected transitionProperties(): string { + return Object.keys(this.attr).join(' '); + } + + protected doTransform() { + this.element.attr(this.attr); + } + +}