tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

sparkplug: fix bug alias

pull/12502/head
nick 2 years ago
parent
d28769ccc8
commit
9873b817e6
7 changed files with 204 additions and 50 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 88
      application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java
  2. 2
      application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java
  3. 32
      application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/AbstractMqttV5RpcSparkplugTest.java
  4. 9
      application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/MqttV5RpcSparkplugTest.java
  5. 16
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java
  6. 86
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java
  7. 21
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugMetricUtil.java

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

@ -29,6 +29,7 @@ import org.eclipse.paho.mqttv5.common.packet.MqttReturnCode;
import org.eclipse.paho.mqttv5.common.packet.MqttWireMessage;
import org.junit.Assert;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.TransportPayloadType;
import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
@ -112,21 +113,29 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
long value = bdSeq = 0;
clientWithCorrectNodeAccessTokenWithNDEATH(ts, value);
}
public void clientWithCorrectNodeAccessTokenWithNDEATH(Long alias) throws Exception {
long ts = calendar.getTimeInMillis();
long value = bdSeq = 0;
clientMqttV5ConnectWithNDEATH(ts, value,alias);
}
public void clientWithCorrectNodeAccessTokenWithNDEATH(long ts, long value) throws Exception {
IMqttToken connectionResult = clientMqttV5ConnectWithNDEATH(ts, value);
IMqttToken connectionResult = clientMqttV5ConnectWithNDEATH(ts, value, -1L);
MqttWireMessage response = connectionResult.getResponse();
Assert.assertEquals(MESSAGE_TYPE_CONNACK, response.getType());
MqttConnAck connAckMsg = (MqttConnAck) response;
Assert.assertEquals(MqttReturnCode.RETURN_CODE_SUCCESS, connAckMsg.getReturnCode());
}
public IMqttToken clientMqttV5ConnectWithNDEATH(long ts, long value, String... nameSpaceBad) throws Exception {
String key = keysBdSeq;
public IMqttToken clientMqttV5ConnectWithNDEATH(long ts, long value, Long alias, String... nameSpaceBad) throws Exception {
return clientMqttV5ConnectWithNDEATH(ts, value, null, alias, nameSpaceBad);
}
public IMqttToken clientMqttV5ConnectWithNDEATH(long ts, long value, String metricName, Long alias, String... nameSpaceBad) throws Exception {
String key = metricName == null ? keysBdSeq : metricName;
MetricDataType metricDataType = Int64;
SparkplugBProto.Payload.Builder deathPayload = SparkplugBProto.Payload.newBuilder()
.setTimestamp(calendar.getTimeInMillis());
deathPayload.addMetrics(createMetric(value, ts, key, metricDataType));
deathPayload.addMetrics(createMetric(value, ts, key, metricDataType, alias));
byte[] deathBytes = deathPayload.build().toByteArray();
this.client = new MqttV5TestClient();
this.mqttCallback = new SparkplugMqttCallback();
@ -153,7 +162,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
MetricDataType metricDataType = Int32;
String key = "Node Metric int32";
int valueDeviceInt32 = 1024;
SparkplugBProto.Payload.Metric metric = createMetric(valueDeviceInt32, ts, key, metricDataType);
SparkplugBProto.Payload.Metric metric = createMetric(valueDeviceInt32, ts, key, metricDataType, -1L);
SparkplugBProto.Payload.Builder payloadBirthNode = SparkplugBProto.Payload.newBuilder()
.setTimestamp(ts)
.setSeq(getBdSeqNum());
@ -165,7 +174,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
}
valueDeviceInt32 = 4024;
metric = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32);
metric = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32, -1L);
for (int i = 0; i < cntDevices; i++) {
SparkplugBProto.Payload.Builder payloadBirthDevice = SparkplugBProto.Payload.newBuilder()
.setTimestamp(ts)
@ -192,6 +201,49 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
return devices;
}
protected List<Device> connectClientWithCorrectAccessTokenWithNDEATHWithAliasCreatedDevices(long ts) throws Exception {
List<Device> devices = new ArrayList<>();
Long alias = 0L;
clientWithCorrectNodeAccessTokenWithNDEATH(alias++);
MetricDataType metricDataType = Int32;
String key = "Node Metric int32";
int valueDeviceInt32 = 1024;
SparkplugBProto.Payload.Metric metric = createMetric(valueDeviceInt32, ts, key, metricDataType, alias++);
SparkplugBProto.Payload.Builder payloadBirthNode = SparkplugBProto.Payload.newBuilder()
.setTimestamp(ts)
.setSeq(getBdSeqNum());
payloadBirthNode.addMetrics(metric);
payloadBirthNode.setTimestamp(ts);
if (client.isConnected()) {
client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode,
payloadBirthNode.build().toByteArray(), 0, false);
}
valueDeviceInt32 = 4024;
metric = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32, alias++);
SparkplugBProto.Payload.Builder payloadBirthDevice = SparkplugBProto.Payload.newBuilder()
.setTimestamp(ts)
.setSeq(getSeqNum());
String deviceName = deviceId + "_" + 1;
payloadBirthDevice.addMetrics(metric);
if (client.isConnected()) {
client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DBIRTH.name() + "/" + edgeNode + "/" + deviceName,
payloadBirthDevice.build().toByteArray(), 0, false);
AtomicReference<Device> device = new AtomicReference<>();
await(alias + "find device [" + deviceName + "] after created")
.atMost(200, TimeUnit.SECONDS)
.until(() -> {
device.set(doGet("/api/tenant/devices?deviceName=" + deviceName, Device.class));
return device.get() != null;
});
devices.add(device.get());
}
Assert.assertEquals(1, devices.size());
return devices;
}
protected long getBdSeqNum() throws Exception {
if (bdSeq == 256) {
bdSeq = 0;
@ -212,12 +264,16 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
.setTimestamp(calendar.getTimeInMillis());
long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS;
long valueBdSec = getBdSeqNum();
payloadBirthNode.addMetrics(createMetric(valueBdSec, ts, keysBdSeq, Int64));
payloadBirthNode.addMetrics(createMetric(valueBdSec, ts, keysBdSeq, Int64, -1L));
listKeys.add(SparkplugMessageType.NBIRTH.name() + " " + keysBdSeq);
payloadBirthNode.addMetrics(createMetric(false, ts, keyNodeRebirth, MetricDataType.Boolean));
payloadBirthNode.addMetrics(createMetric(false, ts, keyNodeRebirth, MetricDataType.Boolean, -1L));
listKeys.add(keyNodeRebirth);
payloadBirthNode.addMetrics(createMetric(metricValue, ts, metricKey, metricDataType));
if (StringUtils.isNotBlank(metricKey)) {
payloadBirthNode.addMetrics(createMetric(metricValue, ts, metricKey, metricDataType, -1L));
} else {
payloadBirthNode.addMetrics(createMetric(metricValue, ts, metricKey, metricDataType, 4L));
}
listKeys.add(metricKey);
if (client.isConnected()) {
@ -302,7 +358,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
private TsKvEntry createdAddMetricTsKvLong(SparkplugBProto.Payload.Builder dataPayload, String key, Object value,
long ts, MetricDataType metricDataType) throws ThingsboardException {
TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new LongDataEntry(key, Long.valueOf(String.valueOf(value))));
dataPayload.addMetrics(createMetric(value, ts, key, metricDataType));
dataPayload.addMetrics(createMetric(value, ts, key, metricDataType, -1L));
return tsKvEntry;
}
@ -310,7 +366,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
long ts, MetricDataType metricDataType) throws ThingsboardException {
Double dd = Double.parseDouble(Float.toString(value));
TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new DoubleDataEntry(key, dd));
dataPayload.addMetrics(createMetric(value, ts, key, metricDataType));
dataPayload.addMetrics(createMetric(value, ts, key, metricDataType, -1L));
return tsKvEntry;
}
@ -318,21 +374,21 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
long ts, MetricDataType metricDataType) throws ThingsboardException {
Long l = Double.valueOf(value).longValue();
TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new LongDataEntry(key, l));
dataPayload.addMetrics(createMetric(value, ts, key, metricDataType));
dataPayload.addMetrics(createMetric(value, ts, key, metricDataType, -1L));
return tsKvEntry;
}
private TsKvEntry createdAddMetricTsKvBoolean(SparkplugBProto.Payload.Builder dataPayload, String key, boolean value,
long ts, MetricDataType metricDataType) throws ThingsboardException {
TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new BooleanDataEntry(key, value));
dataPayload.addMetrics(createMetric(value, ts, key, metricDataType));
dataPayload.addMetrics(createMetric(value, ts, key, metricDataType, -1L));
return tsKvEntry;
}
private TsKvEntry createdAddMetricTsKvString(SparkplugBProto.Payload.Builder dataPayload, String key, String value,
long ts, MetricDataType metricDataType) throws ThingsboardException {
TsKvEntry tsKvEntry = new BasicTsKvEntry(ts, new StringDataEntry(key, value));
dataPayload.addMetrics(createMetric(value, ts, key, metricDataType));
dataPayload.addMetrics(createMetric(value, ts, key, metricDataType, -1L));
return tsKvEntry;
}
@ -353,7 +409,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
if (nodeArray.size() > 0) {
Optional<TsKvEntry> tsKvEntryOptional = Optional.of(new BasicTsKvEntry(ts, new JsonDataEntry(key, nodeArray.toString())));
if (tsKvEntryOptional.isPresent()) {
dataPayload.addMetrics(createMetric(values, ts, key, metricDataType));
dataPayload.addMetrics(createMetric(values, ts, key, metricDataType, -1L));
listTsKvEntry.add(tsKvEntryOptional.get());
listKeys.add(key);
}
@ -421,7 +477,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
return java.util.UUID.randomUUID().toString();
}
public class SparkplugMqttCallback implements MqttCallback {
public class SparkplugMqttCallback implements MqttCallback {
private final List<SparkplugBProto.Payload.Metric> messageArrivedMetrics = new ArrayList<>();
@Override

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

@ -80,7 +80,7 @@ public abstract class AbstractMqttV5ClientSparkplugConnectionTest extends Abstra
protected void processClientWithCorrectNodeAccessTokenNameSpaceInvalid_Test() throws Exception {
long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS;
long value = bdSeq = 0;
MqttException actualException = Assert.assertThrows(MqttException.class, () -> clientMqttV5ConnectWithNDEATH(ts, value, "spBv1.2"));
MqttException actualException = Assert.assertThrows(MqttException.class, () -> clientMqttV5ConnectWithNDEATH(ts, value, -1L,"spBv1.2"));
String expectedMessage = "Server unavailable.";
int expectedReasonCode = 136;
Assert.assertEquals(expectedMessage, actualException.getMessage());

32
application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/AbstractMqttV5RpcSparkplugTest.java
View File

@ -15,8 +15,8 @@
*/
package org.thingsboard.server.transport.mqtt.sparkplug.rpc;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.junit.Assert;
import org.junit.Test;
import org.thingsboard.server.common.data.Device;
@ -74,6 +74,36 @@ public abstract class AbstractMqttV5RpcSparkplugTest extends AbstractMqttV5Clie
Assert.assertTrue(metricBirthValue_Int32 == mqttCallback.getMessageArrivedMetrics().get(0).getIntValue());
}
@Test
public void processClientDeviceWithCorrectAccessTokenPublishWithAlias_TwoWayRpc_Success() throws Exception {
long ts = calendar.getTimeInMillis();
List<Device> devices = connectClientWithCorrectAccessTokenWithNDEATHWithAliasCreatedDevices(ts);
awaitForDeviceActorToReceiveSubscription(devices.get(0).getId(), FeatureType.RPC, 1);
String expected = "{\"result\":\"Success: " + DCMD.name() + "\"}";
String actual = sendRPCSparkplug(DCMD.name() , sparkplugRpcRequest, devices.get(0));
await(alias + DCMD.name())
.atMost(40, TimeUnit.SECONDS)
.until(() -> {
return mqttCallback.getMessageArrivedMetrics().size() == 1;
});
Assert.assertEquals(expected, actual);
Assert.assertFalse(mqttCallback.getMessageArrivedMetrics().get(0).hasName());
Assert.assertTrue(mqttCallback.getMessageArrivedMetrics().get(0).hasAlias());
Assert.assertTrue(2L == mqttCallback.getMessageArrivedMetrics().get(0).getAlias());
Assert.assertTrue(metricBirthValue_Int32 == mqttCallback.getMessageArrivedMetrics().get(0).getIntValue());
}
@Test
public void processClientNodeWithCorrectAccessTokenPublishWithAliasWithoutMetricName_TwoWayRpc_BAD_REQUEST_PARAMS() throws Exception {
long ts = calendar.getTimeInMillis() - PUBLISH_TS_DELTA_MS;
long value = bdSeq = 0;
MqttException actualException = Assert.assertThrows(MqttException.class, () -> clientMqttV5ConnectWithNDEATH(ts, value, "",4L));
String expectedMessage = "Server unavailable.";
int expectedReasonCode = 136;
Assert.assertEquals(expectedMessage, actualException.getMessage());
Assert.assertEquals(expectedReasonCode, actualException.getReasonCode());
}
@Test
public void processClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_InvalidTypeMessage_INVALID_ARGUMENTS() throws Exception {
clientWithCorrectNodeAccessTokenWithNDEATH();

9
application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/rpc/MqttV5RpcSparkplugTest.java
View File

@ -47,6 +47,15 @@ public class MqttV5RpcSparkplugTest extends AbstractMqttV5RpcSparkplugTest {
public void testClientDeviceWithCorrectAccessTokenPublish_TwoWayRpc_Success() throws Exception {
processClientDeviceWithCorrectAccessTokenPublish_TwoWayRpc_Success();
}
@Test
public void testClientDeviceWithCorrectAccessTokenPublishWithAlias_TwoWayRpc_Success() throws Exception {
processClientDeviceWithCorrectAccessTokenPublishWithAlias_TwoWayRpc_Success();
}
@Test
public void testClientNodeWithCorrectAccessTokenPublishWithAliasWithoutMetricName_TwoWayRpc_BAD_REQUEST_PARAMS() throws Exception {
processClientNodeWithCorrectAccessTokenPublishWithAliasWithoutMetricName_TwoWayRpc_BAD_REQUEST_PARAMS();
}
@Test
public void testClientNodeWithCorrectAccessTokenPublish_TwoWayRpc_InvalidTypeMessage_INVALID_ARGUMENTS() throws Exception {

16
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java
View File

@ -16,6 +16,7 @@
package org.thingsboard.server.transport.mqtt.session;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DuplicateKeyException;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.DeviceProfile;
import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
@ -33,7 +34,6 @@ import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.DCMD;
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.getTsKvProto;
@ -57,9 +57,17 @@ public class SparkplugDeviceSessionContext extends AbstractGatewayDeviceSessionC
return deviceBirthMetrics;
}
public void setDeviceBirthMetrics(java.util.List<org.thingsboard.server.gen.transport.mqtt.SparkplugBProto.Payload.Metric> metrics) {
this.deviceBirthMetrics.putAll(metrics.stream()
.collect(Collectors.toMap(SparkplugBProto.Payload.Metric::getName, metric -> metric)));
public void setDeviceBirthMetrics(java.util.List<org.thingsboard.server.gen.transport.mqtt.SparkplugBProto.Payload.Metric> metrics) {
for (var metric : metrics) {
if (metric.hasName()) {
this.deviceBirthMetrics.put(metric.getName(), metric);
} else {
throw new IllegalArgumentException("The metric name of device: '" + this.getDeviceInfo().getDeviceName() + "' must not be empty or null! Metric: [" + metric + "]");
}
if (metric.hasAlias() && this.parent.getNodeAlias().putIfAbsent(metric.getAlias(), metric.getName()) != null) {
throw new DuplicateKeyException("The alias '" + metric.getAlias() + "' already exists in device: '" + this.getDeviceInfo().getDeviceName() + "'");
}
}
}

86
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java
View File

@ -24,6 +24,7 @@ import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.util.CollectionUtils;
import org.thingsboard.server.common.adaptor.AdaptorException;
import org.thingsboard.server.common.adaptor.ProtoConverter;
@ -47,12 +48,10 @@ import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.ONLINE;
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.DBIRTH;
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.NBIRTH;
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.STATE;
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.parseMessageType;
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.SPARKPLUG_BD_SEQUENCE_NUMBER_KEY;
import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.SPARKPLUG_SEQUENCE_NUMBER_KEY;
@ -73,6 +72,8 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
private final SparkplugTopic sparkplugTopicNode;
@Getter
private final Map<String, SparkplugBProto.Payload.Metric> nodeBirthMetrics;
@Getter
private final Map<Long, String> nodeAlias;
private final MqttTransportHandler parent;
public SparkplugNodeSessionHandler(MqttTransportHandler parent, DeviceSessionCtx deviceSessionCtx, UUID sessionId,
@ -81,11 +82,20 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
this.parent = parent;
this.sparkplugTopicNode = sparkplugTopicNode;
this.nodeBirthMetrics = new ConcurrentHashMap<>();
this.nodeAlias = new ConcurrentHashMap<>();
}
public void setNodeBirthMetrics(java.util.List<org.thingsboard.server.gen.transport.mqtt.SparkplugBProto.Payload.Metric> metrics) {
this.nodeBirthMetrics.putAll(metrics.stream()
.collect(Collectors.toMap(SparkplugBProto.Payload.Metric::getName, metric -> metric)));
public void setNodeBirthMetrics(java.util.List<org.thingsboard.server.gen.transport.mqtt.SparkplugBProto.Payload.Metric> metrics) throws AdaptorException {
for (var metric : metrics) {
if (metric.hasName()) {
this.nodeBirthMetrics.put(metric.getName(), metric);
} else {
throw new AdaptorException("The metric name of edgeNode: '" + this.sparkplugTopicNode.getEdgeNodeId() + "' must not be empty or null! Metric: [" + metric + "]");
}
if (metric.hasAlias() && this.nodeAlias.putIfAbsent(metric.getAlias(), metric.getName()) != null) {
throw new AdaptorException("The alias '" + metric.getAlias() + "' already exists in edgeNode: '" + this.sparkplugTopicNode.getEdgeNodeId() + "'");
}
}
}
@ -113,7 +123,11 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
if (topic.isType(DBIRTH)) {
sendSparkplugStateOnTelemetry(ctx.getSessionInfo(), deviceName, ONLINE,
sparkplugBProto.getTimestamp());
ctx.setDeviceBirthMetrics(sparkplugBProto.getMetricsList());
try {
ctx.setDeviceBirthMetrics(sparkplugBProto.getMetricsList());
} catch (IllegalArgumentException | DuplicateKeyException e) {
throw new RuntimeException(e);
}
}
return ctx;
}, MoreExecutors.directExecutor());
@ -203,17 +217,39 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
}
}
/**
* Sparkplug 3.0.0 -> 6.4.6. Metric
* https://sparkplug.eclipse.org/specification/version/3.0/documents/sparkplug-specification-3.0.0.pdf#%5B%7B%22num%22%3A339%2C%22gen%22%3A0%7D%2C%7B%22name%22%3A%22XYZ%22%7D%2C0%2C455.52%2Cnull%5D
* [tck-id-payloads-name-requirement] The name MUST be included with every metric unless aliases are being used. ◦ All UTF-8 characters are allowed in the metric name. However, special characters including but not limited to the following are discouraged: . , \ @ # $ % ^ & * ( ) [ ] { } | ! ` ~ : ; ' " < > ?. This is because many Sparkplug Host Applications may have issues handling them.
* • alias (are optional and not required):
* - This is an unsigned 64-bit integer representing an optional alias for a Sparkplug B payload.
* - If aliases are used, the following rules apply:
* -- [tck-id-payloads-alias-uniqueness] If supplied in an NBIRTH or BIRTH it MUST be a unique number across this Edge Node’s entire set of metrics.
* -- no two metrics for the same Edge Node can have the same alias.
* -- [tck-id-payloads-alias-birth-requirement] NBIRTH and DBIRTH messages MUST include both a metric name and alias.
* -- [tck-id-payloads-alias-data-cmd-requirement] NDATA, DDATA, NCMD, and DCMD messages MUST only include an alias and the metric name MUST be excluded.
* @param sparkplugBProto
* @param attributesMetricNames
* @param topicTypeName
* @return
* @throws AdaptorException
*/
private List<TransportProtos.PostTelemetryMsg> convertToPostTelemetry(SparkplugBProto.Payload
sparkplugBProto, Set<String> attributesMetricNames, String topicTypeName) throws AdaptorException {
try {
List<TransportProtos.PostTelemetryMsg> msgs = new ArrayList<>();
for (SparkplugBProto.Payload.Metric protoMetric : sparkplugBProto.getMetricsList()) {
if (attributesMetricNames == null || !matches(attributesMetricNames, protoMetric)) {
long ts = protoMetric.getTimestamp();
String key = SPARKPLUG_BD_SEQUENCE_NUMBER_KEY.equals(protoMetric.getName()) ?
topicTypeName + " " + protoMetric.getName() : protoMetric.getName();
Optional<TransportProtos.KeyValueProto> keyValueProtoOpt = fromSparkplugBMetricToKeyValueProto(key, protoMetric);
keyValueProtoOpt.ifPresent(kvProto -> msgs.add(postTelemetryMsgCreated(kvProto, ts)));
String metricName = protoMetric.hasName() ? protoMetric.getName() : protoMetric.hasAlias() ? this.nodeAlias.get(protoMetric.getAlias()) : null;
if (metricName == null) {
throw new ThingsboardException("Metric without metricName and alias.", ThingsboardErrorCode.INVALID_ARGUMENTS);
} else {
if (attributesMetricNames == null || !matches(attributesMetricNames, metricName)) {
long ts = protoMetric.getTimestamp();
String key = SPARKPLUG_BD_SEQUENCE_NUMBER_KEY.equals(protoMetric.getName()) ?
topicTypeName + " " + protoMetric.getName() : protoMetric.getName();
Optional<TransportProtos.KeyValueProto> keyValueProtoOpt = fromSparkplugBMetricToKeyValueProto(key, protoMetric);
keyValueProtoOpt.ifPresent(kvProto -> msgs.add(postTelemetryMsgCreated(kvProto, ts)));
}
}
}
@ -237,13 +273,18 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
try {
List<TransportApiProtos.AttributesMsg> msgs = new ArrayList<>();
for (SparkplugBProto.Payload.Metric protoMetric : sparkplugBProto.getMetricsList()) {
if (matches(attributesMetricNames, protoMetric)) {
TransportApiProtos.AttributesMsg.Builder deviceAttributesMsgBuilder = TransportApiProtos.AttributesMsg.newBuilder();
Optional<TransportProtos.PostAttributeMsg> msgOpt = getPostAttributeMsg(protoMetric);
if (msgOpt.isPresent()) {
deviceAttributesMsgBuilder.setDeviceName(deviceName);
deviceAttributesMsgBuilder.setMsg(msgOpt.get());
msgs.add(deviceAttributesMsgBuilder.build());
String metricName = protoMetric.hasName() ? protoMetric.getName() : protoMetric.hasAlias() ? this.nodeAlias.get(protoMetric.getAlias()) : null;
if (metricName == null) {
throw new ThingsboardException("Metric without metricName and alias.", ThingsboardErrorCode.INVALID_ARGUMENTS);
} else {
if (matches(attributesMetricNames, metricName)) {
TransportApiProtos.AttributesMsg.Builder deviceAttributesMsgBuilder = TransportApiProtos.AttributesMsg.newBuilder();
Optional<TransportProtos.PostAttributeMsg> msgOpt = getPostAttributeMsg(protoMetric);
if (msgOpt.isPresent()) {
deviceAttributesMsgBuilder.setDeviceName(deviceName);
deviceAttributesMsgBuilder.setMsg(msgOpt.get());
msgs.add(deviceAttributesMsgBuilder.build());
}
}
}
}
@ -254,8 +295,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
}
}
private boolean matches(Set<String> attributesMetricNames, SparkplugBProto.Payload.Metric protoMetric) {
String metricName = protoMetric.getName();
private boolean matches(Set<String> attributesMetricNames, String metricName) {
for (String attributeMetricFilter : attributesMetricNames) {
if (metricName.equals(attributeMetricFilter) ||
(attributeMetricFilter.endsWith("*") && metricName.startsWith(
@ -288,7 +328,9 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
if (value.isPresent()) {
SparkplugBProto.Payload.Builder cmdPayload = SparkplugBProto.Payload.newBuilder()
.setTimestamp(ts);
cmdPayload.addMetrics(createMetric(value.get(), ts, tsKvProto.getKv().getKey(), metricDataType));
String metricName = tsKvProto.getKv().getKey();
Long alias = metricBirth.hasAlias() ? metricBirth.getAlias() : -1;
cmdPayload.addMetrics(createMetric(value.get(), ts, alias == -1 ? metricName : null, metricDataType, alias));
byte[] payloadInBytes = cmdPayload.build().toByteArray();
return Optional.of(getPayloadAdaptor().createMqttPublishMsg(deviceSessionCtx, sparkplugTopic, payloadInBytes));
} else {

21
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugMetricUtil.java
View File

@ -29,6 +29,8 @@ import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.mqtt.SparkplugBProto;
import org.thingsboard.server.gen.transport.mqtt.SparkplugBProto.Payload.Metric;
import org.thingsboard.server.gen.transport.mqtt.SparkplugBProto.Payload.Metric.Builder;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
@ -144,13 +146,20 @@ public class SparkplugMetricUtil {
return Optional.empty();
}
}
public static SparkplugBProto.Payload.Metric createMetric(Object value, long ts, String key, MetricDataType metricDataType, Long alias) throws ThingsboardException {
Builder metric = Metric.newBuilder();
metric.setTimestamp(ts)
.setDatatype(metricDataType.toIntValue());
if (alias >= 0) {
metric.setAlias(alias);
}
if (StringUtils.isNotBlank(key)) {
metric.setName(key);
}
return addToMetricValue(value, metric.build(), metricDataType);
}
public static SparkplugBProto.Payload.Metric createMetric(Object value, long ts, String key, MetricDataType metricDataType) throws ThingsboardException {
SparkplugBProto.Payload.Metric metric = SparkplugBProto.Payload.Metric.newBuilder()
.setTimestamp(ts)
.setName(key)
.setDatatype(metricDataType.toIntValue())
.build();
public static SparkplugBProto.Payload.Metric addToMetricValue(Object value, SparkplugBProto.Payload.Metric metric, MetricDataType metricDataType) throws ThingsboardException {
switch (metricDataType) {
case Int8: // (byte)
return metric.toBuilder().setIntValue(((Byte) value).intValue()).build();

Write Preview
Loading…
Cancel
Save