From c1d547265eb845beb289e5f622de8e82f30654d5 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Fri, 13 Oct 2023 15:42:12 +0300 Subject: [PATCH] Minor refactoring: changes to be in sync with edge, refactor objectmapper calls --- .../edge/rpc/processor/alarm/AlarmEdgeProcessor.java | 8 +++++--- .../edge/rpc/processor/alarm/BaseAlarmProcessor.java | 6 +++--- .../edge/rpc/processor/device/DeviceEdgeProcessor.java | 6 ++---- .../service/edge/rpc/processor/edge/EdgeProcessor.java | 8 ++++---- .../rpc/processor/relation/BaseRelationProcessor.java | 2 +- .../rpc/processor/relation/RelationEdgeProcessor.java | 8 +++----- .../rpc/processor/telemetry/BaseTelemetryProcessor.java | 6 +++--- 7 files changed, 21 insertions(+), 23 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/AlarmEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/AlarmEdgeProcessor.java index b76e8f187d..f002b5c04e 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/AlarmEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/AlarmEdgeProcessor.java @@ -15,7 +15,6 @@ */ package org.thingsboard.server.service.edge.rpc.processor.alarm; -import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; @@ -69,13 +68,16 @@ public class AlarmEdgeProcessor extends BaseAlarmProcessor { return null; } - public ListenableFuture processAlarmNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException { + public ListenableFuture processAlarmNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()); AlarmId alarmId = new AlarmId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); EdgeId sourceEdgeId = safeGetEdgeId(edgeNotificationMsg.getSourceEdgeIdMSB(), edgeNotificationMsg.getSourceEdgeIdLSB()); switch (actionType) { case DELETED: - Alarm deletedAlarm = JacksonUtil.OBJECT_MAPPER.readValue(edgeNotificationMsg.getBody(), Alarm.class); + Alarm deletedAlarm = JacksonUtil.fromString(edgeNotificationMsg.getBody(), Alarm.class); + if (deletedAlarm == null) { + return Futures.immediateFuture(null); + } List> delFutures = pushEventToAllRelatedEdges(tenantId, deletedAlarm.getOriginator(), alarmId, actionType, JacksonUtil.valueToTree(deletedAlarm), sourceEdgeId); return Futures.transform(Futures.allAsList(delFutures), voids -> null, dbCallbackExecutorService); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/BaseAlarmProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/BaseAlarmProcessor.java index 03a91ae2f0..3fb086cd52 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/BaseAlarmProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/alarm/BaseAlarmProcessor.java @@ -45,7 +45,7 @@ import java.util.UUID; @Slf4j public abstract class BaseAlarmProcessor extends BaseEdgeProcessor { - public ListenableFuture processAlarmMsg(TenantId tenantId, AlarmUpdateMsg alarmUpdateMsg) { + protected ListenableFuture processAlarmMsg(TenantId tenantId, AlarmUpdateMsg alarmUpdateMsg) { EntityId originatorId = getAlarmOriginator(tenantId, alarmUpdateMsg.getOriginatorName(), EntityType.valueOf(alarmUpdateMsg.getOriginatorType())); AlarmId alarmId = new AlarmId(new UUID(alarmUpdateMsg.getIdMSB(), alarmUpdateMsg.getIdLSB())); @@ -72,7 +72,7 @@ public abstract class BaseAlarmProcessor extends BaseEdgeProcessor { alarm.setAcknowledged(alarmStatus.isAck()); alarm.setAckTs(alarmUpdateMsg.getAckTs()); alarm.setEndTs(alarmUpdateMsg.getEndTs()); - alarm.setDetails(JacksonUtil.OBJECT_MAPPER.readTree(alarmUpdateMsg.getDetails())); + alarm.setDetails(JacksonUtil.toJsonNode(alarmUpdateMsg.getDetails())); if (UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE.equals(alarmUpdateMsg.getMsgType())) { alarmService.createAlarm(AlarmCreateOrUpdateActiveRequest.fromAlarm(alarm, null, alarmId)); } else { @@ -89,7 +89,7 @@ public abstract class BaseAlarmProcessor extends BaseEdgeProcessor { Alarm alarmToClear = alarmService.findAlarmById(tenantId, alarmId); if (alarmToClear != null) { alarmService.clearAlarm(tenantId, alarmId, alarmUpdateMsg.getClearTs(), - JacksonUtil.OBJECT_MAPPER.readTree(alarmUpdateMsg.getDetails())); + JacksonUtil.toJsonNode(alarmUpdateMsg.getDetails())); } break; case ENTITY_DELETED_RPC_MESSAGE: diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/DeviceEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/DeviceEdgeProcessor.java index ad152f754b..c2f18daf0d 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/DeviceEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/DeviceEdgeProcessor.java @@ -15,7 +15,6 @@ */ package org.thingsboard.server.service.edge.rpc.processor.device; -import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; @@ -40,7 +39,6 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.rpc.RpcError; import org.thingsboard.server.common.data.security.DeviceCredentials; -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; @@ -196,7 +194,7 @@ public class DeviceEdgeProcessor extends BaseDeviceProcessor { data.put("method", deviceRpcCallMsg.getRequestMsg().getMethod()); data.put("params", deviceRpcCallMsg.getRequestMsg().getParams()); TbMsg tbMsg = TbMsg.newMsg(TbMsgType.TO_SERVER_RPC_REQUEST, deviceId, null, metaData, - TbMsgDataType.JSON, JacksonUtil.OBJECT_MAPPER.writeValueAsString(data)); + TbMsgDataType.JSON, JacksonUtil.toString(data)); tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, new TbQueueCallback() { @Override public void onSuccess(TbQueueMsgMetadata metadata) { @@ -210,7 +208,7 @@ public class DeviceEdgeProcessor extends BaseDeviceProcessor { tenantId, device, deviceRpcCallMsg, t); } }); - } catch (JsonProcessingException | IllegalArgumentException e) { + } catch (IllegalArgumentException e) { log.warn("[{}][{}] Failed to push TO_SERVER_RPC_REQUEST to rule engine. deviceRpcCallMsg {}", tenantId, deviceId, deviceRpcCallMsg, e); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/edge/EdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/edge/EdgeProcessor.java index b38659c881..daffefe477 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/edge/EdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/edge/EdgeProcessor.java @@ -72,9 +72,9 @@ public class EdgeProcessor extends BaseEdgeProcessor { EdgeId edgeId = new EdgeId(new UUID(edgeNotificationMsg.getEntityIdMSB(), edgeNotificationMsg.getEntityIdLSB())); switch (actionType) { case ASSIGNED_TO_CUSTOMER: - CustomerId customerId = JacksonUtil.OBJECT_MAPPER.readValue(edgeNotificationMsg.getBody(), CustomerId.class); + CustomerId customerId = JacksonUtil.fromString(edgeNotificationMsg.getBody(), CustomerId.class); Edge edge = edgeService.findEdgeById(tenantId, edgeId); - if (edge == null || customerId.isNullUid()) { + if (customerId != null && (edge == null || customerId.isNullUid())) { return Futures.immediateFuture(null); } List> futures = new ArrayList<>(); @@ -96,9 +96,9 @@ public class EdgeProcessor extends BaseEdgeProcessor { } while (pageData != null && pageData.hasNext()); return Futures.transform(Futures.allAsList(futures), voids -> null, dbCallbackExecutorService); case UNASSIGNED_FROM_CUSTOMER: - CustomerId customerIdToDelete = JacksonUtil.OBJECT_MAPPER.readValue(edgeNotificationMsg.getBody(), CustomerId.class); + CustomerId customerIdToDelete = JacksonUtil.fromString(edgeNotificationMsg.getBody(), CustomerId.class); edge = edgeService.findEdgeById(tenantId, edgeId); - if (edge == null || customerIdToDelete.isNullUid()) { + if (customerIdToDelete != null && (edge == null || customerIdToDelete.isNullUid())) { return Futures.immediateFuture(null); } return Futures.transformAsync(saveEdgeEvent(edge.getTenantId(), edge.getId(), EdgeEventType.EDGE, EdgeEventActionType.UNASSIGNED_FROM_CUSTOMER, edgeId, null), diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/BaseRelationProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/BaseRelationProcessor.java index 9729c861ee..952affdc3d 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/BaseRelationProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/BaseRelationProcessor.java @@ -33,7 +33,7 @@ import java.util.UUID; @Slf4j public abstract class BaseRelationProcessor extends BaseEdgeProcessor { - public ListenableFuture processRelationMsg(TenantId tenantId, RelationUpdateMsg relationUpdateMsg) { + protected ListenableFuture processRelationMsg(TenantId tenantId, RelationUpdateMsg relationUpdateMsg) { try { EntityRelation entityRelation = new EntityRelation(); diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/RelationEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/RelationEdgeProcessor.java index 2c23fd42b2..fe76fed545 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/RelationEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/relation/RelationEdgeProcessor.java @@ -15,7 +15,6 @@ */ package org.thingsboard.server.service.edge.rpc.processor.relation; -import com.fasterxml.jackson.core.JsonProcessingException; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; @@ -67,10 +66,9 @@ public class RelationEdgeProcessor extends BaseRelationProcessor { .build(); } - public ListenableFuture processRelationNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) throws JsonProcessingException { - EntityRelation relation = JacksonUtil.OBJECT_MAPPER.readValue(edgeNotificationMsg.getBody(), EntityRelation.class); - if (relation.getFrom().getEntityType().equals(EntityType.EDGE) || - relation.getTo().getEntityType().equals(EntityType.EDGE)) { + public ListenableFuture processRelationNotification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { + EntityRelation relation = JacksonUtil.fromString(edgeNotificationMsg.getBody(), EntityRelation.class); + if (relation == null || (relation.getFrom().getEntityType().equals(EntityType.EDGE) || relation.getTo().getEntityType().equals(EntityType.EDGE))) { return Futures.immediateFuture(null); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java index c2d172ede0..f68d668329 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java @@ -22,7 +22,6 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.SettableFuture; import com.google.gson.Gson; -import com.google.gson.JsonElement; import com.google.gson.JsonObject; import com.google.gson.JsonParser; import lombok.extern.slf4j.Slf4j; @@ -346,8 +345,9 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor { log.warn("[{}] Unsupported edge event type [{}]", tenantId, entityType); return null; } - JsonElement entityData = JsonParser.parseString(JacksonUtil.OBJECT_MAPPER.writeValueAsString(body)); - return entityDataMsgConstructor.constructEntityDataMsg(tenantId, entityId, actionType, entityData); + String bodyJackson = JacksonUtil.toString(body); + return bodyJackson == null ? null : + entityDataMsgConstructor.constructEntityDataMsg(tenantId, entityId, actionType, JsonParser.parseString(bodyJackson)); } }