sparkplug: Test connect

application/src/test/java/org/thingsboard/server/controller/AbstractControllerTest.java

@@ -19,6 +19,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootContextLoader;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.web.server.LocalServerPort;
@@ -29,6 +30,7 @@ import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.web.socket.config.annotation.EnableWebSocket;
+import org.thingsboard.server.dao.timeseries.TimeseriesService;
 
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -49,6 +51,9 @@ public abstract class AbstractControllerTest extends AbstractNotifyEntityTest {
 
     public static final String WS_URL = "ws://localhost:";
 
+    @Autowired
+    public TimeseriesService tsService;
+
     @LocalServerPort
     protected int wsPort;
 

application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java

@@ -103,6 +103,7 @@ public abstract class AbstractMqttIntegrationTest extends AbstractTransportInteg
             if (StringUtils.hasLength(config.getAttributesTopicFilter())) {
                 mqttDeviceProfileTransportConfiguration.setDeviceAttributesTopic(config.getAttributesTopicFilter());
             }
+            mqttDeviceProfileTransportConfiguration.setSparkPlug(config.isSparkPlug());
             mqttDeviceProfileTransportConfiguration.setSendAckOnValidationException(config.isSendAckOnValidationException());
             TransportPayloadTypeConfiguration transportPayloadTypeConfiguration;
             if (TransportPayloadType.JSON.equals(transportPayloadType)) {

application/src/test/java/org/thingsboard/server/transport/mqtt/MqttTestConfigProperties.java

@@ -26,6 +26,7 @@ public class MqttTestConfigProperties {
 
     String deviceName;
     String gatewayName;
+    boolean isSparkPlug;
 
     TransportPayloadType transportPayloadType;
 

application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv5/MqttV5TestClient.java

@@ -89,7 +89,9 @@ public class MqttV5TestClient { // We should copy part of MqttV3TestClient, due
         if (client == null) {
             throw new RuntimeException("Failed to connect! MqttAsyncClient is not initialized!");
         }
-        return client.connect(options);
+        IMqttToken connect = client.connect(options);
+        connect.waitForCompletion(TIMEOUT_MS);
+        return connect;
     }
 
     public void disconnectAndWait() throws MqttException {

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java

@@ -0,0 +1,264 @@
+/**
+ * Copyright © 2016-2022 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.sparkplug;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.protobuf.ByteString;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.mqttv5.client.IMqttToken;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.eclipse.paho.mqttv5.common.packet.MqttConnAck;
+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.TransportPayloadType;
+import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
+import org.thingsboard.server.common.data.exception.ThingsboardException;
+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.AbstractMqttIntegrationTest;
+import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties;
+import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient;
+import org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType;
+import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType;
+import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.nio.ByteBuffer;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.awaitility.Awaitility.await;
+import static org.eclipse.paho.mqttv5.common.packet.MqttWireMessage.MESSAGE_TYPE_CONNACK;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.Int64;
+
+/**
+ * Created by nickAS21 on 12.01.23
+ */
+@Slf4j
+public  abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttIntegrationTest {
+
+    protected MqttV5TestClient client;
+    protected Calendar calendar = Calendar.getInstance();
+
+    protected static final String NAMESPACE = "spBv1.0";
+    protected static final String groupId = "SparkplugBGroupId";
+    protected static final String edgeNode = "SparkpluBNode";
+    protected static final String keysBdSeq = "bdSeq";
+    protected int bdSeq = 0;
+    protected int seq = 0;
+    protected static final long PUBLISH_TS_DELTA_MS = 86400000;// Publish start TS <-> 24h
+
+
+    public void beforeSparkplugTest() throws Exception {
+        MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder()
+                .gatewayName("Test Connect Sparkplug client node")
+                .isSparkPlug(true)
+                .transportPayloadType(TransportPayloadType.PROTOBUF)
+                .build();
+        processBeforeTest(configProperties);
+    }
+
+    public void processClientWithCorrectNodeAccess () throws Exception {
+        this.client = new MqttV5TestClient();
+        MqttWireMessage response = clientWithCorrectNodeAccessToken(client);
+        Assert.assertEquals(MESSAGE_TYPE_CONNACK, response.getType());
+        MqttConnAck connAckMsg = (MqttConnAck) response;
+        Assert.assertEquals(MqttReturnCode.RETURN_CODE_SUCCESS, connAckMsg.getReturnCode());
+    }
+
+    protected SparkplugBProto.Payload.Metric createMetric(TsKvEntry tsKvEntry, MetricDataType metricDataType) throws ThingsboardException {
+        SparkplugBProto.Payload.Metric metric = SparkplugBProto.Payload.Metric.newBuilder()
+                .setTimestamp(tsKvEntry.getTs())
+                .setName(tsKvEntry.getKey())
+                .setDatatype(metricDataType.toIntValue())
+                .build();
+        Object value = tsKvEntry.getValue();
+        switch (metricDataType) {
+            case Int8:
+            case Int16:
+            case Int32:
+            case UInt8:
+            case UInt16:
+            case UInt32:
+                return metric.toBuilder().setIntValue(Integer.parseInt(String.valueOf(value))).build();
+            case Int64:
+            case UInt64:
+            case DateTime:
+                return metric.toBuilder().setLongValue(Long.parseLong(String.valueOf(value))).build();
+            case Float:
+                return metric.toBuilder().setFloatValue(Float.parseFloat(String.valueOf(value))).build();
+            case Double:
+                return metric.toBuilder().setDoubleValue(Double.parseDouble(String.valueOf(value))).build();
+            case Boolean:
+                return metric.toBuilder().setBooleanValue(Boolean.parseBoolean(String.valueOf(value))).build();
+            case String:
+            case Text:
+            case UUID:
+                return metric.toBuilder().setDatasetValue((SparkplugBProto.Payload.DataSet) value).build();
+            case DataSet:
+                return metric.toBuilder().setStringValue(String.valueOf(value)).build();
+            case Bytes:
+            case Int8Array:
+                ByteString byteString = ByteString.copyFrom((byte[]) value);
+                return metric.toBuilder().setBytesValue(byteString).build();
+            case Int16Array:
+            case UInt8Array:
+                byte[] int16Array = shortToByte_ByteBuffer_Method((short[]) value);
+                ByteString byteInt16Array = ByteString.copyFrom((int16Array));
+                return metric.toBuilder().setBytesValue(byteInt16Array).build();
+            case Int32Array:
+            case UInt16Array:
+                byte[] int32Array = integerToByte_ByteBuffer_Method((int[]) value);
+                ByteString byteInt32Array = ByteString.copyFrom((int32Array));
+                return metric.toBuilder().setBytesValue(byteInt32Array).build();
+            case Int64Array:
+            case UInt32Array:
+                byte[] int64Array = longToByte_ByteBuffer_Method((long[]) value);
+                ByteString byteInt64Array = ByteString.copyFrom((int64Array));
+                return metric.toBuilder().setBytesValue(byteInt64Array).build();
+            case UInt64Array:
+            case DoubleArray:
+                byte[] doubleArray = doubleToByte_ByteBuffer_Method((double[]) value);
+                ByteString byteDoubleArray = ByteString.copyFrom(doubleArray);
+                return metric.toBuilder().setBytesValue(byteDoubleArray).build();
+            case FloatArray:
+                byte[] floatArray = floatToByte_ByteBuffer_Method((float[]) value);
+                ByteString byteFloatArray = ByteString.copyFrom(floatArray);
+                return metric.toBuilder().setBytesValue(byteFloatArray).build();
+            case BooleanArray:
+                byte[] booleanArray = booleanToByte_ByteBuffer_Method((boolean[]) value);
+                ByteString byteBooleanArray = ByteString.copyFrom(booleanArray);
+                return metric.toBuilder().setBytesValue(byteBooleanArray).build();
+            case StringArray:
+                byte[] stringArray = stringToByte_ByteBuffer_Method((String[]) value);
+                ByteString byteStringArray = ByteString.copyFrom(stringArray);
+                return metric.toBuilder().setBytesValue(byteStringArray).build();
+            case DateTimeArray:
+                byte[] dateArray = dateToByte_ByteBuffer_Method((Date[]) value);
+                ByteString byteDateArray = ByteString.copyFrom(dateArray);
+                return metric.toBuilder().setBytesValue(byteDateArray).build();
+            case File:
+                SparkplugMetricUtil.File file = (SparkplugMetricUtil.File) value;
+                ByteString byteFileString = ByteString.copyFrom(file.getBytes());
+                return metric.toBuilder().setBytesValue(byteFileString).build();
+            case Template:
+                return metric.toBuilder().setTemplateValue((SparkplugBProto.Payload.Template) value).build();
+            case Unknown:
+                throw new ThingsboardException("Invalid value for MetricDataType " + metricDataType.name(), ThingsboardErrorCode.INVALID_ARGUMENTS);
+        }
+        return metric;
+    }
+
+    private byte[] shortToByte_ByteBuffer_Method(short[] inputs) {
+        ByteBuffer bb = ByteBuffer.allocate(inputs.length * 2);
+        for (short d : inputs) {
+            bb.putShort(d);
+        }
+        return bb.array();
+    }
+
+    private byte[] integerToByte_ByteBuffer_Method(int[] inputs) {
+        ByteBuffer bb = ByteBuffer.allocate(inputs.length * 4);
+        for (int d : inputs) {
+            bb.putLong(d);
+        }
+        return bb.array();
+    }
+
+    private byte[] longToByte_ByteBuffer_Method(long[] inputs) {
+        ByteBuffer bb = ByteBuffer.allocate(inputs.length * 8);
+        for (long d : inputs) {
+            bb.putLong(d);
+        }
+        return bb.array();
+    }
+
+    private byte[] doubleToByte_ByteBuffer_Method(double[] inputs) {
+        ByteBuffer bb = ByteBuffer.allocate(inputs.length * 8);
+        for (double d : inputs) {
+            bb.putDouble(d);
+        }
+        return bb.array();
+    }
+
+    private byte[] floatToByte_ByteBuffer_Method(float[] inputs) throws ThingsboardException {
+        ByteArrayOutputStream bas = new ByteArrayOutputStream();
+        DataOutputStream ds = new DataOutputStream(bas);
+        for (float f : inputs) {
+            try {
+                ds.writeFloat(f);
+            } catch (IOException e) {
+                throw new ThingsboardException("Invalid value float ", ThingsboardErrorCode.INVALID_ARGUMENTS);
+            }
+        }
+        return bas.toByteArray();
+    }
+
+    private byte[] booleanToByte_ByteBuffer_Method(boolean[] inputs) {
+        byte[] toReturn = new byte[inputs.length / 8];
+        for (int entry = 0; entry < toReturn.length; entry++) {
+            for (int bit = 0; bit < 8; bit++) {
+                if (inputs[entry * 8 + bit]) {
+                    toReturn[entry] |= (128 >> bit);
+                }
+            }
+        }
+        return toReturn;
+    }
+
+    private byte[] stringToByte_ByteBuffer_Method(String[] inputs) throws ThingsboardException {
+        final ByteArrayOutputStream bas = new ByteArrayOutputStream();
+        try {
+            final ObjectOutputStream os = new ObjectOutputStream(bas);
+            os.writeObject(inputs);
+            os.flush();
+            os.close();
+        } catch (Exception e) {
+            throw new ThingsboardException("Invalid value float ", ThingsboardErrorCode.INVALID_ARGUMENTS);
+        }
+        return bas.toByteArray();
+    }
+
+    private byte[] dateToByte_ByteBuffer_Method(Date[] inputs) {
+        long[] ll = new long[inputs.length];
+        int i = 0;
+        for (Date date : inputs) {
+            ll[i] = date.getTime();
+            i++;
+        }
+        return longToByte_ByteBuffer_Method(ll);
+    }
+
+
+
+    private MqttWireMessage clientWithCorrectNodeAccessToken(MqttV5TestClient client) throws Exception {
+        IMqttToken connectionResult = client.connectAndWait(gatewayAccessToken);
+        return connectionResult.getResponse();
+    }
+
+}

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java

@@ -0,0 +1,125 @@
+/**
+ * Copyright © 2016-2022 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.sparkplug.connection;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.protobuf.ByteString;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.mqttv5.client.IMqttToken;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.eclipse.paho.mqttv5.common.packet.MqttConnAck;
+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.exception.ThingsboardErrorCode;
+import org.thingsboard.server.common.data.exception.ThingsboardException;
+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.AbstractMqttIntegrationTest;
+import org.thingsboard.server.transport.mqtt.mqttv5.MqttV5TestClient;
+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 org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.nio.ByteBuffer;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.awaitility.Awaitility.await;
+import static org.eclipse.paho.mqttv5.common.packet.MqttWireMessage.MESSAGE_TYPE_CONNACK;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.Int64;
+
+/**
+ * Created by nickAS21 on 12.01.23
+ */
+@Slf4j
+public  abstract class AbstractMqttV5ClientSparkplugConnectionTest extends AbstractMqttV5ClientSparkplugTest {
+
+    protected void processClientWithCorrectNodeAccessTokenTest() throws Exception {
+        processClientWithCorrectNodeAccess();
+    }
+
+    protected void processClientWithCorrectNodeAccessTokenWithNdeathTest() throws Exception {
+        long ts = calendar.getTimeInMillis()-PUBLISH_TS_DELTA_MS;
+        int value = bdSeq;
+        MetricDataType metricDataType = Int64;
+        TsKvEntry expectedTsKvEntryOriginal = new BasicTsKvEntry(ts, new LongDataEntry(keysBdSeq, Integer.toUnsignedLong(value)));
+
+        SparkplugBProto.Payload.Builder deathPayload = SparkplugBProto.Payload.newBuilder()
+                .setTimestamp(calendar.getTimeInMillis());
+        deathPayload.addMetrics(createMetric(expectedTsKvEntryOriginal, metricDataType));
+
+        byte[] deathBytes = deathPayload.build().toByteArray();
+
+        MqttWireMessage response = clientWithCorrectNodeAccessTokenWithNdeath(deathBytes);
+
+        Assert.assertEquals(MESSAGE_TYPE_CONNACK, response.getType());
+
+        MqttConnAck connAckMsg = (MqttConnAck) response;
+
+        Assert.assertEquals(MqttReturnCode.RETURN_CODE_SUCCESS, connAckMsg.getReturnCode());
+
+        String keys = SparkplugMessageType.NDEATH.name() + " " + keysBdSeq;
+        TsKvEntry expectedTsKvEntry = new BasicTsKvEntry(ts, new LongDataEntry(keys, Integer.toUnsignedLong(value)));
+        ListenableFuture<Optional<TsKvEntry>> future = tsService.findLatest(tenantId, savedGateway.getId(), keys);
+        AtomicReference<ListenableFuture<Optional<TsKvEntry>>> finalFuture = new AtomicReference<>(future);
+        await("Failed Post Telemetry node proto payload. SparkplugMessageType NDEATH")
+                .atMost(40, TimeUnit.SECONDS)
+                .until(() -> {
+                    finalFuture.set(tsService.findLatest(tenantId, savedGateway.getId(), keys));
+                    return finalFuture.get().get().isPresent();
+                });
+        TsKvEntry actualTsKvEntry = finalFuture.get().get().get();
+        Assert.assertEquals(expectedTsKvEntry, actualTsKvEntry);
+        client.disconnect();
+    }
+
+    protected void processClientWithCorrectAccessTokenCreatedDevices(int cntDevices) throws Exception {
+        processClientWithCorrectNodeAccess();
+
+
+
+    }
+
+    private MqttWireMessage clientWithCorrectNodeAccessTokenWithNdeath(byte[] deathBytes) throws Exception {
+        this.client = new MqttV5TestClient();
+        MqttConnectionOptions options = new MqttConnectionOptions();
+        options.setUserName(gatewayAccessToken);
+        if (deathBytes != null) {
+            String topic = NAMESPACE + "/" + groupId + "/" + SparkplugMessageType.NDEATH.name() + "/" + edgeNode;
+            MqttMessage msg = new MqttMessage();
+            msg.setId(0);
+            msg.setPayload(deathBytes);
+            options.setWill(topic, msg);
+        }
+        IMqttToken connectionResult = client.connect(options);
+        return connectionResult.getResponse();
+    }
+
+}

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/MqttV5ClientSparkplugBConnectionTest.java

@@ -0,0 +1,62 @@
+/**
+ * Copyright © 2016-2022 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.sparkplug.connection;
+
+import org.eclipse.paho.mqttv5.common.MqttException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.thingsboard.server.dao.service.DaoSqlTest;
+
+/**
+ * Created by nickAS21 on 12.01.23
+ */
+@DaoSqlTest
+public class MqttV5ClientSparkplugBConnectionTest extends AbstractMqttV5ClientSparkplugConnectionTest {
+
+    @Before
+    public void beforeTest() throws Exception {
+        beforeSparkplugTest();
+    }
+
+    @After
+    public void afterTest() throws MqttException {
+        if (client.isConnected()) {
+            client.disconnect();
+        }
+    }
+
+    @Test
+    public void testClientWithCorrectAccessToken() throws Exception {
+        processClientWithCorrectNodeAccessTokenTest();
+    }
+
+    @Test
+    public void testClientWithCorrectAccessTokenWithNDEATH() throws Exception {
+        processClientWithCorrectNodeAccessTokenWithNdeathTest();
+    }
+
+    @Test
+    public void testClientWithCorrectAccessTokenCreatedOneDevice() throws Exception {
+        processClientWithCorrectAccessTokenCreatedDevices(1);
+    }
+
+    @Test
+    public void testClientWithCorrectAccessTokenCreatedTwoDevice() throws Exception {
+        processClientWithCorrectAccessTokenCreatedDevices(2);
+    }
+
+}

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java

@@ -0,0 +1,65 @@
+/**
+ * Copyright © 2016-2022 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.sparkplug.timeseries;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.protobuf.ByteString;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.mqttv5.client.IMqttToken;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.eclipse.paho.mqttv5.common.packet.MqttConnAck;
+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.exception.ThingsboardErrorCode;
+import org.thingsboard.server.common.data.exception.ThingsboardException;
+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.mqttv5.MqttV5TestClient;
+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 org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.nio.ByteBuffer;
+import java.util.Date;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.awaitility.Awaitility.await;
+import static org.eclipse.paho.mqttv5.common.packet.MqttWireMessage.MESSAGE_TYPE_CONNACK;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.Int64;
+
+/**
+ * Created by nickAS21 on 12.01.23
+ */
+@Slf4j
+public  abstract class AbstractMqttV5ClientSparkplugTelemetryTest extends AbstractMqttV5ClientSparkplugTest {
+
+    protected void processPushTelemetry() throws Exception {
+        processClientWithCorrectNodeAccess();
+
+    }
+
+}

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/MqttV5ClientSparkplugBTelemetryTest.java

@@ -0,0 +1,47 @@
+/**
+ * Copyright © 2016-2022 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.sparkplug.timeseries;
+
+import org.eclipse.paho.mqttv5.common.MqttException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.thingsboard.server.dao.service.DaoSqlTest;
+import org.thingsboard.server.transport.mqtt.sparkplug.connection.AbstractMqttV5ClientSparkplugConnectionTest;
+
+/**
+ * Created by nickAS21 on 12.01.23
+ */
+@DaoSqlTest
+public class MqttV5ClientSparkplugBTelemetryTest extends AbstractMqttV5ClientSparkplugTelemetryTest {
+
+    @Before
+    public void beforeTest() throws Exception {
+        beforeSparkplugTest();
+    }
+
+    @After
+    public void afterTest () throws MqttException {
+        if (client.isConnected()) {
+            client.disconnect();        }
+    }
+
+    @Test
+    public void testPushTelemetry() throws Exception {
+        processPushTelemetry();
+    }
+
+}