sparkplug - add group to name device

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

@@ -68,6 +68,7 @@ import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor;
 import org.thingsboard.server.transport.mqtt.adaptors.ProtoMqttAdaptor;
 import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService;
 import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState;
+import org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopic;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -296,6 +297,15 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
         }
     }
 
+    ListenableFuture<T> onDeviceConnectSparkplug(SparkplugTopic topic, String deviceType) {
+        T result = devices.get(topic.getNodeDeviceName());
+        if (result == null) {
+            return onDeviceConnect(topic.getNodeDeviceNameAllPath(), deviceType);
+        } else {
+            return Futures.immediateFuture(result);
+        }
+    }
+
     private ListenableFuture<T> getDeviceCreationFuture(String deviceName, String deviceType) {
         final SettableFuture<T> futureToSet = SettableFuture.create();
         ListenableFuture<T> future = deviceFutures.putIfAbsent(deviceName, futureToSet);
@@ -844,7 +854,7 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
                                 deviceSessionCtx, msgId, MqttReasonCodes.PubAck.UNSPECIFIED_ERROR.byteValue()));
                     }
                 }
-                if (msgId <= 0) {
+                if (!deviceSessionCtx.isSparkplug() && msgId <= 0) {
                     closeDeviceSession(deviceName, MqttReasonCodes.Disconnect.MALFORMED_PACKET);
                 }
             }

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

@@ -21,7 +21,6 @@ import com.google.common.util.concurrent.MoreExecutors;
 import com.google.gson.JsonSyntaxException;
 import io.netty.handler.codec.mqtt.MqttMessage;
 import io.netty.handler.codec.mqtt.MqttPublishMessage;
-import io.netty.handler.codec.mqtt.MqttReasonCodes;
 import io.netty.handler.codec.mqtt.MqttTopicSubscription;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
@@ -49,6 +48,7 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -106,31 +106,32 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
     }
 
     public void onAttributesTelemetryProto(int msgId, SparkplugBProto.Payload sparkplugBProto, SparkplugTopic topic) throws AdaptorException, ThingsboardException {
-        String deviceName = topic.getNodeDeviceName();
-        checkDeviceName(deviceName);
-
+        String deviceName = checkDeviceName(this.deviceSessionCtx.getDeviceInfo().getDeviceName());
         ListenableFuture<MqttDeviceAwareSessionContext> contextListenableFuture;
+        TransportProtos.SessionInfoProto sessionInfo = this.deviceSessionCtx.getSessionInfo();
         if (topic.isNode()) {
             if (topic.isType(NBIRTH)) {
-                sendSparkplugStateOnTelemetry(this.deviceSessionCtx.getSessionInfo(), deviceName, ONLINE,
+                sendSparkplugStateOnTelemetry(sessionInfo, deviceName, ONLINE,
                         sparkplugBProto.getTimestamp());
                 setNodeBirthMetrics(sparkplugBProto.getMetricsList());
             }
             contextListenableFuture = Futures.immediateFuture(this.deviceSessionCtx);
         } else {
-            ListenableFuture<SparkplugDeviceSessionContext> deviceCtx = onDeviceConnectProto(topic);
-            contextListenableFuture = Futures.transform(deviceCtx, ctx -> {
-                if (topic.isType(DBIRTH)) {
-                    sendSparkplugStateOnTelemetry(ctx.getSessionInfo(), deviceName, ONLINE,
-                            sparkplugBProto.getTimestamp());
-                    try {
-                        ctx.setDeviceBirthMetrics(sparkplugBProto.getMetricsList());
-                    } catch (IllegalArgumentException | DuplicateKeyException e) {
-                            throw new RuntimeException(e);
+            try {
+                ListenableFuture<SparkplugDeviceSessionContext> deviceCtx = this.onDeviceConnectProto(topic);
+                deviceName = checkDeviceName(deviceCtx.get().getDeviceInfo().getDeviceName());
+                String finalDeviceName = deviceName;
+                contextListenableFuture = Futures.transform(deviceCtx, ctx -> {
+                    if (topic.isType(DBIRTH)) {
+                        sendSparkplugStateOnTelemetry(sessionInfo, finalDeviceName, ONLINE,
+                                sparkplugBProto.getTimestamp());
+                            ctx.setDeviceBirthMetrics(sparkplugBProto.getMetricsList());
                     }
-                }
-                return ctx;
-            }, MoreExecutors.directExecutor());
+                    return ctx;
+                }, MoreExecutors.directExecutor());
+            }  catch (IllegalArgumentException | DuplicateKeyException | ExecutionException | InterruptedException e) {
+                throw new RuntimeException(e);
+            }
         }
         Set<String> attributesMetricNames = ((MqttDeviceProfileTransportConfiguration) deviceSessionCtx
                 .getDeviceProfile().getProfileData().getTransportConfiguration()).getSparkplugAttributesMetricNames();
@@ -222,7 +223,7 @@ public class SparkplugNodeSessionHandler extends AbstractGatewaySessionHandler<S
             ThingsboardException {
         try {
             String deviceType = this.gateway.getDeviceType() + " device";
-            return onDeviceConnect(topic.getNodeDeviceName(), deviceType);
+            return onDeviceConnectSparkplug(topic, 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

@@ -329,6 +329,17 @@ public class SparkplugTopic {
         return isNode() ? edgeNodeId : deviceId;
     }
 
+    public String getNodeDeviceNameAllPath() {
+        StringBuilder sb = new StringBuilder();
+        if (hostApplicationId == null) {
+            sb.append(getGroupId()).append(":").append(getEdgeNodeId());
+            if (getDeviceId() != null) {
+                sb.append(":").append(getDeviceId());
+            }
+        }
+        return sb.toString();
+    }
+
     public static boolean isValidIdElementToUTF8(String deviceIdElement) {
         if (deviceIdElement == null) {
             return false;