From 22755e87637d8f49aa3bf661a894ab7e31b0b30d Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Wed, 2 Sep 2020 18:52:59 +0300 Subject: [PATCH] Device Profile implementation for Transport --- application/pom.xml | 4 - .../server/actors/ActorSystemContext.java | 4 +- .../controller/DeviceProfileController.java | 2 + .../queue/DefaultTbClusterService.java | 18 ++- .../queue/DefaultTbCoreConsumerService.java | 2 +- .../DefaultTbRuleEngineConsumerService.java | 2 +- .../service/queue/TbClusterService.java | 3 + .../processing/AbstractConsumerService.java | 2 +- .../transport/DefaultTransportApiService.java | 108 +++++++++---- .../BaseDeviceProfileControllerTest.java | 7 +- .../server/common/data/DeviceProfile.java | 1 + .../server/common/data/DeviceProfileInfo.java | 8 +- .../server/common/data/DeviceProfileType.java | 3 +- .../common/data/DeviceTransportType.java | 22 +++ .../DefaultDeviceTransportConfiguration.java | 30 ++++ .../data/device/data/DeviceConfiguration.java | 3 +- .../common/data/device/data/DeviceData.java | 1 + .../data/DeviceTransportConfiguration.java | 38 +++++ ...=> Lwm2mDeviceTransportConfiguration.java} | 7 +- .../MqttDeviceTransportConfiguration.java | 29 ++++ ...ltDeviceProfileTransportConfiguration.java | 30 ++++ .../profile/DeviceProfileConfiguration.java | 3 +- .../device/profile/DeviceProfileData.java | 1 + .../DeviceProfileTransportConfiguration.java | 39 +++++ ...mDeviceProfileTransportConfiguration.java} | 7 +- ...ttDeviceProfileTransportConfiguration.java | 29 ++++ common/queue/src/main/proto/queue.proto | 23 +++ .../transport/mqtt/MqttTransportHandler.java | 2 + common/transport/transport-api/pom.xml | 4 + .../common/transport/SessionMsgListener.java | 5 + .../common/transport/TransportService.java | 6 + .../service/DefaultTransportService.java | 145 ++++++++++++++++-- .../util}/DataDecodingEncodingService.java | 6 +- .../transport/util}/ProtoWithFSTService.java | 10 +- .../dao/device/DeviceProfileServiceImpl.java | 2 + .../server/dao/device/DeviceServiceImpl.java | 12 +- .../server/dao/model/ModelConstants.java | 1 + .../dao/model/sql/DeviceProfileEntity.java | 7 + .../sql/device/DeviceProfileRepository.java | 6 +- .../resources/sql/schema-entities-hsql.sql | 1 + .../main/resources/sql/schema-entities.sql | 2 +- .../service/BaseDeviceProfileServiceTest.java | 10 +- 42 files changed, 557 insertions(+), 88 deletions(-) create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/DeviceTransportType.java create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/device/data/DefaultDeviceTransportConfiguration.java create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/device/data/DeviceTransportConfiguration.java rename common/data/src/main/java/org/thingsboard/server/common/data/device/data/{Lwm2mDeviceConfiguration.java => Lwm2mDeviceTransportConfiguration.java} (76%) create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/device/data/MqttDeviceTransportConfiguration.java create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DefaultDeviceProfileTransportConfiguration.java create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileTransportConfiguration.java rename common/data/src/main/java/org/thingsboard/server/common/data/device/profile/{Lwm2mDeviceProfileConfiguration.java => Lwm2mDeviceProfileTransportConfiguration.java} (75%) create mode 100644 common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttDeviceProfileTransportConfiguration.java rename {application/src/main/java/org/thingsboard/server/service/encoding => common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/util}/DataDecodingEncodingService.java (84%) rename {application/src/main/java/org/thingsboard/server/service/encoding => common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/util}/ProtoWithFSTService.java (82%) diff --git a/application/pom.xml b/application/pom.xml index aad62c41ba..c2ae7eccc2 100644 --- a/application/pom.xml +++ b/application/pom.xml @@ -46,10 +46,6 @@ - - de.ruedigermoeller - fst - io.netty netty-transport-native-epoll diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index d2611f2dda..0b6198ebbb 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -44,7 +44,6 @@ import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.msg.tools.TbRateLimits; -import org.thingsboard.server.dao.alarm.AlarmService; import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.audit.AuditLogService; @@ -65,7 +64,7 @@ import org.thingsboard.server.dao.user.UserService; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.service.component.ComponentDiscoveryService; -import org.thingsboard.server.service.encoding.DataDecodingEncodingService; +import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; import org.thingsboard.server.service.executors.DbCallbackExecutorService; import org.thingsboard.server.service.executors.ExternalCallExecutorService; import org.thingsboard.server.service.executors.SharedEventLoopGroupService; @@ -90,7 +89,6 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; @Slf4j @Component diff --git a/application/src/main/java/org/thingsboard/server/controller/DeviceProfileController.java b/application/src/main/java/org/thingsboard/server/controller/DeviceProfileController.java index 44509c8adf..1636e2724c 100644 --- a/application/src/main/java/org/thingsboard/server/controller/DeviceProfileController.java +++ b/application/src/main/java/org/thingsboard/server/controller/DeviceProfileController.java @@ -92,6 +92,8 @@ public class DeviceProfileController extends BaseController { DeviceProfile savedDeviceProfile = checkNotNull(deviceProfileService.saveDeviceProfile(deviceProfile)); + tbClusterService.onDeviceProfileChange(savedDeviceProfile, null); + logEntityAction(savedDeviceProfile.getId(), savedDeviceProfile, null, savedDeviceProfile.getId() == null ? ActionType.ADDED : ActionType.UPDATED, null); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java index 926b08f38b..4a56b94ac9 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java @@ -21,6 +21,7 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; import org.thingsboard.rule.engine.api.msg.ToDeviceActorNotificationMsg; +import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; @@ -29,6 +30,7 @@ import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.FromDeviceRPCResponseProto; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; @@ -40,7 +42,7 @@ import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; -import org.thingsboard.server.service.encoding.DataDecodingEncodingService; +import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; import org.thingsboard.server.service.rpc.FromDeviceRpcResponse; import java.util.HashSet; @@ -163,6 +165,20 @@ public class DefaultTbClusterService implements TbClusterService { broadcast(new ComponentLifecycleMsg(tenantId, entityId, state)); } + @Override + public void onDeviceProfileChange(DeviceProfile deviceProfile, TbQueueCallback callback) { + log.trace("[{}][{}] Processing device profile [{}] event", deviceProfile.getTenantId(), deviceProfile.getId(), deviceProfile.getName()); + TbQueueProducer> toTransportNfProducer = producerProvider.getTransportNotificationsMsgProducer(); + Set tbTransportServices = partitionService.getAllServiceIds(ServiceType.TB_TRANSPORT); + TransportProtos.DeviceProfileUpdateMsg profileUpdateMsg = TransportProtos.DeviceProfileUpdateMsg.newBuilder().setData(ByteString.copyFrom(encodingService.encode(deviceProfile))).build(); + ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setDeviceProfileUpdateMsg(profileUpdateMsg).build(); + for (String transportServiceId : tbTransportServices) { + TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_TRANSPORT, transportServiceId); + toTransportNfProducer.send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), transportMsg), null); + toTransportNfs.incrementAndGet(); + } + } + private void broadcast(ComponentLifecycleMsg msg) { byte[] msgBytes = encodingService.encode(msg); TbQueueProducer> toRuleEngineProducer = producerProvider.getRuleEngineNotificationsMsgProducer(); diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index 6e1eca4a7d..1dba860fcd 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -47,7 +47,7 @@ import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.provider.TbCoreQueueFactory; import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.queue.util.TbCoreComponent; -import org.thingsboard.server.service.encoding.DataDecodingEncodingService; +import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; import org.thingsboard.server.service.queue.processing.AbstractConsumerService; import org.thingsboard.server.service.rpc.FromDeviceRpcResponse; import org.thingsboard.server.service.rpc.TbCoreDeviceRpcService; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index 0f3efb6f05..ae43eff9e1 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -37,7 +37,7 @@ import org.thingsboard.server.queue.provider.TbRuleEngineQueueFactory; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import org.thingsboard.server.queue.util.TbRuleEngineComponent; -import org.thingsboard.server.service.encoding.DataDecodingEncodingService; +import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; import org.thingsboard.server.service.queue.processing.*; import org.thingsboard.server.service.rpc.FromDeviceRpcResponse; import org.thingsboard.server.service.rpc.TbRuleEngineDeviceRpcService; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java index cc722720d5..5f72cfa693 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java @@ -16,6 +16,7 @@ package org.thingsboard.server.service.queue; import org.thingsboard.rule.engine.api.msg.ToDeviceActorNotificationMsg; +import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; @@ -49,4 +50,6 @@ public interface TbClusterService { void onEntityStateChange(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent state); + void onDeviceProfileChange(DeviceProfile deviceProfile, TbQueueCallback callback); + } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index 4007c9e17d..5c2ec700f8 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -26,7 +26,7 @@ import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; -import org.thingsboard.server.service.encoding.DataDecodingEncodingService; +import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; import org.thingsboard.server.service.queue.TbPackCallback; import org.thingsboard.server.service.queue.TbPackProcessingContext; 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 58e36db6aa..dcab1114fd 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 @@ -21,16 +21,17 @@ import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; +import com.google.protobuf.ByteString; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.springframework.util.StringUtils; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; -import org.thingsboard.server.common.data.Tenant; +import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.security.DeviceCredentials; @@ -38,11 +39,14 @@ import org.thingsboard.server.common.data.security.DeviceCredentialsType; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; import org.thingsboard.server.dao.device.DeviceCredentialsService; +import org.thingsboard.server.dao.device.DeviceProfileService; import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.dao.tenant.TenantProfileService; import org.thingsboard.server.dao.tenant.TenantService; +import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.DeviceInfoProto; import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayResponseMsg; @@ -76,47 +80,60 @@ public class DefaultTransportApiService implements TransportApiService { private static final ObjectMapper mapper = new ObjectMapper(); //TODO: Constructor dependencies; - @Autowired - private TenantService tenantService; - - @Autowired - private TenantProfileService tenantProfileService; - - @Autowired - private DeviceService deviceService; - - @Autowired - private RelationService relationService; - - @Autowired - private DeviceCredentialsService deviceCredentialsService; - - @Autowired - private DeviceStateService deviceStateService; - - @Autowired - private DbCallbackExecutorService dbCallbackExecutorService; - - @Autowired - protected TbClusterService tbClusterService; + private final DeviceProfileService deviceProfileService; + private final TenantService tenantService; + private final TenantProfileService tenantProfileService; + private final DeviceService deviceService; + private final RelationService relationService; + private final DeviceCredentialsService deviceCredentialsService; + private final DeviceStateService deviceStateService; + private final DbCallbackExecutorService dbCallbackExecutorService; + private final TbClusterService tbClusterService; + private final DataDecodingEncodingService dataDecodingEncodingService; private final ConcurrentMap deviceCreationLocks = new ConcurrentHashMap<>(); + public DefaultTransportApiService(DeviceProfileService deviceProfileService, TenantService tenantService, + TenantProfileService tenantProfileService, DeviceService deviceService, + RelationService relationService, DeviceCredentialsService deviceCredentialsService, + DeviceStateService deviceStateService, DbCallbackExecutorService dbCallbackExecutorService, + TbClusterService tbClusterService, DataDecodingEncodingService dataDecodingEncodingService) { + this.deviceProfileService = deviceProfileService; + this.tenantService = tenantService; + this.tenantProfileService = tenantProfileService; + this.deviceService = deviceService; + this.relationService = relationService; + this.deviceCredentialsService = deviceCredentialsService; + this.deviceStateService = deviceStateService; + this.dbCallbackExecutorService = dbCallbackExecutorService; + this.tbClusterService = tbClusterService; + this.dataDecodingEncodingService = dataDecodingEncodingService; + } + @Override public ListenableFuture> handle(TbProtoQueueMsg tbProtoQueueMsg) { TransportApiRequestMsg transportApiRequestMsg = tbProtoQueueMsg.getValue(); if (transportApiRequestMsg.hasValidateTokenRequestMsg()) { ValidateDeviceTokenRequestMsg msg = transportApiRequestMsg.getValidateTokenRequestMsg(); - return Futures.transform(validateCredentials(msg.getToken(), DeviceCredentialsType.ACCESS_TOKEN), value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor()); + return Futures.transform(validateCredentials(msg.getToken(), DeviceCredentialsType.ACCESS_TOKEN), + value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor()); } else if (transportApiRequestMsg.hasValidateX509CertRequestMsg()) { ValidateDeviceX509CertRequestMsg msg = transportApiRequestMsg.getValidateX509CertRequestMsg(); - return Futures.transform(validateCredentials(msg.getHash(), DeviceCredentialsType.X509_CERTIFICATE), value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor()); + return Futures.transform(validateCredentials(msg.getHash(), DeviceCredentialsType.X509_CERTIFICATE), + value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor()); } else if (transportApiRequestMsg.hasGetOrCreateDeviceRequestMsg()) { - return Futures.transform(handle(transportApiRequestMsg.getGetOrCreateDeviceRequestMsg()), value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor()); + return Futures.transform(handle(transportApiRequestMsg.getGetOrCreateDeviceRequestMsg()), + value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor()); } else if (transportApiRequestMsg.hasGetTenantRoutingInfoRequestMsg()) { - return Futures.transform(handle(transportApiRequestMsg.getGetTenantRoutingInfoRequestMsg()), value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor()); + return Futures.transform(handle(transportApiRequestMsg.getGetTenantRoutingInfoRequestMsg()), + value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor()); + } else if (transportApiRequestMsg.hasGetDeviceProfileRequestMsg()) { + return Futures.transform(handle(transportApiRequestMsg.getGetDeviceProfileRequestMsg()), + value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor()); + } - return Futures.transform(getEmptyTransportApiResponseFuture(), value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor()); + return Futures.transform(getEmptyTransportApiResponseFuture(), + value -> new TbProtoQueueMsg<>(tbProtoQueueMsg.getKey(), value, tbProtoQueueMsg.getHeaders()), MoreExecutors.directExecutor()); } private ListenableFuture validateCredentials(String credentialsId, DeviceCredentialsType credentialsType) { @@ -160,10 +177,19 @@ public class DefaultTransportApiService implements TransportApiService { TbMsg tbMsg = TbMsg.newMsg(DataConstants.ENTITY_CREATED, deviceId, metaData, TbMsgDataType.JSON, mapper.writeValueAsString(entityNode)); tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, null); } + GetOrCreateDeviceFromGatewayResponseMsg.Builder builder = GetOrCreateDeviceFromGatewayResponseMsg.newBuilder() + .setDeviceInfo(getDeviceInfoProto(device)); + DeviceProfile deviceProfile = deviceProfileService.findDeviceProfileById(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() - .setGetOrCreateDeviceResponseMsg(GetOrCreateDeviceFromGatewayResponseMsg.newBuilder().setDeviceInfo(getDeviceInfoProto(device)).build()).build(); + .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 lookup device by gateway id and name: [{}]", gatewayId, requestMsg.getDeviceName(), e); throw new RuntimeException(e); } finally { deviceCreationLock.unlock(); @@ -182,6 +208,16 @@ public class DefaultTransportApiService implements TransportApiService { .setIsolatedTbRuleEngine(tenantProfile.isIsolatedTbRuleEngine()).build()).build(), dbCallbackExecutorService); } + private ListenableFuture handle(TransportProtos.GetDeviceProfileRequestMsg requestMsg) { + DeviceProfileId profileId = new DeviceProfileId(new UUID(requestMsg.getProfileIdMSB(), requestMsg.getProfileIdLSB())); + DeviceProfile deviceProfile = deviceProfileService.findDeviceProfileById(TenantId.SYS_TENANT_ID, profileId); + return Futures.immediateFuture(TransportApiResponseMsg.newBuilder() + .setGetDeviceProfileResponseMsg( + TransportProtos.GetDeviceProfileResponseMsg.newBuilder() + .setData(ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfile))) + .build()).build()); + } + private ListenableFuture getDeviceInfo(DeviceId deviceId, DeviceCredentials credentials) { return Futures.transform(deviceService.findDeviceByIdAsync(TenantId.SYS_TENANT_ID, deviceId), device -> { if (device == null) { @@ -191,6 +227,12 @@ public class DefaultTransportApiService implements TransportApiService { try { ValidateDeviceCredentialsResponseMsg.Builder builder = ValidateDeviceCredentialsResponseMsg.newBuilder(); builder.setDeviceInfo(getDeviceInfoProto(device)); + DeviceProfile deviceProfile = deviceProfileService.findDeviceProfileById(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()); + } if (!StringUtils.isEmpty(credentials.getCredentialsValue())) { builder.setCredentialsBody(credentials.getCredentialsValue()); } @@ -211,6 +253,8 @@ public class DefaultTransportApiService implements TransportApiService { .setDeviceIdLSB(device.getId().getId().getLeastSignificantBits()) .setDeviceName(device.getName()) .setDeviceType(device.getType()) + .setDeviceProfileIdMSB(device.getDeviceProfileId().getId().getMostSignificantBits()) + .setDeviceProfileIdLSB(device.getDeviceProfileId().getId().getLeastSignificantBits()) .setAdditionalInfo(mapper.writeValueAsString(device.getAdditionalInfo())) .build(); } diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseDeviceProfileControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseDeviceProfileControllerTest.java index d4b3ec5977..f1a268996d 100644 --- a/application/src/test/java/org/thingsboard/server/controller/BaseDeviceProfileControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/BaseDeviceProfileControllerTest.java @@ -19,6 +19,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import org.junit.After; import org.junit.Assert; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; @@ -151,11 +152,13 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController .andExpect(statusReason(containsString("Device profile with such name already exists"))); } + @Ignore @Test public void testSaveSameDeviceProfileWithDifferentType() throws Exception { DeviceProfile deviceProfile = this.createDeviceProfile("Device Profile"); DeviceProfile savedDeviceProfile = doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class); - savedDeviceProfile.setType(DeviceProfileType.LWM2M); + //TODO uncomment once we have other device types; + //savedDeviceProfile.setType(DeviceProfileType.LWM2M); doPost("/api/deviceProfile", savedDeviceProfile).andExpect(status().isBadRequest()) .andExpect(statusReason(containsString("Changing type of device profile is prohibited"))); } @@ -265,7 +268,7 @@ public abstract class BaseDeviceProfileControllerTest extends AbstractController Collections.sort(loadedDeviceProfileInfos, deviceProfileInfoIdComparator); List deviceProfileInfos = deviceProfiles.stream().map(deviceProfile -> new DeviceProfileInfo(deviceProfile.getId(), - deviceProfile.getName(), deviceProfile.getType())).collect(Collectors.toList()); + deviceProfile.getName(), deviceProfile.getType(), deviceProfile.getTransportType())).collect(Collectors.toList()); Assert.assertEquals(deviceProfileInfos, loadedDeviceProfileInfos); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java b/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java index 79620c2d9f..097d64b198 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfile.java @@ -40,6 +40,7 @@ public class DeviceProfile extends SearchTextBased implements H private String description; private boolean isDefault; private DeviceProfileType type; + private DeviceTransportType transportType; private RuleChainId defaultRuleChainId; private transient DeviceProfileData profileData; @JsonIgnore diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfileInfo.java b/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfileInfo.java index b8d157bd1d..310c9ece60 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfileInfo.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfileInfo.java @@ -31,18 +31,22 @@ import java.util.UUID; public class DeviceProfileInfo extends EntityInfo { private final DeviceProfileType type; + private final DeviceTransportType transportType; @JsonCreator public DeviceProfileInfo(@JsonProperty("id") EntityId id, @JsonProperty("name") String name, - @JsonProperty("type") DeviceProfileType type) { + @JsonProperty("type") DeviceProfileType type, + @JsonProperty("transportType") DeviceTransportType transportType) { super(id, name); this.type = type; + this.transportType = transportType; } - public DeviceProfileInfo(UUID uuid, String name, DeviceProfileType type) { + public DeviceProfileInfo(UUID uuid, String name, DeviceProfileType type, DeviceTransportType transportType) { super(EntityIdFactory.getByTypeAndUuid(EntityType.DEVICE_PROFILE, uuid), name); this.type = type; + this.transportType = transportType; } } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfileType.java b/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfileType.java index 19da934bf7..93ca102082 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfileType.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DeviceProfileType.java @@ -16,6 +16,5 @@ package org.thingsboard.server.common.data; public enum DeviceProfileType { - DEFAULT, - LWM2M + DEFAULT } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/DeviceTransportType.java b/common/data/src/main/java/org/thingsboard/server/common/data/DeviceTransportType.java new file mode 100644 index 0000000000..f4a0f99f69 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/DeviceTransportType.java @@ -0,0 +1,22 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data; + +public enum DeviceTransportType { + DEFAULT, + MQTT, + LWM2M +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DefaultDeviceTransportConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DefaultDeviceTransportConfiguration.java new file mode 100644 index 0000000000..1825193e01 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DefaultDeviceTransportConfiguration.java @@ -0,0 +1,30 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.device.data; + +import lombok.Data; +import org.thingsboard.server.common.data.DeviceProfileType; +import org.thingsboard.server.common.data.DeviceTransportType; + +@Data +public class DefaultDeviceTransportConfiguration implements DeviceTransportConfiguration { + + @Override + public DeviceTransportType getType() { + return DeviceTransportType.DEFAULT; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DeviceConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DeviceConfiguration.java index 4794f1592e..1ea2ee4f97 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DeviceConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DeviceConfiguration.java @@ -27,8 +27,7 @@ import org.thingsboard.server.common.data.DeviceProfileType; include = JsonTypeInfo.As.PROPERTY, property = "type") @JsonSubTypes({ - @JsonSubTypes.Type(value = DefaultDeviceConfiguration.class, name = "DEFAULT"), - @JsonSubTypes.Type(value = Lwm2mDeviceConfiguration.class, name = "LWM2M")}) + @JsonSubTypes.Type(value = DefaultDeviceConfiguration.class, name = "DEFAULT")}) public interface DeviceConfiguration { @JsonIgnore diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DeviceData.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DeviceData.java index eea7491e23..6c24ba5e28 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DeviceData.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DeviceData.java @@ -21,5 +21,6 @@ import lombok.Data; public class DeviceData { private DeviceConfiguration configuration; + private DeviceTransportConfiguration transportConfiguration; } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DeviceTransportConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DeviceTransportConfiguration.java new file mode 100644 index 0000000000..e9bd1a3245 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/DeviceTransportConfiguration.java @@ -0,0 +1,38 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.device.data; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; +import org.thingsboard.server.common.data.DeviceTransportType; + +@JsonIgnoreProperties(ignoreUnknown = true) +@JsonTypeInfo( + use = JsonTypeInfo.Id.NAME, + include = JsonTypeInfo.As.PROPERTY, + property = "type") +@JsonSubTypes({ + @JsonSubTypes.Type(value = DefaultDeviceTransportConfiguration.class, name = "DEFAULT"), + @JsonSubTypes.Type(value = MqttDeviceTransportConfiguration.class, name = "MQTT"), + @JsonSubTypes.Type(value = Lwm2mDeviceTransportConfiguration.class, name = "LWM2M")}) +public interface DeviceTransportConfiguration { + + @JsonIgnore + DeviceTransportType getType(); + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/data/Lwm2mDeviceConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/Lwm2mDeviceTransportConfiguration.java similarity index 76% rename from common/data/src/main/java/org/thingsboard/server/common/data/device/data/Lwm2mDeviceConfiguration.java rename to common/data/src/main/java/org/thingsboard/server/common/data/device/data/Lwm2mDeviceTransportConfiguration.java index af3e4ac310..1c6022e12f 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/data/Lwm2mDeviceConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/Lwm2mDeviceTransportConfiguration.java @@ -17,13 +17,14 @@ package org.thingsboard.server.common.data.device.data; import lombok.Data; import org.thingsboard.server.common.data.DeviceProfileType; +import org.thingsboard.server.common.data.DeviceTransportType; @Data -public class Lwm2mDeviceConfiguration implements DeviceConfiguration { +public class Lwm2mDeviceTransportConfiguration implements DeviceTransportConfiguration { @Override - public DeviceProfileType getType() { - return DeviceProfileType.LWM2M; + public DeviceTransportType getType() { + return DeviceTransportType.LWM2M; } } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/data/MqttDeviceTransportConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/MqttDeviceTransportConfiguration.java new file mode 100644 index 0000000000..6cbdee4a65 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/data/MqttDeviceTransportConfiguration.java @@ -0,0 +1,29 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.device.data; + +import lombok.Data; +import org.thingsboard.server.common.data.DeviceTransportType; + +@Data +public class MqttDeviceTransportConfiguration implements DeviceTransportConfiguration { + + @Override + public DeviceTransportType getType() { + return DeviceTransportType.MQTT; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DefaultDeviceProfileTransportConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DefaultDeviceProfileTransportConfiguration.java new file mode 100644 index 0000000000..5610e2555f --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DefaultDeviceProfileTransportConfiguration.java @@ -0,0 +1,30 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.device.profile; + +import lombok.Data; +import org.thingsboard.server.common.data.DeviceProfileType; +import org.thingsboard.server.common.data.DeviceTransportType; + +@Data +public class DefaultDeviceProfileTransportConfiguration implements DeviceProfileTransportConfiguration { + + @Override + public DeviceTransportType getType() { + return DeviceTransportType.DEFAULT; + } + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileConfiguration.java index 9a15aba144..3bb3d29c34 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileConfiguration.java @@ -27,8 +27,7 @@ import org.thingsboard.server.common.data.DeviceProfileType; include = JsonTypeInfo.As.PROPERTY, property = "type") @JsonSubTypes({ - @JsonSubTypes.Type(value = DefaultDeviceProfileConfiguration.class, name = "DEFAULT"), - @JsonSubTypes.Type(value = Lwm2mDeviceProfileConfiguration.class, name = "LWM2M")}) + @JsonSubTypes.Type(value = DefaultDeviceProfileConfiguration.class, name = "DEFAULT")}) public interface DeviceProfileConfiguration { @JsonIgnore diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileData.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileData.java index 2665d9a102..6f9870e84e 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileData.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileData.java @@ -21,5 +21,6 @@ import lombok.Data; public class DeviceProfileData { private DeviceProfileConfiguration configuration; + private DeviceProfileTransportConfiguration transportConfiguration; } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileTransportConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileTransportConfiguration.java new file mode 100644 index 0000000000..34854958d1 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/DeviceProfileTransportConfiguration.java @@ -0,0 +1,39 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.device.profile; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonSubTypes; +import com.fasterxml.jackson.annotation.JsonTypeInfo; +import org.thingsboard.server.common.data.DeviceProfileType; +import org.thingsboard.server.common.data.DeviceTransportType; + +@JsonIgnoreProperties(ignoreUnknown = true) +@JsonTypeInfo( + use = JsonTypeInfo.Id.NAME, + include = JsonTypeInfo.As.PROPERTY, + property = "type") +@JsonSubTypes({ + @JsonSubTypes.Type(value = DefaultDeviceProfileTransportConfiguration.class, name = "DEFAULT"), + @JsonSubTypes.Type(value = MqttDeviceProfileTransportConfiguration.class, name = "MQTT"), + @JsonSubTypes.Type(value = Lwm2mDeviceProfileTransportConfiguration.class, name = "LWM2M")}) +public interface DeviceProfileTransportConfiguration { + + @JsonIgnore + DeviceTransportType getType(); + +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/Lwm2mDeviceProfileConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/Lwm2mDeviceProfileTransportConfiguration.java similarity index 75% rename from common/data/src/main/java/org/thingsboard/server/common/data/device/profile/Lwm2mDeviceProfileConfiguration.java rename to common/data/src/main/java/org/thingsboard/server/common/data/device/profile/Lwm2mDeviceProfileTransportConfiguration.java index 3ad18f35bc..83e1247e1c 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/Lwm2mDeviceProfileConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/Lwm2mDeviceProfileTransportConfiguration.java @@ -17,13 +17,14 @@ package org.thingsboard.server.common.data.device.profile; import lombok.Data; import org.thingsboard.server.common.data.DeviceProfileType; +import org.thingsboard.server.common.data.DeviceTransportType; @Data -public class Lwm2mDeviceProfileConfiguration implements DeviceProfileConfiguration { +public class Lwm2mDeviceProfileTransportConfiguration implements DeviceProfileTransportConfiguration { @Override - public DeviceProfileType getType() { - return DeviceProfileType.LWM2M; + public DeviceTransportType getType() { + return DeviceTransportType.LWM2M; } } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttDeviceProfileTransportConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttDeviceProfileTransportConfiguration.java new file mode 100644 index 0000000000..6a65d70b43 --- /dev/null +++ b/common/data/src/main/java/org/thingsboard/server/common/data/device/profile/MqttDeviceProfileTransportConfiguration.java @@ -0,0 +1,29 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.data.device.profile; + +import lombok.Data; +import org.thingsboard.server.common.data.DeviceTransportType; + +@Data +public class MqttDeviceProfileTransportConfiguration implements DeviceProfileTransportConfiguration { + + @Override + public DeviceTransportType getType() { + return DeviceTransportType.MQTT; + } + +} diff --git a/common/queue/src/main/proto/queue.proto b/common/queue/src/main/proto/queue.proto index f5e7328131..898208ac48 100644 --- a/common/queue/src/main/proto/queue.proto +++ b/common/queue/src/main/proto/queue.proto @@ -51,6 +51,8 @@ message SessionInfoProto { string deviceType = 9; int64 gwSessionIdMSB = 10; int64 gwSessionIdLSB = 11; + int64 deviceProfileIdMSB = 12; + int64 deviceProfileIdLSB = 13; } enum SessionEvent { @@ -99,6 +101,8 @@ message DeviceInfoProto { string deviceName = 5; string deviceType = 6; string additionalInfo = 7; + int64 deviceProfileIdMSB = 8; + int64 deviceProfileIdLSB = 9; } /** @@ -147,6 +151,7 @@ message ValidateDeviceX509CertRequestMsg { message ValidateDeviceCredentialsResponseMsg { DeviceInfoProto deviceInfo = 1; string credentialsBody = 2; + bytes profileBody = 3; } message GetOrCreateDeviceFromGatewayRequestMsg { @@ -158,6 +163,7 @@ message GetOrCreateDeviceFromGatewayRequestMsg { message GetOrCreateDeviceFromGatewayResponseMsg { DeviceInfoProto deviceInfo = 1; + bytes profileBody = 2; } message GetTenantRoutingInfoRequestMsg { @@ -170,6 +176,19 @@ message GetTenantRoutingInfoResponseMsg { bool isolatedTbRuleEngine = 2; } +message GetDeviceProfileRequestMsg { + int64 profileIdMSB = 1; + int64 profileIdLSB = 2; +} + +message GetDeviceProfileResponseMsg { + bytes data = 1; +} + +message DeviceProfileUpdateMsg { + bytes data = 1; +} + message SessionCloseNotificationProto { string message = 1; } @@ -399,6 +418,7 @@ message FromDeviceRPCResponseProto { string response = 3; int32 error = 4; } + /** * Main messages; */ @@ -409,6 +429,7 @@ message TransportApiRequestMsg { ValidateDeviceX509CertRequestMsg validateX509CertRequestMsg = 2; GetOrCreateDeviceFromGatewayRequestMsg getOrCreateDeviceRequestMsg = 3; GetTenantRoutingInfoRequestMsg getTenantRoutingInfoRequestMsg = 4; + GetDeviceProfileRequestMsg getDeviceProfileRequestMsg = 5; } /* Response from ThingsBoard Core Service to Transport Service */ @@ -416,6 +437,7 @@ message TransportApiResponseMsg { ValidateDeviceCredentialsResponseMsg validateTokenResponseMsg = 1; GetOrCreateDeviceFromGatewayResponseMsg getOrCreateDeviceResponseMsg = 2; GetTenantRoutingInfoResponseMsg getTenantRoutingInfoResponseMsg = 4; + GetDeviceProfileResponseMsg getDeviceProfileResponseMsg = 5; } /* Messages that are handled by ThingsBoard Core Service */ @@ -456,4 +478,5 @@ message ToTransportMsg { AttributeUpdateNotificationMsg attributeUpdateNotification = 5; ToDeviceRpcRequestMsg toDeviceRequest = 6; ToServerRpcResponseMsg toServerResponse = 7; + DeviceProfileUpdateMsg deviceProfileUpdateMsg = 8; } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index d62ab6a7ee..8724990e66 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -520,6 +520,8 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement .setTenantIdLSB(msg.getDeviceInfo().getTenantIdLSB()) .setDeviceName(msg.getDeviceInfo().getDeviceName()) .setDeviceType(msg.getDeviceInfo().getDeviceType()) + .setDeviceProfileIdMSB(msg.getDeviceInfo().getDeviceProfileIdMSB()) + .setDeviceProfileIdLSB(msg.getDeviceInfo().getDeviceProfileIdLSB()) .build(); transportService.process(sessionInfo, DefaultTransportService.getSessionEventMsg(SessionEvent.OPEN), new TransportServiceCallback() { @Override diff --git a/common/transport/transport-api/pom.xml b/common/transport/transport-api/pom.xml index f323ad4fe5..41813bb006 100644 --- a/common/transport/transport-api/pom.xml +++ b/common/transport/transport-api/pom.xml @@ -60,6 +60,10 @@ com.google.code.gson gson + + de.ruedigermoeller + fst + org.slf4j slf4j-api diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/SessionMsgListener.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/SessionMsgListener.java index 8517bc9390..ccd63ca430 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/SessionMsgListener.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/SessionMsgListener.java @@ -15,6 +15,7 @@ */ package org.thingsboard.server.common.transport; +import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcResponseMsg; import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg; import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeResponseMsg; @@ -35,4 +36,8 @@ public interface SessionMsgListener { void onToDeviceRpcRequest(ToDeviceRpcRequestMsg toDeviceRequest); void onToServerRpcResponse(ToServerRpcResponseMsg toServerResponse); + + default void onProfileUpdate(DeviceProfile deviceProfile) { + } + } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java index 2af55c9206..da9a3d7167 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java @@ -15,6 +15,8 @@ */ package org.thingsboard.server.common.transport; +import org.thingsboard.server.common.data.DeviceProfile; +import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg; import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg; @@ -50,6 +52,10 @@ public interface TransportService { void process(GetOrCreateDeviceFromGatewayRequestMsg msg, TransportServiceCallback callback); + void getDeviceProfile(DeviceProfileId deviceProfileId, TransportServiceCallback callback); + + void onProfileUpdate(DeviceProfile deviceProfile); + boolean checkLimits(SessionInfoProto sessionInfo, Object msg, TransportServiceCallback callback); void process(SessionInfoProto sessionInfo, SessionEventMsg msg, TransportServiceCallback callback); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index 438c3b9ff6..6f0d41fd5e 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -15,19 +15,26 @@ */ package org.thingsboard.server.common.transport.service; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; import com.google.gson.Gson; import com.google.gson.JsonObject; +import com.google.protobuf.ByteString; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.DeviceProfileId; +import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.msg.queue.ServiceQueue; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.msg.session.SessionMsgType; @@ -36,6 +43,7 @@ import org.thingsboard.server.common.msg.tools.TbRateLimitsException; import org.thingsboard.server.common.transport.SessionMsgListener; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; +import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; import org.thingsboard.server.common.transport.util.JsonUtils; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; @@ -61,9 +69,11 @@ import org.thingsboard.server.common.stats.StatsType; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Random; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; @@ -75,6 +85,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; /** * Created by ashvayka on 17.10.18. @@ -105,6 +116,8 @@ public class DefaultTransportService implements TransportService { private final PartitionService partitionService; private final TbServiceInfoProvider serviceInfoProvider; private final StatsFactory statsFactory; + private final DataDecodingEncodingService dataDecodingEncodingService; + protected TbQueueRequestTemplate, TbProtoQueueMsg> transportApiRequestTemplate; protected TbQueueProducer> ruleEngineMsgProducer; @@ -120,19 +133,26 @@ public class DefaultTransportService implements TransportService { private final ConcurrentMap sessions = new ConcurrentHashMap<>(); private final Map toServerRpcPendingMap = new ConcurrentHashMap<>(); - //TODO: Implement cleanup of this maps. + //TODO 3.2: @ybondarenko Implement cleanup of this maps. private final ConcurrentMap perTenantLimits = new ConcurrentHashMap<>(); private final ConcurrentMap perDeviceLimits = new ConcurrentHashMap<>(); + private final ConcurrentMap deviceProfiles = new ConcurrentHashMap<>(); private ExecutorService mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("transport-consumer")); private volatile boolean stopped = false; - public DefaultTransportService(TbServiceInfoProvider serviceInfoProvider, TbTransportQueueFactory queueProvider, TbQueueProducerProvider producerProvider, PartitionService partitionService, StatsFactory statsFactory) { + public DefaultTransportService(TbServiceInfoProvider serviceInfoProvider, + TbTransportQueueFactory queueProvider, + TbQueueProducerProvider producerProvider, + PartitionService partitionService, + StatsFactory statsFactory, + DataDecodingEncodingService dataDecodingEncodingService) { this.serviceInfoProvider = serviceInfoProvider; this.queueProvider = queueProvider; this.producerProvider = producerProvider; this.partitionService = partitionService; this.statsFactory = statsFactory; + this.dataDecodingEncodingService = dataDecodingEncodingService; } @PostConstruct @@ -231,15 +251,22 @@ public class DefaultTransportService implements TransportService { public void process(TransportProtos.ValidateDeviceTokenRequestMsg msg, TransportServiceCallback callback) { log.trace("Processing msg: {}", msg); TbProtoQueueMsg protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setValidateTokenRequestMsg(msg).build()); - AsyncCallbackTemplate.withCallback(transportApiRequestTemplate.send(protoMsg), - response -> callback.onSuccess(response.getValue().getValidateTokenResponseMsg()), callback::onError, transportCallbackExecutor); + process(callback, protoMsg); } @Override public void process(TransportProtos.ValidateDeviceX509CertRequestMsg msg, TransportServiceCallback callback) { log.trace("Processing msg: {}", msg); TbProtoQueueMsg protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setValidateX509CertRequestMsg(msg).build()); - AsyncCallbackTemplate.withCallback(transportApiRequestTemplate.send(protoMsg), + process(callback, protoMsg); + } + + private void process(TransportServiceCallback callback, TbProtoQueueMsg protoMsg) { + ListenableFuture> result = extractProfile(transportApiRequestTemplate.send(protoMsg), + response -> response.getValidateTokenResponseMsg().hasDeviceInfo(), + response -> response.getValidateTokenResponseMsg().getDeviceInfo(), + response -> response.getValidateTokenResponseMsg().getProfileBody()); + AsyncCallbackTemplate.withCallback(result, response -> callback.onSuccess(response.getValue().getValidateTokenResponseMsg()), callback::onError, transportCallbackExecutor); } @@ -247,7 +274,11 @@ public class DefaultTransportService implements TransportService { public void process(TransportProtos.GetOrCreateDeviceFromGatewayRequestMsg msg, TransportServiceCallback callback) { log.trace("Processing msg: {}", msg); TbProtoQueueMsg protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setGetOrCreateDeviceRequestMsg(msg).build()); - AsyncCallbackTemplate.withCallback(transportApiRequestTemplate.send(protoMsg), + ListenableFuture> result = extractProfile(transportApiRequestTemplate.send(protoMsg), + response -> response.getGetOrCreateDeviceResponseMsg().hasDeviceInfo(), + response -> response.getGetOrCreateDeviceResponseMsg().getDeviceInfo(), + response -> response.getGetOrCreateDeviceResponseMsg().getProfileBody()); + AsyncCallbackTemplate.withCallback(result, response -> callback.onSuccess(response.getValue().getGetOrCreateDeviceResponseMsg()), callback::onError, transportCallbackExecutor); } @@ -282,7 +313,9 @@ public class DefaultTransportService implements TransportService { metaData.putValue("deviceType", sessionInfo.getDeviceType()); metaData.putValue("ts", tsKv.getTs() + ""); JsonObject json = JsonUtils.getJsonObject(tsKv.getKvList()); - TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), deviceId, metaData, gson.toJson(json)); + RuleChainId ruleChainId = resolveRuleChainId(sessionInfo); + TbMsg tbMsg = TbMsg.newMsg(ServiceQueue.MAIN, SessionMsgType.POST_TELEMETRY_REQUEST.name(), + deviceId, metaData, gson.toJson(json), ruleChainId, null); sendToRuleEngine(tenantId, tbMsg, packCallback); } } @@ -298,7 +331,9 @@ public class DefaultTransportService implements TransportService { TbMsgMetaData metaData = new TbMsgMetaData(); metaData.putValue("deviceName", sessionInfo.getDeviceName()); metaData.putValue("deviceType", sessionInfo.getDeviceType()); - TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), deviceId, metaData, gson.toJson(json)); + RuleChainId ruleChainId = resolveRuleChainId(sessionInfo); + TbMsg tbMsg = TbMsg.newMsg(ServiceQueue.MAIN, SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), + deviceId, metaData, gson.toJson(json), ruleChainId, null); sendToRuleEngine(tenantId, tbMsg, new TransportTbQueueCallback(callback)); } } @@ -380,9 +415,10 @@ public class DefaultTransportService implements TransportService { metaData.putValue("requestId", Integer.toString(msg.getRequestId())); metaData.putValue("serviceId", serviceInfoProvider.getServiceId()); metaData.putValue("sessionId", sessionId.toString()); - TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.TO_SERVER_RPC_REQUEST.name(), deviceId, metaData, TbMsgDataType.JSON, gson.toJson(json)); + RuleChainId ruleChainId = resolveRuleChainId(sessionInfo); + TbMsg tbMsg = TbMsg.newMsg(ServiceQueue.MAIN, SessionMsgType.TO_SERVER_RPC_REQUEST.name(), + deviceId, metaData, gson.toJson(json), ruleChainId, null); sendToRuleEngine(tenantId, tbMsg, new TransportTbQueueCallback(callback)); - String requestId = sessionId + "-" + msg.getRequestId(); toServerRpcPendingMap.put(requestId, new RpcRequestMetadata(sessionId, msg.getRequestId())); schedulerExecutor.schedule(() -> processTimeout(requestId), clientSideRpcTimeout, TimeUnit.MILLISECONDS); @@ -538,11 +574,62 @@ public class DefaultTransportService implements TransportService { deregisterSession(md.getSessionInfo()); } } else { - //TODO: should we notify the device actor about missed session? - log.debug("[{}] Missing session.", sessionId); + if (toSessionMsg.hasDeviceProfileUpdateMsg()) { + Optional deviceProfile = dataDecodingEncodingService.decode(toSessionMsg.getDeviceProfileUpdateMsg().getData().toByteArray()); + deviceProfile.ifPresent(this::onProfileUpdate); + } else { + //TODO: should we notify the device actor about missed session? + log.debug("[{}] Missing session.", sessionId); + } + } + } + + @Override + public void getDeviceProfile(DeviceProfileId deviceProfileId, TransportServiceCallback callback) { + DeviceProfile deviceProfile = deviceProfiles.get(deviceProfileId); + if (deviceProfile != null) { + callback.onSuccess(deviceProfile); + } else { + log.trace("Processing device profile request: [{}]", deviceProfileId); + TransportProtos.GetDeviceProfileRequestMsg msg = TransportProtos.GetDeviceProfileRequestMsg.newBuilder() + .setProfileIdMSB(deviceProfileId.getId().getMostSignificantBits()) + .setProfileIdLSB(deviceProfileId.getId().getLeastSignificantBits()) + .build(); + TbProtoQueueMsg protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), + TransportApiRequestMsg.newBuilder().setGetDeviceProfileRequestMsg(msg).build()); + AsyncCallbackTemplate.withCallback(transportApiRequestTemplate.send(protoMsg), + response -> { + byte[] devProfileBody = response.getValue().getGetDeviceProfileResponseMsg().getData().toByteArray(); + if (devProfileBody != null && devProfileBody.length > 0) { + Optional deviceProfileOpt = dataDecodingEncodingService.decode(devProfileBody); + if (deviceProfileOpt.isPresent()) { + deviceProfiles.put(deviceProfileOpt.get().getId(), deviceProfile); + callback.onSuccess(deviceProfileOpt.get()); + } else { + log.warn("Failed to decode device profile: {}", Arrays.toString(devProfileBody)); + callback.onError(new IllegalArgumentException("Failed to decode device profile!")); + } + } else { + log.warn("Failed to find device profile: [{}]", deviceProfileId); + callback.onError(new IllegalArgumentException("Failed to find device profile!")); + } + }, callback::onError, transportCallbackExecutor); } } + @Override + public void onProfileUpdate(DeviceProfile deviceProfile) { + deviceProfiles.put(deviceProfile.getId(), deviceProfile); + long deviceProfileIdMSB = deviceProfile.getId().getId().getMostSignificantBits(); + long deviceProfileIdLSB = deviceProfile.getId().getId().getLeastSignificantBits(); + sessions.forEach((id, md) -> { + if (md.getSessionInfo().getDeviceProfileIdMSB() == deviceProfileIdMSB + && md.getSessionInfo().getDeviceProfileIdLSB() == deviceProfileIdLSB) { + transportCallbackExecutor.submit(() -> md.getListener().onProfileUpdate(deviceProfile)); + } + }); + } + protected UUID toSessionId(TransportProtos.SessionInfoProto sessionInfo) { return new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB()); } @@ -593,6 +680,40 @@ public class DefaultTransportService implements TransportService { ruleEngineMsgProducer.send(tpi, new TbProtoQueueMsg<>(tbMsg.getId(), msg), wrappedCallback); } + private RuleChainId resolveRuleChainId(TransportProtos.SessionInfoProto sessionInfo) { + DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(sessionInfo.getDeviceProfileIdMSB(), sessionInfo.getDeviceProfileIdLSB())); + DeviceProfile deviceProfile = deviceProfiles.get(deviceProfileId); + RuleChainId ruleChainId; + if (deviceProfile == null) { + log.warn("[{}] Device profile is null!", deviceProfileId); + ruleChainId = null; + } else { + ruleChainId = deviceProfile.getDefaultRuleChainId(); + } + return ruleChainId; + } + + private ListenableFuture> extractProfile(ListenableFuture> send, + Function hasDeviceInfo, + Function deviceInfoF, + Function profileBodyF) { + return Futures.transform(send, response -> { + T value = response.getValue(); + if (hasDeviceInfo.apply(value)) { + TransportProtos.DeviceInfoProto deviceInfo = deviceInfoF.apply(value); + ByteString profileBody = profileBodyF.apply(value); + if (profileBody != null && !profileBody.isEmpty()) { + DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(deviceInfo.getDeviceProfileIdMSB(), deviceInfo.getDeviceProfileIdLSB())); + if (!deviceProfiles.containsKey(deviceProfileId)) { + Optional deviceProfile = dataDecodingEncodingService.decode(profileBody.toByteArray()); + deviceProfile.ifPresent(profile -> deviceProfiles.put(deviceProfileId, profile)); + } + } + } + return response; + }, transportCallbackExecutor); + } + private class TransportTbQueueCallback implements TbQueueCallback { private final TransportServiceCallback callback; diff --git a/application/src/main/java/org/thingsboard/server/service/encoding/DataDecodingEncodingService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/util/DataDecodingEncodingService.java similarity index 84% rename from application/src/main/java/org/thingsboard/server/service/encoding/DataDecodingEncodingService.java rename to common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/util/DataDecodingEncodingService.java index 4a781b8673..1b10cb5dc3 100644 --- a/application/src/main/java/org/thingsboard/server/service/encoding/DataDecodingEncodingService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/util/DataDecodingEncodingService.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.service.encoding; +package org.thingsboard.server.common.transport.util; import org.thingsboard.server.common.msg.TbActorMsg; @@ -21,9 +21,9 @@ import java.util.Optional; public interface DataDecodingEncodingService { - Optional decode(byte[] byteArray); + Optional decode(byte[] byteArray); - byte[] encode(TbActorMsg msq); + byte[] encode(T msq); } diff --git a/application/src/main/java/org/thingsboard/server/service/encoding/ProtoWithFSTService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/util/ProtoWithFSTService.java similarity index 82% rename from application/src/main/java/org/thingsboard/server/service/encoding/ProtoWithFSTService.java rename to common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/util/ProtoWithFSTService.java index 8d89059488..eee9dacfe4 100644 --- a/application/src/main/java/org/thingsboard/server/service/encoding/ProtoWithFSTService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/util/ProtoWithFSTService.java @@ -13,12 +13,13 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.service.encoding; +package org.thingsboard.server.common.transport.util; import lombok.extern.slf4j.Slf4j; import org.nustaq.serialization.FSTConfiguration; import org.springframework.stereotype.Service; import org.thingsboard.server.common.msg.TbActorMsg; +import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; import java.util.Optional; @@ -29,11 +30,10 @@ public class ProtoWithFSTService implements DataDecodingEncodingService { private final FSTConfiguration config = FSTConfiguration.createDefaultConfiguration(); @Override - public Optional decode(byte[] byteArray) { + public Optional decode(byte[] byteArray) { try { - TbActorMsg msg = (TbActorMsg) config.asObject(byteArray); + T msg = (T) config.asObject(byteArray); return Optional.of(msg); - } catch (IllegalArgumentException e) { log.error("Error during deserialization message, [{}]", e.getMessage()); return Optional.empty(); @@ -41,7 +41,7 @@ public class ProtoWithFSTService implements DataDecodingEncodingService { } @Override - public byte[] encode(TbActorMsg msq) { + public byte[] encode(T msq) { return config.asByteArray(msq); } diff --git a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java index 10f308984d..df33337a11 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java @@ -26,6 +26,7 @@ import org.springframework.stereotype.Service; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.DeviceProfileInfo; import org.thingsboard.server.common.data.DeviceProfileType; +import org.thingsboard.server.common.data.DeviceTransportType; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.device.profile.DefaultDeviceProfileConfiguration; import org.thingsboard.server.common.data.device.profile.DeviceProfileData; @@ -166,6 +167,7 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D deviceProfile.setDefault(true); deviceProfile.setName("Default"); deviceProfile.setType(DeviceProfileType.DEFAULT); + deviceProfile.setTransportType(DeviceTransportType.DEFAULT); deviceProfile.setDescription("Default device profile"); DeviceProfileData deviceProfileData = new DeviceProfileData(); DefaultDeviceProfileConfiguration configuration = new DefaultDeviceProfileConfiguration(); diff --git a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java index b6602d1702..21537de5ac 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java @@ -41,8 +41,10 @@ import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.device.DeviceSearchQuery; import org.thingsboard.server.common.data.device.data.DefaultDeviceConfiguration; +import org.thingsboard.server.common.data.device.data.DefaultDeviceTransportConfiguration; import org.thingsboard.server.common.data.device.data.DeviceData; -import org.thingsboard.server.common.data.device.data.Lwm2mDeviceConfiguration; +import org.thingsboard.server.common.data.device.data.Lwm2mDeviceTransportConfiguration; +import org.thingsboard.server.common.data.device.data.MqttDeviceTransportConfiguration; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; @@ -175,8 +177,14 @@ public class DeviceServiceImpl extends AbstractEntityService implements DeviceSe case DEFAULT: deviceData.setConfiguration(new DefaultDeviceConfiguration()); break; + } + switch (deviceProfile.getTransportType()){ + case DEFAULT: + deviceData.setTransportConfiguration(new DefaultDeviceTransportConfiguration()); + case MQTT: + deviceData.setTransportConfiguration(new MqttDeviceTransportConfiguration()); case LWM2M: - deviceData.setConfiguration(new Lwm2mDeviceConfiguration()); + deviceData.setTransportConfiguration(new Lwm2mDeviceTransportConfiguration()); break; } device.setDeviceData(deviceData); diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java b/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java index 4b58e81e03..b62dac2d44 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java @@ -168,6 +168,7 @@ public class ModelConstants { public static final String DEVICE_PROFILE_TENANT_ID_PROPERTY = TENANT_ID_PROPERTY; public static final String DEVICE_PROFILE_NAME_PROPERTY = "name"; public static final String DEVICE_PROFILE_TYPE_PROPERTY = "type"; + public static final String DEVICE_PROFILE_TRANSPORT_TYPE_PROPERTY = "transport_type"; public static final String DEVICE_PROFILE_PROFILE_DATA_PROPERTY = "profile_data"; public static final String DEVICE_PROFILE_DESCRIPTION_PROPERTY = "description"; public static final String DEVICE_PROFILE_IS_DEFAULT_PROPERTY = "is_default"; diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/DeviceProfileEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/DeviceProfileEntity.java index 7381ddf638..fa481e1f80 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/sql/DeviceProfileEntity.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/DeviceProfileEntity.java @@ -23,6 +23,7 @@ import org.hibernate.annotations.Type; import org.hibernate.annotations.TypeDef; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.DeviceProfileType; +import org.thingsboard.server.common.data.DeviceTransportType; import org.thingsboard.server.common.data.device.profile.DeviceProfileData; import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.data.id.RuleChainId; @@ -57,6 +58,10 @@ public final class DeviceProfileEntity extends BaseSqlEntity impl @Column(name = ModelConstants.DEVICE_PROFILE_TYPE_PROPERTY) private DeviceProfileType type; + @Enumerated(EnumType.STRING) + @Column(name = ModelConstants.DEVICE_PROFILE_TRANSPORT_TYPE_PROPERTY) + private DeviceTransportType transportType; + @Column(name = ModelConstants.DEVICE_PROFILE_DESCRIPTION_PROPERTY) private String description; @@ -87,6 +92,7 @@ public final class DeviceProfileEntity extends BaseSqlEntity impl this.setCreatedTime(deviceProfile.getCreatedTime()); this.name = deviceProfile.getName(); this.type = deviceProfile.getType(); + this.transportType = deviceProfile.getTransportType(); this.description = deviceProfile.getDescription(); this.isDefault = deviceProfile.isDefault(); this.profileData = JacksonUtil.convertValue(deviceProfile.getProfileData(), ObjectNode.class); @@ -118,6 +124,7 @@ public final class DeviceProfileEntity extends BaseSqlEntity impl } deviceProfile.setName(name); deviceProfile.setType(type); + deviceProfile.setTransportType(transportType); deviceProfile.setDescription(description); deviceProfile.setDefault(isDefault); deviceProfile.setProfileData(JacksonUtil.convertValue(profileData, DeviceProfileData.class)); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/device/DeviceProfileRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/device/DeviceProfileRepository.java index 162a915d88..6c11c328d8 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/device/DeviceProfileRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/device/DeviceProfileRepository.java @@ -27,7 +27,7 @@ import java.util.UUID; public interface DeviceProfileRepository extends PagingAndSortingRepository { - @Query("SELECT new org.thingsboard.server.common.data.DeviceProfileInfo(d.id, d.name, d.type) " + + @Query("SELECT new org.thingsboard.server.common.data.DeviceProfileInfo(d.id, d.name, d.type, d.transportType) " + "FROM DeviceProfileEntity d " + "WHERE d.id = :deviceProfileId") DeviceProfileInfo findDeviceProfileInfoById(@Param("deviceProfileId") UUID deviceProfileId); @@ -38,7 +38,7 @@ public interface DeviceProfileRepository extends PagingAndSortingRepository findDeviceProfileInfos(@Param("tenantId") UUID tenantId, @@ -49,7 +49,7 @@ public interface DeviceProfileRepository extends PagingAndSortingRepository deviceProfileInfos = deviceProfiles.stream().map(deviceProfile -> new DeviceProfileInfo(deviceProfile.getId(), - deviceProfile.getName(), deviceProfile.getType())).collect(Collectors.toList()); + List deviceProfileInfos = deviceProfiles.stream() + .map(deviceProfile -> new DeviceProfileInfo(deviceProfile.getId(), + deviceProfile.getName(), deviceProfile.getType(), deviceProfile.getTransportType())).collect(Collectors.toList()); Assert.assertEquals(deviceProfileInfos, loadedDeviceProfileInfos);