Fixed alarm delete functionality on edge

Volodymyr Babak
1 parent 0ce509e8
Showing 6 changed files with 90 additions and 45 deletions
application/src/main/java/org/thingsboard/server/controller/AlarmController.java
application/src/main/java/org/thingsboard/server/controller/BaseController.java
application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java
rest-client/src/main/java/org/thingsboard/rest/client/RestClient.java
--- a/application/src/main/java/org/thingsboard/server/controller/AlarmController.java
View file @1297fef
+++ b/application/src/main/java/org/thingsboard/server/controller/AlarmController.java
View file @1297fef
@@ -38,6 +38,7 @@ import org.thingsboard.server.common.data.edge.EdgeEventActionType;
 import org.thingsboard.server.common.data.exception.ThingsboardErrorCode;
 import org.thingsboard.server.common.data.exception.ThingsboardException;
 import org.thingsboard.server.common.data.id.AlarmId;
+import org.thingsboard.server.common.data.id.EdgeId;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.EntityIdFactory;
 import org.thingsboard.server.common.data.page.PageData;
@@ -46,6 +47,8 @@ import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.security.permission.Operation;
 import org.thingsboard.server.service.security.permission.Resource;
+import java.util.List;
+
 @RestController
 @TbCoreComponent
 @RequestMapping("/api")
@@ -112,10 +115,13 @@ public class AlarmController extends BaseController {
             AlarmId alarmId = new AlarmId(toUUID(strAlarmId));
             Alarm alarm = checkAlarmId(alarmId, Operation.WRITE);
+            List<EdgeId> relatedEdgeIds = findRelatedEdgeIds(getTenantId(), alarm.getOriginator());
+
             logEntityAction(alarm.getOriginator(), alarm,
                     getCurrentUser().getCustomerId(),
                     ActionType.ALARM_DELETE, null);
-            sendEntityNotificationMsg(getTenantId(), alarmId, EdgeEventActionType.DELETED);
+
+            sendAlarmDeleteNotificationMsg(getTenantId(), alarmId, relatedEdgeIds, alarm);
             return alarmService.deleteAlarm(getTenantId(), alarmId);
          } catch (Exception e) {
--- a/application/src/main/java/org/thingsboard/server/controller/BaseController.java
View file @1297fef
+++ b/application/src/main/java/org/thingsboard/server/controller/BaseController.java
View file @1297fef
@@ -852,13 +852,25 @@ public abstract class BaseController {
     }
     protected void sendDeleteNotificationMsg(TenantId tenantId, EntityId entityId, List<EdgeId> edgeIds) {
+        sendDeleteNotificationMsg(tenantId, entityId, edgeIds, null);
+    }
+
+    protected void sendDeleteNotificationMsg(TenantId tenantId, EntityId entityId, List<EdgeId> edgeIds, String body) {
         if (edgeIds != null && !edgeIds.isEmpty()) {
             for (EdgeId edgeId : edgeIds) {
-                sendNotificationMsgToEdgeService(tenantId, edgeId, entityId, null, null, EdgeEventActionType.DELETED);
+                sendNotificationMsgToEdgeService(tenantId, edgeId, entityId, body, null, EdgeEventActionType.DELETED);
             }
         }
     }
+    protected void sendAlarmDeleteNotificationMsg(TenantId tenantId, EntityId entityId, List<EdgeId> edgeIds, Alarm alarm) {
+        try {
+            sendDeleteNotificationMsg(tenantId, entityId, edgeIds, json.writeValueAsString(alarm));
+        } catch (Exception e) {
+            log.warn("Failed to push delete alarm msg to core: {}", alarm, e);
+        }
+    }
+
     protected void sendEntityAssignToCustomerNotificationMsg(TenantId tenantId, EntityId entityId, CustomerId customerId, EdgeEventActionType action) {
         try {
             sendNotificationMsgToEdgeService(tenantId, null, entityId, json.writeValueAsString(customerId), null, action);
--- a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
View file @1297fef
+++ b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
View file @1297fef
@@ -121,6 +121,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
     @Override
     public void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) {
+        log.trace("Pushing notification to edge {}", edgeNotificationMsg);
         try {
             TenantId tenantId = new TenantId(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB()));
             EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType());
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View file @1297fef
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View file @1297fef
@@ -527,7 +527,7 @@ public final class EdgeGrpcSession implements Closeable {
             case RULE_CHAIN_METADATA:
                 return ctx.getRuleChainProcessor().processRuleChainMetadataToEdge(edgeEvent, msgType);
             case ALARM:
-                return ctx.getAlarmProcessor().processAlarmToEdge(edge, edgeEvent, msgType);
+                return ctx.getAlarmProcessor().processAlarmToEdge(edge, edgeEvent, msgType, action);
             case USER:
                 return ctx.getUserProcessor().processUserToEdge(edge, edgeEvent, msgType, action);
             case RELATION:
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java
View file @1297fef
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/AlarmEdgeProcessor.java
View file @1297fef
@@ -15,6 +15,7 @@
  */
 package org.thingsboard.server.service.edge.rpc.processor;
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
@@ -114,59 +115,84 @@ public class AlarmEdgeProcessor extends BaseEdgeProcessor {
         }
     }
-    public DownlinkMsg processAlarmToEdge(Edge edge, EdgeEvent edgeEvent, UpdateMsgType msgType) {
+    public DownlinkMsg processAlarmToEdge(Edge edge, EdgeEvent edgeEvent, UpdateMsgType msgType, EdgeEventActionType action) {
+        AlarmId alarmId = new AlarmId(edgeEvent.getEntityId());
         DownlinkMsg downlinkMsg = null;
-        try {
-            AlarmId alarmId = new AlarmId(edgeEvent.getEntityId());
-            Alarm alarm = alarmService.findAlarmByIdAsync(edgeEvent.getTenantId(), alarmId).get();
-            if (alarm != null) {
+        switch (action) {
+            case ADDED:
+            case UPDATED:
+            case ALARM_ACK:
+            case ALARM_CLEAR:
+                try {
+                    Alarm alarm = alarmService.findAlarmByIdAsync(edgeEvent.getTenantId(), alarmId).get();
+                    if (alarm != null) {
+                        downlinkMsg = DownlinkMsg.newBuilder()
+                                .setDownlinkMsgId(EdgeUtils.nextPositiveInt())
+                                .addAlarmUpdateMsg(alarmMsgConstructor.constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm))
+                                .build();
+                    }
+                } catch (Exception e) {
+                    log.error("Can't process alarm msg [{}] [{}]", edgeEvent, msgType, e);
+                }
+                break;
+            case DELETED:
+                Alarm alarm = mapper.convertValue(edgeEvent.getBody(), Alarm.class);
+                AlarmUpdateMsg alarmUpdateMsg =
+                        alarmMsgConstructor.constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm);
                 downlinkMsg = DownlinkMsg.newBuilder()
                         .setDownlinkMsgId(EdgeUtils.nextPositiveInt())
-                        .addAlarmUpdateMsg(alarmMsgConstructor.constructAlarmUpdatedMsg(edge.getTenantId(), msgType, alarm))
+                        .addAlarmUpdateMsg(alarmUpdateMsg)
                         .build();
-            }
-        } catch (Exception e) {
-            log.error("Can't process alarm msg [{}] [{}]", edgeEvent, msgType, e);
+                break;
         }
         return downlinkMsg;
     }
-    public void processAlarmNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) {
+    public void processAlarmNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException {
+        EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction());
         AlarmId alarmId = new AlarmId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB()));
-        ListenableFuture<Alarm> alarmFuture = alarmService.findAlarmByIdAsync(tenantId, alarmId);
-        Futures.addCallback(alarmFuture, new FutureCallback<Alarm>() {
-            @Override
-            public void onSuccess(@Nullable Alarm alarm) {
-                if (alarm != null) {
-                    EdgeEventType type = EdgeUtils.getEdgeEventTypeByEntityType(alarm.getOriginator().getEntityType());
-                    if (type != null) {
-                        PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
-                        PageData<EdgeId> pageData;
-                        do {
-                            pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, alarm.getOriginator(), pageLink);
-                            if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
-                                for (EdgeId edgeId : pageData.getData()) {
-                                    saveEdgeEvent(tenantId,
-                                            edgeId,
-                                            EdgeEventType.ALARM,
-                                            EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()),
-                                            alarmId,
-                                            null);
-                                }
-                                if (pageData.hasNext()) {
-                                    pageLink = pageLink.nextPageLink();
-                                }
+        switch (actionType) {
+            case DELETED:
+                EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEdgeIdMSB(), edgeNotificationMsg.getEdgeIdLSB()));
+                Alarm alarm = mapper.readValue(edgeNotificationMsg.getBody(), Alarm.class);
+                saveEdgeEvent(tenantId, edgeId, EdgeEventType.ALARM, actionType, alarmId, mapper.valueToTree(alarm));
+                break;
+            default:
+                ListenableFuture<Alarm> alarmFuture = alarmService.findAlarmByIdAsync(tenantId, alarmId);
+                Futures.addCallback(alarmFuture, new FutureCallback<Alarm>() {
+                    @Override
+                    public void onSuccess(@Nullable Alarm alarm) {
+                        if (alarm != null) {
+                            EdgeEventType type = EdgeUtils.getEdgeEventTypeByEntityType(alarm.getOriginator().getEntityType());
+                            if (type != null) {
+                                PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
+                                PageData<EdgeId> pageData;
+                                do {
+                                    pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, alarm.getOriginator(), pageLink);
+                                    if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
+                                        for (EdgeId edgeId : pageData.getData()) {
+                                            saveEdgeEvent(tenantId,
+                                                    edgeId,
+                                                    EdgeEventType.ALARM,
+                                                    EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()),
+                                                    alarmId,
+                                                    null);
+                                        }
+                                        if (pageData.hasNext()) {
+                                            pageLink = pageLink.nextPageLink();
+                                        }
+                                    }
+                                } while (pageData != null && pageData.hasNext());
                             }
-                        } while (pageData != null && pageData.hasNext());
+                        }
                     }
-                }
-            }
-            @Override
-            public void onFailure(Throwable t) {
-                log.warn("[{}] can't find alarm by id [{}] {}", tenantId.getId(), alarmId.getId(), t);
-            }
-        }, dbCallbackExecutorService);
+                    @Override
+                    public void onFailure(Throwable t) {
+                        log.warn("[{}] can't find alarm by id [{}] {}", tenantId.getId(), alarmId.getId(), t);
+                    }
+                }, dbCallbackExecutorService);
+        }
     }
 }
--- a/rest-client/src/main/java/org/thingsboard/rest/client/RestClient.java
View file @1297fef
+++ b/rest-client/src/main/java/org/thingsboard/rest/client/RestClient.java
View file @1297fef
@@ -337,7 +337,7 @@ public class RestClient implements ClientHttpRequestInterceptor, Closeable {
         addTimePageLinkToParam(params, pageLink);
         return restTemplate.exchange(
-                baseURL +  urlSecondPart + getTimeUrlParams(pageLink),
+                baseURL +  urlSecondPart + "&" + getTimeUrlParams(pageLink),
                 HttpMethod.GET,
                 HttpEntity.EMPTY,
                 new ParameterizedTypeReference<PageData<AlarmInfo>>() {