sparkplug - add tests with devices when devices are already created

application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java

@@ -114,6 +114,7 @@ import java.util.stream.Collectors;
 
 import static org.thingsboard.server.service.transport.BasicCredentialsValidationResult.PASSWORD_MISMATCH;
 import static org.thingsboard.server.service.transport.BasicCredentialsValidationResult.VALID;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.DEVICE_NAME_SPLIT_REGEXP;
 
 /**
  * Created by ashvayka on 05.10.18.
@@ -330,86 +331,153 @@ public class DefaultTransportApiService implements TransportApiService {
     }
 
     private TransportApiResponseMsg handle(GetOrCreateDeviceFromGatewayRequestMsg requestMsg) {
-        DeviceId gatewayId = new DeviceId(new UUID(requestMsg.getGatewayIdMSB(), requestMsg.getGatewayIdLSB()));
+        DeviceId gatewayId = toDeviceId(requestMsg);
         Device gateway = deviceService.findDeviceById(TenantId.SYS_TENANT_ID, gatewayId);
-        Lock deviceCreationLock = deviceCreationLocks.computeIfAbsent(requestMsg.getDeviceName(), id -> new ReentrantLock());
+        String deviceName = requestMsg.getDeviceName();
-        deviceCreationLock.lock();
+        Lock lock = deviceCreationLocks.computeIfAbsent(deviceName, k -> new ReentrantLock());
+        lock.lock();
         try {
-            Device device = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), requestMsg.getDeviceName());
+            Device device = findOrCreateDevice(requestMsg, gateway, gatewayId);
-            if (device == null) {
+            updateLastConnectedGateway(device, gatewayId);
-                TenantId tenantId = gateway.getTenantId();
+            return buildResponse(device);
-                device = new Device();
-                device.setTenantId(tenantId);
-                device.setName(requestMsg.getDeviceName());
-                device.setType(requestMsg.getDeviceType());
-                device.setCustomerId(gateway.getCustomerId());
-                DeviceProfile deviceProfile = deviceProfileCache.findOrCreateDeviceProfile(gateway.getTenantId(), requestMsg.getDeviceType());
-
-                device.setDeviceProfileId(deviceProfile.getId());
-                ObjectNode additionalInfo = JacksonUtil.newObjectNode();
-                additionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString());
-                device.setAdditionalInfo(additionalInfo);
-                device = deviceService.saveDevice(device);
-
-                relationService.saveRelation(tenantId, new EntityRelation(gateway.getId(), device.getId(), "Created"));
-
-                TbMsgMetaData metaData = new TbMsgMetaData();
-                CustomerId customerId = gateway.getCustomerId();
-                if (customerId != null && !customerId.isNullUid()) {
-                    metaData.putValue("customerId", customerId.toString());
-                }
-                metaData.putValue("gatewayId", gatewayId.toString());
-
-                DeviceId deviceId = device.getId();
-                JsonNode entityNode = JacksonUtil.valueToTree(device);
-                TbMsg tbMsg = TbMsg.newMsg()
-                        .type(TbMsgType.ENTITY_CREATED)
-                        .originator(deviceId)
-                        .customerId(customerId)
-                        .copyMetaData(metaData)
-                        .dataType(TbMsgDataType.JSON)
-                        .data(JacksonUtil.toString(entityNode))
-                        .build();
-                tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, null);
-            } else {
-                JsonNode deviceAdditionalInfo = device.getAdditionalInfo();
-                if (deviceAdditionalInfo == null) {
-                    deviceAdditionalInfo = JacksonUtil.newObjectNode();
-                }
-                if (deviceAdditionalInfo.isObject() &&
-                        (!deviceAdditionalInfo.has(DataConstants.LAST_CONNECTED_GATEWAY)
-                                || !gatewayId.toString().equals(deviceAdditionalInfo.get(DataConstants.LAST_CONNECTED_GATEWAY).asText()))) {
-                    ObjectNode newDeviceAdditionalInfo = (ObjectNode) deviceAdditionalInfo;
-                    newDeviceAdditionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString());
-                    deviceService.saveDevice(device);
-                }
-            }
-            GetOrCreateDeviceFromGatewayResponseMsg.Builder builder = GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
-                    .setDeviceInfo(ProtoUtils.toDeviceInfoProto(device));
-            DeviceProfile deviceProfile = deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId());
-            if (deviceProfile != null) {
-                builder.setDeviceProfile(ProtoUtils.toProto(deviceProfile));
-            } else {
-                log.warn("[{}] Failed to find device profile [{}] for device. ", device.getId(), device.getDeviceProfileId());
-            }
-            return TransportApiResponseMsg.newBuilder()
-                    .setGetOrCreateDeviceResponseMsg(builder.build())
-                    .build();
         } catch (JsonProcessingException e) {
-            log.warn("[{}] Failed to lookup device by gateway id and name: [{}]", gatewayId, requestMsg.getDeviceName(), e);
+            log.warn("[{}] Failed to process device [{}]", gatewayId, deviceName, e);
             throw new RuntimeException(e);
         } catch (EntitiesLimitExceededException e) {
-            log.warn("[{}][{}] API limit exception: [{}]", e.getTenantId(), gatewayId, e.getMessage());
+            return buildLimitErrorResponse(e, gatewayId);
-            return TransportApiResponseMsg.newBuilder()
-                    .setGetOrCreateDeviceResponseMsg(
-                            GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
-                                    .setError(TransportProtos.TransportApiRequestErrorCode.ENTITY_LIMIT))
-                    .build();
         } finally {
-            deviceCreationLock.unlock();
+            lock.unlock();
+            deviceCreationLocks.remove(deviceName, lock);
+        }
+    }
+
+    private DeviceId toDeviceId(GetOrCreateDeviceFromGatewayRequestMsg requestMsg) {
+        return new DeviceId(new UUID(
+                requestMsg.getGatewayIdMSB(),
+                requestMsg.getGatewayIdLSB()
+        ));
+    }
+
+    private Device findOrCreateDevice(GetOrCreateDeviceFromGatewayRequestMsg requestMsg,
+                                      Device gateway,
+                                      DeviceId gatewayId) throws JsonProcessingException {
+        TenantId tenantId = gateway.getTenantId();
+        String deviceName = requestMsg.getDeviceName();
+        Device device = deviceService.findDeviceByTenantIdAndName(tenantId, deviceName);
+        if (device != null) {
+            return device;
+        }
+        device = tryRenameSparkplugDevice(requestMsg, gateway);
+        if (device != null) {
+            return device;
+        }
+        device = createNewDevice(requestMsg, gateway, gatewayId);
+        pushCreatedEvent(device, gateway);
+        return device;
+    }
+
+    private Device tryRenameSparkplugDevice(GetOrCreateDeviceFromGatewayRequestMsg requestMsg,
+                                            Device gateway) {
+        if (!requestMsg.getIsSparkplug()) {
+            return null;
+        }
+        String[] topicPath = requestMsg.getDeviceName().split(DEVICE_NAME_SPLIT_REGEXP);
+        if (topicPath.length != 3) {
+            return null;
+        }
+        String deviceId = topicPath[2];
+        Device existingDevice =
+                deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), deviceId);
+        if (existingDevice == null) {
+            return null;
+        }
+        existingDevice.setName(requestMsg.getDeviceName());
+        return deviceService.saveDevice(existingDevice);
+    }
+
+    private Device createNewDevice(GetOrCreateDeviceFromGatewayRequestMsg requestMsg,
+                                   Device gateway,
+                                   DeviceId gatewayId) {
+        TenantId tenantId = gateway.getTenantId();
+        Device device = new Device();
+        device.setTenantId(tenantId);
+        device.setName(requestMsg.getDeviceName());
+        device.setType(requestMsg.getDeviceType());
+        device.setCustomerId(gateway.getCustomerId());
+        DeviceProfile profile =
+                deviceProfileCache.findOrCreateDeviceProfile(tenantId, requestMsg.getDeviceType());
+        device.setDeviceProfileId(profile.getId());
+        ObjectNode additionalInfo = JacksonUtil.newObjectNode();
+        additionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString());
+        device.setAdditionalInfo(additionalInfo);
+        device = deviceService.saveDevice(device);
+        relationService.saveRelation(
+                tenantId,
+                new EntityRelation(gateway.getId(), device.getId(), "Created")
+        );
+        return device;
+    }
+
+    private void updateLastConnectedGateway(Device device, DeviceId gatewayId) {
+        String gatewayIdStr = gatewayId.toString();
+        JsonNode info = device.getAdditionalInfo();
+        ObjectNode objectNode = (info instanceof ObjectNode)
+                ? (ObjectNode) info
+                : JacksonUtil.newObjectNode();
+        if (!objectNode.has(DataConstants.LAST_CONNECTED_GATEWAY)
+                || !gatewayIdStr.equals(objectNode.get(DataConstants.LAST_CONNECTED_GATEWAY).asText())) {
+            objectNode.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayIdStr);
+            device.setAdditionalInfo(objectNode);
+            deviceService.saveDevice(device);
         }
     }
 
+    private void pushCreatedEvent(Device device, Device gateway) {
+        TenantId tenantId = gateway.getTenantId();
+        CustomerId customerId = gateway.getCustomerId();
+        TbMsgMetaData metaData = new TbMsgMetaData();
+        metaData.putValue("gatewayId", gateway.getId().toString());
+        if (customerId != null && !customerId.isNullUid()) {
+            metaData.putValue("customerId", customerId.toString());
+        }
+        JsonNode entityNode = JacksonUtil.valueToTree(device);
+        TbMsg msg = TbMsg.newMsg()
+                .type(TbMsgType.ENTITY_CREATED)
+                .originator(device.getId())
+                .customerId(customerId)
+                .copyMetaData(metaData)
+                .dataType(TbMsgDataType.JSON)
+                .data(JacksonUtil.toString(entityNode))
+                .build();
+        tbClusterService.pushMsgToRuleEngine(tenantId, device.getId(), msg, null);
+    }
+
+    private TransportApiResponseMsg buildResponse(Device device) throws JsonProcessingException {
+        GetOrCreateDeviceFromGatewayResponseMsg.Builder builder =
+                GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
+                        .setDeviceInfo(ProtoUtils.toDeviceInfoProto(device));
+        DeviceProfile profile =
+                deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId());
+        if (profile != null) {
+            builder.setDeviceProfile(ProtoUtils.toProto(profile));
+        }
+        return TransportApiResponseMsg.newBuilder()
+                .setGetOrCreateDeviceResponseMsg(builder.build())
+                .build();
+    }
+
+    private TransportApiResponseMsg buildLimitErrorResponse(EntitiesLimitExceededException e,
+                                                            DeviceId gatewayId) {
+        log.warn("[{}][{}] API limit exception: [{}]",
+                e.getTenantId(), gatewayId, e.getMessage());
+        return TransportApiResponseMsg.newBuilder()
+                .setGetOrCreateDeviceResponseMsg(
+                        GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
+                                .setError(TransportProtos.TransportApiRequestErrorCode.ENTITY_LIMIT)
+                )
+                .build();
+    }
+
     private TransportApiResponseMsg handle(ProvisionDeviceRequestMsg requestMsg) {
         ProvisionResponse provisionResponse;
         try {

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

@@ -71,7 +71,9 @@ import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConn
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.STATE;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.messageName;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMetricUtil.createMetric;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.DEVICE_NAME_SPLIT_REGEXP;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_ROOT_SPB_V_1_0;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_REGEXP;
 
 /**
  * Created by nickAS21 on 12.01.23
@@ -103,37 +105,18 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
     protected Set<String> sparkplugAttributesMetricNames;
 
     public void beforeSparkplugTest(boolean isCreateDevices) throws Exception {
+        MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder()
+                .gatewayName(edgeNodeDeviceName)
+                .isSparkplug(true)
+                .sparkplugAttributesMetricNames(sparkplugAttributesMetricNames)
+                .transportPayloadType(TransportPayloadType.PROTOBUF)
+                .build();
+        processBeforeTest(configProperties);
         if (isCreateDevices) {
-            MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder()
+            String deviceName = deviceId + "_1";
-                    .gatewayName(edgeNodeDeviceName)
+            createDevice(deviceName, deviceProfile.getName(), false);
-                    .isSparkplug(true)
+            deviceName = groupId + DEVICE_NAME_SPLIT_REGEXP + edgeNode + DEVICE_NAME_SPLIT_REGEXP + deviceId + "_2";
-                    .sparkplugAttributesMetricNames(sparkplugAttributesMetricNames)
+            createDevice(deviceName, deviceProfile.getName(), false);
-                    .transportPayloadType(TransportPayloadType.PROTOBUF)
-                    .build();
-            processBeforeTest(configProperties);
-            configProperties = MqttTestConfigProperties.builder()
-                    .gatewayName(deviceId)
-                    .isSparkplug(true)
-                    .sparkplugAttributesMetricNames(sparkplugAttributesMetricNames)
-                    .transportPayloadType(TransportPayloadType.PROTOBUF)
-                    .build();
-            processBeforeTest(configProperties);
-            configProperties = MqttTestConfigProperties.builder()
-                    .gatewayName(groupId +  ":" + edgeNode +  ":" + deviceId)
-                    .isSparkplug(true)
-                    .sparkplugAttributesMetricNames(sparkplugAttributesMetricNames)
-                    .transportPayloadType(TransportPayloadType.PROTOBUF)
-                    .build();
-            processBeforeTest(configProperties);
-
-        } else {
-            MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder()
-                    .gatewayName(edgeNodeDeviceName)
-                    .isSparkplug(true)
-                    .sparkplugAttributesMetricNames(sparkplugAttributesMetricNames)
-                    .transportPayloadType(TransportPayloadType.PROTOBUF)
-                    .build();
-            processBeforeTest(configProperties);
         }
     }
 
@@ -175,7 +158,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
         options.setSessionExpiryInterval(0L);
         options.setUserName(gatewayAccessToken);
         String nameSpace = nameSpaceBad.length == 0 ? TOPIC_ROOT_SPB_V_1_0 : nameSpaceBad[0];
-        String topic = nameSpace + "/" + groupId + "/" + SparkplugMessageType.NDEATH.name() + "/" + edgeNode;
+        String topic = nameSpace + TOPIC_SPLIT_REGEXP + groupId + TOPIC_SPLIT_REGEXP + SparkplugMessageType.NDEATH.name() + TOPIC_SPLIT_REGEXP + edgeNode;
         // The NDEATH message MUST set the MQTT Will QoS to 1 and Retained flag to false
         MqttMessage msg = new MqttMessage();
         msg.setId(0);
@@ -198,7 +181,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
         payloadBirthNode.addMetrics(metric);
         payloadBirthNode.setTimestamp(ts);
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + groupId + TOPIC_SPLIT_REGEXP + SparkplugMessageType.NBIRTH.name() + TOPIC_SPLIT_REGEXP + edgeNode,
                     payloadBirthNode.build().toByteArray(), 0, false);
         }
 
@@ -212,7 +195,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
             String deviceName =  groupId +  ":" + edgeNode +  ":" + deviceIdName;
             payloadBirthDevice.addMetrics(metric);
             if (client.isConnected()) {
-                client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DBIRTH.name() + "/" + edgeNode + "/" + deviceIdName,
+                client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + groupId + TOPIC_SPLIT_REGEXP + SparkplugMessageType.DBIRTH.name() + TOPIC_SPLIT_REGEXP + edgeNode + TOPIC_SPLIT_REGEXP + deviceIdName,
                         payloadBirthDevice.build().toByteArray(), 0, false);
                 AtomicReference<Device> device = new AtomicReference<>();
                 await(alias + "find device [" + deviceIdName + "] after created")
@@ -243,52 +226,53 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
         payloadBirthNode.addMetrics(metric);
         payloadBirthNode.setTimestamp(ts);
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + groupId + TOPIC_SPLIT_REGEXP + SparkplugMessageType.NBIRTH.name() + TOPIC_SPLIT_REGEXP + edgeNode,
                     payloadBirthNode.build().toByteArray(), 0, false);
         }
 
         valueDeviceInt32 = 4024;
         metric = createMetric(valueDeviceInt32, ts, metricBirthName_Int32, metricBirthDataType_Int32, -1L);
         // as old device name -> deviceId
-        String deviceName = deviceId;
+        String deviceiDName1 = deviceId + "_1";
-        AtomicReference<Device> device1 = new AtomicReference<>();
-        String finalDeviceName1 = deviceName;
-        await(alias + "find device [" + deviceId + "] before connecting")
-                .atMost(200, TimeUnit.SECONDS)
-                .until(() -> {
-                    device1.set(doGet("/api/tenant/devices?deviceName=" + finalDeviceName1, Device.class));
-                    return device1.get() != null;
-                });
 
         if (client.isConnected()) {
             SparkplugBProto.Payload.Builder payloadBirthDevice1 = SparkplugBProto.Payload.newBuilder()
                     .setTimestamp(ts)
                     .setSeq(getSeqNum());
             payloadBirthDevice1.addMetrics(metric);
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DBIRTH.name() + "/" + edgeNode + "/" + deviceId,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + groupId + TOPIC_SPLIT_REGEXP + SparkplugMessageType.DBIRTH.name() + TOPIC_SPLIT_REGEXP + edgeNode + TOPIC_SPLIT_REGEXP + deviceiDName1,
                     payloadBirthDevice1.build().toByteArray(), 0, false);
-            devices.add(device1.get());
+
         }
-        // as new device name ->  groupId +  ":" + edgeNode +  ":" + deviceId;
+        String deviceName1 = groupId + DEVICE_NAME_SPLIT_REGEXP + edgeNode + DEVICE_NAME_SPLIT_REGEXP + deviceiDName1;;
-        deviceName = groupId + ":" + edgeNode + ":" + deviceId;
+        AtomicReference<Device> device1 = new AtomicReference<>();
-        AtomicReference<Device> device2 = new AtomicReference<>();
+        await(alias + "find device [" + deviceName1 + "] before connecting")
-        String finalDeviceName2 = deviceName;
-        await(alias + "find device [" + deviceName + "] before connecting")
                 .atMost(200, TimeUnit.SECONDS)
                 .until(() -> {
-                    device2.set(doGet("/api/tenant/devices?deviceName=" + finalDeviceName2, Device.class));
+                    device1.set(doGet("/api/tenant/devices?deviceName=" + deviceName1, Device.class));
-                    return device2.get() != null;
+                    return device1.get() != null;
                 });
+        devices.add(device1.get());
 
+        // as new device name ->  groupId +  ":" + edgeNode +  ":" + deviceId;
+        String deviceiDName2 = deviceId + "_2";
         if (client.isConnected()) {
             SparkplugBProto.Payload.Builder payloadBirthDevice2 = SparkplugBProto.Payload.newBuilder()
                     .setTimestamp(ts)
                     .setSeq(getSeqNum());
             payloadBirthDevice2.addMetrics(metric);
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DBIRTH.name() + "/" + edgeNode + "/" + deviceId,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + groupId + TOPIC_SPLIT_REGEXP + SparkplugMessageType.DBIRTH.name() + TOPIC_SPLIT_REGEXP + edgeNode + TOPIC_SPLIT_REGEXP + deviceiDName2,
                     payloadBirthDevice2.build().toByteArray(), 0, false);
-            devices.add(device2.get());
         }
+        String deviceName2 = groupId + ":" + edgeNode + ":" + deviceiDName2;
+        AtomicReference<Device> device2 = new AtomicReference<>();
+        await(alias + "find device [" + deviceName2 + "] before connecting")
+                .atMost(200, TimeUnit.SECONDS)
+                .until(() -> {
+                    device2.set(doGet("/api/tenant/devices?deviceName=" + deviceName2, Device.class));
+                    return device2.get() != null;
+                });
+        devices.add(device2.get());
         Assert.assertEquals(cntDevices, devices.size());
         state_ONLINE_ALL (devices, calendar.getTimeInMillis());
     }
@@ -334,7 +318,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
         payloadBirthNode.addMetrics(metric);
         payloadBirthNode.setTimestamp(ts);
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + groupId + TOPIC_SPLIT_REGEXP + SparkplugMessageType.NBIRTH.name() + TOPIC_SPLIT_REGEXP + edgeNode,
                     payloadBirthNode.build().toByteArray(), 0, false);
         }
 
@@ -348,7 +332,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
 
         payloadBirthDevice.addMetrics(metric);
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DBIRTH.name() + "/" + edgeNode + "/" + deviceIdName,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + groupId + TOPIC_SPLIT_REGEXP + SparkplugMessageType.DBIRTH.name() + TOPIC_SPLIT_REGEXP + edgeNode + TOPIC_SPLIT_REGEXP + deviceIdName,
                     payloadBirthDevice.build().toByteArray(), 0, false);
             AtomicReference<Device> device = new AtomicReference<>();
             await(alias + "find device [" + deviceName + "] after created")
@@ -397,7 +381,7 @@ public abstract class AbstractMqttV5ClientSparkplugTest extends AbstractMqttInte
         listKeys.add(metricKey);
 
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.NBIRTH.name() + "/" + edgeNode,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + groupId + TOPIC_SPLIT_REGEXP + SparkplugMessageType.NBIRTH.name() + TOPIC_SPLIT_REGEXP + edgeNode,
                     payloadBirthNode.build().toByteArray(), 0, false);
         }
         return listKeys;

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

@@ -37,10 +37,10 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import static org.awaitility.Awaitility.await;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.OFFLINE;
-import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugConnectionState.ONLINE;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.STATE;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.messageName;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_ROOT_SPB_V_1_0;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_REGEXP;
 
 /**
  * Created by nickAS21 on 12.01.23
@@ -111,7 +111,7 @@ public abstract class AbstractMqttV5ClientSparkplugConnectionTest extends Abstra
         if (client.isConnected()) {
             List<Device> devicesList = new ArrayList<>(devices);
             Device device =  devicesList.get(indexDeviceDisconnect);
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + SparkplugMessageType.DDEATH.name() + "/" + edgeNode + "/" + device.getName(),
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + groupId + TOPIC_SPLIT_REGEXP + SparkplugMessageType.DDEATH.name() + TOPIC_SPLIT_REGEXP + edgeNode + TOPIC_SPLIT_REGEXP + device.getName(),
                     payloadDeathDevice.build().toByteArray(), 0, false);
             await(alias + messageName(STATE) + ", device: " + device.getName())
                     .atMost(40, TimeUnit.SECONDS)

application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java

@@ -30,6 +30,7 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import static org.awaitility.Awaitility.await;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_ROOT_SPB_V_1_0;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_REGEXP;
 
 /**
  * Created by nickAS21 on 12.01.23
@@ -67,7 +68,7 @@ public abstract class AbstractMqttV5ClientSparkplugTelemetryTest extends Abstrac
         createdAddMetricValuePrimitiveTsKv(listTsKvEntry, listKeys, ndataPayload, ts);
 
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + messageTypeName + "/" + edgeNode,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + groupId + TOPIC_SPLIT_REGEXP + messageTypeName + TOPIC_SPLIT_REGEXP + edgeNode,
                     ndataPayload.build().toByteArray(), 0, false);
         }
 
@@ -96,7 +97,7 @@ public abstract class AbstractMqttV5ClientSparkplugTelemetryTest extends Abstrac
         createdAddMetricValueArraysPrimitiveTsKv(listTsKvEntry, listKeys, ndataPayload, ts);
 
         if (client.isConnected()) {
-            client.publish(TOPIC_ROOT_SPB_V_1_0 + "/" + groupId + "/" + messageTypeName + "/" + edgeNode,
+            client.publish(TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + groupId + TOPIC_SPLIT_REGEXP + messageTypeName + TOPIC_SPLIT_REGEXP + edgeNode,
                     ndataPayload.build().toByteArray(), 0, false);
         }
 

common/proto/src/main/proto/queue.proto

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

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

@@ -261,7 +261,7 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
         log.trace("[{}][{}][{}] onDeviceConnect: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName);
         int msgId = getMsgId(msg);
         AtomicBoolean ackSent = new AtomicBoolean(false);
-        process(onDeviceConnect(deviceName, deviceType),
+        process(onDeviceConnect(deviceName, deviceType, false),
                 result -> {
                     ack(msg, MqttReasonCodes.PubAck.SUCCESS);
                     log.trace("[{}][{}][{}] onDeviceConnectOk: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName);
@@ -277,7 +277,7 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
         }
     }
 
-    ListenableFuture<T> onDeviceConnect(String deviceName, String deviceType) {
+    ListenableFuture<T> onDeviceConnect(String deviceName, String deviceType, boolean isSparkplug) {
         T result = devices.get(deviceName);
         if (result == null) {
             Lock deviceCreationLock = deviceCreationLockMap.computeIfAbsent(deviceName, s -> new ReentrantLock());
@@ -285,7 +285,7 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
             try {
                 result = devices.get(deviceName);
                 if (result == null) {
-                    return getDeviceCreationFuture(deviceName, deviceType);
+                    return getDeviceCreationFuture(deviceName, deviceType, isSparkplug);
                 } else {
                     return Futures.immediateFuture(result);
                 }
@@ -300,13 +300,13 @@ 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);
+            return onDeviceConnect(topic.getNodeDeviceNameAllPath(), deviceType, true);
         } else {
             return Futures.immediateFuture(result);
         }
     }
 
-    private ListenableFuture<T> getDeviceCreationFuture(String deviceName, String deviceType) {
+    private ListenableFuture<T> getDeviceCreationFuture(String deviceName, String deviceType, boolean isSparkplug) {
         final SettableFuture<T> futureToSet = SettableFuture.create();
         ListenableFuture<T> future = deviceFutures.putIfAbsent(deviceName, futureToSet);
         if (future != null) {
@@ -319,6 +319,7 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
                             .setDeviceType(deviceType)
                             .setGatewayIdMSB(gateway.getDeviceId().getId().getMostSignificantBits())
                             .setGatewayIdLSB(gateway.getDeviceId().getId().getLeastSignificantBits())
+                            .setIsSparkplug(isSparkplug)
                             .build(),
                     new TransportServiceCallback<>() {
                         @Override
@@ -882,7 +883,7 @@ public abstract class AbstractGatewaySessionHandler<T extends AbstractGatewayDev
     }
 
     protected void process(String deviceName, Consumer<T> onSuccess, Consumer<Throwable> onFailure) {
-        ListenableFuture<T> deviceCtxFuture = onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE);
+        ListenableFuture<T> deviceCtxFuture = onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE, false);
         process(deviceCtxFuture, onSuccess, onFailure);
     }
 

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

@@ -128,5 +128,4 @@ public class SparkplugDeviceSessionContext extends AbstractGatewayDeviceSessionC
                             rpcRequest.getMethodName() + ". " + e.getMessage());
         }
     }
-
 }

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

@@ -21,6 +21,7 @@ import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
 import org.thingsboard.server.common.data.exception.ThingsboardException;
 
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugMessageType.parseMessageType;
+import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.DEVICE_NAME_SPLIT_REGEXP;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_ROOT_SPB_V_1_0;
 import static org.thingsboard.server.transport.mqtt.util.sparkplug.SparkplugTopicService.TOPIC_SPLIT_REGEXP;
 
@@ -332,9 +333,9 @@ public class SparkplugTopic {
     public String getNodeDeviceNameAllPath() {
         StringBuilder sb = new StringBuilder();
         if (hostApplicationId == null) {
-            sb.append(getGroupId()).append(":").append(getEdgeNodeId());
+            sb.append(getGroupId()).append(DEVICE_NAME_SPLIT_REGEXP).append(getEdgeNodeId());
             if (getDeviceId() != null) {
-                sb.append(":").append(getDeviceId());
+                sb.append(DEVICE_NAME_SPLIT_REGEXP).append(getDeviceId());
             }
         }
         return sb.toString();

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

@@ -35,6 +35,7 @@ public class SparkplugTopicService {
     public static final String TOPIC_ROOT_SPB_V_1_0 = "spBv1.0";
     public static final String TOPIC_ROOT_CERT_SP = "$sparkplug/certificates/";
     public static final String TOPIC_SPLIT_REGEXP = "/";
+    public static final String DEVICE_NAME_SPLIT_REGEXP = ":";
     public static final String TOPIC_STATE_REGEXP = TOPIC_ROOT_SPB_V_1_0 + TOPIC_SPLIT_REGEXP + STATE.name() + TOPIC_SPLIT_REGEXP;
 
     public static SparkplugTopic getSplitTopic(String topic) throws ThingsboardException {