Merge pull request #8222 from thingsboard/sparkplug-transport-proto

[fix_bug][3.5]Sparkplug-transport-proto

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

@@ -137,6 +137,12 @@ public abstract class AbstractMqttV5ClientSparkplugConnectionTest extends Abstra
         }
     }
 
+    /**
+     * OFFLINE_All
+     * @param cntDevices
+     * @throws Exception
+     */
+
     protected void processConnectClientWithCorrectAccessTokenWithNDEATH_State_ONLINE_All_Then_OFFLINE_All(int cntDevices) throws Exception {
         long ts = calendar.getTimeInMillis();
         List<Device> devices = connectClientWithCorrectAccessTokenWithNDEATHCreatedDevices(cntDevices, ts);
@@ -166,6 +172,7 @@ public abstract class AbstractMqttV5ClientSparkplugConnectionTest extends Abstra
         }
     }
 
+
     private boolean findEqualsKeyValueInKvEntrys(List<TsKvEntry> finalFuture, TsKvEntry tsKvEntry) {
         for (TsKvEntry kvEntry : finalFuture) {
             if (kvEntry.getKey().equals(tsKvEntry.getKey()) && kvEntry.getValue().equals(tsKvEntry.getValue())) {
@@ -174,5 +181,5 @@ public abstract class AbstractMqttV5ClientSparkplugConnectionTest extends Abstra
         }
         return false;
     }
-
 }
+

common/cluster-api/src/main/proto/queue.proto

@@ -186,7 +186,6 @@ message GetOrCreateDeviceFromGatewayRequestMsg {
   int64 gatewayIdLSB = 2;
   string deviceName = 3;
   string deviceType = 4;
-  bool sparkplug = 5;
 }
 
 message GetOrCreateDeviceFromGatewayResponseMsg {

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

@@ -399,7 +399,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
                     case NBIRTH:
                     case NCMD:
                     case NDATA:
-                        sparkplugSessionHandler.onAttributesTelemetryProto(msgId, sparkplugBProtoNode, deviceSessionCtx.getDeviceInfo().getDeviceName(), sparkplugTopic);
+                        sparkplugSessionHandler.onAttributesTelemetryProto(msgId, sparkplugBProtoNode, sparkplugTopic);
                         break;
                     default:
                 }
@@ -410,7 +410,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
                     case DBIRTH:
                     case DCMD:
                     case DDATA:
-                        sparkplugSessionHandler.onAttributesTelemetryProto(msgId, sparkplugBProtoDevice, sparkplugTopic.getDeviceId(), sparkplugTopic);
+                        sparkplugSessionHandler.onAttributesTelemetryProto(msgId, sparkplugBProtoDevice, sparkplugTopic);
                         break;
                     case DDEATH:
                         sparkplugSessionHandler.onDeviceDisconnect(mqttMsg, sparkplugTopic.getDeviceId());
@@ -1067,8 +1067,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
                 if (sparkplugTopicNode != null) {
                     SparkplugBProto.Payload sparkplugBProtoNode = SparkplugBProto.Payload.parseFrom(connectMessage.payload().willMessageInBytes());
                     sparkplugSessionHandler = new SparkplugNodeSessionHandler(this, deviceSessionCtx, sessionId, sparkplugTopicNode);
-                    sparkplugSessionHandler.onAttributesTelemetryProto(0, sparkplugBProtoNode,
+                    sparkplugSessionHandler.onAttributesTelemetryProto(0, sparkplugBProtoNode, sparkplugTopicNode);
-                            deviceSessionCtx.getDeviceInfo().getDeviceName(), sparkplugTopicNode);
                     sessionMetaData.setOverwriteActivityTime(true);
                 } else {
                     log.trace("[{}][{}] Failed to fetch sparkplugDevice connect:  sparkplugTopicName without SparkplugMessageType.NDEATH.", sessionId, deviceSessionCtx.getDeviceInfo().getDeviceName());

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

@@ -253,7 +253,6 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
                             .setDeviceType(deviceType)
                             .setGatewayIdMSB(gateway.getDeviceId().getId().getMostSignificantBits())
                             .setGatewayIdLSB(gateway.getDeviceId().getId().getLeastSignificantBits())
-                            .setSparkplug(this.deviceSessionCtx.isSparkplug())
                             .build(),
                     new TransportServiceCallback<>() {
                         @Override

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

@@ -101,7 +101,8 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
         }
     }
 
-    public void onAttributesTelemetryProto(int msgId, SparkplugBProto.Payload sparkplugBProto, String deviceName, SparkplugTopic topic) throws AdaptorException, ThingsboardException {
+    public void onAttributesTelemetryProto(int msgId, SparkplugBProto.Payload sparkplugBProto, SparkplugTopic topic) throws AdaptorException, ThingsboardException {
+        String deviceName = topic.getNodeDeviceName();
         checkDeviceName(deviceName);
 
         ListenableFuture<MqttDeviceAwareSessionContext> contextListenableFuture;
@@ -113,7 +114,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
             }
             contextListenableFuture = Futures.immediateFuture(this.deviceSessionCtx);
         } else {
-            ListenableFuture<SparkplugDeviceSessionContext> deviceCtx = onDeviceConnectProto(deviceName);
+            ListenableFuture<SparkplugDeviceSessionContext> deviceCtx = onDeviceConnectProto(topic);
             contextListenableFuture = Futures.transform(deviceCtx, ctx -> {
                 if (topic.isType(DBIRTH)) {
                     sendSparkplugStateOnTelemetry(ctx.getSessionInfo(), deviceName, ONLINE,
@@ -218,10 +219,10 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
         }
     }
 
-    private ListenableFuture<SparkplugDeviceSessionContext> onDeviceConnectProto(String deviceName) throws ThingsboardException {
+    private ListenableFuture<SparkplugDeviceSessionContext> onDeviceConnectProto(SparkplugTopic topic) throws ThingsboardException {
         try {
             String deviceType = this.gateway.getDeviceType() + "-node";
-            return onDeviceConnect(deviceName, deviceType);
+            return onDeviceConnect(topic.getNodeDeviceName(), deviceType);
         } catch (RuntimeException e) {
             log.error("Failed Sparkplug Device connect proto!", e);
             throw new ThingsboardException(e, ThingsboardErrorCode.BAD_REQUEST_PARAMS);

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

@@ -161,4 +161,9 @@ public class SparkplugTopic {
     public boolean isNode() {
         return this.deviceId == null;
     }
+
+    public String getNodeDeviceName() {
+        return isNode() ? edgeNodeId : deviceId;
+    }
 }
+