diff --git a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java index 48bd497c6e..1baa3cdccf 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java +++ b/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()); - deviceCreationLock.lock(); + String deviceName = requestMsg.getDeviceName(); + Lock lock = deviceCreationLocks.computeIfAbsent(deviceName, k -> new ReentrantLock()); + lock.lock(); try { - Device device = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), requestMsg.getDeviceName()); - if (device == null) { - TenantId tenantId = gateway.getTenantId(); - 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(); + Device device = findOrCreateDevice(requestMsg, gateway, gatewayId); + updateLastConnectedGateway(device, gatewayId); + return buildResponse(device); } 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 TransportApiResponseMsg.newBuilder() - .setGetOrCreateDeviceResponseMsg( - GetOrCreateDeviceFromGatewayResponseMsg.newBuilder() - .setError(TransportProtos.TransportApiRequestErrorCode.ENTITY_LIMIT)) - .build(); + return buildLimitErrorResponse(e, gatewayId); } 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 { diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java index 7d8770bbcf..9cca1a1a54 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/AbstractMqttV5ClientSparkplugTest.java +++ b/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 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() - .gatewayName(edgeNodeDeviceName) - .isSparkplug(true) - .sparkplugAttributesMetricNames(sparkplugAttributesMetricNames) - .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); + String deviceName = deviceId + "_1"; + createDevice(deviceName, deviceProfile.getName(), false); + deviceName = groupId + DEVICE_NAME_SPLIT_REGEXP + edgeNode + DEVICE_NAME_SPLIT_REGEXP + deviceId + "_2"; + createDevice(deviceName, deviceProfile.getName(), false); } } @@ -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 = 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; - AtomicReference 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; - }); + String deviceiDName1 = deviceId + "_1"; 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; - deviceName = groupId + ":" + edgeNode + ":" + deviceId; - AtomicReference device2 = new AtomicReference<>(); - String finalDeviceName2 = deviceName; - await(alias + "find device [" + deviceName + "] before connecting") + String deviceName1 = groupId + DEVICE_NAME_SPLIT_REGEXP + edgeNode + DEVICE_NAME_SPLIT_REGEXP + deviceiDName1;; + AtomicReference device1 = new AtomicReference<>(); + await(alias + "find device [" + deviceName1 + "] before connecting") .atMost(200, TimeUnit.SECONDS) .until(() -> { - device2.set(doGet("/api/tenant/devices?deviceName=" + finalDeviceName2, Device.class)); - return device2.get() != null; + device1.set(doGet("/api/tenant/devices?deviceName=" + deviceName1, Device.class)); + 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 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 = 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; diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java index 89b17f6b48..d0f3430fe7 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/connection/AbstractMqttV5ClientSparkplugConnectionTest.java +++ b/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 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) diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java index 319b61b7f3..dc513421e6 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/sparkplug/timeseries/AbstractMqttV5ClientSparkplugTelemetryTest.java +++ b/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); } diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 8a44a19e17..70b623f6c6 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/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 { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index cbb805731e..799a0fdf21 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -261,7 +261,7 @@ public abstract class AbstractGatewaySessionHandler { ack(msg, MqttReasonCodes.PubAck.SUCCESS); log.trace("[{}][{}][{}] onDeviceConnectOk: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName); @@ -277,7 +277,7 @@ public abstract class AbstractGatewaySessionHandler onDeviceConnect(String deviceName, String deviceType) { + ListenableFuture 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 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 getDeviceCreationFuture(String deviceName, String deviceType) { + private ListenableFuture getDeviceCreationFuture(String deviceName, String deviceType, boolean isSparkplug) { final SettableFuture futureToSet = SettableFuture.create(); ListenableFuture future = deviceFutures.putIfAbsent(deviceName, futureToSet); if (future != null) { @@ -319,6 +319,7 @@ public abstract class AbstractGatewaySessionHandler() { @Override @@ -882,7 +883,7 @@ public abstract class AbstractGatewaySessionHandler onSuccess, Consumer onFailure) { - ListenableFuture deviceCtxFuture = onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE); + ListenableFuture deviceCtxFuture = onDeviceConnect(deviceName, DEFAULT_DEVICE_TYPE, false); process(deviceCtxFuture, onSuccess, onFailure); } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java index 2bd0d7702f..2653ed8166 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/SparkplugDeviceSessionContext.java +++ b/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()); } } - } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopic.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopic.java index 910fed0f3a..b27b36e837 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopic.java +++ b/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(); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopicService.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopicService.java index 74ea6858fe..6558ba618f 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/util/sparkplug/SparkplugTopicService.java +++ b/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 {