Merge pull request #5968 from volodymyr-babak/edge-rpc-fix

[3.3.3] Fix RPC call from cloud to edge

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

@@ -38,7 +38,6 @@ import org.springframework.web.bind.annotation.ResponseStatus;
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.context.request.async.DeferredResult;
 import org.thingsboard.rule.engine.api.msg.DeviceCredentialsUpdateNotificationMsg;
-import org.thingsboard.rule.engine.api.msg.DeviceEdgeUpdateMsg;
 import org.thingsboard.server.common.data.ClaimRequest;
 import org.thingsboard.server.common.data.Customer;
 import org.thingsboard.server.common.data.DataConstants;
@@ -875,9 +874,6 @@ public class DeviceController extends BaseController {
 
             Device savedDevice = checkNotNull(deviceService.assignDeviceToEdge(getCurrentUser().getTenantId(), deviceId, edgeId));
 
-            tbClusterService.pushMsgToCore(new DeviceEdgeUpdateMsg(savedDevice.getTenantId(),
-                    savedDevice.getId(), edgeId), null);
-
             logEntityAction(deviceId, savedDevice,
                     savedDevice.getCustomerId(),
                     ActionType.ASSIGNED_TO_EDGE, null, strDeviceId, strEdgeId, edge.getName());
@@ -918,9 +914,6 @@ public class DeviceController extends BaseController {
 
             Device savedDevice = checkNotNull(deviceService.unassignDeviceFromEdge(getCurrentUser().getTenantId(), deviceId, edgeId));
 
-            tbClusterService.pushMsgToCore(new DeviceEdgeUpdateMsg(savedDevice.getTenantId(),
-                    savedDevice.getId(), null), null);
-
             logEntityAction(deviceId, device,
                     device.getCustomerId(),
                     ActionType.UNASSIGNED_FROM_EDGE, null, strDeviceId, strEdgeId, edge.getName());

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

@@ -21,6 +21,7 @@ import lombok.extern.slf4j.Slf4j;
 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.DeviceEdgeUpdateMsg;
 import org.thingsboard.rule.engine.api.msg.DeviceNameOrTypeUpdateMsg;
 import org.thingsboard.server.cluster.TbClusterService;
 import org.thingsboard.server.common.data.EdgeUtils;
@@ -455,6 +456,21 @@ public class DefaultTbClusterService implements TbClusterService {
         TransportProtos.EdgeNotificationMsgProto msg = builder.build();
         log.trace("[{}] sending notification to edge service {}", tenantId.getId(), msg);
         pushMsgToCore(tenantId, entityId != null ? entityId : tenantId, TransportProtos.ToCoreMsg.newBuilder().setEdgeNotificationMsg(msg).build(), null);
+
+        if (entityId != null && EntityType.DEVICE.equals(entityId.getEntityType())) {
+            pushDeviceUpdateMessage(tenantId, edgeId, entityId, action);
+        }
     }
 
+    private void pushDeviceUpdateMessage(TenantId tenantId, EdgeId edgeId, EntityId entityId, EdgeEventActionType action) {
+        log.trace("{} Going to send edge update notification for device actor, device id {}, edge id {}", tenantId, entityId, edgeId);
+        switch (action) {
+            case ASSIGNED_TO_EDGE:
+                pushMsgToCore(new DeviceEdgeUpdateMsg(tenantId, new DeviceId(entityId.getId()), edgeId), null);
+                break;
+            case UNASSIGNED_FROM_EDGE:
+                pushMsgToCore(new DeviceEdgeUpdateMsg(tenantId, new DeviceId(entityId.getId()), null), null);
+                break;
+        }
+    }
 }

common/data/src/main/java/org/thingsboard/server/common/data/id/EdgeId.java

@@ -19,6 +19,8 @@ import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.swagger.annotations.ApiModelProperty;
+import org.springframework.util.ConcurrentReferenceHashMap;
+import org.springframework.util.ConcurrentReferenceHashMap.ReferenceType;
 import org.thingsboard.server.common.data.EntityType;
 
 import java.util.UUID;
@@ -27,6 +29,9 @@ public class EdgeId extends UUIDBased implements EntityId {
 
     private static final long serialVersionUID = 1L;
 
+    @JsonIgnore
+    static final ConcurrentReferenceHashMap<UUID, EdgeId> edges = new ConcurrentReferenceHashMap<>(16, ReferenceType.SOFT);
+
     @JsonCreator
     public EdgeId(@JsonProperty("id") UUID id) {
         super(id);
@@ -41,4 +46,9 @@ public class EdgeId extends UUIDBased implements EntityId {
     public EntityType getEntityType() {
         return EntityType.EDGE;
     }
+
+    @JsonCreator
+    public static EdgeId fromUUID(@JsonProperty("id") UUID id) {
+        return edges.computeIfAbsent(id, EdgeId::new);
+    }
 }