Improved transport cache invalidation for device updates

@@ -72,6 +72,7 @@ import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -701,12 +702,12 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
     }
 
     @Override
-    public void onProfileUpdate(DeviceProfile deviceProfile) {
-        deviceSessionCtx.onProfileUpdate(deviceProfile);
+    public void onDeviceProfileUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceProfile deviceProfile) {
+        deviceSessionCtx.onDeviceProfileUpdate(sessionInfo, deviceProfile);
     }
 
     @Override
-    public void onDeviceProfileUpdate(Device device, TransportProtos.SessionInfoProto sessionInfo) {
-        deviceSessionCtx.onDeviceProfileUpdate(device, sessionInfo);
+    public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional<DeviceProfile> deviceProfileOpt) {
+        deviceSessionCtx.onDeviceUpdate(sessionInfo, device, deviceProfileOpt);
     }
 }

@@ -27,6 +27,7 @@ import org.thingsboard.server.common.data.device.profile.DeviceProfileTransportC
 import org.thingsboard.server.common.data.device.profile.MqttDeviceProfileTransportConfiguration;
 import org.thingsboard.server.common.data.device.profile.ProtoTransportPayloadConfiguration;
 import org.thingsboard.server.common.data.device.profile.TransportPayloadTypeConfiguration;
+import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.transport.mqtt.MqttTransportContext;
 import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor;
 import org.thingsboard.server.transport.mqtt.util.MqttTopicFilter;
@@ -108,8 +109,8 @@ public class DeviceSessionCtx extends MqttDeviceAwareSessionContext {
     }
 
     @Override
-    public void onProfileUpdate(DeviceProfile deviceProfile) {
-        super.onProfileUpdate(deviceProfile);
+    public void onDeviceProfileUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceProfile deviceProfile) {
+        super.onDeviceProfileUpdate(sessionInfo, deviceProfile);
         updateTopicFilters(deviceProfile);
     }
 

@@ -24,6 +24,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeResponse
 import org.thingsboard.server.gen.transport.TransportProtos.SessionCloseNotificationProto;
 import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcRequestMsg;
 
+import java.util.Optional;
+
 /**
  * Created by ashvayka on 04.10.18.
  */
@@ -39,9 +41,9 @@ public interface SessionMsgListener {
 
     void onToServerRpcResponse(ToServerRpcResponseMsg toServerResponse);
 
-    default void onProfileUpdate(DeviceProfile deviceProfile) {
+    default void onDeviceProfileUpdate(TransportProtos.SessionInfoProto newSessionInfo, DeviceProfile deviceProfile) {
     }
 
-    default void onDeviceProfileUpdate(Device device, TransportProtos.SessionInfoProto sessionInfo) {
+    default void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional<DeviceProfile> deviceProfileOpt) {
     }
 }

@@ -672,9 +672,17 @@ public class DefaultTransportService implements TransportService {
         long deviceProfileIdMSB = deviceProfile.getId().getId().getMostSignificantBits();
         long deviceProfileIdLSB = deviceProfile.getId().getId().getLeastSignificantBits();
         sessions.forEach((id, md) -> {
+            //TODO: if transport types are different - we should close the session.
             if (md.getSessionInfo().getDeviceProfileIdMSB() == deviceProfileIdMSB
                     && md.getSessionInfo().getDeviceProfileIdLSB() == deviceProfileIdLSB) {
-                transportCallbackExecutor.submit(() -> md.getListener().onProfileUpdate(deviceProfile));
+                TransportProtos.SessionInfoProto newSessionInfo = TransportProtos.SessionInfoProto.newBuilder()
+                        .mergeFrom(md.getSessionInfo())
+                        .setDeviceProfileIdMSB(deviceProfileIdMSB)
+                        .setDeviceProfileIdLSB(deviceProfileIdLSB)
+                        .setDeviceType(deviceProfile.getName())
+                        .build();
+                md.setSessionInfo(newSessionInfo);
+                transportCallbackExecutor.submit(() -> md.getListener().onDeviceProfileUpdate(newSessionInfo, deviceProfile));
             }
         });
     }
@@ -684,28 +692,27 @@ public class DefaultTransportService implements TransportService {
         long deviceIdLSB = device.getId().getId().getLeastSignificantBits();
         long deviceProfileIdMSB = device.getDeviceProfileId().getId().getMostSignificantBits();
         long deviceProfileIdLSB = device.getDeviceProfileId().getId().getLeastSignificantBits();
-        for (Map.Entry<UUID, SessionMetaData> entry : sessions.entrySet()) {
-            SessionMetaData md = entry.getValue();
-            if ((md.getSessionInfo().getDeviceIdMSB() == deviceIdMSB
-                    && md.getSessionInfo().getDeviceIdLSB() == deviceIdLSB)
-                    && (md.getSessionInfo().getDeviceProfileIdMSB() != deviceProfileIdMSB
-                    && md.getSessionInfo().getDeviceProfileIdLSB() != deviceProfileIdLSB)) {
-                updateSessionMetadata(device, entry, md);
+        sessions.forEach((id, md) -> {
+            if ((md.getSessionInfo().getDeviceIdMSB() == deviceIdMSB && md.getSessionInfo().getDeviceIdLSB() == deviceIdLSB)) {
+                DeviceProfile newDeviceProfile;
+                if (md.getSessionInfo().getDeviceProfileIdMSB() != deviceProfileIdMSB
+                        && md.getSessionInfo().getDeviceProfileIdLSB() != deviceProfileIdLSB) {
+                    //TODO: if transport types are different - we should close the session.
+                    newDeviceProfile = deviceProfileCache.get(new DeviceProfileId(new UUID(deviceProfileIdMSB, deviceProfileIdLSB)));
+                } else {
+                    newDeviceProfile = null;
+                }
+                TransportProtos.SessionInfoProto newSessionInfo = TransportProtos.SessionInfoProto.newBuilder()
+                        .mergeFrom(md.getSessionInfo())
+                        .setDeviceProfileIdMSB(deviceProfileIdMSB)
+                        .setDeviceProfileIdLSB(deviceProfileIdLSB)
+                        .setDeviceName(device.getName())
+                        .setDeviceType(device.getType())
+                        .build();
+                md.setSessionInfo(newSessionInfo);
+                transportCallbackExecutor.submit(() -> md.getListener().onDeviceUpdate(newSessionInfo, device, Optional.ofNullable(newDeviceProfile)));
             }
-        }
-    }
-
-    private void updateSessionMetadata(Device device, Map.Entry<UUID, SessionMetaData> entry, SessionMetaData md) {
-        TransportProtos.SessionInfoProto newSessionInfo = TransportProtos.SessionInfoProto.newBuilder()
-                .mergeFrom(md.getSessionInfo())
-                .setDeviceProfileIdMSB(device.getDeviceProfileId().getId().getMostSignificantBits())
-                .setDeviceProfileIdLSB(device.getDeviceProfileId().getId().getLeastSignificantBits())
-                .setDeviceType(device.getType())
-                .build();
-        SessionMetaData newSessionMetaData = new SessionMetaData(newSessionInfo, md.getSessionType(), md.getListener());
-        entry.setValue(newSessionMetaData);
-        transportCallbackExecutor.submit(() -> newSessionMetaData.getListener().onDeviceProfileUpdate(device,
-                newSessionMetaData.getSessionInfo()));
+        });
     }
 
     protected UUID toSessionId(TransportProtos.SessionInfoProto sessionInfo) {

@@ -27,22 +27,17 @@ import java.util.concurrent.ScheduledFuture;
 @Data
 class SessionMetaData {
 
-    private final TransportProtos.SessionInfoProto sessionInfo;
+    private volatile TransportProtos.SessionInfoProto sessionInfo;
     private final TransportProtos.SessionType sessionType;
     private final SessionMsgListener listener;
 
-    private ScheduledFuture scheduledFuture;
-
+    private volatile ScheduledFuture scheduledFuture;
     private volatile long lastActivityTime;
     private volatile long lastReportedActivityTime;
     private volatile boolean subscribedToAttributes;
     private volatile boolean subscribedToRPC;
 
-    SessionMetaData(
-            TransportProtos.SessionInfoProto sessionInfo,
-            TransportProtos.SessionType sessionType,
-            SessionMsgListener listener
-    ) {
+    SessionMetaData(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.SessionType sessionType, SessionMsgListener listener) {
         this.sessionInfo = sessionInfo;
         this.sessionType = sessionType;
         this.listener = listener;
@@ -54,11 +49,15 @@ class SessionMetaData {
         this.lastActivityTime = System.currentTimeMillis();
     }
 
-    void setScheduledFuture(ScheduledFuture scheduledFuture) { this.scheduledFuture = scheduledFuture; }
+    void setScheduledFuture(ScheduledFuture scheduledFuture) {
+        this.scheduledFuture = scheduledFuture;
+    }
 
     public ScheduledFuture getScheduledFuture() {
         return scheduledFuture;
     }
 
-    public boolean hasScheduledFuture() { return null != this.scheduledFuture; }
+    public boolean hasScheduledFuture() {
+        return null != this.scheduledFuture;
+    }
 }

@@ -24,6 +24,7 @@ import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.transport.auth.TransportDeviceInfo;
 import org.thingsboard.server.gen.transport.TransportProtos;
 
+import java.util.Optional;
 import java.util.UUID;
 
 /**
@@ -58,16 +59,19 @@ public abstract class DeviceAwareSessionContext implements SessionContext {
     }
 
     @Override
-    public void onProfileUpdate(DeviceProfile deviceProfile) {
+    public void onDeviceProfileUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceProfile deviceProfile) {
+        this.sessionInfo = sessionInfo;
         this.deviceProfile = deviceProfile;
         this.deviceInfo.setDeviceType(deviceProfile.getName());
-        this.sessionInfo = TransportProtos.SessionInfoProto.newBuilder().mergeFrom(sessionInfo).setDeviceType(deviceProfile.getName()).build();
+
     }
 
-    public void onDeviceProfileUpdate(Device device, TransportProtos.SessionInfoProto sessionInfo) {
+    @Override
+    public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional<DeviceProfile> deviceProfileOpt) {
+        this.sessionInfo = sessionInfo;
         this.deviceInfo.setDeviceProfileId(device.getDeviceProfileId());
         this.deviceInfo.setDeviceType(device.getType());
-        this.sessionInfo = sessionInfo;
+        deviceProfileOpt.ifPresent(profile -> this.deviceProfile = profile);
     }
 
     public boolean isConnected() {

@@ -19,6 +19,7 @@ import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.gen.transport.TransportProtos;
 
+import java.util.Optional;
 import java.util.UUID;
 
 public interface SessionContext {
@@ -27,7 +28,7 @@ public interface SessionContext {
 
     int nextMsgId();
 
-    void onProfileUpdate(DeviceProfile deviceProfile);
+    void onDeviceProfileUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceProfile deviceProfile);
 
-    void onDeviceProfileUpdate(Device device, TransportProtos.SessionInfoProto sessionInfo);
+    void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional<DeviceProfile> deviceProfileOpt);
 }