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

Merge branch 'feature/sparkplug-shared-attributes'

pull/8463/head
Andrii Shvaika 3 years ago
parent
cbb3575e4a 8e20fd8800
commit
6bf957cc5b
6 changed files with 28 additions and 8 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. 1
      common/cluster-api/src/main/proto/queue.proto
  2. 2
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
  3. 19
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugNodeSessionHandler.java
  4. 10
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/ProtoConverter.java
  5. 3
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
  6. 1
      ui-ngx/src/app/shared/models/device.models.ts

1
common/cluster-api/src/main/proto/queue.proto
View File

@ -148,6 +148,7 @@ message PostTelemetryMsg {
message PostAttributeMsg {
repeated KeyValueProto kv = 1;
bool shared = 2;
}
message GetAttributeRequestMsg {

2
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java
View File

@ -535,7 +535,7 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
throw new IllegalArgumentException("Attributes List for device: " + deviceName + " is empty!");
}
try {
TransportProtos.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributeMsg(kvListProto.toByteArray());
TransportProtos.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributeMsg(kvListProto);
processPostAttributesMsg(deviceCtx, postAttributeMsg, deviceName, msgId);
} catch (Throwable e) {
log.warn("[{}][{}] Failed to process device attributes command: {}", gateway.getDeviceId(), deviceName, kvListProto, e);

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

@ -173,7 +173,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
public void onSuccess(@Nullable MqttDeviceAwareSessionContext deviceCtx) {
TransportProtos.PostAttributeMsg kvListProto = attributesMsg.getMsg();
try {
TransportProtos.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributeMsg(kvListProto.toByteArray());
TransportProtos.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributeMsg(kvListProto);
processPostAttributesMsg(deviceCtx, postAttributeMsg, deviceName, msgId);
} catch (Throwable e) {
log.warn("[{}][{}] Failed to process device attributes command: {}", gateway.getDeviceId(), deviceName, kvListProto, e);
@ -233,7 +233,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
try {
List<TransportProtos.PostTelemetryMsg> msgs = new ArrayList<>();
for (SparkplugBProto.Payload.Metric protoMetric : sparkplugBProto.getMetricsList()) {
if (attributesMetricNames == null || !attributesMetricNames.contains(protoMetric.getName())) {
if (attributesMetricNames == null || !matches(attributesMetricNames, protoMetric)) {
long ts = protoMetric.getTimestamp();
String key = "bdSeq".equals(protoMetric.getName()) ?
topicTypeName + " " + protoMetric.getName() : protoMetric.getName();
@ -264,7 +264,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
try {
List<TransportApiProtos.AttributesMsg> msgs = new ArrayList<>();
for (SparkplugBProto.Payload.Metric protoMetric : sparkplugBProto.getMetricsList()) {
if (attributesMetricNames.contains(protoMetric.getName())) {
if (matches(attributesMetricNames, protoMetric)) {
TransportApiProtos.AttributesMsg.Builder deviceAttributesMsgBuilder = TransportApiProtos.AttributesMsg.newBuilder();
Optional<TransportProtos.PostAttributeMsg> msgOpt = getPostAttributeMsg(protoMetric);
if (msgOpt.isPresent()) {
@ -281,11 +281,24 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
}
}
private boolean matches(Set<String> attributesMetricNames, SparkplugBProto.Payload.Metric protoMetric) {
String metricName = protoMetric.getName();
for (String attributeMetricFilter : attributesMetricNames) {
if (metricName.equals(attributeMetricFilter) ||
(attributeMetricFilter.endsWith("*") && metricName.startsWith(
attributeMetricFilter.substring(0, attributeMetricFilter.length() - 1)))) {
return true;
}
}
return false;
}
private Optional<TransportProtos.PostAttributeMsg> getPostAttributeMsg(SparkplugBProto.Payload.Metric protoMetric) throws ThingsboardException {
Optional<TransportProtos.KeyValueProto> keyValueProtoOpt = fromSparkplugBMetricToKeyValueProto(protoMetric.getName(), protoMetric);
if (keyValueProtoOpt.isPresent()) {
TransportProtos.PostAttributeMsg.Builder builder = TransportProtos.PostAttributeMsg.newBuilder();
builder.addKv(keyValueProtoOpt.get());
builder.setShared(true);
return Optional.of(builder.build());
}
return Optional.empty();

10
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/ProtoConverter.java
View File

@ -67,13 +67,15 @@ public class ProtoConverter {
}
}
public static TransportProtos.PostAttributeMsg validatePostAttributeMsg(byte[] bytes) throws IllegalArgumentException, InvalidProtocolBufferException {
TransportProtos.PostAttributeMsg proto = TransportProtos.PostAttributeMsg.parseFrom(bytes);
List<TransportProtos.KeyValueProto> kvList = proto.getKvList();
if (!CollectionUtils.isEmpty(kvList)) {
public static TransportProtos.PostAttributeMsg validatePostAttributeMsg(TransportProtos.PostAttributeMsg msg) throws IllegalArgumentException, InvalidProtocolBufferException {
if (!CollectionUtils.isEmpty(msg.getKvList())) {
byte[] bytes = msg.toByteArray();
TransportProtos.PostAttributeMsg proto = TransportProtos.PostAttributeMsg.parseFrom(bytes);
List<TransportProtos.KeyValueProto> kvList = proto.getKvList();
List<TransportProtos.KeyValueProto> keyValueProtos = validateKeyValueProtos(kvList);
TransportProtos.PostAttributeMsg.Builder result = TransportProtos.PostAttributeMsg.newBuilder();
result.addAllKv(keyValueProtos);
result.setShared(msg.getShared());
return result.build();
} else {
throw new IllegalArgumentException("KeyValue list is empty!");

3
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
View File

@ -593,6 +593,9 @@ public class DefaultTransportService implements TransportService {
TbMsgMetaData metaData = new TbMsgMetaData();
metaData.putValue("deviceName", sessionInfo.getDeviceName());
metaData.putValue("deviceType", sessionInfo.getDeviceType());
if (msg.getShared()) {
metaData.putValue(DataConstants.SCOPE, DataConstants.SHARED_SCOPE);
}
metaData.putValue(DataConstants.NOTIFY_DEVICE_METADATA_KEY, "false");
CustomerId customerId = getCustomerId(sessionInfo);
sendToRuleEngine(tenantId, deviceId, customerId, sessionInfo, json, metaData, SessionMsgType.POST_ATTRIBUTES_REQUEST,

1
ui-ngx/src/app/shared/models/device.models.ts
View File

@ -371,6 +371,7 @@ export const createDeviceProfileTransportConfiguration = (type: DeviceTransportT
deviceAttributesTopic: 'v1/devices/me/attributes',
deviceAttributesSubscribeTopic: 'v1/devices/me/attributes',
sparkplug: false,
sparkplugAttributesMetricNames: ['Node Control/*', 'Device Control/*', 'Properties/*'],
sendAckOnValidationException: false,
transportPayloadTypeConfiguration: {
transportPayloadType: TransportPayloadType.JSON,

Write Preview
Loading…
Cancel
Save