From 731fb1eae893006d5c9a5b5deee527d8fb071447 Mon Sep 17 00:00:00 2001
From: Volodymyr Babak <volodymyr.babak@gmail.com>
Date: Wed, 22 Jun 2022 19:42:00 +0300
Subject: [PATCH] Code review changes: avoid creating device actor if not
 needed

---
 .../device/DeviceActorMessageProcessor.java    |  7 -------
 .../rpc/processor/TelemetryEdgeProcessor.java  | 11 +++--------
 .../queue/DefaultTbCoreConsumerService.java    | 18 ++++++++++++++++++
 .../service/queue/TbCoreConsumerStats.java     |  5 +++++
 common/cluster-api/src/main/proto/queue.proto  |  8 ++++++--
 5 files changed, 32 insertions(+), 17 deletions(-)

diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
index 28c7a6a499..b0830317a7 100644
--- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
@@ -418,9 +418,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         if (msg.hasUplinkNotificationMsg()) {
             processUplinkNotificationMsg(context, sessionInfo, msg.getUplinkNotificationMsg());
         }
-        if (msg.hasDeviceActivity()) {
-            handleDeviceActivity(context, msg.getDeviceActivity());
-        }
         callback.onSuccess();
     }
 
@@ -732,10 +729,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         }
     }
 
-    private void handleDeviceActivity(TbActorCtx context, TransportProtos.DeviceActivityProto deviceActivityProto) {
-        systemContext.getDeviceStateService().onDeviceActivity(tenantId, deviceId, deviceActivityProto.getLastActivityTime());
-    }
-
     void processCredentialsUpdate(TbActorMsg msg) {
         if (((DeviceCredentialsUpdateNotificationMsg) msg).getDeviceCredentials().getCredentialsType() == DeviceCredentialsType.LWM2M_CREDENTIALS) {
             sessions.forEach((k, v) -> {
diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java
index 88ea417c24..f30111e1ef 100644
--- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/TelemetryEdgeProcessor.java
@@ -109,21 +109,16 @@ public class TelemetryEdgeProcessor extends BaseEdgeProcessor {
             if (EntityType.DEVICE.equals(entityId.getEntityType())) {
                 DeviceId deviceId = new DeviceId(entityId.getId());
 
-                TransportProtos.SessionInfoProto sessionInfo = TransportProtos.SessionInfoProto.newBuilder()
+                TransportProtos.DeviceActivityProto deviceActivityMsg = TransportProtos.DeviceActivityProto.newBuilder()
                         .setTenantIdMSB(tenantId.getId().getMostSignificantBits())
                         .setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
                         .setDeviceIdMSB(deviceId.getId().getMostSignificantBits())
-                        .setDeviceIdLSB(deviceId.getId().getLeastSignificantBits()).build();
-
-                TransportProtos.DeviceActivityProto deviceActivityProto = TransportProtos.DeviceActivityProto.newBuilder()
+                        .setDeviceIdLSB(deviceId.getId().getLeastSignificantBits())
                         .setLastActivityTime(System.currentTimeMillis()).build();
 
-                TransportProtos.TransportToDeviceActorMsg msg = TransportProtos.TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo)
-                        .setDeviceActivity(deviceActivityProto).build();
-
                 TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, deviceId);
                 tbCoreMsgProducer.send(tpi, new TbProtoQueueMsg<>(deviceId.getId(),
-                        TransportProtos.ToCoreMsg.newBuilder().setToDeviceActorMsg(msg).build()), null);
+                        TransportProtos.ToCoreMsg.newBuilder().setDeviceActivityMsg(deviceActivityMsg).build()), null);
             }
         }
         if (entityData.hasAttributeDeleteMsg()) {
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 d071d591a1..a3bae8bc4e 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
@@ -28,6 +28,7 @@ import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.common.data.alarm.Alarm;
+import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.rpc.RpcError;
 import org.thingsboard.server.common.msg.MsgType;
@@ -236,6 +237,9 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
                                 } else if (toCoreMsg.hasEdgeNotificationMsg()) {
                                     log.trace("[{}] Forwarding message to edge service {}", id, toCoreMsg.getEdgeNotificationMsg());
                                     forwardToEdgeNotificationService(toCoreMsg.getEdgeNotificationMsg(), callback);
+                                } else if (toCoreMsg.hasDeviceActivityMsg()) {
+                                    log.trace("[{}] Forwarding message to device state service {}", id, toCoreMsg.getDeviceActivityMsg());
+                                    forwardToStateService(toCoreMsg.getDeviceActivityMsg(), callback);
                                 } else if (!toCoreMsg.getToDeviceActorNotificationMsg().isEmpty()) {
                                     Optional<TbActorMsg> actorMsg = encodingService.decode(toCoreMsg.getToDeviceActorNotificationMsg().toByteArray());
                                     if (actorMsg.isPresent()) {
@@ -520,6 +524,20 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
         stateService.onQueueMsg(deviceStateServiceMsg, callback);
     }
 
+    private void forwardToStateService(TransportProtos.DeviceActivityProto deviceActivityMsg, TbCallback callback) {
+        if (statsEnabled) {
+            stats.log(deviceActivityMsg);
+        }
+        TenantId tenantId = TenantId.fromUUID(new UUID(deviceActivityMsg.getTenantIdMSB(), deviceActivityMsg.getTenantIdLSB()));
+        DeviceId deviceId = new DeviceId(new UUID(deviceActivityMsg.getDeviceIdMSB(), deviceActivityMsg.getDeviceIdLSB()));
+        try {
+            stateService.onDeviceActivity(tenantId, deviceId, deviceActivityMsg.getLastActivityTime());
+            callback.onSuccess();
+        } catch (Exception e) {
+            callback.onFailure(new RuntimeException("Failed update device activity for device [" + deviceId.getId() + "]!", e));
+        }
+    }
+
     private void forwardToEdgeNotificationService(EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) {
         if (statsEnabled) {
             stats.log(edgeNotificationMsg);
diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java b/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java
index 2432bca426..8f12cbb78a 100644
--- a/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java
+++ b/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java
@@ -112,6 +112,11 @@ public class TbCoreConsumerStats {
         edgeNotificationsCounter.increment();
     }
 
+    public void log(TransportProtos.DeviceActivityProto msg) {
+        totalCounter.increment();
+        deviceStateCounter.increment();
+    }
+
     public void log(TransportProtos.SubscriptionMgrMsgProto msg) {
         totalCounter.increment();
         subscriptionMsgCounter.increment();
diff --git a/common/cluster-api/src/main/proto/queue.proto b/common/cluster-api/src/main/proto/queue.proto
index 7fefe18d4a..28d4073c87 100644
--- a/common/cluster-api/src/main/proto/queue.proto
+++ b/common/cluster-api/src/main/proto/queue.proto
@@ -455,7 +455,11 @@ message GetOtaPackageResponseMsg {
 }
 
 message DeviceActivityProto {
-  int64 lastActivityTime = 1;
+  int64 tenantIdMSB = 1;
+  int64 tenantIdLSB = 2;
+  int64 deviceIdMSB = 3;
+  int64 deviceIdLSB = 4;
+  int64 lastActivityTime = 5;
 }
 
 //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level.
@@ -487,7 +491,6 @@ message TransportToDeviceActorMsg {
   ToDeviceRpcResponseStatusMsg rpcResponseStatusMsg = 10;
   SendPendingRPCMsg sendPendingRPC = 11;
   UplinkNotificationMsg uplinkNotificationMsg = 12;
-  DeviceActivityProto deviceActivity = 13;
 }
 
 message TransportToRuleEngineMsg {
@@ -907,6 +910,7 @@ message ToCoreMsg {
   SubscriptionMgrMsgProto toSubscriptionMgrMsg = 3;
   bytes toDeviceActorNotificationMsg = 4;
   EdgeNotificationMsgProto edgeNotificationMsg = 5;
+  DeviceActivityProto deviceActivityMsg = 6;
 }
 
 /* High priority messages with low latency are handled by ThingsBoard Core Service separately */