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 a9bdbb55db..2c050ea1d1 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 @@ -288,6 +288,33 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte return listKeys; } + protected List connectionWithNBirthMetricNameAndAlias(MetricDataType metricDataType, String metricKey, Object metricValue, Long alias) throws Exception { + List listKeys = new ArrayList<>(); + SparkplugBProto.Payload.Builder payloadBirthNode = SparkplugBProto.Payload.newBuilder() + .setTimestamp(calendar.getTimeInMillis()); + long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS; + long valueBdSec = getBdSeqNum(); + payloadBirthNode.addMetrics(createMetric(valueBdSec, ts, keysBdSeq, Int64, -1L)); + listKeys.add(SparkplugMessageType.NBIRTH.name() + " " + keysBdSeq); + payloadBirthNode.addMetrics(createMetric(false, ts, keyNodeRebirth, MetricDataType.Boolean, -1L)); + listKeys.add(keyNodeRebirth); + + payloadBirthNode.addMetrics(createMetric(metricValue, ts, metricKey, metricDataType, alias)); + + listKeys.add(metricKey); + + if (client.isConnected()) { + client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode, + payloadBirthNode.build().toByteArray(), 0, false); + } + return listKeys; + } + + protected void createdAddMetricValueWithAliasTsKv(SparkplugBProto.Payload.Builder dataPayload, Object value, MetricDataType metricDataType, + long ts) throws ThingsboardException { + dataPayload.addMetrics(createMetric(value, ts, null, metricDataType, 4L)); + } + protected void createdAddMetricValuePrimitiveTsKv(List listTsKvEntry, List listKeys, SparkplugBProto.Payload.Builder dataPayload, long ts) throws ThingsboardException { diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java index 319b61b7f3..5125bef09f 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java @@ -18,9 +18,12 @@ package org.thingsboard.server.transport.mqtt.sparkplug.timeseries; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.junit.Assert; +import org.thingsboard.server.common.data.kv.BasicTsKvEntry; +import org.thingsboard.server.common.data.kv.LongDataEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.gen.transport.mqtt.SparkplugBProto; import org.thingsboard.server.transport.mqtt.sparkplug.AbstractMqttV5ClientSparkplugTest; +import org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType; import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType; import java.util.ArrayList; @@ -51,6 +54,52 @@ public abstract class AbstractMqttV5ClientSparkplugTelemetryTest extends Abstrac Assert.assertEquals(listKeys.size(), finalFuture.get().get().size()); } + protected void processClientWithCorrectAccessTokenPublishNBIRTH_NDATA_AliasWithoutMetricName() throws Exception { + clientWithCorrectNodeAccessTokenWithNDEATH(); + String metricBirthName_Alias = "Node Metric int32 Only Alias"; + MetricDataType metricBirthDataType = metricBirthDataType_Int32; + List listKeys = connectionWithNBirthMetricNameAndAlias(metricBirthDataType, metricBirthName_Alias, nextInt32(), 4L); + Assert.assertTrue("Connection node is failed", client.isConnected()); + AtomicReference>> finalFuture = new AtomicReference<>(); + await(alias + SparkplugMessageType.NBIRTH.name()) + .atMost(40, TimeUnit.SECONDS) + .until(() -> { + finalFuture.set(tsService.findLatest(tenantId, savedGateway.getId(), listKeys)); + return !finalFuture.get().get().isEmpty(); + }); + Assert.assertEquals(listKeys.size(), finalFuture.get().get().size()); + + String messageTypeName = SparkplugMessageType.NDATA.name(); + + List listTsKvEntry = new ArrayList<>(); + + SparkplugBProto.Payload.Builder ndataPayload = SparkplugBProto.Payload.newBuilder() + .setTimestamp(calendar.getTimeInMillis()) + .setSeq(getSeqNum()); + long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS; + int valueKey = nextInt32(); + + TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new LongDataEntry(metricBirthName_Alias, Long.valueOf(String.valueOf(valueKey)))); + listTsKvEntry.add(tsKvEntry); + createdAddMetricValueWithAliasTsKv(ndataPayload, valueKey, metricBirthDataType, ts); + + if (client.isConnected()) { + client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + messageTypeName + "/" + edgeNode, + ndataPayload.build().toByteArray(), 0, false); + } + + AtomicReference>> finalFutureAlias = new AtomicReference<>(); + await(alias + SparkplugMessageType.NDATA.name()) + .atMost(40, TimeUnit.SECONDS) + .until(() -> { + finalFutureAlias.set(tsService.findAllLatest(tenantId, savedGateway.getId())); + return finalFutureAlias.get().get().size() == (listKeys.size() + listTsKvEntry.size() + 1); + }); + Assert.assertTrue("Actual tsKvEntries is not containsAll Expected tsKvEntries", containsIgnoreVersion(finalFutureAlias .get().get(), listTsKvEntry)); + + + } + protected void processClientWithCorrectAccessTokenPushNodeMetricBuildPrimitiveSimple() throws Exception { List listKeys = new ArrayList<>(); clientWithCorrectNodeAccessTokenWithNDEATH(); diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/MqttV5ClientSparkplugBTelemetryTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/MqttV5ClientSparkplugBTelemetryTest.java index bafe2d81d2..da6949d02c 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/MqttV5ClientSparkplugBTelemetryTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/MqttV5ClientSparkplugBTelemetryTest.java @@ -43,6 +43,11 @@ public class MqttV5ClientSparkplugBTelemetryTest extends AbstractMqttV5ClientSpa processClientWithCorrectAccessTokenPublishNBIRTH(); } + @Test + public void testClientWithCorrectAccessTokenPublishNBIRTH_NDATA_AliasWithoutMetricName() throws Exception { + processClientWithCorrectAccessTokenPublishNBIRTH_NDATA_AliasWithoutMetricName(); + } + @Test public void testClientWithCorrectAccessTokenPushNodeMetricBuildPrimitiveSimple() throws Exception { processClientWithCorrectAccessTokenPushNodeMetricBuildPrimitiveSimple(); 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 d7cea8c61b..390e2453d6 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 @@ -257,8 +257,8 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler keyValueProtoOpt = fromSparkplugBMetricToKeyValueProto(key, protoMetric); keyValueProtoOpt.ifPresent(kvProto -> msgs.add(postTelemetryMsgCreated(kvProto, ts))); }