Merge pull request #8070 from thingsboard/attributes

[3.4.5] sparkplug: add attributes by Metric`s name

common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DefaultDeviceProfileTransportConfiguration.java

@@ -16,7 +16,6 @@
 package org.thingsboard.server.common.data.device.profile;
 
 import lombok.Data;
-import org.thingsboard.server.common.data.DeviceProfileType;
 import org.thingsboard.server.common.data.DeviceTransportType;
 
 @Data

common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttDeviceProfileTransportConfiguration.java

@@ -19,6 +19,8 @@ import lombok.Data;
 import org.thingsboard.server.common.data.DeviceTransportType;
 import org.thingsboard.server.common.data.validation.NoXss;
 
+import java.util.Set;
+
 @Data
 public class MqttDeviceProfileTransportConfiguration implements DeviceProfileTransportConfiguration {
 
@@ -28,6 +30,7 @@ public class MqttDeviceProfileTransportConfiguration implements DeviceProfileTra
     private String deviceAttributesTopic = MqttTopics.DEVICE_ATTRIBUTES_TOPIC;
     private TransportPayloadTypeConfiguration transportPayloadTypeConfiguration;
     private boolean sparkPlug;
+    private Set<String> sparkPlugAttributesMetricNames;
     private boolean sendAckOnValidationException;
 
     @Override

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java

@@ -91,6 +91,7 @@ import java.security.cert.Certificate;
 import java.security.cert.X509Certificate;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Date;
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
@@ -110,6 +111,7 @@ import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
 import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_CLOSED;
 import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_OPEN;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.NDEATH;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageTypeSate.OFFLINE;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicUtil.parseTopicPublish;
 
 /**
@@ -412,10 +414,6 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
                         // TODO
                         break;
                     case DBIRTH:
-                        sparkplugSessionHandler.onTelemetryProto(msgId, sparkplugBProtoDevice, deviceName, sparkplugTopic);
-
-                        System.out.println();
-                        break;
                     case DCMD:
                     case DDATA:
                         sparkplugSessionHandler.onTelemetryProto(msgId, sparkplugBProtoDevice, deviceName, sparkplugTopic);
@@ -1128,7 +1126,13 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
             transportService.process(deviceSessionCtx.getSessionInfo(), SESSION_EVENT_MSG_CLOSED, null);
             transportService.deregisterSession(deviceSessionCtx.getSessionInfo());
             if (gatewaySessionHandler != null) {
-                gatewaySessionHandler.onGatewayDisconnect();
+                gatewaySessionHandler.onDevicesDisconnect();
+            }
+            if (sparkplugSessionHandler != null) {
+                    // add Msg Telemetry node: key STATE type: String value: OFFLINE ts: sparkplugBProto.getTimestamp()
+                sparkplugSessionHandler.stateSparkplugtSendOnTelemetry(deviceSessionCtx.getSessionInfo(),
+                        deviceSessionCtx.getDeviceInfo().getDeviceName(), OFFLINE, new Date().getTime());
+                sparkplugSessionHandler.onDevicesDisconnect();
             }
             deviceSessionCtx.setDisconnected();
         }
@@ -1228,6 +1232,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
     @Override
     public void onRemoteSessionCloseCommand(UUID sessionId, TransportProtos.SessionCloseNotificationProto sessionCloseNotification) {
         log.trace("[{}] Received the remote command to close the session: {}", sessionId, sessionCloseNotification.getMessage());
+        transportService.deregisterSession(deviceSessionCtx.getSessionInfo());
         deviceSessionCtx.getChannel().close();
     }
 

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

@@ -54,9 +54,12 @@ import org.thingsboard.server.transport.mqtt.adaptors.JsonMqttAdaptor;
 import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor;
 import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor;
 import org.thingsboard.server.transport.mqtt.util.ReturnCode;
+import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageTypeSate;
 
 import javax.annotation.Nullable;
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Date;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -72,6 +75,8 @@ import static org.thingsboard.server.common.transport.service.DefaultTransportSe
 import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_OPEN;
 import static org.thingsboard.server.common.transport.service.DefaultTransportService.SUBSCRIBE_TO_ATTRIBUTE_UPDATES_ASYNC_MSG;
 import static org.thingsboard.server.common.transport.service.DefaultTransportService.SUBSCRIBE_TO_RPC_ASYNC_MSG;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.STATE;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageTypeSate.OFFLINE;
 
 /**
  * Created by ashvayka on 19.01.17.
@@ -157,7 +162,7 @@ public abstract class AbstractGatewaySessionHandler {
         }
     }
 
-    public void onGatewayDisconnect() {
+    public void onDevicesDisconnect() {
         devices.forEach(this::deregisterSession);
     }
 
@@ -395,10 +400,21 @@ public abstract class AbstractGatewaySessionHandler {
         }
     }
 
-    protected void processPostTelemetryMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.PostTelemetryMsg postTelemetryMsg, String deviceName, int msgId) {
+    public void processPostTelemetryMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.PostTelemetryMsg postTelemetryMsg, String deviceName, int msgId) {
         transportService.process(deviceCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(channel, deviceName, msgId, postTelemetryMsg));
     }
 
+    public TransportProtos.PostTelemetryMsg postTelemetryMsgCreated(TransportProtos.KeyValueProto keyValueProto, long ts) {
+        List<TransportProtos.KeyValueProto> result = new ArrayList<>();
+        result.add(keyValueProto);
+        TransportProtos.PostTelemetryMsg.Builder request = TransportProtos.PostTelemetryMsg.newBuilder();
+        TransportProtos.TsKvListProto.Builder builder = TransportProtos.TsKvListProto.newBuilder();
+        builder.setTs(ts);
+        builder.addAllKv(result);
+        request.addTsKvList(builder.build());
+        return request.build();
+    }
+
     private void onDeviceClaimJson(int msgId, ByteBuf payload) throws AdaptorException {
         JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, payload);
         if (json.isJsonObject()) {
@@ -711,12 +727,26 @@ public abstract class AbstractGatewaySessionHandler {
     }
 
     private void deregisterSession(String deviceName, MqttDeviceAwareSessionContext deviceSessionCtx) {
+        if (this.deviceSessionCtx.isSparkplug()){
+            // add Msg Telemetry: key STATE type: String value: OFFLINE ts: sparkplugBProto.getTimestamp()
+            stateSparkplugtSendOnTelemetry (deviceSessionCtx.getSessionInfo(),
+                    deviceSessionCtx.getDeviceInfo().getDeviceName(), OFFLINE, new Date().getTime());
+        }
         transportService.deregisterSession(deviceSessionCtx.getSessionInfo());
         transportService.process(deviceSessionCtx.getSessionInfo(), SESSION_EVENT_MSG_CLOSED, null);
         System.out.println("Removed device " + deviceName + " from the gateway session");
         log.debug("[{}] Removed device [{}] from the gateway session", sessionId, deviceName);
     }
 
+    public void stateSparkplugtSendOnTelemetry (TransportProtos.SessionInfoProto sessionInfo, String deviceName, SparkplugMessageTypeSate typeSate, long ts) {
+        TransportProtos.KeyValueProto.Builder keyValueProtoBuilder = TransportProtos.KeyValueProto.newBuilder();
+        keyValueProtoBuilder.setKey(STATE.name());
+        keyValueProtoBuilder.setType(TransportProtos.KeyValueType.STRING_V);
+        keyValueProtoBuilder.setStringV(typeSate.name());
+        TransportProtos.PostTelemetryMsg postTelemetryMsg =  postTelemetryMsgCreated(keyValueProtoBuilder.build(), ts);
+        transportService.process(sessionInfo, postTelemetryMsg, getPubAckCallback(channel, deviceName, -1, postTelemetryMsg));
+    }
+
     private <T> TransportServiceCallback<Void> getPubAckCallback(final ChannelHandlerContext ctx, final String deviceName, final int msgId, final T msg) {
         return new TransportServiceCallback<Void>() {
             @Override
@@ -734,4 +764,5 @@ public abstract class AbstractGatewaySessionHandler {
             }
         };
     }
+
 }

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

@@ -51,6 +51,7 @@ import java.util.stream.Collectors;
 
 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.SparkplugMessageTypeSate.ONLINE;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.createMetric;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.fromSparkplugBMetricToKeyValueProto;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.validatedValueByTypeMetric;
@@ -102,6 +103,9 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler {
         List<TransportProtos.PostTelemetryMsg> msgs = convertToPostTelemetry(sparkplugBProto, topic.getType().name());
         if (topic.isType(NBIRTH) || topic.isType(DBIRTH)) {
             try {
+                // add Msg Telemetry: key STATE type: String value: ONLINE ts: sparkplugBProto.getTimestamp()
+                stateSparkplugtSendOnTelemetry(contextListenableFuture.get().getSessionInfo(), deviceName, ONLINE,
+                        sparkplugBProto.getTimestamp());
                 contextListenableFuture.get().setDeviceBirthMetrics(sparkplugBProto.getMetricsList());
             } catch (InterruptedException | ExecutionException e) {
                 log.error("Failed add Metrics. MessageType *BIRTH.", e);
@@ -182,30 +186,16 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler {
                         topicTypeName + " " + protoMetric.getName() : protoMetric.getName();
                 Optional<TransportProtos.KeyValueProto> keyValueProtoOpt = fromSparkplugBMetricToKeyValueProto(key, protoMetric);
                 if (keyValueProtoOpt.isPresent()) {
-                    List<TransportProtos.KeyValueProto> result = new ArrayList<>();
+                    msgs.add(postTelemetryMsgCreated(keyValueProtoOpt.get(), ts));
-                    result.add(keyValueProtoOpt.get());
-                    TransportProtos.PostTelemetryMsg.Builder request = TransportProtos.PostTelemetryMsg.newBuilder();
-                    TransportProtos.TsKvListProto.Builder builder = TransportProtos.TsKvListProto.newBuilder();
-                    builder.setTs(ts);
-                    builder.addAllKv(result);
-                    request.addTsKvList(builder.build());
-                    msgs.add(request.build());
                 }
             }
 
             if (DBIRTH.name().equals(topicTypeName)) {
-                List<TransportProtos.KeyValueProto> result = new ArrayList<>();
                 TransportProtos.KeyValueProto.Builder keyValueProtoBuilder = TransportProtos.KeyValueProto.newBuilder();
                 keyValueProtoBuilder.setKey(topicTypeName + " " + "seq");
                 keyValueProtoBuilder.setType(TransportProtos.KeyValueType.LONG_V);
                 keyValueProtoBuilder.setLongV(sparkplugBProto.getSeq());
-                result.add(keyValueProtoBuilder.build());
+                msgs.add(postTelemetryMsgCreated(keyValueProtoBuilder.build(), sparkplugBProto.getTimestamp()));
-                TransportProtos.PostTelemetryMsg.Builder request = TransportProtos.PostTelemetryMsg.newBuilder();
-                TransportProtos.TsKvListProto.Builder builder = TransportProtos.TsKvListProto.newBuilder();
-                builder.setTs(sparkplugBProto.getTimestamp());
-                builder.addAllKv(result);
-                request.addTsKvList(builder.build());
-                msgs.add(request.build());
             }
             return msgs;
         } catch (IllegalStateException | JsonSyntaxException | ThingsboardException e) {
@@ -233,11 +223,11 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler {
                 return Optional.of(getPayloadAdaptor().createMqttPublishMsg(deviceSessionCtx, sparkplugTopic, payloadInBytes));
             } else {
                 log.trace("DeviceId: [{}] tenantId: [{}] sessionId:[{}] Failed to convert device attributes [{}] response to MQTT sparkplug  msg",
-                        deviceSessionCtx.getDeviceInfo().getDeviceId(), deviceSessionCtx.getDeviceInfo().getTenantId(),  sessionId, tsKvProto.getKv());
+                        deviceSessionCtx.getDeviceInfo().getDeviceId(), deviceSessionCtx.getDeviceInfo().getTenantId(), sessionId, tsKvProto.getKv());
             }
         } catch (Exception e) {
             log.trace("DeviceId: [{}] tenantId: [{}] sessionId:[{}] Failed to convert device attributes response to MQTT sparkplug  msg",
-                    deviceSessionCtx.getDeviceInfo().getDeviceId(), deviceSessionCtx.getDeviceInfo().getTenantId(),  sessionId, e);
+                    deviceSessionCtx.getDeviceInfo().getDeviceId(), deviceSessionCtx.getDeviceInfo().getTenantId(), sessionId, e);
             return Optional.empty();
         }
         return Optional.empty();

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugMessageTypeSate.java

@@ -0,0 +1,30 @@
+/**
+ * Copyright © 2016-2022 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.transport.mqtt.util.sparkplug;
+
+public enum SparkplugMessageTypeSate {
+    /**
+     * The EoN node should examine the payload of this
+     * message to ensure that it is a value of “ONLINE”
+     */
+    OFFLINE,
+    /**
+     * If the value is “OFFLINE”, this indicates the Primary Application
+     * has lost its MQTT Session to this particular MQTT Server.
+     */
+    ONLINE
+
+}