diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index 091c656462..cabc61cab6 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -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 deviceProfileOpt) { + deviceSessionCtx.onDeviceUpdate(sessionInfo, device, deviceProfileOpt); } } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java index c2c2d258fc..487aa8a207 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/DeviceSessionCtx.java @@ -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); } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/SessionMsgListener.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/SessionMsgListener.java index 85c162930d..9eb0fbcd6c 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/SessionMsgListener.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/SessionMsgListener.java @@ -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 deviceProfileOpt) { } } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index 70a4817079..3e619f8d6a 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -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 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 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) { diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/SessionMetaData.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/SessionMetaData.java index 5181d8e781..ac293307a9 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/SessionMetaData.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/SessionMetaData.java @@ -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; + } } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java index 5d33dccff6..168a20ba67 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java @@ -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 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() { diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionContext.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionContext.java index 9b5870cace..fc82974764 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionContext.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionContext.java @@ -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 deviceProfileOpt); }