diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractControllerTest.java index ed8d8632da..f3e632b4b4 100644 --- a/application/src/test/java/org/thingsboard/server/controller/AbstractControllerTest.java +++ b/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; diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java index 917e41d1b4..28e458d971 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/AbstractMqttIntegrationTest.java +++ b/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)) { diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/MqttTestConfigProperties.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/MqttTestConfigProperties.java index bc535b424b..1a2b6eefa4 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/MqttTestConfigProperties.java +++ b/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; diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv5/MqttV5TestClient.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv5/MqttV5TestClient.java index 0671dd6f95..0af79c5f06 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv5/MqttV5TestClient.java +++ b/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 { 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 new file mode 100644 index 0000000000..5a58c0372c --- /dev/null +++ b/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(); + } + +} 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 new file mode 100644 index 0000000000..09a0a2ef78 --- /dev/null +++ b/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> future = tsService.findLatest(tenantId, savedGateway.getId(), keys); + AtomicReference>> 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(); + } + +} diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/MqttV5ClientSparkplugBConnectionTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/MqttV5ClientSparkplugBConnectionTest.java new file mode 100644 index 0000000000..85f708d0f8 --- /dev/null +++ b/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); + } + +} 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 new file mode 100644 index 0000000000..70f3cdf9eb --- /dev/null +++ b/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(); + + } + +} 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 new file mode 100644 index 0000000000..042cbae51e --- /dev/null +++ b/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(); + } + +}