Merge branch 'feature/lwm2m-refactoring-downlink' of https://github.com/thingsbo…

…ard/thingsboard into feature/lwm2m-refactoring-downlink

# Conflicts:
#	common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/downlink/TbLwM2MReadCallback.java

@@ -18,11 +18,11 @@ package org.thingsboard.server.transport.lwm2m.server.client;
 import org.eclipse.leshan.server.registration.Registration;
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.device.profile.Lwm2mDeviceProfileTransportConfiguration;
-import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
 import org.thingsboard.server.gen.transport.TransportProtos;
 
 import java.util.Collection;
+import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 
@@ -34,7 +34,7 @@ public interface LwM2mClientContext {
 
     LwM2mClient getClientBySessionInfo(TransportProtos.SessionInfoProto sessionInfo);
 
-    void register(LwM2mClient lwM2MClient, Registration registration) throws LwM2MClientStateException;
+    Optional<TransportProtos.SessionInfoProto> register(LwM2mClient lwM2MClient, Registration registration) throws LwM2MClientStateException;
 
     void updateRegistration(LwM2mClient client, Registration registration) throws LwM2MClientStateException;
 

@@ -60,12 +60,14 @@ public class LwM2mClientContextImpl implements LwM2mClientContext {
     }
 
     @Override
-    public void register(LwM2mClient lwM2MClient, Registration registration) throws LwM2MClientStateException {
+    public Optional<TransportProtos.SessionInfoProto> register(LwM2mClient lwM2MClient, Registration registration) throws LwM2MClientStateException {
+        TransportProtos.SessionInfoProto oldSession = null;
         lwM2MClient.lock();
         try {
             if (LwM2MClientState.UNREGISTERED.equals(lwM2MClient.getState())) {
                 throw new LwM2MClientStateException(lwM2MClient.getState(), "Client is in invalid state.");
             }
+            oldSession = lwM2MClient.getSession();
             TbLwM2MSecurityInfo securityInfo = securityStore.getTbLwM2MSecurityInfoByEndpoint(lwM2MClient.getEndpoint());
             if (securityInfo.getSecurityMode() != null) {
                 if (securityInfo.getDeviceProfile() != null) {
@@ -89,6 +91,7 @@ public class LwM2mClientContextImpl implements LwM2mClientContext {
         } finally {
             lwM2MClient.unlock();
         }
+        return Optional.ofNullable(oldSession);
     }
 
     @Override

@@ -15,12 +15,13 @@
  */
 package org.thingsboard.server.transport.lwm2m.server.downlink;
 
-import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
+import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient;
+import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
 
 import static org.thingsboard.server.transport.lwm2m.server.LwM2mTransportUtil.LOG_LWM2M_INFO;
-import static org.thingsboard.server.transport.lwm2m.server.LwM2mOperationType.OBSERVE_CANCEL_ALL;
 
+@Slf4j
 public class TbLwM2MCancelAllObserveCallback extends AbstractTbLwM2MRequestCallback<TbLwM2MCancelAllRequest, Integer> {
 
     public TbLwM2MCancelAllObserveCallback(LwM2mUplinkMsgHandler handler, LwM2mClient client) {
@@ -29,7 +30,8 @@ public class TbLwM2MCancelAllObserveCallback extends AbstractTbLwM2MRequestCallb
 
     @Override
     public void onSuccess(TbLwM2MCancelAllRequest request, Integer canceledSubscriptionsCount) {
-        String observeCancelMsg = String.format("%s: type operation %s paths: count: %d", LOG_LWM2M_INFO, OBSERVE_CANCEL_ALL.name(), canceledSubscriptionsCount);
+        log.trace("[{}] Cancel of all observations was successful: {}", client.getEndpoint(),  canceledSubscriptionsCount);
+        handler.logToTelemetry(client, String.format("[%s]: Cancel of all observations was successful. Result: [%s]", LOG_LWM2M_INFO, canceledSubscriptionsCount));
     }
 
 }

@@ -15,12 +15,14 @@
  */
 package org.thingsboard.server.transport.lwm2m.server.downlink;
 
+import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient;
 
 import static org.thingsboard.server.transport.lwm2m.server.LwM2mTransportUtil.LOG_LWM2M_INFO;
 import static org.thingsboard.server.transport.lwm2m.server.LwM2mOperationType.OBSERVE_CANCEL;
 
+@Slf4j
 public class TbLwM2MCancelObserveCallback extends AbstractTbLwM2MRequestCallback<TbLwM2MCancelObserveRequest, Integer> {
 
     private final String versionedId;
@@ -32,7 +34,8 @@ public class TbLwM2MCancelObserveCallback extends AbstractTbLwM2MRequestCallback
 
     @Override
     public void onSuccess(TbLwM2MCancelObserveRequest request, Integer canceledSubscriptionsCount) {
-        String observeCancelMsg = String.format("%s: type operation %s paths: %s count: %d", LOG_LWM2M_INFO, OBSERVE_CANCEL.name(), versionedId, canceledSubscriptionsCount);
+        log.trace("[{}] Cancel observation of [{}] successful: {}", client.getEndpoint(),  versionedId, canceledSubscriptionsCount);
+        handler.logToTelemetry(client, String.format("[%s]: Cancel Observe for [%s] successful. Result: [%s]", LOG_LWM2M_INFO, versionedId, canceledSubscriptionsCount));
     }
 
 }

@@ -20,18 +20,10 @@ import org.eclipse.leshan.core.response.ExecuteResponse;
 import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient;
 
-public class TbLwM2MExecuteCallback extends AbstractTbLwM2MRequestCallback<ExecuteRequest, ExecuteResponse> {
-
-    private final String targetId;
+public class TbLwM2MExecuteCallback extends TbLwM2MTargetedCallback<ExecuteRequest, ExecuteResponse> {
 
     public TbLwM2MExecuteCallback(LwM2mUplinkMsgHandler handler, LwM2mClient client, String targetId) {
-        super(handler, client);
-        this.targetId = targetId;
-    }
-
-    @Override
-    public void onSuccess(ExecuteRequest request, ExecuteResponse response) {
-        //TODO: separate callback wrapper for the RPC calls.
+        super(handler, client, targetId);
     }
 
 }

@@ -17,14 +17,10 @@ package org.thingsboard.server.transport.lwm2m.server.downlink;
 
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.leshan.core.request.ObserveRequest;
-import org.eclipse.leshan.core.request.ReadRequest;
 import org.eclipse.leshan.core.response.ObserveResponse;
-import org.eclipse.leshan.core.response.ReadResponse;
 import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient;
 
-import static org.thingsboard.server.transport.lwm2m.server.LwM2mTransportUtil.LOG_LWM2M_INFO;
-
 @Slf4j
 public class TbLwM2MObserveCallback extends TbLwM2MTargetedCallback<ObserveRequest, ObserveResponse> {
 
@@ -35,6 +31,6 @@ public class TbLwM2MObserveCallback extends TbLwM2MTargetedCallback<ObserveReque
     @Override
     public void onSuccess(ObserveRequest request, ObserveResponse response) {
         super.onSuccess(request, response);
-        handler.onUpdateValueAfterReadResponse(client.getRegistration(), targetId, response, null);
+        handler.onUpdateValueAfterReadResponse(client.getRegistration(), versionedId, response, null);
     }
 }

@@ -31,7 +31,7 @@ public class TbLwM2MReadCallback extends TbLwM2MTargetedCallback<ReadRequest, Re
     @Override
     public void onSuccess(ReadRequest request, ReadResponse response) {
         super.onSuccess(request, response);
-        handler.onUpdateValueAfterReadResponse(client.getRegistration(), targetId, response, null);
+        handler.onUpdateValueAfterReadResponse(client.getRegistration(), versionedId, response, null);
     }
 
 }

@@ -16,8 +16,6 @@
 package org.thingsboard.server.transport.lwm2m.server.downlink;
 
 import lombok.extern.slf4j.Slf4j;
-import org.eclipse.leshan.core.request.ReadRequest;
-import org.eclipse.leshan.core.response.ReadResponse;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient;
 import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
 
@@ -26,19 +24,19 @@ import static org.thingsboard.server.transport.lwm2m.server.LwM2mTransportUtil.L
 @Slf4j
 public abstract class TbLwM2MTargetedCallback<R, T> extends AbstractTbLwM2MRequestCallback<R, T> {
 
-    protected final String targetId;
+    protected final String versionedId;
 
-    public TbLwM2MTargetedCallback(LwM2mUplinkMsgHandler handler, LwM2mClient client, String targetId) {
+    public TbLwM2MTargetedCallback(LwM2mUplinkMsgHandler handler, LwM2mClient client, String versionedId) {
         super(handler, client);
-        this.targetId = targetId;
+        this.versionedId = versionedId;
     }
 
     @Override
     public void onSuccess(R request, T response) {
         //TODO convert camelCase to "camel case" using .split("(?<!(^|[A-Z]))(?=[A-Z])|(?<!^)(?=[A-Z][a-z])")
         String requestName = request.getClass().getSimpleName();
-        log.trace("[{}] {} [{}] successful: {}", client.getEndpoint(), requestName, targetId, response);
-        handler.logToTelemetry(client, String.format("[%s]: %s [%s] successful. Result: [%s]", LOG_LWM2M_INFO, requestName, targetId, response));
+        log.trace("[{}] {} [{}] successful: {}", client.getEndpoint(), requestName, versionedId, response);
+        handler.logToTelemetry(client, String.format("[%s]: %s [%s] successful. Result: [%s]", LOG_LWM2M_INFO, requestName, versionedId, response));
     }
 
 }

@@ -17,21 +17,13 @@ package org.thingsboard.server.transport.lwm2m.server.downlink;
 
 import org.eclipse.leshan.core.request.WriteAttributesRequest;
 import org.eclipse.leshan.core.response.WriteAttributesResponse;
-import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient;
+import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
 
-public class TbLwM2MWriteAttributesCallback extends AbstractTbLwM2MRequestCallback<WriteAttributesRequest, WriteAttributesResponse> {
-
-    private final String targetId;
+public class TbLwM2MWriteAttributesCallback extends TbLwM2MTargetedCallback<WriteAttributesRequest, WriteAttributesResponse> {
 
     public TbLwM2MWriteAttributesCallback(LwM2mUplinkMsgHandler handler, LwM2mClient client, String targetId) {
-        super(handler, client);
-        this.targetId = targetId;
-    }
-
-    @Override
-    public void onSuccess(WriteAttributesRequest request, WriteAttributesResponse response) {
-        //TODO: separate callback wrapper for the RPC calls.
+        super(handler, client, targetId);
     }
 
 }

@@ -15,25 +15,21 @@
  */
 package org.thingsboard.server.transport.lwm2m.server.downlink;
 
-import lombok.Setter;
 import org.eclipse.leshan.core.request.WriteRequest;
 import org.eclipse.leshan.core.response.WriteResponse;
-import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient;
+import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
 
-public class TbLwM2MWriteResponseCallback extends AbstractTbLwM2MRequestCallback<WriteRequest, WriteResponse> {
-
-    private final String targetId;
+public class TbLwM2MWriteResponseCallback extends TbLwM2MTargetedCallback<WriteRequest, WriteResponse> {
 
     public TbLwM2MWriteResponseCallback(LwM2mUplinkMsgHandler handler, LwM2mClient client, String targetId) {
-        super(handler, client);
-        this.targetId = targetId;
+        super(handler, client, targetId);
     }
 
     @Override
     public void onSuccess(WriteRequest request, WriteResponse response) {
-        handler.onWriteResponseOk(client, targetId, request);
-        //TODO: separate callback wrapper for the RPC calls.
+        super.onSuccess(request, response);
+        handler.onWriteResponseOk(client, versionedId, request);
     }
 
 }

@@ -201,7 +201,11 @@ public class DefaultLwM2MUplinkMsgHandler implements LwM2mUplinkMsgHandler {
             try {
                 log.warn("[{}] [{{}] Client: create after Registration", registration.getEndpoint(), registration.getId());
                 if (lwM2MClient != null) {
-                    this.clientContext.register(lwM2MClient, registration);
+                    Optional<SessionInfoProto> oldSessionInfo = this.clientContext.register(lwM2MClient, registration);
+                    if (oldSessionInfo.isPresent()) {
+                        log.info("[{}] Closing old session: {}", registration.getEndpoint(), new UUID(oldSessionInfo.get().getSessionIdMSB(), oldSessionInfo.get().getSessionIdLSB()));
+                        closeSession(oldSessionInfo.get());
+                    }
                     this.logToTelemetry(lwM2MClient, LOG_LWM2M_INFO + ": Client registered with registration id: " + registration.getId());
                     SessionInfoProto sessionInfo = lwM2MClient.getSession();
                     transportService.registerAsyncSession(sessionInfo, new LwM2mSessionMsgListener(this, rpcHandler, sessionInfo));
@@ -278,8 +282,7 @@ public class DefaultLwM2MUplinkMsgHandler implements LwM2mUplinkMsgHandler {
                 clientContext.unregister(client, registration);
                 SessionInfoProto sessionInfo = client.getSession();
                 if (sessionInfo != null) {
-                    transportService.process(sessionInfo, DefaultTransportService.getSessionEventMsg(SessionEvent.CLOSED), null);
-                    transportService.deregisterSession(sessionInfo);
+                    closeSession(sessionInfo);
                     sessionStore.remove(registration.getEndpoint());
                     log.info("Client close session: [{}] unReg [{}] name  [{}] profile ", registration.getId(), registration.getEndpoint(), sessionInfo.getDeviceType());
                 } else {
@@ -294,6 +297,11 @@ public class DefaultLwM2MUplinkMsgHandler implements LwM2mUplinkMsgHandler {
         });
     }
 
+    public void closeSession(SessionInfoProto sessionInfo) {
+        transportService.process(sessionInfo, DefaultTransportService.getSessionEventMsg(SessionEvent.CLOSED), null);
+        transportService.deregisterSession(sessionInfo);
+    }
+
     @Override
     public void onSleepingDev(Registration registration) {
         log.info("[{}] [{}] Received endpoint Sleeping version event", registration.getId(), registration.getEndpoint());
@@ -447,8 +455,7 @@ public class DefaultLwM2MUplinkMsgHandler implements LwM2mUplinkMsgHandler {
      */
     @Override
     public void doDisconnect(SessionInfoProto sessionInfo) {
-        transportService.process(sessionInfo, DefaultTransportService.getSessionEventMsg(SessionEvent.CLOSED), null);
-        transportService.deregisterSession(sessionInfo);
+        closeSession(sessionInfo);
     }
 
     /**
@@ -613,7 +620,7 @@ public class DefaultLwM2MUplinkMsgHandler implements LwM2mUplinkMsgHandler {
      * #3 If fr_update -> UpdateFirmware
      * #4 updateAttrTelemetry
      *
-     * @param lwM2MClient  - Registration LwM2M Client
+     * @param lwM2MClient   - Registration LwM2M Client
      * @param lwM2mResource - LwM2mSingleResource response.getContent()
      * @param path          - resource
      */
@@ -811,13 +818,6 @@ public class DefaultLwM2MUplinkMsgHandler implements LwM2mUplinkMsgHandler {
         return lwm2mResourceValue;
     }
 
-    /**
-     * Update resource (attribute) value  on thingsboard after update value in client
-     *
-     * @param registration -
-     * @param path         -
-     * @param request      -
-     */
     @Override
     public void onWriteResponseOk(LwM2mClient client, String path, WriteRequest request) {
         if (request.getNode() instanceof LwM2mResource) {
@@ -982,7 +982,7 @@ public class DefaultLwM2MUplinkMsgHandler implements LwM2mUplinkMsgHandler {
      * Get path to resource from profile equal keyName
      *
      * @param sessionInfo -
-     * @param keyName        -
+     * @param keyName     -
      * @return -
      */
     @Override