sparkplug: add attributes by Metric`s name, STATE ONLINE

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

@@ -412,10 +412,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);

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

@@ -180,6 +180,14 @@ public abstract class AbstractGatewaySessionHandler {
     void deregisterSession(String deviceName) {
         MqttDeviceAwareSessionContext deviceSessionCtx = devices.remove(deviceName);
         if (deviceSessionCtx != null) {
+            if (deviceSessionCtx.isSparkplug()){
+//                onDeviceTelemetryProto(int msgId, ByteBuf payload)
+                // add Metric name: STATE type: String value: OFFLINE
+//                SparkplugBProto.Payload.Metric metricState = createMetric(SparkplugMessageTypeSate.OFFLINE.name(),
+//                        sparkplugBProtoDevice.getTimestamp(), STATE.name(), MetricDataType.Text);
+//                sparkplugBProtoDevice.getMetricsList().add(metricState);
+//                onTelemetryProto(-1, sparkplugBProtoDevice, deviceName, sparkplugTopic);
+            }
             deregisterSession(deviceName, deviceSessionCtx);
         } else {
             log.debug("[{}] Device [{}] was already removed from the gateway session", sessionId, deviceName);

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

@@ -51,6 +51,8 @@ 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.SparkplugMessageType.STATE;
+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 +104,12 @@ 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()
+                TransportProtos.KeyValueProto.Builder keyValueProtoBuilder = TransportProtos.KeyValueProto.newBuilder();
+                keyValueProtoBuilder.setKey(STATE.name());
+                keyValueProtoBuilder.setType(TransportProtos.KeyValueType.STRING_V);
+                keyValueProtoBuilder.setStringV(ONLINE.name());
+                msgs.add(postTelemetryMsgCreated(keyValueProtoBuilder.build(), sparkplugBProto.getTimestamp()));
                 contextListenableFuture.get().setDeviceBirthMetrics(sparkplugBProto.getMetricsList());
             } catch (InterruptedException | ExecutionException e) {
                 log.error("Failed add Metrics. MessageType *BIRTH.", e);
@@ -182,30 +190,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<>();
-                    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());
+                    msgs.add(postTelemetryMsgCreated(keyValueProtoOpt.get(), ts));
                 }
             }
 
             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());
-                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());
+                msgs.add(postTelemetryMsgCreated(keyValueProtoBuilder.build(), sparkplugBProto.getTimestamp()));
             }
             return msgs;
         } catch (IllegalStateException | JsonSyntaxException | ThingsboardException e) {
@@ -233,11 +227,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();
@@ -248,4 +242,15 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler {
         return new SparkplugDeviceSessionContext(this, msg.getDeviceInfo(), msg.getDeviceProfile(), mqttQoSMap, transportService);
     }
 
+    protected 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();
+    }
+
 }

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
+
+}