From 9873b817e6100d499021f143e08ccfb594988327 Mon Sep 17 00:00:00 2001 From: nick Date: Tue, 25 Feb 2025 15:53:26 +0200 Subject: [PATCH] sparkplug: fix bug alias --- .../AbstractMqttV5ClientSparkplugTest.java | 88 +++++++++++++++---- ...ctMqttV5ClientSparkplugConnectionTest.java | 2 +- .../rpc/AbstractMqttV5RpcSparkplugTest.java | 32 ++++++- .../sparkplug/rpc/MqttV5RpcSparkplugTest.java | 9 ++ .../SparkplugDeviceSessionContext.java | 16 +++- .../session/SparkplugNodeSessionHandler.java | 86 +++++++++++++----- .../util/sparkplug/SparkplugMetricUtil.java | 21 +++-- 7 files changed, 204 insertions(+), 50 deletions(-) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java index f10621842f..c512527066 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java @@ -29,6 +29,7 @@ import org.eclipse.paho.mqttv5.common.packet.MqttReturnCode; import org.eclipse.paho.mqttv5.common.packet.MqttWireMessage; import org.junit.Assert; import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.TransportPayloadType; import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.data.kv.BasicTsKvEntry; @@ -112,21 +113,29 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte long value = bdSeq = 0; clientWithCorrectNodeAccessTokenWithNDEATH(ts, value); } + public void clientWithCorrectNodeAccessTokenWithNDEATH(Long alias) throws Exception { + long ts = calendar.getTimeInMillis(); + long value = bdSeq = 0; + clientMqttV5ConnectWithNDEATH(ts, value,alias); + } public void clientWithCorrectNodeAccessTokenWithNDEATH(long ts, long value) throws Exception { - IMqttToken connectionResult = clientMqttV5ConnectWithNDEATH(ts, value); + IMqttToken connectionResult = clientMqttV5ConnectWithNDEATH(ts, value, -1L); MqttWireMessage response = connectionResult.getResponse(); Assert.assertEquals(MESSAGE_TYPE_CONNACK, response.getType()); MqttConnAck connAckMsg = (MqttConnAck) response; Assert.assertEquals(MqttReturnCode.RETURN_CODE_SUCCESS, connAckMsg.getReturnCode()); } - public IMqttToken clientMqttV5ConnectWithNDEATH(long ts, long value, String... nameSpaceBad) throws Exception { - String key = keysBdSeq; + public IMqttToken clientMqttV5ConnectWithNDEATH(long ts, long value, Long alias, String... nameSpaceBad) throws Exception { + return clientMqttV5ConnectWithNDEATH(ts, value, null, alias, nameSpaceBad); + } + public IMqttToken clientMqttV5ConnectWithNDEATH(long ts, long value, String metricName, Long alias, String... nameSpaceBad) throws Exception { + String key = metricName == null ? keysBdSeq : metricName; MetricDataType metricDataType = Int64; SparkplugBProto.Payload.Builder deathPayload = SparkplugBProto.Payload.newBuilder() .setTimestamp(calendar.getTimeInMillis()); - deathPayload.addMetrics(createMetric(value, ts, key, metricDataType)); + deathPayload.addMetrics(createMetric(value, ts, key, metricDataType, alias)); byte[] deathBytes = deathPayload.build().toByteArray(); this.client = new MqttV5TestClient(); this.mqttCallback = new SparkplugMqttCallback(); @@ -153,7 +162,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte MetricDataType metricDataType = Int32; String key = "Node Metric int32"; int valueDeviceInt32 = 1024; - SparkplugBProto.Payload.Metric metric = createMetric(valueDeviceInt32, ts, key, metricDataType); + SparkplugBProto.Payload.Metric metric = createMetric(valueDeviceInt32, ts, key, metricDataType, -1L); SparkplugBProto.Payload.Builder payloadBirthNode = SparkplugBProto.Payload.newBuilder() .setTimestamp(ts) .setSeq(getBdSeqNum()); @@ -165,7 +174,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte } valueDeviceInt32 = 4024; - metric = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32); + metric = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32, -1L); for (int i = 0; i < cntDevices; i++) { SparkplugBProto.Payload.Builder payloadBirthDevice = SparkplugBProto.Payload.newBuilder() .setTimestamp(ts) @@ -192,6 +201,49 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte return devices; } + protected List connectClientWithCorrectAccessTokenWithNDEATHWithAliasCreatedDevices(long ts) throws Exception { + List devices = new ArrayList<>(); + Long alias = 0L; + clientWithCorrectNodeAccessTokenWithNDEATH(alias++); + MetricDataType metricDataType = Int32; + String key = "Node Metric int32"; + int valueDeviceInt32 = 1024; + SparkplugBProto.Payload.Metric metric = createMetric(valueDeviceInt32, ts, key, metricDataType, alias++); + SparkplugBProto.Payload.Builder payloadBirthNode = SparkplugBProto.Payload.newBuilder() + .setTimestamp(ts) + .setSeq(getBdSeqNum()); + payloadBirthNode.addMetrics(metric); + payloadBirthNode.setTimestamp(ts); + if (client.isConnected()) { + client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode, + payloadBirthNode.build().toByteArray(), 0, false); + } + + valueDeviceInt32 = 4024; + metric = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32, alias++); + SparkplugBProto.Payload.Builder payloadBirthDevice = SparkplugBProto.Payload.newBuilder() + .setTimestamp(ts) + .setSeq(getSeqNum()); + String deviceName = deviceId + "_" + 1; + + payloadBirthDevice.addMetrics(metric); + if (client.isConnected()) { + client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DBIRTH.name() + "/" + edgeNode + "/" + deviceName, + payloadBirthDevice.build().toByteArray(), 0, false); + AtomicReference device = new AtomicReference<>(); + await(alias + "find device [" + deviceName + "] after created") + .atMost(200, TimeUnit.SECONDS) + .until(() -> { + device.set(doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class)); + return device.get() != null; + }); + devices.add(device.get()); + } + + Assert.assertEquals(1, devices.size()); + return devices; + } + protected long getBdSeqNum() throws Exception { if (bdSeq == 256) { bdSeq = 0; @@ -212,12 +264,16 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte .setTimestamp(calendar.getTimeInMillis()); long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS; long valueBdSec = getBdSeqNum(); - payloadBirthNode.addMetrics(createMetric(valueBdSec, ts, keysBdSeq, Int64)); + payloadBirthNode.addMetrics(createMetric(valueBdSec, ts, keysBdSeq, Int64, -1L)); listKeys.add(SparkplugMessageType.NBIRTH.name() + " " + keysBdSeq); - payloadBirthNode.addMetrics(createMetric(false, ts, keyNodeRebirth, MetricDataType.Boolean)); + payloadBirthNode.addMetrics(createMetric(false, ts, keyNodeRebirth, MetricDataType.Boolean, -1L)); listKeys.add(keyNodeRebirth); - payloadBirthNode.addMetrics(createMetric(metricValue, ts, metricKey, metricDataType)); + if (StringUtils.isNotBlank(metricKey)) { + payloadBirthNode.addMetrics(createMetric(metricValue, ts, metricKey, metricDataType, -1L)); + } else { + payloadBirthNode.addMetrics(createMetric(metricValue, ts, metricKey, metricDataType, 4L)); + } listKeys.add(metricKey); if (client.isConnected()) { @@ -302,7 +358,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte private TsKvEntry createdAddMetricTsKvLong(SparkplugBProto.Payload.Builder dataPayload, String key, Object value, long ts, MetricDataType metricDataType) throws ThingsboardException { TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new LongDataEntry(key, Long.valueOf(String.valueOf(value)))); - dataPayload.addMetrics(createMetric(value, ts, key, metricDataType)); + dataPayload.addMetrics(createMetric(value, ts, key, metricDataType, -1L)); return tsKvEntry; } @@ -310,7 +366,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte long ts, MetricDataType metricDataType) throws ThingsboardException { Double dd = Double.parseDouble(Float.toString(value)); TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new DoubleDataEntry(key, dd)); - dataPayload.addMetrics(createMetric(value, ts, key, metricDataType)); + dataPayload.addMetrics(createMetric(value, ts, key, metricDataType, -1L)); return tsKvEntry; } @@ -318,21 +374,21 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte long ts, MetricDataType metricDataType) throws ThingsboardException { Long l = Double.valueOf(value).longValue(); TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new LongDataEntry(key, l)); - dataPayload.addMetrics(createMetric(value, ts, key, metricDataType)); + dataPayload.addMetrics(createMetric(value, ts, key, metricDataType, -1L)); return tsKvEntry; } private TsKvEntry createdAddMetricTsKvBoolean(SparkplugBProto.Payload.Builder dataPayload, String key, boolean value, long ts, MetricDataType metricDataType) throws ThingsboardException { TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new BooleanDataEntry(key, value)); - dataPayload.addMetrics(createMetric(value, ts, key, metricDataType)); + dataPayload.addMetrics(createMetric(value, ts, key, metricDataType, -1L)); return tsKvEntry; } private TsKvEntry createdAddMetricTsKvString(SparkplugBProto.Payload.Builder dataPayload, String key, String value, long ts, MetricDataType metricDataType) throws ThingsboardException { TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new StringDataEntry(key, value)); - dataPayload.addMetrics(createMetric(value, ts, key, metricDataType)); + dataPayload.addMetrics(createMetric(value, ts, key, metricDataType, -1L)); return tsKvEntry; } @@ -353,7 +409,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte if (nodeArray.size() > 0) { Optional tsKvEntryOptional = Optional.of(new BasicTsKvEntry(ts, new JsonDataEntry(key, nodeArray.toString()))); if (tsKvEntryOptional.isPresent()) { - dataPayload.addMetrics(createMetric(values, ts, key, metricDataType)); + dataPayload.addMetrics(createMetric(values, ts, key, metricDataType, -1L)); listTsKvEntry.add(tsKvEntryOptional.get()); listKeys.add(key); } @@ -421,7 +477,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte return java.util.UUID.randomUUID().toString(); } - public class SparkplugMqttCallback implements MqttCallback { + public class SparkplugMqttCallback implements MqttCallback { private final List messageArrivedMetrics = new ArrayList<>(); @Override diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java index dee295604c..941b7c0c69 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java @@ -80,7 +80,7 @@ public abstract class AbstractMqttV5ClientSparkplugConnectionTest extends Abstra protected void processClientWithCorrectNodeAccessTokenNameSpaceInvalid_Test() throws Exception { long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS; long value = bdSeq = 0; - MqttException actualException = Assert.assertThrows(MqttException.class, () -> clientMqttV5ConnectWithNDEATH(ts, value, "spBv1.2")); + MqttException actualException = Assert.assertThrows(MqttException.class, () -> clientMqttV5ConnectWithNDEATH(ts, value, -1L,"spBv1.2")); String expectedMessage = "Server unavailable."; int expectedReasonCode = 136; Assert.assertEquals(expectedMessage, actualException.getMessage()); diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/AbstractMqttV5RpcSparkplugTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/AbstractMqttV5RpcSparkplugTest.java index 4bec661c7d..47a07abdbe 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/AbstractMqttV5RpcSparkplugTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/AbstractMqttV5RpcSparkplugTest.java @@ -15,8 +15,8 @@ */ package org.thingsboard.server.transport.mqtt.sparkplug.rpc; -import io.netty.handler.codec.mqtt.MqttQoS; import lombok.extern.slf4j.Slf4j; +import org.eclipse.paho.mqttv5.common.MqttException; import org.junit.Assert; import org.junit.Test; import org.thingsboard.server.common.data.Device; @@ -74,6 +74,36 @@ public abstract class AbstractMqttV5RpcSparkplugTest extends AbstractMqttV5Clie Assert.assertTrue(metricBirthValue_Int32 == mqttCallback.getMessageArrivedMetrics().get(0).getIntValue()); } + @Test + public void processClientDeviceWithCorrectAccessTokenPublishWithAlias_TwoWayRpc_Success() throws Exception { + long ts = calendar.getTimeInMillis(); + List devices = connectClientWithCorrectAccessTokenWithNDEATHWithAliasCreatedDevices(ts); + awaitForDeviceActorToReceiveSubscription(devices.get(0).getId(), FeatureType.RPC, 1); + String expected = "{\"result\":\"Success: " + DCMD.name() + "\"}"; + String actual = sendRPCSparkplug(DCMD.name() , sparkplugRpcRequest, devices.get(0)); + await(alias + DCMD.name()) + .atMost(40, TimeUnit.SECONDS) + .until(() -> { + return mqttCallback.getMessageArrivedMetrics().size() == 1; + }); + Assert.assertEquals(expected, actual); + Assert.assertFalse(mqttCallback.getMessageArrivedMetrics().get(0).hasName()); + Assert.assertTrue(mqttCallback.getMessageArrivedMetrics().get(0).hasAlias()); + Assert.assertTrue(2L == mqttCallback.getMessageArrivedMetrics().get(0).getAlias()); + Assert.assertTrue(metricBirthValue_Int32 == mqttCallback.getMessageArrivedMetrics().get(0).getIntValue()); + } + + @Test + public void processClientNodeWithCorrectAccessTokenPublishWithAliasWithoutMetricName_TwoWayRpc_BAD_REQUEST_PARAMS() throws Exception { + long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS; + long value = bdSeq = 0; + MqttException actualException = Assert.assertThrows(MqttException.class, () -> clientMqttV5ConnectWithNDEATH(ts, value, "",4L)); + String expectedMessage = "Server unavailable."; + int expectedReasonCode = 136; + Assert.assertEquals(expectedMessage, actualException.getMessage()); + Assert.assertEquals(expectedReasonCode, actualException.getReasonCode()); + } + @Test public void processClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_InvalidTypeMessage_INVALID_ARGUMENTS() throws Exception { clientWithCorrectNodeAccessTokenWithNDEATH(); diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/MqttV5RpcSparkplugTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/MqttV5RpcSparkplugTest.java index fec60ad122..0c97bca36b 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/MqttV5RpcSparkplugTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/MqttV5RpcSparkplugTest.java @@ -47,6 +47,15 @@ public class MqttV5RpcSparkplugTest extends AbstractMqttV5RpcSparkplugTest { public void testClientDeviceWithCorrectAccessTokenPublish_TwoWayRpc_Success() throws Exception { processClientDeviceWithCorrectAccessTokenPublish_TwoWayRpc_Success(); } + @Test + public void testClientDeviceWithCorrectAccessTokenPublishWithAlias_TwoWayRpc_Success() throws Exception { + processClientDeviceWithCorrectAccessTokenPublishWithAlias_TwoWayRpc_Success(); + } + + @Test + public void testClientNodeWithCorrectAccessTokenPublishWithAliasWithoutMetricName_TwoWayRpc_BAD_REQUEST_PARAMS() throws Exception { + processClientNodeWithCorrectAccessTokenPublishWithAliasWithoutMetricName_TwoWayRpc_BAD_REQUEST_PARAMS(); + } @Test public void testClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_InvalidTypeMessage_INVALID_ARGUMENTS() throws Exception { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java index ccbb0e19b4..e854402815 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java @@ -16,6 +16,7 @@ package org.thingsboard.server.transport.mqtt.session; import lombok.extern.slf4j.Slf4j; +import org.springframework.dao.DuplicateKeyException; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; @@ -33,7 +34,6 @@ import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.stream.Collectors; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.DCMD; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.getTsKvProto; @@ -57,9 +57,17 @@ public class SparkplugDeviceSessionContext extends AbstractGatewayDeviceSessionC return deviceBirthMetrics; } - public void setDeviceBirthMetrics(java.util.List metrics) { - this.deviceBirthMetrics.putAll(metrics.stream() - .collect(Collectors.toMap(SparkplugBProto.Payload.Metric::getName, metric -> metric))); + public void setDeviceBirthMetrics(java.util.List metrics) { + for (var metric : metrics) { + if (metric.hasName()) { + this.deviceBirthMetrics.put(metric.getName(), metric); + } else { + throw new IllegalArgumentException("The metric name of device: '" + this.getDeviceInfo().getDeviceName() + "' must not be empty or null! Metric: [" + metric + "]"); + } + if (metric.hasAlias() && this.parent.getNodeAlias().putIfAbsent(metric.getAlias(), metric.getName()) != null) { + throw new DuplicateKeyException("The alias '" + metric.getAlias() + "' already exists in device: '" + this.getDeviceInfo().getDeviceName() + "'"); + } + } } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java index 2a3270ab5e..434d4ec486 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java @@ -24,6 +24,7 @@ import io.netty.handler.codec.mqtt.MqttPublishMessage; import io.netty.handler.codec.mqtt.MqttTopicSubscription; import lombok.Getter; import lombok.extern.slf4j.Slf4j; +import org.springframework.dao.DuplicateKeyException; import org.springframework.util.CollectionUtils; import org.thingsboard.server.common.adaptor.AdaptorException; import org.thingsboard.server.common.adaptor.ProtoConverter; @@ -47,12 +48,10 @@ import java.util.Optional; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; -import java.util.stream.Collectors; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.ONLINE; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.DBIRTH; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.NBIRTH; -import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.STATE; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.parseMessageType; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.SPARKPLUG_BD_SEQUENCE_NUMBER_KEY; import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.SPARKPLUG_SEQUENCE_NUMBER_KEY; @@ -73,6 +72,8 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler nodeBirthMetrics; + @Getter + private final Map nodeAlias; private final MqttTransportHandler parent; public SparkplugNodeSessionHandler(MqttTransportHandler parent, DeviceSessionCtx deviceSessionCtx, UUID sessionId, @@ -81,11 +82,20 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler(); + this.nodeAlias = new ConcurrentHashMap<>(); } - public void setNodeBirthMetrics(java.util.List metrics) { - this.nodeBirthMetrics.putAll(metrics.stream() - .collect(Collectors.toMap(SparkplugBProto.Payload.Metric::getName, metric -> metric))); + public void setNodeBirthMetrics(java.util.List metrics) throws AdaptorException { + for (var metric : metrics) { + if (metric.hasName()) { + this.nodeBirthMetrics.put(metric.getName(), metric); + } else { + throw new AdaptorException("The metric name of edgeNode: '" + this.sparkplugTopicNode.getEdgeNodeId() + "' must not be empty or null! Metric: [" + metric + "]"); + } + if (metric.hasAlias() && this.nodeAlias.putIfAbsent(metric.getAlias(), metric.getName()) != null) { + throw new AdaptorException("The alias '" + metric.getAlias() + "' already exists in edgeNode: '" + this.sparkplugTopicNode.getEdgeNodeId() + "'"); + } + } } @@ -113,7 +123,11 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler 6.4.6. Metric + * https://sparkplug.eclipse.org/specification/version/3.0/documents/sparkplug-specification-3.0.0.pdf#%5B%7B%22num%22%3A339%2C%22gen%22%3A0%7D%2C%7B%22name%22%3A%22XYZ%22%7D%2C0%2C455.52%2Cnull%5D + * [tck-id-payloads-name-requirement] The name MUST be included with every metric unless aliases are being used. ◦ All UTF-8 characters are allowed in the metric name. However, special characters including but not limited to the following are discouraged: . , \ @ # $ % ^ & * ( ) [ ] { } | ! ` ~ : ; ' " < > ?. This is because many Sparkplug Host Applications may have issues handling them. + * • alias (are optional and not required): + * - This is an unsigned 64-bit integer representing an optional alias for a Sparkplug B payload. + * - If aliases are used, the following rules apply: + * -- [tck-id-payloads-alias-uniqueness] If supplied in an NBIRTH or BIRTH it MUST be a unique number across this Edge Node’s entire set of metrics. + * -- no two metrics for the same Edge Node can have the same alias. + * -- [tck-id-payloads-alias-birth-requirement] NBIRTH and DBIRTH messages MUST include both a metric name and alias. + * -- [tck-id-payloads-alias-data-cmd-requirement] NDATA, DDATA, NCMD, and DCMD messages MUST only include an alias and the metric name MUST be excluded. + * @param sparkplugBProto + * @param attributesMetricNames + * @param topicTypeName + * @return + * @throws AdaptorException + */ private List convertToPostTelemetry(SparkplugBProto.Payload sparkplugBProto, Set attributesMetricNames, String topicTypeName) throws AdaptorException { try { List msgs = new ArrayList<>(); for (SparkplugBProto.Payload.Metric protoMetric : sparkplugBProto.getMetricsList()) { - if (attributesMetricNames == null || !matches(attributesMetricNames, protoMetric)) { - long ts = protoMetric.getTimestamp(); - String key = SPARKPLUG_BD_SEQUENCE_NUMBER_KEY.equals(protoMetric.getName()) ? - topicTypeName + " " + protoMetric.getName() : protoMetric.getName(); - Optional keyValueProtoOpt = fromSparkplugBMetricToKeyValueProto(key, protoMetric); - keyValueProtoOpt.ifPresent(kvProto -> msgs.add(postTelemetryMsgCreated(kvProto, ts))); + String metricName = protoMetric.hasName() ? protoMetric.getName() : protoMetric.hasAlias() ? this.nodeAlias.get(protoMetric.getAlias()) : null; + if (metricName == null) { + throw new ThingsboardException("Metric without metricName and alias.", ThingsboardErrorCode.INVALID_ARGUMENTS); + } else { + if (attributesMetricNames == null || !matches(attributesMetricNames, metricName)) { + long ts = protoMetric.getTimestamp(); + String key = SPARKPLUG_BD_SEQUENCE_NUMBER_KEY.equals(protoMetric.getName()) ? + topicTypeName + " " + protoMetric.getName() : protoMetric.getName(); + Optional keyValueProtoOpt = fromSparkplugBMetricToKeyValueProto(key, protoMetric); + keyValueProtoOpt.ifPresent(kvProto -> msgs.add(postTelemetryMsgCreated(kvProto, ts))); + } } } @@ -237,13 +273,18 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler msgs = new ArrayList<>(); for (SparkplugBProto.Payload.Metric protoMetric : sparkplugBProto.getMetricsList()) { - if (matches(attributesMetricNames, protoMetric)) { - TransportApiProtos.AttributesMsg.Builder deviceAttributesMsgBuilder = TransportApiProtos.AttributesMsg.newBuilder(); - Optional msgOpt = getPostAttributeMsg(protoMetric); - if (msgOpt.isPresent()) { - deviceAttributesMsgBuilder.setDeviceName(deviceName); - deviceAttributesMsgBuilder.setMsg(msgOpt.get()); - msgs.add(deviceAttributesMsgBuilder.build()); + String metricName = protoMetric.hasName() ? protoMetric.getName() : protoMetric.hasAlias() ? this.nodeAlias.get(protoMetric.getAlias()) : null; + if (metricName == null) { + throw new ThingsboardException("Metric without metricName and alias.", ThingsboardErrorCode.INVALID_ARGUMENTS); + } else { + if (matches(attributesMetricNames, metricName)) { + TransportApiProtos.AttributesMsg.Builder deviceAttributesMsgBuilder = TransportApiProtos.AttributesMsg.newBuilder(); + Optional msgOpt = getPostAttributeMsg(protoMetric); + if (msgOpt.isPresent()) { + deviceAttributesMsgBuilder.setDeviceName(deviceName); + deviceAttributesMsgBuilder.setMsg(msgOpt.get()); + msgs.add(deviceAttributesMsgBuilder.build()); + } } } } @@ -254,8 +295,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler attributesMetricNames, SparkplugBProto.Payload.Metric protoMetric) { - String metricName = protoMetric.getName(); + private boolean matches(Set attributesMetricNames, String metricName) { for (String attributeMetricFilter : attributesMetricNames) { if (metricName.equals(attributeMetricFilter) || (attributeMetricFilter.endsWith("*") && metricName.startsWith( @@ -288,7 +328,9 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler= 0) { + metric.setAlias(alias); + } + if (StringUtils.isNotBlank(key)) { + metric.setName(key); + } + return addToMetricValue(value, metric.build(), metricDataType); + } - public static SparkplugBProto.Payload.Metric createMetric(Object value, long ts, String key, MetricDataType metricDataType) throws ThingsboardException { - SparkplugBProto.Payload.Metric metric = SparkplugBProto.Payload.Metric.newBuilder() - .setTimestamp(ts) - .setName(key) - .setDatatype(metricDataType.toIntValue()) - .build(); + public static SparkplugBProto.Payload.Metric addToMetricValue(Object value, SparkplugBProto.Payload.Metric metric, MetricDataType metricDataType) throws ThingsboardException { switch (metricDataType) { case Int8: // (byte) return metric.toBuilder().setIntValue(((Byte) value).intValue()).build();