272 changed files with 11707 additions and 2823 deletions
@ -0,0 +1,462 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.fasterxml.jackson.databind.node.ArrayNode; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.eclipse.paho.mqttv5.client.IMqttToken; |
||||
|
import org.eclipse.paho.mqttv5.client.MqttCallback; |
||||
|
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions; |
||||
|
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse; |
||||
|
import org.eclipse.paho.mqttv5.common.MqttException; |
||||
|
import org.eclipse.paho.mqttv5.common.MqttMessage; |
||||
|
import org.eclipse.paho.mqttv5.common.packet.MqttConnAck; |
||||
|
import org.eclipse.paho.mqttv5.common.packet.MqttProperties; |
||||
|
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.TransportPayloadType; |
||||
|
import org.thingsboard.server.common.data.exception.ThingsboardException; |
||||
|
import org.thingsboard.server.common.data.kv.BasicTsKvEntry; |
||||
|
import org.thingsboard.server.common.data.kv.BooleanDataEntry; |
||||
|
import org.thingsboard.server.common.data.kv.DoubleDataEntry; |
||||
|
import org.thingsboard.server.common.data.kv.JsonDataEntry; |
||||
|
import org.thingsboard.server.common.data.kv.LongDataEntry; |
||||
|
import org.thingsboard.server.common.data.kv.StringDataEntry; |
||||
|
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 java.util.ArrayList; |
||||
|
import java.util.Calendar; |
||||
|
import java.util.List; |
||||
|
import java.util.Optional; |
||||
|
import java.util.Set; |
||||
|
import java.util.concurrent.ThreadLocalRandom; |
||||
|
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.common.util.JacksonUtil.newArrayNode; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.Bytes; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.Int16; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.Int32; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.Int64; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.Int8; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.UInt16; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.UInt32; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.UInt64; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.UInt8; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.createMetric; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicUtil.NAMESPACE; |
||||
|
|
||||
|
/** |
||||
|
* Created by nickAS21 on 12.01.23 |
||||
|
*/ |
||||
|
@Slf4j |
||||
|
public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttIntegrationTest { |
||||
|
|
||||
|
protected MqttV5TestClient client; |
||||
|
protected SparkplugMqttCallback mqttCallback; |
||||
|
protected Calendar calendar = Calendar.getInstance(); |
||||
|
protected ThreadLocalRandom random = ThreadLocalRandom.current(); |
||||
|
|
||||
|
protected static final String groupId = "SparkplugBGroupId"; |
||||
|
protected static final String edgeNode = "SparkpluBNode"; |
||||
|
protected static final String keysBdSeq = "bdSeq"; |
||||
|
protected static final String alias = "Failed Telemetry/Attribute proto sparkplug payload. SparkplugMessageType "; |
||||
|
protected String deviceId = "Test Sparkplug B Device"; |
||||
|
protected int bdSeq = 0; |
||||
|
protected int seq = 0; |
||||
|
protected static final long PUBLISH_TS_DELTA_MS = 86400000;// Publish start TS <-> 24h
|
||||
|
|
||||
|
// NBIRTH
|
||||
|
protected static final String keyNodeRebirth = "Node Control/Rebirth"; |
||||
|
|
||||
|
//*BIRTH
|
||||
|
protected static final MetricDataType metricBirthDataType_Int32 = Int32; |
||||
|
protected static final String metricBirthName_Int32 = "Device Metric int32"; |
||||
|
protected Set<String> sparkplugAttributesMetricNames; |
||||
|
|
||||
|
public void beforeSparkplugTest() throws Exception { |
||||
|
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
||||
|
.gatewayName("Test Connect Sparkplug client node") |
||||
|
.isSparkplug(true) |
||||
|
.sparkplugAttributesMetricNames(sparkplugAttributesMetricNames) |
||||
|
.transportPayloadType(TransportPayloadType.PROTOBUF) |
||||
|
.build(); |
||||
|
processBeforeTest(configProperties); |
||||
|
} |
||||
|
|
||||
|
public void clientWithCorrectNodeAccessTokenWithNDEATH() throws Exception { |
||||
|
long ts = calendar.getTimeInMillis(); |
||||
|
long value = bdSeq = 0; |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(ts, value); |
||||
|
} |
||||
|
|
||||
|
public void clientWithCorrectNodeAccessTokenWithNDEATH(long ts, long value) throws Exception { |
||||
|
IMqttToken connectionResult = clientConnectWithNDEATH(ts, value); |
||||
|
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 clientConnectWithNDEATH(long ts, long value, String... nameSpaceBad) throws Exception { |
||||
|
String key = keysBdSeq; |
||||
|
MetricDataType metricDataType = Int64; |
||||
|
SparkplugBProto.Payload.Builder deathPayload = SparkplugBProto.Payload.newBuilder() |
||||
|
.setTimestamp(calendar.getTimeInMillis()); |
||||
|
deathPayload.addMetrics(createMetric(value, ts, key, metricDataType)); |
||||
|
byte[] deathBytes = deathPayload.build().toByteArray(); |
||||
|
this.client = new MqttV5TestClient(); |
||||
|
this.mqttCallback = new SparkplugMqttCallback(); |
||||
|
this.client.setCallback(this.mqttCallback); |
||||
|
MqttConnectionOptions options = new MqttConnectionOptions(); |
||||
|
options.setUserName(gatewayAccessToken); |
||||
|
String nameSpace = nameSpaceBad.length == 0 ? NAMESPACE : nameSpaceBad[0]; |
||||
|
String topic = nameSpace + "/" + groupId + "/" + SparkplugMessageType.NDEATH.name() + "/" + edgeNode; |
||||
|
MqttMessage msg = new MqttMessage(); |
||||
|
msg.setId(0); |
||||
|
msg.setPayload(deathBytes); |
||||
|
options.setWill(topic, msg); |
||||
|
return client.connect(options); |
||||
|
} |
||||
|
|
||||
|
protected List<Device> connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(int cntDevices, long ts) throws Exception { |
||||
|
List<Device> devices = new ArrayList<>(); |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
MetricDataType metricDataType = Int32; |
||||
|
String key = "Node Metric int32"; |
||||
|
int valueDeviceInt32 = 1024; |
||||
|
SparkplugBProto.Payload.Metric metric = createMetric(valueDeviceInt32, ts, key, metricDataType); |
||||
|
SparkplugBProto.Payload.Builder payloadBirthNode = SparkplugBProto.Payload.newBuilder() |
||||
|
.setTimestamp(ts) |
||||
|
.setSeq(getBdSeqNum()); |
||||
|
payloadBirthNode.addMetrics(metric); |
||||
|
payloadBirthNode.setTimestamp(ts); |
||||
|
if (client.isConnected()) { |
||||
|
client.publish(NAMESPACE + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode, |
||||
|
payloadBirthNode.build().toByteArray(), 0, false); |
||||
|
} |
||||
|
|
||||
|
valueDeviceInt32 = 4024; |
||||
|
metric = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32); |
||||
|
for (int i = 0; i < cntDevices; i++) { |
||||
|
SparkplugBProto.Payload.Builder payloadBirthDevice = SparkplugBProto.Payload.newBuilder() |
||||
|
.setTimestamp(ts) |
||||
|
.setSeq(getSeqNum()); |
||||
|
String deviceName = deviceId + "_" + i; |
||||
|
|
||||
|
payloadBirthDevice.addMetrics(metric); |
||||
|
if (client.isConnected()) { |
||||
|
client.publish(NAMESPACE + "/" + groupId + "/" + SparkplugMessageType.DBIRTH.name() + "/" + edgeNode + "/" + deviceName, |
||||
|
payloadBirthDevice.build().toByteArray(), 0, false); |
||||
|
AtomicReference<Device> 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(cntDevices, devices.size()); |
||||
|
return devices; |
||||
|
} |
||||
|
|
||||
|
protected long getBdSeqNum() throws Exception { |
||||
|
if (bdSeq == 256) { |
||||
|
bdSeq = 0; |
||||
|
} |
||||
|
return bdSeq++; |
||||
|
} |
||||
|
|
||||
|
protected long getSeqNum() throws Exception { |
||||
|
if (seq == 256) { |
||||
|
seq = 0; |
||||
|
} |
||||
|
return seq++; |
||||
|
} |
||||
|
|
||||
|
protected List<String> connectionWithNBirth(MetricDataType metricDataType, String metricKey, Object metricValue) throws Exception { |
||||
|
List<String> 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)); |
||||
|
listKeys.add(SparkplugMessageType.NBIRTH.name() + " " + keysBdSeq); |
||||
|
payloadBirthNode.addMetrics(createMetric(false, ts, keyNodeRebirth, MetricDataType.Boolean)); |
||||
|
listKeys.add(keyNodeRebirth); |
||||
|
|
||||
|
payloadBirthNode.addMetrics(createMetric(metricValue, ts, metricKey, metricDataType)); |
||||
|
listKeys.add(metricKey); |
||||
|
|
||||
|
if (client.isConnected()) { |
||||
|
client.publish(NAMESPACE + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode, |
||||
|
payloadBirthNode.build().toByteArray(), 0, false); |
||||
|
} |
||||
|
return listKeys; |
||||
|
} |
||||
|
|
||||
|
protected void createdAddMetricValuePrimitiveTsKv(List<TsKvEntry> listTsKvEntry, List<String> listKeys, |
||||
|
SparkplugBProto.Payload.Builder dataPayload, long ts) throws ThingsboardException { |
||||
|
|
||||
|
String keys = "MyInt8"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvLong(dataPayload, keys, nextInt8(), ts, Int8)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyInt16"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvLong(dataPayload, keys, nextInt16(), ts, Int16)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyInt32"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvLong(dataPayload, keys, nextInt32(), ts, Int32)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyInt64"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvLong(dataPayload, keys, nextInt64(), ts, Int64)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyUInt8"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvLong(dataPayload, keys, nextUInt8(), ts, UInt8)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyUInt16"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvLong(dataPayload, keys, nextUInt16(), ts, UInt16)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyUInt32"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvLong(dataPayload, keys, nextUInt32(), ts, UInt32)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyUInt64"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvLong(dataPayload, keys, nextUInt64(), ts, UInt64)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyFloat"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvFloat(dataPayload, keys, nextFloat(0, 100), ts, MetricDataType.Float)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyDateTime"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvLong(dataPayload, keys, nextDateTime(), ts, MetricDataType.DateTime)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyDouble"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvDouble(dataPayload, keys, nextDouble(), ts, MetricDataType.Double)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyBoolean"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvBoolean(dataPayload, keys, nextBoolean(), ts, MetricDataType.Boolean)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyString"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvString(dataPayload, keys, nextString(), ts, MetricDataType.String)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyText"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvString(dataPayload, keys, nextString(), ts, MetricDataType.Text)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
keys = "MyUUID"; |
||||
|
listTsKvEntry.add(createdAddMetricTsKvString(dataPayload, keys, nextString(), ts, MetricDataType.UUID)); |
||||
|
listKeys.add(keys); |
||||
|
|
||||
|
} |
||||
|
|
||||
|
protected void createdAddMetricValueArraysPrimitiveTsKv(List<TsKvEntry> listTsKvEntry, List<String> listKeys, |
||||
|
SparkplugBProto.Payload.Builder dataPayload, long ts) throws ThingsboardException { |
||||
|
String keys = "MyBytesArray"; |
||||
|
byte[] bytes = {nextInt8(), nextInt8(), nextInt8()}; |
||||
|
createdAddMetricTsKvJson(dataPayload, keys, bytes, ts, Bytes, listTsKvEntry, listKeys); |
||||
|
} |
||||
|
|
||||
|
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)); |
||||
|
return tsKvEntry; |
||||
|
} |
||||
|
|
||||
|
private TsKvEntry createdAddMetricTsKvFloat(SparkplugBProto.Payload.Builder dataPayload, String key, float value, |
||||
|
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)); |
||||
|
return tsKvEntry; |
||||
|
} |
||||
|
|
||||
|
private TsKvEntry createdAddMetricTsKvDouble(SparkplugBProto.Payload.Builder dataPayload, String key, double value, |
||||
|
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)); |
||||
|
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)); |
||||
|
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)); |
||||
|
return tsKvEntry; |
||||
|
} |
||||
|
|
||||
|
private void createdAddMetricTsKvJson(SparkplugBProto.Payload.Builder dataPayload, String key, |
||||
|
Object values, long ts, MetricDataType metricDataType, |
||||
|
List<TsKvEntry> listTsKvEntry, |
||||
|
List<String> listKeys) throws ThingsboardException { |
||||
|
ArrayNode nodeArray = newArrayNode(); |
||||
|
switch (metricDataType) { |
||||
|
case Bytes: |
||||
|
for (byte b : (byte[]) values) { |
||||
|
nodeArray.add(b); |
||||
|
} |
||||
|
break; |
||||
|
default: |
||||
|
throw new IllegalStateException("Unexpected value: " + metricDataType); |
||||
|
} |
||||
|
if (nodeArray.size() > 0) { |
||||
|
Optional<TsKvEntry> tsKvEntryOptional = Optional.of(new BasicTsKvEntry(ts, new JsonDataEntry(key, nodeArray.toString()))); |
||||
|
if (tsKvEntryOptional.isPresent()) { |
||||
|
dataPayload.addMetrics(createMetric(values, ts, key, metricDataType)); |
||||
|
listTsKvEntry.add(tsKvEntryOptional.get()); |
||||
|
listKeys.add(key); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private byte nextInt8() { |
||||
|
return (byte) random.nextInt(Byte.MIN_VALUE, Byte.MAX_VALUE); |
||||
|
} |
||||
|
|
||||
|
private short nextUInt8() { |
||||
|
return (short) random.nextInt(0, Byte.MAX_VALUE * 2 + 1); |
||||
|
} |
||||
|
|
||||
|
private short nextInt16() { |
||||
|
return (short) random.nextInt(Short.MIN_VALUE, Short.MAX_VALUE); |
||||
|
} |
||||
|
|
||||
|
private int nextUInt16() { |
||||
|
return random.nextInt(0, Short.MAX_VALUE * 2 + 1); |
||||
|
} |
||||
|
|
||||
|
protected int nextInt32() { |
||||
|
return random.nextInt(Integer.MIN_VALUE, Integer.MAX_VALUE); |
||||
|
} |
||||
|
|
||||
|
protected long nextUInt32() { |
||||
|
long l = Integer.MAX_VALUE; |
||||
|
return random.nextLong(0, l * 2 + 1); |
||||
|
} |
||||
|
|
||||
|
private long nextInt64() { |
||||
|
return random.nextLong(Long.MIN_VALUE, Long.MAX_VALUE); |
||||
|
} |
||||
|
|
||||
|
private long nextUInt64() { |
||||
|
double d = Long.MAX_VALUE; |
||||
|
return random.nextLong(0, (long) (d * 2 + 1)); |
||||
|
} |
||||
|
|
||||
|
protected double nextDouble() { |
||||
|
return random.nextDouble(Long.MIN_VALUE, Long.MAX_VALUE); |
||||
|
} |
||||
|
|
||||
|
private long nextDateTime() { |
||||
|
long min = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS; |
||||
|
long max = calendar.getTimeInMillis(); |
||||
|
return random.nextLong(min, max); |
||||
|
} |
||||
|
|
||||
|
protected float nextFloat(float min, float max) { |
||||
|
if (min >= max) |
||||
|
throw new IllegalArgumentException("max must be greater than min"); |
||||
|
float result = ThreadLocalRandom.current().nextFloat() * (max - min) + min; |
||||
|
if (result >= max) // correct for rounding
|
||||
|
result = Float.intBitsToFloat(Float.floatToIntBits(max) - 1); |
||||
|
return result; |
||||
|
} |
||||
|
|
||||
|
protected boolean nextBoolean() { |
||||
|
return random.nextBoolean(); |
||||
|
} |
||||
|
|
||||
|
protected String nextString() { |
||||
|
return java.util.UUID.randomUUID().toString(); |
||||
|
} |
||||
|
|
||||
|
public class SparkplugMqttCallback implements MqttCallback { |
||||
|
private final List<SparkplugBProto.Payload.Metric> messageArrivedMetrics = new ArrayList<>(); |
||||
|
|
||||
|
@Override |
||||
|
public void disconnected(MqttDisconnectResponse mqttDisconnectResponse) { |
||||
|
|
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void mqttErrorOccurred(MqttException e) { |
||||
|
|
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void messageArrived(String topic, MqttMessage mqttMsg) throws Exception { |
||||
|
SparkplugBProto.Payload sparkplugBProtoNode = SparkplugBProto.Payload.parseFrom(mqttMsg.getPayload()); |
||||
|
messageArrivedMetrics.addAll(sparkplugBProtoNode.getMetricsList()); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void deliveryComplete(IMqttToken iMqttToken) { |
||||
|
|
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void connectComplete(boolean b, String s) { |
||||
|
|
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void authPacketArrived(int i, MqttProperties mqttProperties) { |
||||
|
|
||||
|
} |
||||
|
|
||||
|
public List<SparkplugBProto.Payload.Metric> getMessageArrivedMetrics() { |
||||
|
return messageArrivedMetrics; |
||||
|
} |
||||
|
|
||||
|
public void deleteMessageArrivedMetrics(int id) { |
||||
|
messageArrivedMetrics.remove(id); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,432 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.attributes; |
||||
|
|
||||
|
import com.fasterxml.jackson.core.type.TypeReference; |
||||
|
import io.netty.handler.codec.mqtt.MqttQoS; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.junit.Assert; |
||||
|
import org.thingsboard.server.common.data.Device; |
||||
|
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.List; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
import java.util.concurrent.atomic.AtomicReference; |
||||
|
|
||||
|
import static org.awaitility.Awaitility.await; |
||||
|
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
||||
|
import static org.thingsboard.server.common.data.DataConstants.CLIENT_SCOPE; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.MetricDataType.UInt32; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.NCMD; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicUtil.NAMESPACE; |
||||
|
|
||||
|
/** |
||||
|
* Created by nickAS21 on 12.01.23 |
||||
|
*/ |
||||
|
@Slf4j |
||||
|
public abstract class AbstractMqttV5ClientSparkplugAttributesTest extends AbstractMqttV5ClientSparkplugTest { |
||||
|
|
||||
|
protected void processClientWithCorrectAccessTokenPublishNCMDReBirth() throws Exception { |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
List<String> listKeys = connectionWithNBirth(metricBirthDataType_Int32, metricBirthName_Int32, nextInt32()); |
||||
|
// Shared attribute "Node Control/Rebirth" = true. type = NCMD.
|
||||
|
boolean value = true; |
||||
|
Assert.assertTrue(listKeys.contains(keyNodeRebirth)); |
||||
|
String SHARED_ATTRIBUTES_PAYLOAD = "{\"" + keyNodeRebirth + "\":" + value + "}"; |
||||
|
Assert.assertTrue("Connection node is failed", client.isConnected()); |
||||
|
client.subscribeAndWait(NAMESPACE + "/" + groupId + "/" + NCMD.name() + "/" + edgeNode + "/#", MqttQoS.AT_MOST_ONCE); |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(keyNodeRebirth, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertTrue(mqttCallback.getMessageArrivedMetrics().get(0).getBooleanValue()); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* If boolean - send long 0 or 1 |
||||
|
* If String - try to parse long |
||||
|
* If double - cast long |
||||
|
* If we can't parse, cast, or JSON there - debug the message with the id of the devise/node, tenant, |
||||
|
* the name and type of the attribute into an error and don't send anything. |
||||
|
*/ |
||||
|
protected void processClientWithCorrectAccessTokenPublishNCMD_BooleanType_IfMetricFailedTypeCheck_SendValueOk() throws Exception { |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
MetricDataType metricDataType = MetricDataType.Boolean; |
||||
|
String metricKey = "MyBoolean"; |
||||
|
Object metricValue = nextBoolean(); |
||||
|
connectionWithNBirth(metricDataType, metricKey, metricValue); |
||||
|
Assert.assertTrue("Connection node is failed", client.isConnected()); |
||||
|
client.subscribeAndWait(NAMESPACE + "/" + groupId + "/" + NCMD.name() + "/" + edgeNode + "/#", MqttQoS.AT_MOST_ONCE); |
||||
|
|
||||
|
// Boolean <-> String
|
||||
|
boolean expectedValue = true; |
||||
|
String valueStr = "123"; |
||||
|
String SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueStr + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getBooleanValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
expectedValue = false; |
||||
|
valueStr = "0"; |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueStr + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getBooleanValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
// Boolean <-> Integer
|
||||
|
expectedValue = true; |
||||
|
Integer valueInt = nextInt32(); |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueInt + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getBooleanValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
expectedValue = false; |
||||
|
valueInt = 0; |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueInt + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getBooleanValue()); |
||||
|
} |
||||
|
|
||||
|
protected void processClientWithCorrectAccessTokenPublishNCMD_LongType_IfMetricFailedTypeCheck_SendValueOk() throws Exception { |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
MetricDataType metricDataType = UInt32; |
||||
|
String metricKey = "MyLong"; |
||||
|
Object metricValue = nextUInt32(); |
||||
|
connectionWithNBirth(metricDataType, metricKey, metricValue); |
||||
|
Assert.assertTrue("Connection node is failed", client.isConnected()); |
||||
|
client.subscribeAndWait(NAMESPACE + "/" + groupId + "/" + NCMD.name() + "/" + edgeNode + "/#", MqttQoS.AT_MOST_ONCE); |
||||
|
|
||||
|
// Long <-> String
|
||||
|
String valueStr = "123"; |
||||
|
long expectedValue = Long.valueOf(valueStr); |
||||
|
|
||||
|
String SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueStr + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getLongValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
// Long <-> Boolean
|
||||
|
Boolean valueBoolean = true; |
||||
|
expectedValue = 1L; |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueBoolean + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getLongValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
valueBoolean = false; |
||||
|
expectedValue = 0L; |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueBoolean + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getLongValue()); |
||||
|
} |
||||
|
|
||||
|
protected void processClientWithCorrectAccessTokenPublishNCMD_FloatType_IfMetricFailedTypeCheck_SendValueOk() throws Exception { |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
MetricDataType metricDataType = MetricDataType.Float; |
||||
|
String metricKey = "MyFloat"; |
||||
|
Object metricValue = nextFloat(30, 400); |
||||
|
connectionWithNBirth(metricDataType, metricKey, metricValue); |
||||
|
Assert.assertTrue("Connection node is failed", client.isConnected()); |
||||
|
client.subscribeAndWait(NAMESPACE + "/" + groupId + "/" + NCMD.name() + "/" + edgeNode + "/#", MqttQoS.AT_MOST_ONCE); |
||||
|
|
||||
|
// Float <-> String
|
||||
|
String valueStr = "123.345"; |
||||
|
float expectedValue = Float.valueOf(valueStr); |
||||
|
|
||||
|
String SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueStr + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertTrue(expectedValue == mqttCallback.getMessageArrivedMetrics().get(0).getFloatValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
// Float <-> Boolean
|
||||
|
Boolean valueBoolean = true; |
||||
|
expectedValue = 1f; |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueBoolean + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertTrue(expectedValue == mqttCallback.getMessageArrivedMetrics().get(0).getFloatValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
valueBoolean = false; |
||||
|
expectedValue = 0f; |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueBoolean + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertTrue(expectedValue == mqttCallback.getMessageArrivedMetrics().get(0).getFloatValue()); |
||||
|
} |
||||
|
|
||||
|
protected void processClientWithCorrectAccessTokenPublishNCMD_DoubleType_IfMetricFailedTypeCheck_SendValueOk() throws Exception { |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
MetricDataType metricDataType = MetricDataType.Double; |
||||
|
String metricKey = "MyDouble"; |
||||
|
Object metricValue = nextDouble(); |
||||
|
connectionWithNBirth(metricDataType, metricKey, metricValue); |
||||
|
Assert.assertTrue("Connection node is failed", client.isConnected()); |
||||
|
client.subscribeAndWait(NAMESPACE + "/" + groupId + "/" + NCMD.name() + "/" + edgeNode + "/#", MqttQoS.AT_MOST_ONCE); |
||||
|
|
||||
|
// Double <-> String
|
||||
|
String valueStr = "123345456"; |
||||
|
double expectedValue = Double.valueOf(valueStr); |
||||
|
|
||||
|
String SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueStr + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertTrue(expectedValue == mqttCallback.getMessageArrivedMetrics().get(0).getDoubleValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
// Double <-> Boolean
|
||||
|
Boolean valueBoolean = true; |
||||
|
expectedValue = 1d; |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueBoolean + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertTrue(expectedValue == mqttCallback.getMessageArrivedMetrics().get(0).getDoubleValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
valueBoolean = false; |
||||
|
expectedValue = 0d; |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueBoolean + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertTrue(expectedValue == mqttCallback.getMessageArrivedMetrics().get(0).getDoubleValue()); |
||||
|
} |
||||
|
|
||||
|
protected void processClientWithCorrectAccessTokenPublishNCMD_StringType_IfMetricFailedTypeCheck_SendValueOk() throws Exception { |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
MetricDataType metricDataType = MetricDataType.String; |
||||
|
String metricKey = "MyString"; |
||||
|
Object metricValue = nextString(); |
||||
|
connectionWithNBirth(metricDataType, metricKey, metricValue); |
||||
|
Assert.assertTrue("Connection node is failed", client.isConnected()); |
||||
|
client.subscribeAndWait(NAMESPACE + "/" + groupId + "/" + NCMD.name() + "/" + edgeNode + "/#", MqttQoS.AT_MOST_ONCE); |
||||
|
|
||||
|
// String <-> Long
|
||||
|
long valueLong = 123345456L; |
||||
|
String expectedValue = String.valueOf(valueLong); |
||||
|
|
||||
|
String SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueLong + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getStringValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
// String <-> Boolean
|
||||
|
Boolean valueBoolean = true; |
||||
|
expectedValue = "true"; |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueBoolean + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getStringValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
valueBoolean = false; |
||||
|
expectedValue = "false"; |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricKey + "\":" + valueBoolean + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricKey, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getStringValue()); |
||||
|
} |
||||
|
|
||||
|
protected void processClientDeviceWithCorrectAccessTokenPublishWithBirth_SharedAttribute() throws Exception { |
||||
|
long ts = calendar.getTimeInMillis(); |
||||
|
List<Device> devices = connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(1, ts); |
||||
|
|
||||
|
// Integer <-> Integer
|
||||
|
int expectedValueInt = 123456; |
||||
|
|
||||
|
String SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricBirthName_Int32 + "\":" + expectedValueInt + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + devices.get(0).getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.DBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricBirthName_Int32, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(metricBirthName_Int32, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValueInt, mqttCallback.getMessageArrivedMetrics().get(0).getIntValue()); |
||||
|
} |
||||
|
|
||||
|
protected void processClientDeviceWithCorrectAccessTokenPublishWithBirth_SharedAttributes_LongType_IfMetricFailedTypeCheck_SendValueOk() throws Exception { |
||||
|
long ts = calendar.getTimeInMillis(); |
||||
|
List<Device> devices = connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(1, ts); |
||||
|
|
||||
|
// Int <-> String
|
||||
|
String valueStr = "123"; |
||||
|
long expectedValue = Long.valueOf(valueStr); |
||||
|
|
||||
|
String SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricBirthName_Int32 + "\":" + valueStr + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + devices.get(0).getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.DBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricBirthName_Int32, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getIntValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
// Int <-> Boolean
|
||||
|
Boolean valueBoolean = true; |
||||
|
expectedValue = 1; |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricBirthName_Int32 + "\":" + valueBoolean + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + devices.get(0).getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricBirthName_Int32, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getIntValue()); |
||||
|
mqttCallback.deleteMessageArrivedMetrics(0); |
||||
|
|
||||
|
valueBoolean = false; |
||||
|
expectedValue = 0; |
||||
|
SHARED_ATTRIBUTES_PAYLOAD = "{\"" + metricBirthName_Int32 + "\":" + valueBoolean + "}"; |
||||
|
doPostAsync("/api/plugins/telemetry/DEVICE/" + devices.get(0).getId().getId() + "/attributes/SHARED_SCOPE", SHARED_ATTRIBUTES_PAYLOAD, String.class, status().isOk()); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricBirthName_Int32, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertEquals(expectedValue, mqttCallback.getMessageArrivedMetrics().get(0).getIntValue()); |
||||
|
} |
||||
|
|
||||
|
protected void processClientNodeWithCorrectAccessTokenPublish_AttributesInProfileContainsKeyAttributes() throws Exception { |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
connectionWithNBirth(metricBirthDataType_Int32, metricBirthName_Int32, nextInt32()); |
||||
|
String urlTemplate = "/api/plugins/telemetry/DEVICE/" + savedGateway.getId().getId() + "/keys/attributes/" + CLIENT_SCOPE; |
||||
|
AtomicReference<List<String>> actualKeys = new AtomicReference<>(); |
||||
|
await(alias + SparkplugMessageType.NBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
actualKeys.set(doGetAsyncTyped(urlTemplate, new TypeReference<>() { |
||||
|
})); |
||||
|
return actualKeys.get().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricBirthName_Int32, actualKeys.get().get(0)); |
||||
|
} |
||||
|
|
||||
|
protected void processClientDeviceWithCorrectAccessTokenPublish_AttributesInProfileContainsKeyAttributes() throws Exception { |
||||
|
long ts = calendar.getTimeInMillis(); |
||||
|
List<Device> devices = connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(1, ts); |
||||
|
String urlTemplate = "/api/plugins/telemetry/DEVICE/" + devices.get(0).getId().getId() + "/keys/attributes/" + CLIENT_SCOPE; |
||||
|
AtomicReference<List<String>> actualKeys = new AtomicReference<>(); |
||||
|
await(alias + SparkplugMessageType.DBIRTH.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
actualKeys.set(doGetAsyncTyped(urlTemplate, new TypeReference<>() { |
||||
|
})); |
||||
|
return actualKeys.get().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(metricBirthName_Int32, actualKeys.get().get(0)); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,55 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.attributes; |
||||
|
|
||||
|
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 java.util.HashSet; |
||||
|
|
||||
|
/** |
||||
|
* Created by nickAS21 on 12.01.23 |
||||
|
*/ |
||||
|
@DaoSqlTest |
||||
|
public class MqttV5ClientSparkplugBAttributesInProfileTest extends AbstractMqttV5ClientSparkplugAttributesTest { |
||||
|
|
||||
|
@Before |
||||
|
public void beforeTest() throws Exception { |
||||
|
sparkplugAttributesMetricNames = new HashSet<>(); |
||||
|
sparkplugAttributesMetricNames.add(metricBirthName_Int32); |
||||
|
beforeSparkplugTest(); |
||||
|
} |
||||
|
|
||||
|
@After |
||||
|
public void afterTest () throws MqttException { |
||||
|
if (client.isConnected()) { |
||||
|
client.disconnect(); } |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientNodeWithCorrectAccessTokenPublish_AttributesInProfileContainsKeyAttributes() throws Exception { |
||||
|
processClientNodeWithCorrectAccessTokenPublish_AttributesInProfileContainsKeyAttributes(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientDeviceWithCorrectAccessTokenPublish_AttributesInProfileContainsKeyAttributes() throws Exception { |
||||
|
processClientDeviceWithCorrectAccessTokenPublish_AttributesInProfileContainsKeyAttributes(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,81 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.attributes; |
||||
|
|
||||
|
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 MqttV5ClientSparkplugBAttributesTest extends AbstractMqttV5ClientSparkplugAttributesTest { |
||||
|
|
||||
|
@Before |
||||
|
public void beforeTest() throws Exception { |
||||
|
beforeSparkplugTest(); |
||||
|
} |
||||
|
|
||||
|
@After |
||||
|
public void afterTest () throws MqttException { |
||||
|
if (client.isConnected()) { |
||||
|
client.disconnect(); } |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectAccessTokenPublishNCMDReBirth() throws Exception { |
||||
|
processClientWithCorrectAccessTokenPublishNCMDReBirth(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectAccessTokenPublishNCMD_BooleanType_IfMetricFailedTypeCheck_SendValueOk() throws Exception { |
||||
|
processClientWithCorrectAccessTokenPublishNCMD_BooleanType_IfMetricFailedTypeCheck_SendValueOk(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectAccessTokenPublishNCMD_LongType_IfMetricFailedTypeCheck_SendValueOk() throws Exception { |
||||
|
processClientWithCorrectAccessTokenPublishNCMD_LongType_IfMetricFailedTypeCheck_SendValueOk(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectAccessTokenPublishNCMD_FloatType_IfMetricFailedTypeCheck_SendValueOk() throws Exception { |
||||
|
processClientWithCorrectAccessTokenPublishNCMD_FloatType_IfMetricFailedTypeCheck_SendValueOk(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectAccessTokenPublishNCMD_DoubleType_IfMetricFailedTypeCheck_SendValueOk() throws Exception { |
||||
|
processClientWithCorrectAccessTokenPublishNCMD_DoubleType_IfMetricFailedTypeCheck_SendValueOk(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectAccessTokenPublishNCMD_StringType_IfMetricFailedTypeCheck_SendValueOk() throws Exception { |
||||
|
processClientWithCorrectAccessTokenPublishNCMD_StringType_IfMetricFailedTypeCheck_SendValueOk(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientDeviceWithCorrectAccessTokenPublishWithBirth_SharedAttribute() throws Exception { |
||||
|
processClientDeviceWithCorrectAccessTokenPublishWithBirth_SharedAttribute(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientDeviceWithCorrectAccessTokenPublishWithBirth_SharedAttributes_LongType_IfMetricFailedTypeCheck_SendValueOk() throws Exception { |
||||
|
processClientDeviceWithCorrectAccessTokenPublishWithBirth_SharedAttributes_LongType_IfMetricFailedTypeCheck_SendValueOk(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,178 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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 lombok.extern.slf4j.Slf4j; |
||||
|
import org.eclipse.paho.mqttv5.common.MqttException; |
||||
|
import org.junit.Assert; |
||||
|
import org.thingsboard.server.common.data.Device; |
||||
|
import org.thingsboard.server.common.data.kv.BasicTsKvEntry; |
||||
|
import org.thingsboard.server.common.data.kv.LongDataEntry; |
||||
|
import org.thingsboard.server.common.data.kv.StringDataEntry; |
||||
|
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.SparkplugMessageType; |
||||
|
|
||||
|
import java.util.ArrayList; |
||||
|
import java.util.List; |
||||
|
import java.util.Optional; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
import java.util.concurrent.atomic.AtomicReference; |
||||
|
|
||||
|
import static org.awaitility.Awaitility.await; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.OFFLINE; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.ONLINE; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.STATE; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.messageName; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicUtil.NAMESPACE; |
||||
|
|
||||
|
/** |
||||
|
* Created by nickAS21 on 12.01.23 |
||||
|
*/ |
||||
|
@Slf4j |
||||
|
public abstract class AbstractMqttV5ClientSparkplugConnectionTest extends AbstractMqttV5ClientSparkplugTest { |
||||
|
|
||||
|
protected void processClientWithCorrectNodeAccessTokenWithNDEATH_Test() throws Exception { |
||||
|
long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS; |
||||
|
long value = bdSeq = 0; |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(ts, value); |
||||
|
|
||||
|
String keys = SparkplugMessageType.NDEATH.name() + " " + keysBdSeq; |
||||
|
TsKvEntry expectedTsKvEntry = new BasicTsKvEntry(ts, new LongDataEntry(keys, value)); |
||||
|
AtomicReference<ListenableFuture<Optional<TsKvEntry>>> finalFuture = new AtomicReference<>(); |
||||
|
await(alias + SparkplugMessageType.NDEATH.name()) |
||||
|
.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); |
||||
|
} |
||||
|
|
||||
|
protected void processClientWithCorrectNodeAccessTokenWithoutNDEATH_Test() throws Exception { |
||||
|
this.client = new MqttV5TestClient(); |
||||
|
MqttException actualException = Assert.assertThrows(MqttException.class, () -> client.connectAndWait(gatewayAccessToken)); |
||||
|
String expectedMessage = "Server unavailable."; |
||||
|
int expectedReasonCode = 136; |
||||
|
Assert.assertEquals(expectedMessage, actualException.getMessage()); |
||||
|
Assert.assertEquals(expectedReasonCode, actualException.getReasonCode()); |
||||
|
} |
||||
|
|
||||
|
protected void processClientWithCorrectNodeAccessTokenNameSpaceInvalid_Test() throws Exception { |
||||
|
long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS; |
||||
|
long value = bdSeq = 0; |
||||
|
MqttException actualException = Assert.assertThrows(MqttException.class, () -> clientConnectWithNDEATH(ts, value, "spBv1.2")); |
||||
|
String expectedMessage = "Server unavailable."; |
||||
|
int expectedReasonCode = 136; |
||||
|
Assert.assertEquals(expectedMessage, actualException.getMessage()); |
||||
|
Assert.assertEquals(expectedReasonCode, actualException.getReasonCode()); |
||||
|
} |
||||
|
|
||||
|
protected void processClientWithCorrectAccessTokenWithNDEATHCreatedDevices(int cntDevices) throws Exception { |
||||
|
long ts = calendar.getTimeInMillis(); |
||||
|
connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(cntDevices, ts); |
||||
|
} |
||||
|
|
||||
|
protected void processConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_ALL(int cntDevices) throws Exception { |
||||
|
long ts = calendar.getTimeInMillis(); |
||||
|
List<Device> devices = connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(cntDevices, ts); |
||||
|
|
||||
|
TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new StringDataEntry(messageName(STATE), ONLINE.name())); |
||||
|
AtomicReference<ListenableFuture<List<TsKvEntry>>> finalFuture = new AtomicReference<>(); |
||||
|
await(alias + messageName(STATE) + ", device: " + savedGateway.getName()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
finalFuture.set(tsService.findAllLatest(tenantId, savedGateway.getId())); |
||||
|
return finalFuture.get().get().contains(tsKvEntry); |
||||
|
}); |
||||
|
|
||||
|
for (Device device : devices) { |
||||
|
await(alias + messageName(STATE) + ", device: " + device.getName()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
finalFuture.set(tsService.findAllLatest(tenantId, device.getId())); |
||||
|
return finalFuture.get().get().contains(tsKvEntry); |
||||
|
}); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
protected void processConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_All_Then_OneDeviceOFFLINE(int cntDevices, int indexDeviceDisconnect) throws Exception { |
||||
|
long ts = calendar.getTimeInMillis(); |
||||
|
List<Device> devices = connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(cntDevices, ts); |
||||
|
|
||||
|
TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new StringDataEntry(messageName(STATE), OFFLINE.name())); |
||||
|
AtomicReference<ListenableFuture<List<TsKvEntry>>> finalFuture = new AtomicReference<>(); |
||||
|
|
||||
|
SparkplugBProto.Payload.Builder payloadDeathDevice = SparkplugBProto.Payload.newBuilder() |
||||
|
.setTimestamp(ts) |
||||
|
.setSeq(getSeqNum()); |
||||
|
if (client.isConnected()) { |
||||
|
List<Device> devicesList = new ArrayList<>(devices); |
||||
|
Device device = devicesList.get(indexDeviceDisconnect); |
||||
|
client.publish(NAMESPACE + "/" + groupId + "/" + SparkplugMessageType.DDEATH.name() + "/" + edgeNode + "/" + device.getName(), |
||||
|
payloadDeathDevice.build().toByteArray(), 0, false); |
||||
|
await(alias + messageName(STATE) + ", device: " + device.getName()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
finalFuture.set(tsService.findAllLatest(tenantId, device.getId())); |
||||
|
return findEqualsKeyValueInKvEntrys(finalFuture.get().get(), tsKvEntry); |
||||
|
}); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
protected void processConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_All_Then_OFFLINE_All(int cntDevices) throws Exception { |
||||
|
long ts = calendar.getTimeInMillis(); |
||||
|
List<Device> devices = connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(cntDevices, ts); |
||||
|
|
||||
|
TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new StringDataEntry(messageName(STATE), OFFLINE.name())); |
||||
|
AtomicReference<ListenableFuture<List<TsKvEntry>>> finalFuture = new AtomicReference<>(); |
||||
|
|
||||
|
if (client.isConnected()) { |
||||
|
client.disconnect(); |
||||
|
|
||||
|
await(alias + messageName(STATE) + ", device: " + savedGateway.getName()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
finalFuture.set(tsService.findAllLatest(tenantId, savedGateway.getId())); |
||||
|
return findEqualsKeyValueInKvEntrys(finalFuture.get().get(), tsKvEntry); |
||||
|
}); |
||||
|
|
||||
|
List<Device> devicesList = new ArrayList<>(devices); |
||||
|
for (Device device : devicesList) { |
||||
|
await(alias + messageName(STATE) + ", device: " + device.getName()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
finalFuture.set(tsService.findAllLatest(tenantId, device.getId())); |
||||
|
return findEqualsKeyValueInKvEntrys(finalFuture.get().get(), tsKvEntry); |
||||
|
}); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private boolean findEqualsKeyValueInKvEntrys(List<TsKvEntry> finalFuture, TsKvEntry tsKvEntry) { |
||||
|
for (TsKvEntry kvEntry : finalFuture) { |
||||
|
if (kvEntry.getKey().equals(tsKvEntry.getKey()) && kvEntry.getValue().equals(tsKvEntry.getValue())) { |
||||
|
return true; |
||||
|
} |
||||
|
} |
||||
|
return false; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,82 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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 testClientWithCorrectAccessTokenWithNDEATH() throws Exception { |
||||
|
processClientWithCorrectNodeAccessTokenWithNDEATH_Test(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectNodeAccessTokenWithoutNDEATH() throws Exception { |
||||
|
processClientWithCorrectNodeAccessTokenWithoutNDEATH_Test(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectNodeAccessTokenNameSpaceInvalid() throws Exception { |
||||
|
processClientWithCorrectNodeAccessTokenNameSpaceInvalid_Test(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectAccessTokenWithNDEATHCreatedOneDevice() throws Exception { |
||||
|
processClientWithCorrectAccessTokenWithNDEATHCreatedDevices(1); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectAccessTokenWithNDEATHCreatedTwoDevice() throws Exception { |
||||
|
processClientWithCorrectAccessTokenWithNDEATHCreatedDevices(2); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_ALL() throws Exception { |
||||
|
processConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_ALL(3); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_All_Then_OneDeviceOFFLINE() throws Exception { |
||||
|
processConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_All_Then_OneDeviceOFFLINE(3, 1); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_All_Then_OFFLINE_All() throws Exception { |
||||
|
processConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_All_Then_OFFLINE_All(3); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,108 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.rpc; |
||||
|
|
||||
|
import io.netty.handler.codec.mqtt.MqttQoS; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.junit.Assert; |
||||
|
import org.junit.Test; |
||||
|
import org.thingsboard.server.common.data.Device; |
||||
|
import org.thingsboard.server.transport.mqtt.sparkplug.AbstractMqttV5ClientSparkplugTest; |
||||
|
import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType; |
||||
|
|
||||
|
import java.util.List; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
import static org.awaitility.Awaitility.await; |
||||
|
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
||||
|
import static org.thingsboard.server.common.data.exception.ThingsboardErrorCode.INVALID_ARGUMENTS; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.DCMD; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.NCMD; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicUtil.NAMESPACE; |
||||
|
|
||||
|
@Slf4j |
||||
|
public abstract class AbstractMqttV5RpcSparkplugTest extends AbstractMqttV5ClientSparkplugTest { |
||||
|
|
||||
|
private static final int metricBirthValue_Int32 = 123456; |
||||
|
private static final String sparkplugRpcRequest = "{\"metricName\":\"" + metricBirthName_Int32 + "\",\"value\":" + metricBirthValue_Int32 + "}"; |
||||
|
|
||||
|
@Test |
||||
|
public void processClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_Success() throws Exception { |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
connectionWithNBirth(metricBirthDataType_Int32, metricBirthName_Int32, nextInt32()); |
||||
|
Assert.assertTrue("Connection node is failed", client.isConnected()); |
||||
|
client.subscribeAndWait(NAMESPACE + "/" + groupId + "/" + NCMD.name() + "/" + edgeNode + "/#", MqttQoS.AT_MOST_ONCE); |
||||
|
String expected = "{\"result\":\"Success: " + SparkplugMessageType.NCMD.name() + "\"}"; |
||||
|
String actual = sendRPCSparkplug(NCMD.name(), sparkplugRpcRequest, savedGateway); |
||||
|
await(alias + SparkplugMessageType.NCMD.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(expected, actual); |
||||
|
Assert.assertEquals(metricBirthName_Int32, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertTrue(metricBirthValue_Int32 == mqttCallback.getMessageArrivedMetrics().get(0).getIntValue()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void processClientDeviceWithCorrectAccessTokenPublish_TwoWayRpc_Success() throws Exception { |
||||
|
long ts = calendar.getTimeInMillis(); |
||||
|
List<Device> devices = connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(1, ts); |
||||
|
String expected = "{\"result\":\"Success: " + DCMD.name() + "\"}"; |
||||
|
String actual = sendRPCSparkplug(DCMD.name() , sparkplugRpcRequest, devices.get(0)); |
||||
|
await(alias + NCMD.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
return mqttCallback.getMessageArrivedMetrics().size() == 1; |
||||
|
}); |
||||
|
Assert.assertEquals(expected, actual); |
||||
|
Assert.assertEquals(metricBirthName_Int32, mqttCallback.getMessageArrivedMetrics().get(0).getName()); |
||||
|
Assert.assertTrue(metricBirthValue_Int32 == mqttCallback.getMessageArrivedMetrics().get(0).getIntValue()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void processClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_InvalidTypeMessage_INVALID_ARGUMENTS() throws Exception { |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
connectionWithNBirth(metricBirthDataType_Int32, metricBirthName_Int32, nextInt32()); |
||||
|
Assert.assertTrue("Connection node is failed", client.isConnected()); |
||||
|
client.subscribeAndWait(NAMESPACE + "/" + groupId + "/" + NCMD.name() + "/" + edgeNode + "/#", MqttQoS.AT_MOST_ONCE); |
||||
|
String invalidateTypeMessageName = "RCMD"; |
||||
|
String expected = "{\"result\":\"" + INVALID_ARGUMENTS + "\",\"error\":\"Failed to convert device RPC command to MQTT msg: " + |
||||
|
invalidateTypeMessageName + "{\\\"metricName\\\":\\\"" + metricBirthName_Int32 + "\\\",\\\"value\\\":" + metricBirthValue_Int32 + "}\"}"; |
||||
|
String actual = sendRPCSparkplug(invalidateTypeMessageName, sparkplugRpcRequest, savedGateway); |
||||
|
Assert.assertEquals(expected, actual); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void processClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_InBirthNotHaveMetric_BAD_REQUEST_PARAMS() throws Exception { |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
connectionWithNBirth(metricBirthDataType_Int32, metricBirthName_Int32, nextInt32()); |
||||
|
Assert.assertTrue("Connection node is failed", client.isConnected()); |
||||
|
client.subscribeAndWait(NAMESPACE + "/" + groupId + "/" + NCMD.name() + "/" + edgeNode + "/#", MqttQoS.AT_MOST_ONCE); |
||||
|
String metricNameBad = metricBirthName_Int32 + "_Bad"; |
||||
|
String sparkplugRpcRequestBad = "{\"metricName\":\"" + metricNameBad + "\",\"value\":" + metricBirthValue_Int32 + "}"; |
||||
|
String expected = "{\"result\":\"BAD_REQUEST_PARAMS\",\"error\":\"Failed send To Node Rpc Request: " + |
||||
|
DCMD.name() + ". This node does not have a metricName: [" + metricNameBad + "]\"}"; |
||||
|
String actual = sendRPCSparkplug(DCMD.name(), sparkplugRpcRequestBad, savedGateway); |
||||
|
Assert.assertEquals(expected, actual); |
||||
|
} |
||||
|
|
||||
|
private String sendRPCSparkplug(String nameTypeMessage, String keyValue, Device device) throws Exception { |
||||
|
String setRpcRequest = "{\"method\": \"" + nameTypeMessage + "\", \"params\": " + keyValue + "}"; |
||||
|
return doPostAsync("/api/plugins/rpc/twoway/" + device.getId().getId(), setRpcRequest, String.class, status().isOk()); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,61 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.rpc; |
||||
|
|
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
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; |
||||
|
|
||||
|
@DaoSqlTest |
||||
|
@Slf4j |
||||
|
public class MqttV5RpcSparkplugTest extends AbstractMqttV5RpcSparkplugTest { |
||||
|
|
||||
|
@Before |
||||
|
public void beforeTest() throws Exception { |
||||
|
beforeSparkplugTest(); |
||||
|
} |
||||
|
|
||||
|
@After |
||||
|
public void afterTest() throws MqttException { |
||||
|
if (client.isConnected()) { |
||||
|
client.disconnect(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_Success() throws Exception { |
||||
|
processClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_Success(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientDeviceWithCorrectAccessTokenPublish_TwoWayRpc_Success() throws Exception { |
||||
|
processClientDeviceWithCorrectAccessTokenPublish_TwoWayRpc_Success(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_InvalidTypeMessage_INVALID_ARGUMENTS() throws Exception { |
||||
|
processClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_InvalidTypeMessage_INVALID_ARGUMENTS(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_InBirthNotHaveMetric_BAD_REQUEST_PARAMS() throws Exception { |
||||
|
processClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_InvalidTypeMessage_INVALID_ARGUMENTS(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,113 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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 lombok.extern.slf4j.Slf4j; |
||||
|
import org.junit.Assert; |
||||
|
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.SparkplugMessageType; |
||||
|
|
||||
|
import java.util.ArrayList; |
||||
|
import java.util.List; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
import java.util.concurrent.atomic.AtomicReference; |
||||
|
|
||||
|
import static org.awaitility.Awaitility.await; |
||||
|
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicUtil.NAMESPACE; |
||||
|
|
||||
|
/** |
||||
|
* Created by nickAS21 on 12.01.23 |
||||
|
*/ |
||||
|
@Slf4j |
||||
|
public abstract class AbstractMqttV5ClientSparkplugTelemetryTest extends AbstractMqttV5ClientSparkplugTest { |
||||
|
|
||||
|
protected void processClientWithCorrectAccessTokenPublishNBIRTH() throws Exception { |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
List<String> listKeys = connectionWithNBirth(metricBirthDataType_Int32, metricBirthName_Int32, nextInt32()); |
||||
|
Assert.assertTrue("Connection node is failed", client.isConnected()); |
||||
|
AtomicReference<ListenableFuture<List<TsKvEntry>>> 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()); |
||||
|
} |
||||
|
|
||||
|
protected void processClientWithCorrectAccessTokenPushNodeMetricBuildPrimitiveSimple() throws Exception { |
||||
|
List<String> listKeys = new ArrayList<>(); |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
|
||||
|
String messageTypeName = SparkplugMessageType.NDATA.name(); |
||||
|
|
||||
|
List<TsKvEntry> listTsKvEntry = new ArrayList<>(); |
||||
|
|
||||
|
SparkplugBProto.Payload.Builder ndataPayload = SparkplugBProto.Payload.newBuilder() |
||||
|
.setTimestamp(calendar.getTimeInMillis()) |
||||
|
.setSeq(getSeqNum()); |
||||
|
long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS; |
||||
|
|
||||
|
createdAddMetricValuePrimitiveTsKv(listTsKvEntry, listKeys, ndataPayload, ts); |
||||
|
|
||||
|
if (client.isConnected()) { |
||||
|
client.publish(NAMESPACE + "/" + groupId + "/" + messageTypeName + "/" + edgeNode, |
||||
|
ndataPayload.build().toByteArray(), 0, false); |
||||
|
} |
||||
|
|
||||
|
AtomicReference<ListenableFuture<List<TsKvEntry>>> finalFuture = new AtomicReference<>(); |
||||
|
await(alias + SparkplugMessageType.NDATA.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
finalFuture.set(tsService.findAllLatest(tenantId, savedGateway.getId())); |
||||
|
return finalFuture.get().get().size() == (listTsKvEntry.size() + 1); |
||||
|
}); |
||||
|
Assert.assertTrue("Actual tsKvEntrys is not containsAll Expected tsKvEntrys", finalFuture.get().get().containsAll(listTsKvEntry)); |
||||
|
} |
||||
|
|
||||
|
protected void processClientWithCorrectAccessTokenPushNodeMetricBuildArraysPrimitiveSimple() throws Exception { |
||||
|
clientWithCorrectNodeAccessTokenWithNDEATH(); |
||||
|
|
||||
|
String messageTypeName = SparkplugMessageType.NDATA.name(); |
||||
|
List<String> listKeys = new ArrayList<>(); |
||||
|
List<TsKvEntry> listTsKvEntry = new ArrayList<>(); |
||||
|
|
||||
|
SparkplugBProto.Payload.Builder ndataPayload = SparkplugBProto.Payload.newBuilder() |
||||
|
.setTimestamp(calendar.getTimeInMillis()) |
||||
|
.setSeq(getSeqNum()); |
||||
|
long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS; |
||||
|
|
||||
|
createdAddMetricValueArraysPrimitiveTsKv(listTsKvEntry, listKeys, ndataPayload, ts); |
||||
|
|
||||
|
if (client.isConnected()) { |
||||
|
client.publish(NAMESPACE + "/" + groupId + "/" + messageTypeName + "/" + edgeNode, |
||||
|
ndataPayload.build().toByteArray(), 0, false); |
||||
|
} |
||||
|
|
||||
|
AtomicReference<ListenableFuture<List<TsKvEntry>>> finalFuture = new AtomicReference<>(); |
||||
|
await(alias + SparkplugMessageType.NDATA.name()) |
||||
|
.atMost(40, TimeUnit.SECONDS) |
||||
|
.until(() -> { |
||||
|
finalFuture.set(tsService.findAllLatest(tenantId, savedGateway.getId())); |
||||
|
return finalFuture.get().get().size() == (listTsKvEntry.size() + 1); |
||||
|
}); |
||||
|
Assert.assertTrue("Actual tsKvEntrys is not containsAll Expected tsKvEntrys", finalFuture.get().get().containsAll(listTsKvEntry)); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,56 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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; |
||||
|
|
||||
|
/** |
||||
|
* 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 testClientWithCorrectAccessTokenPublishNBIRTH() throws Exception { |
||||
|
processClientWithCorrectAccessTokenPublishNBIRTH(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectAccessTokenPushNodeMetricBuildPrimitiveSimple() throws Exception { |
||||
|
processClientWithCorrectAccessTokenPushNodeMetricBuildPrimitiveSimple(); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testClientWithCorrectAccessTokenPushNodeMetricBuildPArraysPrimitiveSimple() throws Exception { |
||||
|
processClientWithCorrectAccessTokenPushNodeMetricBuildArraysPrimitiveSimple(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,85 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.dao.alarm; |
||||
|
|
||||
|
import lombok.Builder; |
||||
|
import lombok.Data; |
||||
|
import org.thingsboard.server.common.data.alarm.Alarm; |
||||
|
import org.thingsboard.server.common.data.alarm.AlarmInfo; |
||||
|
import org.thingsboard.server.common.data.alarm.AlarmSeverity; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
|
||||
|
import java.util.List; |
||||
|
|
||||
|
|
||||
|
@Data |
||||
|
public class AlarmApiCallResult { |
||||
|
|
||||
|
private final boolean successful; |
||||
|
private final boolean created; |
||||
|
private final boolean modified; |
||||
|
private final boolean cleared; |
||||
|
private final AlarmInfo alarm; |
||||
|
private final Alarm old; |
||||
|
private final List<EntityId> propagatedEntitiesList; |
||||
|
|
||||
|
@Builder |
||||
|
private AlarmApiCallResult(boolean successful, boolean created, boolean modified, boolean cleared, AlarmInfo alarm, Alarm old, List<EntityId> propagatedEntitiesList) { |
||||
|
this.successful = successful; |
||||
|
this.created = created; |
||||
|
this.modified = modified; |
||||
|
this.cleared = cleared; |
||||
|
this.alarm = alarm; |
||||
|
this.old = old; |
||||
|
this.propagatedEntitiesList = propagatedEntitiesList; |
||||
|
} |
||||
|
|
||||
|
public AlarmApiCallResult(AlarmApiCallResult other, List<EntityId> propagatedEntitiesList) { |
||||
|
this.successful = other.successful; |
||||
|
this.created = other.created; |
||||
|
this.modified = other.modified; |
||||
|
this.cleared = other.cleared; |
||||
|
this.alarm = other.alarm; |
||||
|
this.old = other.old; |
||||
|
this.propagatedEntitiesList = propagatedEntitiesList; |
||||
|
} |
||||
|
|
||||
|
public boolean isSeverityChanged() { |
||||
|
if (alarm == null || old == null) { |
||||
|
return false; |
||||
|
} else { |
||||
|
return !alarm.getSeverity().equals(old.getSeverity()); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public AlarmSeverity getOldSeverity() { |
||||
|
return isSeverityChanged() ? old.getSeverity() : null; |
||||
|
} |
||||
|
|
||||
|
public boolean isPropagationChanged() { |
||||
|
if (created) { |
||||
|
return true; |
||||
|
} |
||||
|
if (alarm == null || old == null) { |
||||
|
return false; |
||||
|
} |
||||
|
return (alarm.isPropagate() != old.isPropagate()) || |
||||
|
(alarm.isPropagateToOwner() != old.isPropagateToOwner()) || |
||||
|
(alarm.isPropagateToTenant() != old.isPropagateToTenant()) || |
||||
|
(!alarm.getPropagateRelationTypes().equals(old.getPropagateRelationTypes())); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,22 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.data; |
||||
|
|
||||
|
public interface HasEmail extends HasName { |
||||
|
|
||||
|
String getEmail(); |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,22 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.data; |
||||
|
|
||||
|
public interface HasLabel extends HasName { |
||||
|
|
||||
|
String getLabel(); |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,22 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.data; |
||||
|
|
||||
|
public interface HasTitle { |
||||
|
|
||||
|
String getTitle(); |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,37 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.data.alarm; |
||||
|
|
||||
|
import lombok.AllArgsConstructor; |
||||
|
import lombok.Builder; |
||||
|
import lombok.Data; |
||||
|
import org.thingsboard.server.common.data.id.UserId; |
||||
|
|
||||
|
import java.io.Serializable; |
||||
|
|
||||
|
@Builder |
||||
|
@AllArgsConstructor |
||||
|
@Data |
||||
|
public class AlarmAssignee implements Serializable { |
||||
|
|
||||
|
private static final long serialVersionUID = 6628286223963972860L; |
||||
|
|
||||
|
private final UserId id; |
||||
|
private final String firstName; |
||||
|
private final String lastName; |
||||
|
private final String email; |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,30 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.data.alarm; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
|
||||
|
import java.io.Serializable; |
||||
|
|
||||
|
@Data |
||||
|
public class AlarmAssigneeUpdate implements Serializable { |
||||
|
|
||||
|
private static final long serialVersionUID = -2391676304697483808L; |
||||
|
|
||||
|
private final boolean deleted; |
||||
|
private final AlarmAssignee assignee; |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,87 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.data.alarm; |
||||
|
|
||||
|
import com.fasterxml.jackson.databind.JsonNode; |
||||
|
import io.swagger.annotations.ApiModelProperty; |
||||
|
import lombok.Builder; |
||||
|
import lombok.Data; |
||||
|
import org.thingsboard.server.common.data.id.CustomerId; |
||||
|
import org.thingsboard.server.common.data.id.EntityId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.id.UserId; |
||||
|
import org.thingsboard.server.common.data.validation.Length; |
||||
|
import org.thingsboard.server.common.data.validation.NoXss; |
||||
|
|
||||
|
import javax.validation.Valid; |
||||
|
import javax.validation.constraints.NotNull; |
||||
|
|
||||
|
@Data |
||||
|
@Builder |
||||
|
public class AlarmCreateOrUpdateActiveRequest implements AlarmModificationRequest { |
||||
|
|
||||
|
@NotNull |
||||
|
@ApiModelProperty(position = 1, value = "JSON object with Tenant Id", accessMode = ApiModelProperty.AccessMode.READ_ONLY) |
||||
|
private TenantId tenantId; |
||||
|
@ApiModelProperty(position = 2, value = "JSON object with Customer Id", accessMode = ApiModelProperty.AccessMode.READ_ONLY) |
||||
|
private CustomerId customerId; |
||||
|
@NotNull |
||||
|
@ApiModelProperty(position = 3, required = true, value = "representing type of the Alarm", example = "High Temperature Alarm") |
||||
|
@Length(fieldName = "type") |
||||
|
private String type; |
||||
|
@NotNull |
||||
|
@ApiModelProperty(position = 4, required = true, value = "JSON object with alarm originator id") |
||||
|
private EntityId originator; |
||||
|
@NotNull |
||||
|
@ApiModelProperty(position = 5, required = true, value = "Alarm severity", example = "CRITICAL") |
||||
|
private AlarmSeverity severity; |
||||
|
@ApiModelProperty(position = 6, value = "Timestamp of the alarm start time, in milliseconds", example = "1634058704565") |
||||
|
private long startTs; |
||||
|
@ApiModelProperty(position = 7, value = "Timestamp of the alarm end time(last time update), in milliseconds", example = "1634111163522") |
||||
|
private long endTs; |
||||
|
@NoXss |
||||
|
@ApiModelProperty(position = 8, value = "JSON object with alarm details") |
||||
|
private JsonNode details; |
||||
|
@Valid |
||||
|
@ApiModelProperty(position = 9, value = "JSON object with propagation details") |
||||
|
private AlarmPropagationInfo propagation; |
||||
|
|
||||
|
private UserId userId; |
||||
|
|
||||
|
public static AlarmCreateOrUpdateActiveRequest fromAlarm(Alarm a) { |
||||
|
return fromAlarm(a, null); |
||||
|
} |
||||
|
|
||||
|
public static AlarmCreateOrUpdateActiveRequest fromAlarm(Alarm a, UserId userId) { |
||||
|
return AlarmCreateOrUpdateActiveRequest.builder() |
||||
|
.tenantId(a.getTenantId()) |
||||
|
.customerId(a.getCustomerId()) |
||||
|
.type(a.getType()) |
||||
|
.originator(a.getOriginator()) |
||||
|
.severity((a.getSeverity())) |
||||
|
.startTs(a.getStartTs()) |
||||
|
.endTs(a.getEndTs()) |
||||
|
.details(a.getDetails()) |
||||
|
.propagation(AlarmPropagationInfo.builder() |
||||
|
.propagate(a.isPropagate()) |
||||
|
.propagateToOwner(a.isPropagateToOwner()) |
||||
|
.propagateToTenant(a.isPropagateToTenant()) |
||||
|
.propagateRelationTypes(a.getPropagateRelationTypes()).build()) |
||||
|
.userId(userId) |
||||
|
.build(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,34 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.data.alarm; |
||||
|
|
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.id.UserId; |
||||
|
|
||||
|
public interface AlarmModificationRequest { |
||||
|
|
||||
|
TenantId getTenantId(); |
||||
|
|
||||
|
long getStartTs(); |
||||
|
|
||||
|
long getEndTs(); |
||||
|
|
||||
|
void setStartTs(long startTs); |
||||
|
|
||||
|
void setEndTs(long endTs); |
||||
|
|
||||
|
UserId getUserId(); |
||||
|
} |
||||
@ -0,0 +1,46 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.data.alarm; |
||||
|
|
||||
|
import io.swagger.annotations.ApiModelProperty; |
||||
|
import lombok.AllArgsConstructor; |
||||
|
import lombok.Builder; |
||||
|
import lombok.Data; |
||||
|
import lombok.NoArgsConstructor; |
||||
|
import org.thingsboard.server.common.data.validation.NoXss; |
||||
|
|
||||
|
import java.util.Collections; |
||||
|
import java.util.List; |
||||
|
|
||||
|
@Builder |
||||
|
@Data |
||||
|
public class AlarmPropagationInfo { |
||||
|
|
||||
|
public static AlarmPropagationInfo EMPTY = new AlarmPropagationInfo(false, false, false, Collections.emptyList()); |
||||
|
|
||||
|
@ApiModelProperty(position = 1, value = "Propagation flag to specify if alarm should be propagated to parent entities of alarm originator", example = "true") |
||||
|
private boolean propagate; |
||||
|
@ApiModelProperty(position = 2, value = "Propagation flag to specify if alarm should be propagated to the owner (tenant or customer) of alarm originator", example = "true") |
||||
|
private boolean propagateToOwner; |
||||
|
@ApiModelProperty(position = 3, value = "Propagation flag to specify if alarm should be propagated to the tenant entity", example = "true") |
||||
|
private boolean propagateToTenant; |
||||
|
@NoXss |
||||
|
@ApiModelProperty(position = 4, value = "JSON array of relation types that should be used for propagation. " + |
||||
|
"By default, 'propagateRelationTypes' array is empty which means that the alarm will be propagated based on any relation type to parent entities. " + |
||||
|
"This parameter should be used only in case when 'propagate' parameter is set to true, otherwise, 'propagateRelationTypes' array will be ignored.") |
||||
|
private List<String> propagateRelationTypes; |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,118 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.data.alarm; |
||||
|
|
||||
|
import java.util.List; |
||||
|
import java.util.Optional; |
||||
|
|
||||
|
public class AlarmStatusFilter { |
||||
|
|
||||
|
private static final AlarmStatusFilter EMPTY = new AlarmStatusFilter(Optional.empty(), Optional.empty()); |
||||
|
|
||||
|
private final Optional<Boolean> clearFilter; |
||||
|
private final Optional<Boolean> ackFilter; |
||||
|
|
||||
|
private AlarmStatusFilter(Optional<Boolean> clearFilter, Optional<Boolean> ackFilter) { |
||||
|
this.clearFilter = clearFilter; |
||||
|
this.ackFilter = ackFilter; |
||||
|
} |
||||
|
|
||||
|
public static AlarmStatusFilter from(AlarmQuery query) { |
||||
|
if (query.getSearchStatus() != null) { |
||||
|
return AlarmStatusFilter.from(query.getSearchStatus()); |
||||
|
} else if (query.getStatus() != null) { |
||||
|
return AlarmStatusFilter.from(query.getStatus()); |
||||
|
} |
||||
|
return AlarmStatusFilter.empty(); |
||||
|
} |
||||
|
|
||||
|
public static AlarmStatusFilter from(AlarmSearchStatus alarmSearchStatus) { |
||||
|
switch (alarmSearchStatus) { |
||||
|
case ACK: |
||||
|
return new AlarmStatusFilter(Optional.empty(), Optional.of(true)); |
||||
|
case UNACK: |
||||
|
return new AlarmStatusFilter(Optional.empty(), Optional.of(false)); |
||||
|
case ACTIVE: |
||||
|
return new AlarmStatusFilter(Optional.of(false), Optional.empty()); |
||||
|
case CLEARED: |
||||
|
return new AlarmStatusFilter(Optional.of(true), Optional.empty()); |
||||
|
default: |
||||
|
return EMPTY; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public static AlarmStatusFilter from(AlarmStatus alarmStatus) { |
||||
|
switch (alarmStatus) { |
||||
|
case ACTIVE_UNACK: |
||||
|
return new AlarmStatusFilter(Optional.of(false), Optional.of(false)); |
||||
|
case ACTIVE_ACK: |
||||
|
return new AlarmStatusFilter(Optional.of(false), Optional.of(true)); |
||||
|
case CLEARED_UNACK: |
||||
|
return new AlarmStatusFilter(Optional.of(true), Optional.of(false)); |
||||
|
case CLEARED_ACK: |
||||
|
return new AlarmStatusFilter(Optional.of(true), Optional.of(true)); |
||||
|
default: |
||||
|
return EMPTY; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public static AlarmStatusFilter empty() { |
||||
|
return EMPTY; |
||||
|
} |
||||
|
|
||||
|
public boolean hasAnyFilter() { |
||||
|
return clearFilter.isPresent() || ackFilter.isPresent(); |
||||
|
} |
||||
|
|
||||
|
public boolean hasClearFilter() { |
||||
|
return clearFilter.isPresent(); |
||||
|
} |
||||
|
|
||||
|
public boolean hasAckFilter() { |
||||
|
return ackFilter.isPresent(); |
||||
|
} |
||||
|
|
||||
|
public boolean getClearFilter() { |
||||
|
return clearFilter.orElseThrow(() -> new RuntimeException("Clear filter is not set! Use `hasClearFilter` to check.")); |
||||
|
} |
||||
|
|
||||
|
public boolean getAckFilter() { |
||||
|
return ackFilter.orElseThrow(() -> new RuntimeException("Ack filter is not set! Use `hasAckFilter` to check.")); |
||||
|
} |
||||
|
|
||||
|
|
||||
|
public static AlarmStatusFilter fromList(List<AlarmSearchStatus> list) { |
||||
|
if (list == null || list.isEmpty() || list.contains(AlarmSearchStatus.ANY)) { |
||||
|
return EMPTY; |
||||
|
} |
||||
|
boolean clearFilter = list.contains(AlarmSearchStatus.CLEARED); |
||||
|
boolean activeFilter = list.contains(AlarmSearchStatus.ACTIVE); |
||||
|
Optional<Boolean> clear = Optional.empty(); |
||||
|
if (clearFilter && !activeFilter || !clearFilter && activeFilter) { |
||||
|
clear = Optional.of(clearFilter); |
||||
|
} |
||||
|
|
||||
|
boolean ackFilter = list.contains(AlarmSearchStatus.ACK); |
||||
|
boolean unackFilter = list.contains(AlarmSearchStatus.UNACK); |
||||
|
Optional<Boolean> ack = Optional.empty(); |
||||
|
if (ackFilter && !unackFilter || !ackFilter && unackFilter) { |
||||
|
ack = Optional.of(ackFilter); |
||||
|
} |
||||
|
return new AlarmStatusFilter(clear, ack); |
||||
|
} |
||||
|
|
||||
|
|
||||
|
} |
||||
@ -0,0 +1,79 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.data.alarm; |
||||
|
|
||||
|
import com.fasterxml.jackson.databind.JsonNode; |
||||
|
import io.swagger.annotations.ApiModelProperty; |
||||
|
import lombok.Builder; |
||||
|
import lombok.Data; |
||||
|
import org.thingsboard.server.common.data.id.AlarmId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.id.UserId; |
||||
|
import org.thingsboard.server.common.data.validation.NoXss; |
||||
|
|
||||
|
import javax.validation.Valid; |
||||
|
import javax.validation.constraints.NotNull; |
||||
|
|
||||
|
@Data |
||||
|
@Builder |
||||
|
public class AlarmUpdateRequest implements AlarmModificationRequest { |
||||
|
|
||||
|
@NotNull |
||||
|
@ApiModelProperty(position = 1, value = "JSON object with Tenant Id", accessMode = ApiModelProperty.AccessMode.READ_ONLY) |
||||
|
private TenantId tenantId; |
||||
|
@NotNull |
||||
|
@ApiModelProperty(position = 2, value = "JSON object with the alarm Id. " + |
||||
|
"Specify this field to update the alarm. " + |
||||
|
"Referencing non-existing alarm Id will cause error. " + |
||||
|
"Omit this field to create new alarm.") |
||||
|
private AlarmId alarmId; |
||||
|
@NotNull |
||||
|
@ApiModelProperty(position = 3, required = true, value = "Alarm severity", example = "CRITICAL") |
||||
|
private AlarmSeverity severity; |
||||
|
@ApiModelProperty(position = 4, value = "Timestamp of the alarm start time, in milliseconds", example = "1634058704565") |
||||
|
private long startTs; |
||||
|
@ApiModelProperty(position = 5, value = "Timestamp of the alarm end time(last time update), in milliseconds", example = "1634111163522") |
||||
|
private long endTs; |
||||
|
@NoXss |
||||
|
@ApiModelProperty(position = 6, value = "JSON object with alarm details") |
||||
|
private JsonNode details; |
||||
|
@Valid |
||||
|
@ApiModelProperty(position = 7, value = "JSON object with propagation details") |
||||
|
private AlarmPropagationInfo propagation; |
||||
|
|
||||
|
private UserId userId; |
||||
|
|
||||
|
public static AlarmUpdateRequest fromAlarm(Alarm a) { |
||||
|
return fromAlarm(a, null); |
||||
|
} |
||||
|
|
||||
|
public static AlarmUpdateRequest fromAlarm(Alarm a, UserId userId) { |
||||
|
return AlarmUpdateRequest.builder() |
||||
|
.tenantId(a.getTenantId()) |
||||
|
.alarmId(a.getId()) |
||||
|
.severity((a.getSeverity())) |
||||
|
.startTs(a.getStartTs()) |
||||
|
.endTs(a.getEndTs()) |
||||
|
.details(a.getDetails()) |
||||
|
.propagation(AlarmPropagationInfo.builder() |
||||
|
.propagate(a.isPropagate()) |
||||
|
.propagateToOwner(a.isPropagateToOwner()) |
||||
|
.propagateToTenant(a.isPropagateToTenant()) |
||||
|
.propagateRelationTypes(a.getPropagateRelationTypes()).build()) |
||||
|
.userId(userId) |
||||
|
.build(); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,27 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2023 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.data.id; |
||||
|
|
||||
|
import lombok.AllArgsConstructor; |
||||
|
import lombok.Getter; |
||||
|
|
||||
|
@Getter |
||||
|
@AllArgsConstructor |
||||
|
public class NameLabelAndCustomerDetails { |
||||
|
private final String name; |
||||
|
private final String label; |
||||
|
private final CustomerId customerId; |
||||
|
} |
||||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue