dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactoring of the initial implementation

Browse Source
This commit is contained in:
Andrii Shvaika 2022-12-21 14:21:09 +02:00
parent ca54838f23
commit 38ef27567c
5 changed files with 26 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -283,13 +283,6 @@ public class DefaultTransportApiService implements TransportApiService {
Lock deviceCreationLock = deviceCreationLocks.computeIfAbsent(requestMsg.getDeviceName(), id -> new ReentrantLock()); Lock deviceCreationLock = deviceCreationLocks.computeIfAbsent(requestMsg.getDeviceName(), id -> new ReentrantLock());
deviceCreationLock.lock(); deviceCreationLock.lock();
try { try {
DeviceProfile deviceProfile = deviceProfileCache.findOrCreateDeviceProfile(gateway.getTenantId(), requestMsg.getDeviceType());
DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration();
boolean isSparkplug = false;
if (transportConfiguration instanceof MqttDeviceProfileTransportConfiguration &&
((MqttDeviceProfileTransportConfiguration) transportConfiguration).isSparkPlug()) {
isSparkplug = true;
}
Device device = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), requestMsg.getDeviceName()); Device device = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), requestMsg.getDeviceName());
if (device == null) { if (device == null) {
TenantId tenantId = gateway.getTenantId(); TenantId tenantId = gateway.getTenantId();
@ -298,6 +291,7 @@ public class DefaultTransportApiService implements TransportApiService {
device.setName(requestMsg.getDeviceName()); device.setName(requestMsg.getDeviceName());
device.setType(requestMsg.getDeviceType()); device.setType(requestMsg.getDeviceType());
device.setCustomerId(gateway.getCustomerId()); device.setCustomerId(gateway.getCustomerId());
DeviceProfile deviceProfile = deviceProfileCache.findOrCreateDeviceProfile(gateway.getTenantId(), requestMsg.getDeviceType());
device.setDeviceProfileId(deviceProfile.getId()); device.setDeviceProfileId(deviceProfile.getId());
ObjectNode additionalInfo = JacksonUtil.newObjectNode(); ObjectNode additionalInfo = JacksonUtil.newObjectNode();
@ -314,7 +308,7 @@ public class DefaultTransportApiService implements TransportApiService {
if (customerId != null && !customerId.isNullUid()) { if (customerId != null && !customerId.isNullUid()) {
metaData.putValue("customerId", customerId.toString()); metaData.putValue("customerId", customerId.toString());
} }
String deviceIdStr = isSparkplug ? "sparkplugId" : "gatewayId"; String deviceIdStr = requestMsg.getSparkplug() ? "sparkplugId" : "gatewayId";
metaData.putValue(deviceIdStr, gatewayId.toString()); metaData.putValue(deviceIdStr, gatewayId.toString());
DeviceId deviceId = device.getId(); DeviceId deviceId = device.getId();
@ -326,7 +320,7 @@ public class DefaultTransportApiService implements TransportApiService {
if (deviceAdditionalInfo == null) { if (deviceAdditionalInfo == null) {
deviceAdditionalInfo = JacksonUtil.newObjectNode(); deviceAdditionalInfo = JacksonUtil.newObjectNode();
} }
String lastConnectedStr = isSparkplug ? DataConstants.LAST_CONNECTED_SPARKPLUG : DataConstants.LAST_CONNECTED_GATEWAY; String lastConnectedStr = requestMsg.getSparkplug() ? DataConstants.LAST_CONNECTED_SPARKPLUG : DataConstants.LAST_CONNECTED_GATEWAY;
if (deviceAdditionalInfo.isObject() && if (deviceAdditionalInfo.isObject() &&
(!deviceAdditionalInfo.has(lastConnectedStr) (!deviceAdditionalInfo.has(lastConnectedStr)
|| !gatewayId.toString().equals(deviceAdditionalInfo.get(lastConnectedStr).asText()))) { || !gatewayId.toString().equals(deviceAdditionalInfo.get(lastConnectedStr).asText()))) {
@ -338,7 +332,12 @@ public class DefaultTransportApiService implements TransportApiService {
} }
GetOrCreateDeviceFromGatewayResponseMsg.Builder builder = GetOrCreateDeviceFromGatewayResponseMsg.newBuilder() GetOrCreateDeviceFromGatewayResponseMsg.Builder builder = GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
.setDeviceInfo(getDeviceInfoProto(device)); .setDeviceInfo(getDeviceInfoProto(device));
builder.setProfileBody(ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfile))); DeviceProfile deviceProfile = deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId());
if (deviceProfile != null) {
builder.setProfileBody(ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfile)));
} else {
log.warn("[{}] Failed to find device profile [{}] for device. ", device.getId(), device.getDeviceProfileId());
}
return TransportApiResponseMsg.newBuilder() return TransportApiResponseMsg.newBuilder()
.setGetOrCreateDeviceResponseMsg(builder.build()) .setGetOrCreateDeviceResponseMsg(builder.build())
.build(); .build();

1
common/cluster-api/src/main/proto/queue.proto
View File

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

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

@ -132,7 +132,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
final DeviceSessionCtx deviceSessionCtx; final DeviceSessionCtx deviceSessionCtx;
volatile InetSocketAddress address; volatile InetSocketAddress address;
volatile GatewaySessionHandler gatewaySessionHandler; volatile GatewaySessionHandler gatewaySessionHandler;
volatile SparkplugNodeSessionHandler sparkPlugSessionHandler; volatile SparkplugNodeSessionHandler sparkplugSessionHandler;
private final ConcurrentHashMap<String, String> otaPackSessions; private final ConcurrentHashMap<String, String> otaPackSessions;
private final ConcurrentHashMap<String, Integer> chunkSizes; private final ConcurrentHashMap<String, Integer> chunkSizes;
@ -325,14 +325,14 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
int msgId = mqttMsg.variableHeader().packetId(); int msgId = mqttMsg.variableHeader().packetId();
log.trace("[{}][{}] Processing publish msg [{}][{}]!", sessionId, deviceSessionCtx.getDeviceId(), topicName, msgId); log.trace("[{}][{}] Processing publish msg [{}][{}]!", sessionId, deviceSessionCtx.getDeviceId(), topicName, msgId);
if (topicName.startsWith(MqttTopics.BASE_GATEWAY_API_TOPIC)) { if (sparkplugSessionHandler != null) {
handleSparkplugPublishMsg(ctx, topicName, msgId, mqttMsg);
transportService.reportActivity(deviceSessionCtx.getSessionInfo());
} else if (topicName.startsWith(MqttTopics.BASE_GATEWAY_API_TOPIC)) {
if (gatewaySessionHandler != null) { if (gatewaySessionHandler != null) {
handleGatewayPublishMsg(ctx, topicName, msgId, mqttMsg); handleGatewayPublishMsg(ctx, topicName, msgId, mqttMsg);
transportService.reportActivity(deviceSessionCtx.getSessionInfo()); transportService.reportActivity(deviceSessionCtx.getSessionInfo());
} }
} else if (sparkPlugSessionHandler != null) {
handleSparkplugPublishMsg(ctx, topicName, msgId, mqttMsg);
transportService.reportActivity(deviceSessionCtx.getSessionInfo());
} else { } else {
processDevicePublish(ctx, mqttMsg, topicName, msgId); processDevicePublish(ctx, mqttMsg, topicName, msgId);
} }
@ -376,7 +376,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
private void handleSparkplugPublishMsg(ChannelHandlerContext ctx, String topicName, int msgId, MqttPublishMessage mqttMsg) { private void handleSparkplugPublishMsg(ChannelHandlerContext ctx, String topicName, int msgId, MqttPublishMessage mqttMsg) {
try { try {
sparkPlugSessionHandler.onPublishMsg(ctx, topicName, msgId, mqttMsg); sparkplugSessionHandler.onPublishMsg(ctx, topicName, msgId, mqttMsg);
} catch (RuntimeException e) { } catch (RuntimeException e) {
log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e); log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
ctx.close(); ctx.close();
@ -643,9 +643,9 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
String topic = subscription.topicName(); String topic = subscription.topicName();
MqttQoS reqQoS = subscription.qualityOfService(); MqttQoS reqQoS = subscription.qualityOfService();
try { try {
if (sparkPlugSessionHandler != null) { if (sparkplugSessionHandler != null) {
SparkplugTopic sparkplugTopic = parseTopic(mqttMsg.payload().topicSubscriptions().get(0).topicName()); SparkplugTopic sparkplugTopic = parseTopic(mqttMsg.payload().topicSubscriptions().get(0).topicName());
sparkPlugSessionHandler.handleSparkplugSubscribeMsg(grantedQoSList, sparkplugTopic, reqQoS); sparkplugSessionHandler.handleSparkplugSubscribeMsg(grantedQoSList, sparkplugTopic, reqQoS);
} else { } else {
switch (topic) { switch (topic) {
case MqttTopics.DEVICE_ATTRIBUTES_TOPIC: { case MqttTopics.DEVICE_ATTRIBUTES_TOPIC: {
@ -978,14 +978,14 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
} }
} }
private void checkSparkPlugSession(MqttConnectMessage connectMessage) { private void checkSparkplugSession(MqttConnectMessage connectMessage) {
try { try {
SparkplugTopic sparkplugTopic = parseTopic(connectMessage.payload().willTopic()); SparkplugTopic sparkplugTopic = parseTopic(connectMessage.payload().willTopic());
// Test proto // Test proto
SparkplugBProto.Payload payloadBProto = SparkplugBProto.Payload.parseFrom(connectMessage.payload().willMessageInBytes()); SparkplugBProto.Payload payloadBProto = SparkplugBProto.Payload.parseFrom(connectMessage.payload().willMessageInBytes());
// //
if (sparkPlugSessionHandler == null) { if (sparkplugSessionHandler == null) {
sparkPlugSessionHandler = new SparkplugNodeSessionHandler(deviceSessionCtx, sessionId, sparkplugTopic.toString()); sparkplugSessionHandler = new SparkplugNodeSessionHandler(deviceSessionCtx, sessionId, sparkplugTopic.toString());
} else { } else {
log.warn("SparkPlugNodeReConnected [{}] [{}]", sparkplugTopic.getDeviceId(), sparkplugTopic.getType()); log.warn("SparkPlugNodeReConnected [{}] [{}]", sparkplugTopic.getDeviceId(), sparkplugTopic.getType());
} }
@ -1029,7 +1029,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
public void onSuccess(Void msg) { public void onSuccess(Void msg) {
SessionMetaData sessionMetaData = transportService.registerAsyncSession(deviceSessionCtx.getSessionInfo(), MqttTransportHandler.this); SessionMetaData sessionMetaData = transportService.registerAsyncSession(deviceSessionCtx.getSessionInfo(), MqttTransportHandler.this);
if (deviceSessionCtx.isSparkplug()) { if (deviceSessionCtx.isSparkplug()) {
checkSparkPlugSession(connectMessage); checkSparkplugSession(connectMessage);
} else { } else {
checkGatewaySession(sessionMetaData); checkGatewaySession(sessionMetaData);
} }

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

@ -360,6 +360,7 @@ public class SparkplugNodeSessionHandler {
.setDeviceType(deviceType) .setDeviceType(deviceType)
.setGatewayIdMSB(nodeSparkplugInfo.getDeviceId().getId().getMostSignificantBits()) .setGatewayIdMSB(nodeSparkplugInfo.getDeviceId().getId().getMostSignificantBits())
.setGatewayIdLSB(nodeSparkplugInfo.getDeviceId().getId().getLeastSignificantBits()) .setGatewayIdLSB(nodeSparkplugInfo.getDeviceId().getId().getLeastSignificantBits())
.setSparkplug(true)
.build(), .build(),
new TransportServiceCallback<>() { new TransportServiceCallback<>() {
@Override @Override

7
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java
View File

@ -87,11 +87,10 @@ public abstract class DeviceAwareSessionContext implements SessionContext {
public boolean isSparkplug() { public boolean isSparkplug() {
DeviceProfileTransportConfiguration transportConfiguration = this.deviceProfile.getProfileData().getTransportConfiguration(); DeviceProfileTransportConfiguration transportConfiguration = this.deviceProfile.getProfileData().getTransportConfiguration();
if (transportConfiguration instanceof MqttDeviceProfileTransportConfiguration) { if (transportConfiguration instanceof MqttDeviceProfileTransportConfiguration) {
if (((MqttDeviceProfileTransportConfiguration) transportConfiguration).isSparkPlug()) { return ((MqttDeviceProfileTransportConfiguration) transportConfiguration).isSparkPlug();
return true; } else {
} return false;
} }
return false;
} }
} }