Init commit for notifications to gateway

application/src/main/java/org/thingsboard/server/controller/DeviceController.java

@@ -74,6 +74,7 @@ import org.thingsboard.server.dao.exception.IncorrectParameterException;
 import org.thingsboard.server.dao.model.ModelConstants;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.device.DeviceBulkImportService;
+import org.thingsboard.server.service.gateway_device.GatewayDeviceStateService;
 import org.thingsboard.server.service.importing.BulkImportRequest;
 import org.thingsboard.server.service.importing.BulkImportResult;
 import org.thingsboard.server.service.security.model.SecurityUser;
@@ -128,6 +129,8 @@ public class DeviceController extends BaseController {
 
     private final DeviceBulkImportService deviceBulkImportService;
 
+    private final GatewayDeviceStateService gatewayDeviceStatusService;
+
     @ApiOperation(value = "Get Device (getDeviceById)",
             notes = "Fetch the Device object based on the provided Device Id. " +
                     "If the user has the authority of 'TENANT_ADMIN', the server checks that the device is owned by the same tenant. " +
@@ -261,6 +264,7 @@ public class DeviceController extends BaseController {
 
             List<EdgeId> relatedEdgeIds = findRelatedEdgeIds(getTenantId(), deviceId);
 
+            gatewayDeviceStatusService.delete(device);
             deviceService.deleteDevice(getCurrentUser().getTenantId(), deviceId);
 
             tbClusterService.onDeviceDeleted(device, null);

application/src/main/java/org/thingsboard/server/service/gateway_device/DefaultGatewayDeviceStateService.java

@@ -0,0 +1,137 @@
+package org.thingsboard.server.service.gateway_device;
+
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.ListenableFuture;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.DonAsynchron;
+import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.server.common.data.DataConstants;
+import org.thingsboard.server.common.data.Device;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
+import org.thingsboard.server.common.data.kv.JsonDataEntry;
+import org.thingsboard.server.common.data.kv.KvEntry;
+import org.thingsboard.server.common.data.relation.EntityRelation;
+import org.thingsboard.server.common.data.relation.RelationTypeGroup;
+import org.thingsboard.server.dao.attributes.AttributesService;
+import org.thingsboard.server.dao.relation.RelationService;
+import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class DefaultGatewayDeviceStateService implements GatewayDeviceStateService {
+
+    private static final String RENAMED_GATEWAY_DEVICES = "renamedGatewayDevices";
+
+    @Lazy
+    @Autowired
+    private TelemetrySubscriptionService tsSubService;
+    private final AttributesService attributesService;
+    private final RelationService relationService;
+
+    @Override
+    public void update(Device device, Device oldDevice) {
+        List<EntityRelation> relationToGatewayList = relationService.findByFromAndType(TenantId.SYS_TENANT_ID, device.getId(), DataConstants.LAST_CONNECTED_GATEWAY, RelationTypeGroup.COMMON);
+        if (!relationToGatewayList.isEmpty()) {
+            EntityRelation relationToGateway = relationToGatewayList.get(0);
+            ListenableFuture<List<AttributeKvEntry>> renamedGatewayDevicesFuture = attributesService.find(device.getTenantId(), relationToGateway.getTo(), DataConstants.SHARED_SCOPE, Collections.singletonList("renamedGatewayDevices"));
+            DonAsynchron.withCallback(renamedGatewayDevicesFuture, renamedGatewayDevicesList -> {
+                        ObjectNode renamedGatewayDevicesNode;
+                        KvEntry renamedGatewayDevicesKvEntry;
+                        String newDeviceName = device.getName();
+                        String oldDeviceName = oldDevice.getName();
+
+                        if (renamedGatewayDevicesList.isEmpty()) {
+                            renamedGatewayDevicesNode = JacksonUtil.newObjectNode();
+                            renamedGatewayDevicesNode.put(oldDeviceName, newDeviceName);
+                        } else {
+                            AttributeKvEntry receivedRenamedGatewayDevicesAttribute = renamedGatewayDevicesList.get(0);
+                            renamedGatewayDevicesNode = (ObjectNode) JacksonUtil.toJsonNode(receivedRenamedGatewayDevicesAttribute.getValueAsString());
+                            if (renamedGatewayDevicesNode.findValue(newDeviceName) != null) {
+                                // If new device name is the same like the first name
+                                renamedGatewayDevicesNode.remove(newDeviceName);
+                            } else if (renamedGatewayDevicesNode.findValue(oldDeviceName) == null) {
+
+                                AtomicBoolean renamedFirstTime = new AtomicBoolean(true);
+
+                                renamedGatewayDevicesNode.fields().forEachRemaining(entry -> {
+                                    // If device was renamed earlier
+                                    if (oldDeviceName.equals(entry.getValue().asText())) {
+                                        renamedGatewayDevicesNode.put(entry.getKey(), newDeviceName);
+                                        renamedFirstTime.set(false);
+                                    }
+                                });
+                                if (renamedFirstTime.get()) {
+                                    renamedGatewayDevicesNode.put(oldDeviceName, newDeviceName);
+                                }
+                            }
+                        }
+
+                        renamedGatewayDevicesKvEntry = new JsonDataEntry(RENAMED_GATEWAY_DEVICES, JacksonUtil.toString(renamedGatewayDevicesNode));
+                        saveGatewayDevicesAttribute(device, relationToGateway, renamedGatewayDevicesKvEntry);
+                    },
+                    e -> log.error("Cannot get gateway renamed devices attribute", e));
+        }
+    }
+
+    @Override
+    public void delete(Device device) {
+        List<EntityRelation> relationToGatewayList = relationService.findByFromAndType(TenantId.SYS_TENANT_ID, device.getId(), DataConstants.LAST_CONNECTED_GATEWAY, RelationTypeGroup.COMMON);
+        if (!relationToGatewayList.isEmpty()) {
+            EntityRelation relationToGateway = relationToGatewayList.get(0);
+            String deletedDeviceName = device.getName();
+            ListenableFuture<List<AttributeKvEntry>> renamedGatewayDevicesFuture = attributesService.find(device.getTenantId(), relationToGateway.getTo(), DataConstants.SHARED_SCOPE, Collections.singletonList("renamedGatewayDevices"));
+            DonAsynchron.withCallback(renamedGatewayDevicesFuture, renamedGatewayDevicesList -> {
+                if (!renamedGatewayDevicesList.isEmpty()) {
+                    ObjectNode renamedGatewayDevicesNode = JacksonUtil.fromString(renamedGatewayDevicesList.get(0).getValueAsString(), ObjectNode.class);
+                    if (renamedGatewayDevicesNode != null && renamedGatewayDevicesNode.findValue(deletedDeviceName) != null) {
+                        renamedGatewayDevicesNode.remove(deletedDeviceName);
+                        KvEntry renamedGatewayDevicesKvEntry = new JsonDataEntry(RENAMED_GATEWAY_DEVICES, JacksonUtil.toString(renamedGatewayDevicesNode));
+                        saveGatewayDevicesAttribute(device, relationToGateway, renamedGatewayDevicesKvEntry);
+                    }
+                }
+            }, e -> log.error("Cannot get gateway renamed devices attribute", e));
+            ListenableFuture<List<AttributeKvEntry>> deletedGatewayDevicesFuture = attributesService.find(device.getTenantId(), relationToGateway.getTo(), DataConstants.SHARED_SCOPE, Collections.singletonList("deletedGatewayDevices"));
+            DonAsynchron.withCallback(deletedGatewayDevicesFuture, deletedGatewayDevicesList -> {
+                ArrayNode deletedGatewayDevicesNode;
+                if (!deletedGatewayDevicesList.isEmpty()) {
+                    deletedGatewayDevicesNode = (ArrayNode) JacksonUtil.toJsonNode(deletedGatewayDevicesList.get(0).getValueAsString());
+                } else {
+                    deletedGatewayDevicesNode = JacksonUtil.OBJECT_MAPPER.createArrayNode();
+                }
+                deletedGatewayDevicesNode.add(deletedDeviceName);
+                KvEntry renamedGatewayDevicesKvEntry = new JsonDataEntry(RENAMED_GATEWAY_DEVICES, JacksonUtil.toString(deletedGatewayDevicesNode));
+                saveGatewayDevicesAttribute(device, relationToGateway, renamedGatewayDevicesKvEntry);
+            }, e -> log.error("Cannot get gateway deleted devices attribute", e));
+        }
+    }
+
+    private void saveGatewayDevicesAttribute(Device device, EntityRelation relationToGateway, KvEntry renamedGatewayDevicesKvEntry) {
+        AttributeKvEntry attrKvEntry = new BaseAttributeKvEntry(System.currentTimeMillis(), renamedGatewayDevicesKvEntry);
+        tsSubService.saveAndNotify(device.getTenantId(), relationToGateway.getTo(), DataConstants.SHARED_SCOPE, List.of(attrKvEntry), true, new FutureCallback<Void>() {
+            @Override
+            public void onSuccess(@Nullable Void unused) {
+                log.trace("Attribute saved for gateway with ID [{}] and data [{}]", relationToGateway.getTo(), renamedGatewayDevicesKvEntry.getJsonValue());
+            }
+
+            @Override
+            public void onFailure(Throwable throwable) {
+                log.error("Cannot save gateway device attribute", throwable);
+            }
+        });
+    }
+}

application/src/main/java/org/thingsboard/server/service/gateway_device/GatewayDeviceStateService.java

@@ -0,0 +1,10 @@
+package org.thingsboard.server.service.gateway_device;
+
+import org.thingsboard.server.common.data.Device;
+
+public interface GatewayDeviceStateService {
+
+    void update(Device device, Device oldDevice);
+
+    void delete(Device device);
+}

application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java

@@ -61,6 +61,7 @@ import org.thingsboard.server.queue.common.MultipleTbQueueCallbackWrapper;
 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.gateway_device.GatewayDeviceStateService;
 import org.thingsboard.server.service.ota.OtaPackageStateService;
 import org.thingsboard.server.service.profile.TbDeviceProfileCache;
 import org.thingsboard.server.common.msg.rpc.FromDeviceRpcResponse;
@@ -91,6 +92,7 @@ public class DefaultTbClusterService implements TbClusterService {
     private final DataDecodingEncodingService encodingService;
     private final TbDeviceProfileCache deviceProfileCache;
     private final OtaPackageStateService otaPackageStateService;
+    private final GatewayDeviceStateService gatewayDeviceStateService;
 
     @Override
     public void pushMsgToCore(TenantId tenantId, EntityId entityId, ToCoreMsg msg, TbQueueCallback callback) {
@@ -405,6 +407,7 @@ public class DefaultTbClusterService implements TbClusterService {
         broadcastEntityStateChangeEvent(device.getTenantId(), device.getId(), created ? ComponentLifecycleEvent.CREATED : ComponentLifecycleEvent.UPDATED);
         sendDeviceStateServiceEvent(device.getTenantId(), device.getId(), created, !created, false);
         otaPackageStateService.update(device, old);
+        gatewayDeviceStateService.update(device, old);
         if (!created && notifyEdge) {
             sendNotificationMsgToEdgeService(device.getTenantId(), null, device.getId(), null, null, EdgeEventActionType.UPDATED);
         }

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

@@ -56,6 +56,7 @@ import org.thingsboard.server.common.data.ota.OtaPackageUtil;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.common.data.relation.EntityRelation;
+import org.thingsboard.server.common.data.relation.RelationTypeGroup;
 import org.thingsboard.server.common.data.security.DeviceCredentials;
 import org.thingsboard.server.common.data.security.DeviceCredentialsType;
 import org.thingsboard.server.common.msg.EncryptionUtil;
@@ -99,6 +100,7 @@ import org.thingsboard.server.cluster.TbClusterService;
 import org.thingsboard.server.service.resource.TbResourceService;
 import org.thingsboard.server.service.state.DeviceStateService;
 
+import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
@@ -304,6 +306,17 @@ public class DefaultTransportApiService implements TransportApiService {
                     TbMsg tbMsg = TbMsg.newMsg(DataConstants.ENTITY_CREATED, deviceId, customerId, metaData, TbMsgDataType.JSON, mapper.writeValueAsString(entityNode));
                     tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, null);
                 }
+
+                List<EntityRelation> currentLastConnectedGatewayRelationList = relationService.findByFromAndType(TenantId.SYS_TENANT_ID, device.getId(), DataConstants.LAST_CONNECTED_GATEWAY, RelationTypeGroup.COMMON);
+                EntityRelation lastConnectedGatewayRelation;
+                if (!currentLastConnectedGatewayRelationList.isEmpty()) {
+                    lastConnectedGatewayRelation = currentLastConnectedGatewayRelationList.get(0);
+                    lastConnectedGatewayRelation.setTo(gateway.getId());
+                } else {
+                    lastConnectedGatewayRelation = new EntityRelation(device.getId(), gateway.getId(), DataConstants.LAST_CONNECTED_GATEWAY);
+                }
+                relationService.saveRelationAsync(TenantId.SYS_TENANT_ID, lastConnectedGatewayRelation);
+
                 GetOrCreateDeviceFromGatewayResponseMsg.Builder builder = GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
                         .setDeviceInfo(getDeviceInfoProto(device));
                 DeviceProfile deviceProfile = deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId());

common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java

@@ -116,4 +116,5 @@ public class DataConstants {
     public static final String EDGE_MSG_SOURCE = "edge";
     public static final String MSG_SOURCE_KEY = "source";
 
+    public static final String LAST_CONNECTED_GATEWAY = "lastConnectedGateway";
 }