LWM2M: fix bug RPC device of line

nickAS21
1 parent dea35dc1
Showing 10 changed files with 326 additions and 33 deletions
application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
common/queue/src/main/proto/queue.proto
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/AbstractLwm2mTransportResource.java
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/DefaultLwM2MTransportMsgHandler.java
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/DefaultLwM2mTransportService.java
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportCoapResource.java
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportUtil.java
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClientContextImpl.java
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
--- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
View file @f0f08fd
+++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
View file @f0f08fd
@@ -192,6 +192,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
                     syncSessionSet.add(key);
                 }
             });
+            log.warn("46) Rpc syncSessionSet [{}] subscription after sent [{}]",syncSessionSet, rpcSubscriptions);
             syncSessionSet.forEach(rpcSubscriptions::remove);
         }
@@ -240,6 +241,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
             log.debug("[{}] Pushing {} pending RPC messages to new async session [{}]", deviceId, toDeviceRpcPendingMap.size(), sessionId);
             if (sessionType == SessionType.SYNC) {
                 log.debug("[{}] Cleanup sync rpc session [{}]", deviceId, sessionId);
+                log.warn("47) Rpc sessionId [{}] Cleanup sync rpc session subscription before [{}]",sessionId,  rpcSubscriptions);
                 rpcSubscriptions.remove(sessionId);
             }
         } else {
@@ -454,7 +456,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         } else {
             SessionInfoMetaData sessionMD = sessions.get(sessionId);
             if (sessionMD == null) {
-                sessionMD = new SessionInfoMetaData(new SessionInfo(SessionType.SYNC, sessionInfo.getNodeId()));
+                sessionMD = new SessionInfoMetaData(new SessionInfo(subscribeCmd.getSessionType(), sessionInfo.getNodeId()));
             }
             sessionMD.setSubscribedToAttributes(true);
             log.debug("[{}] Registering attributes subscription for session [{}]", deviceId, sessionId);
@@ -471,15 +473,17 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         UUID sessionId = getSessionId(sessionInfo);
         if (subscribeCmd.getUnsubscribe()) {
             log.debug("[{}] Canceling rpc subscription for session [{}]", deviceId, sessionId);
+            log.warn("48) Rpc sessionId [{}] Canceling rpc subscription for session subscription before [{}]",sessionId,  rpcSubscriptions);
             rpcSubscriptions.remove(sessionId);
         } else {
             SessionInfoMetaData sessionMD = sessions.get(sessionId);
             if (sessionMD == null) {
-                sessionMD = new SessionInfoMetaData(new SessionInfo(SessionType.SYNC, sessionInfo.getNodeId()));
+                sessionMD = new SessionInfoMetaData(new SessionInfo(subscribeCmd.getSessionType(), sessionInfo.getNodeId()));
             }
             sessionMD.setSubscribedToRPC(true);
             log.debug("[{}] Registering rpc subscription for session [{}]", deviceId, sessionId);
             rpcSubscriptions.put(sessionId, sessionMD.getSessionInfo());
+            log.warn("45) sessionId [{}] Registering rpc subscription for session [{}]",sessionId, rpcSubscriptions);
             sendPendingRequests(context, sessionId, sessionInfo);
             dumpSessions();
         }
@@ -509,6 +513,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
             log.debug("[{}] Canceling subscriptions for closed session [{}]", deviceId, sessionId);
             sessions.remove(sessionId);
             attributeSubscriptions.remove(sessionId);
+            log.warn("49) Rpc sessionId [{}] Canceling subscriptions for closed session subscription before [{}]",sessionId,  rpcSubscriptions);
             rpcSubscriptions.remove(sessionId);
             if (sessions.isEmpty()) {
                 reportSessionClose();
@@ -764,6 +769,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         Map<UUID, SessionInfoMetaData> sessionsToRemove = sessions.entrySet().stream().filter(kv -> kv.getValue().getLastActivityTime() < expTime).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
         sessionsToRemove.forEach((sessionId, sessionMD) -> {
             sessions.remove(sessionId);
+            log.warn("50) Rpc sessionId [{}] checkSessionsTimeout subscription before [{}]",sessionId,  rpcSubscriptions);
             rpcSubscriptions.remove(sessionId);
             attributeSubscriptions.remove(sessionId);
             notifyTransportAboutClosedSession(sessionId, sessionMD, "session timeout!");
--- a/common/queue/src/main/proto/queue.proto
View file @f0f08fd
+++ b/common/queue/src/main/proto/queue.proto
View file @f0f08fd
@@ -310,10 +310,12 @@ message SessionCloseNotificationProto {
 message SubscribeToAttributeUpdatesMsg {
   bool unsubscribe = 1;
+  SessionType sessionType = 2;
 }
 message SubscribeToRPCMsg {
   bool unsubscribe = 1;
+  SessionType sessionType = 2;
 }
 message ToDeviceRpcRequestMsg {
--- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/AbstractLwm2mTransportResource.java 0 → 100644
View file @f0f08fd
+++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/AbstractLwm2mTransportResource.java 0 → 100644
View file @f0f08fd
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.transport.lwm2m.server;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.californium.core.coap.CoAP;
+import org.eclipse.californium.core.coap.Response;
+import org.eclipse.californium.core.server.resources.CoapExchange;
+import org.eclipse.leshan.core.californium.LwM2mCoapResource;
+import org.thingsboard.server.common.transport.TransportServiceCallback;
+
+@Slf4j
+public abstract class AbstractLwm2mTransportResource extends LwM2mCoapResource {
+    protected final LwM2mTransportMsgHandler handler;
+
+    public AbstractLwm2mTransportResource(LwM2mTransportMsgHandler handler, String name) {
+        super(name);
+        this.handler = handler;
+    }
+
+    @Override
+    public void handleGET(CoapExchange exchange) {
+        processHandleGet(exchange);
+    }
+
+    @Override
+    public void handlePOST(CoapExchange exchange) {
+        processHandlePost(exchange);
+    }
+
+    protected abstract void processHandleGet(CoapExchange exchange);
+
+    protected abstract void processHandlePost(CoapExchange exchange);
+
+    public static class CoapOkCallback implements TransportServiceCallback<Void> {
+        private final CoapExchange exchange;
+        private final CoAP.ResponseCode onSuccessResponse;
+        private final CoAP.ResponseCode onFailureResponse;
+
+        public CoapOkCallback(CoapExchange exchange, CoAP.ResponseCode onSuccessResponse, CoAP.ResponseCode onFailureResponse) {
+            this.exchange = exchange;
+            this.onSuccessResponse = onSuccessResponse;
+            this.onFailureResponse = onFailureResponse;
+        }
+
+        @Override
+        public void onSuccess(Void msg) {
+            Response response = new Response(onSuccessResponse);
+            response.setAcknowledged(isConRequest());
+            exchange.respond(response);
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            exchange.respond(onFailureResponse);
+        }
+
+        private boolean isConRequest() {
+            return exchange.advanced().getRequest().isConfirmable();
+        }
+    }
+
+    public static class CoapNoOpCallback implements TransportServiceCallback<Void> {
+        private final CoapExchange exchange;
+
+        CoapNoOpCallback(CoapExchange exchange) {
+            this.exchange = exchange;
+        }
+
+        @Override
+        public void onSuccess(Void msg) {
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR);
+        }
+    }
+}
--- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/DefaultLwM2MTransportMsgHandler.java
View file @f0f08fd
+++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/DefaultLwM2MTransportMsgHandler.java
View file @f0f08fd
@@ -42,10 +42,10 @@ import org.thingsboard.common.util.ThingsBoardExecutors;
 import org.thingsboard.server.cache.ota.OtaPackageDataCache;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.DeviceProfile;
+import org.thingsboard.server.common.data.id.OtaPackageId;
 import org.thingsboard.server.common.data.ota.OtaPackageKey;
 import org.thingsboard.server.common.data.ota.OtaPackageType;
 import org.thingsboard.server.common.data.ota.OtaPackageUtil;
-import org.thingsboard.server.common.data.id.OtaPackageId;
 import org.thingsboard.server.common.transport.TransportService;
 import org.thingsboard.server.common.transport.TransportServiceCallback;
 import org.thingsboard.server.common.transport.adaptor.AdaptorException;
@@ -87,9 +87,9 @@ import java.util.stream.Collectors;
 import static org.eclipse.californium.core.coap.CoAP.ResponseCode.BAD_REQUEST;
 import static org.eclipse.leshan.core.attributes.Attribute.OBJECT_VERSION;
+import static org.thingsboard.server.common.data.lwm2m.LwM2mConstants.LWM2M_SEPARATOR_PATH;
 import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.DOWNLOADED;
 import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.UPDATING;
-import static org.thingsboard.server.common.data.lwm2m.LwM2mConstants.LWM2M_SEPARATOR_PATH;
 import static org.thingsboard.server.transport.lwm2m.server.LwM2mTransportServerHelper.getValueFromKvProto;
 import static org.thingsboard.server.transport.lwm2m.server.LwM2mTransportUtil.CLIENT_NOT_AUTHORIZED;
 import static org.thingsboard.server.transport.lwm2m.server.LwM2mTransportUtil.DEVICE_ATTRIBUTES_REQUEST;
@@ -189,13 +189,15 @@ public class DefaultLwM2MTransportMsgHandler implements LwM2mTransportMsgHandler
                     SessionInfoProto sessionInfo = this.getSessionInfoOrCloseSession(lwM2MClient);
                     if (sessionInfo != null) {
                         transportService.registerAsyncSession(sessionInfo, new LwM2mSessionMsgListener(this, sessionInfo));
-                        TransportProtos.TransportToDeviceActorMsg msg = TransportProtos.TransportToDeviceActorMsg.newBuilder()
+                        log.warn("40) sessionId [{}] Registering rpc subscription after Registration client",new UUID (sessionInfo.getSessionIdMSB(),sessionInfo.getSessionIdLSB()));
+                        transportService.process(TransportProtos.TransportToDeviceActorMsg.newBuilder()
                                 .setSessionInfo(sessionInfo)
                                 .setSessionEvent(DefaultTransportService.getSessionEventMsg(SessionEvent.OPEN))
-                                .setSubscribeToAttributes(TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().build())
-                                .setSubscribeToRPC(TransportProtos.SubscribeToRPCMsg.newBuilder().build())
-                                .build();
-                        transportService.process(msg, null);
+                                .setSubscribeToAttributes(TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder()
+                                        .setSessionType(TransportProtos.SessionType.ASYNC).build())
+                                .setSubscribeToRPC(TransportProtos.SubscribeToRPCMsg.newBuilder()
+                                        .setSessionType(TransportProtos.SessionType.ASYNC).build())
+                                .build(), null);
                         this.getInfoFirmwareUpdate(lwM2MClient, null);
                         this.getInfoSoftwareUpdate(lwM2MClient, null);
                         this.initLwM2mFromClientValue(registration, lwM2MClient);
@@ -250,7 +252,7 @@ public class DefaultLwM2MTransportMsgHandler implements LwM2mTransportMsgHandler
         unRegistrationExecutor.submit(() -> {
             try {
                 this.sendLogsToThingsboard(LOG_LW2M_INFO + ": Client unRegistration", registration.getId());
-                this.closeClientSession(registration);
+//                this.closeClientSession(registration);
             } catch (Throwable t) {
                 log.error("[{}] endpoint [{}] error Unable un registration.", registration.getEndpoint(), t);
                 this.sendLogsToThingsboard(LOG_LW2M_ERROR + String.format(": Client Unable un Registration, %s", t.getMessage()), registration.getId());
@@ -262,10 +264,11 @@ public class DefaultLwM2MTransportMsgHandler implements LwM2mTransportMsgHandler
         SessionInfoProto sessionInfo = this.getSessionInfoOrCloseSession(registration);
         if (sessionInfo != null) {
             transportService.deregisterSession(sessionInfo);
+            // TO DO may problem, better by registrationId
             sessionStore.remove(registration.getEndpoint());
             this.doCloseSession(sessionInfo);
             clientContext.removeClientByRegistrationId(registration.getId());
-            log.info("Client close session: [{}] unReg [{}] name  [{}] profile ", registration.getId(), registration.getEndpoint(), sessionInfo.getDeviceType());
+            log.warn("52) Client close session [{}]: [{}] unReg [{}] name  [{}] profile ", new UUID (sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB()), registration.getId(), registration.getEndpoint(), sessionInfo.getDeviceType());
         } else {
             log.error("Client close session: [{}] unReg [{}] name  [{}] sessionInfo ", registration.getId(), registration.getEndpoint(), null);
         }
@@ -446,10 +449,10 @@ public class DefaultLwM2MTransportMsgHandler implements LwM2mTransportMsgHandler
     public void onToDeviceRpcRequest(TransportProtos.ToDeviceRpcRequestMsg toDeviceRpcRequestMsg, SessionInfoProto sessionInfo) {
         // #1
         this.checkRpcRequestTimeout();
-        log.warn ("4) toDeviceRpcRequestMsg: [{}], sessionUUID: [{}]", toDeviceRpcRequestMsg,  new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB()));
+        UUID requestUUID = new UUID(toDeviceRpcRequestMsg.getRequestIdMSB(), toDeviceRpcRequestMsg.getRequestIdLSB());
+        log.warn ("4) toDeviceRpcRequestMsg: [{}], sessionUUID: [{}]", toDeviceRpcRequestMsg,  requestUUID);
         String bodyParams = StringUtils.trimToNull(toDeviceRpcRequestMsg.getParams()) != null ? toDeviceRpcRequestMsg.getParams() : "null";
         LwM2mTypeOper lwM2mTypeOper = setValidTypeOper(toDeviceRpcRequestMsg.getMethodName());
-        UUID requestUUID = new UUID(toDeviceRpcRequestMsg.getRequestIdMSB(), toDeviceRpcRequestMsg.getRequestIdLSB());
         if (!this.rpcSubscriptions.containsKey(requestUUID)) {
             this.rpcSubscriptions.put(requestUUID, toDeviceRpcRequestMsg.getExpirationTime());
             Lwm2mClientRpcRequest lwm2mClientRpcRequest = null;
@@ -479,8 +482,14 @@ public class DefaultLwM2MTransportMsgHandler implements LwM2mTransportMsgHandler
     }
     private void checkRpcRequestTimeout() {
-        Set<UUID> rpcSubscriptionsToRemove = rpcSubscriptions.entrySet().stream().filter(kv -> System.currentTimeMillis() > kv.getValue()).map(Map.Entry::getKey).collect(Collectors.toSet());
-        rpcSubscriptionsToRemove.forEach(rpcSubscriptions::remove);
+        log.warn ("4.1) before rpcSubscriptions.size(): [{}]", rpcSubscriptions.size());
+        if (rpcSubscriptions.size() > 0) {
+            Set<UUID> rpcSubscriptionsToRemove = rpcSubscriptions.entrySet().stream().filter(kv -> System.currentTimeMillis() > kv.getValue()).map(Map.Entry::getKey).collect(Collectors.toSet());
+            log.warn ("4.2) System.currentTimeMillis(): [{}]", System.currentTimeMillis());
+            log.warn ("4.3) rpcSubscriptionsToRemove: [{}]", rpcSubscriptionsToRemove);
+            rpcSubscriptionsToRemove.forEach(rpcSubscriptions::remove);
+        }
+        log.warn ("4.4) after rpcSubscriptions.size(): [{}]", rpcSubscriptions.size());
     }
     public void sentRpcResponse(Lwm2mClientRpcRequest rpcRequest, String requestCode, String msg, String typeMsg) {
@@ -1372,7 +1381,8 @@ public class DefaultLwM2MTransportMsgHandler implements LwM2mTransportMsgHandler
                                     lwM2MClient.getFwUpdate().setRpcRequest(rpcRequest);
                                     lwM2MClient.getFwUpdate().setCurrentVersion(response.getVersion());
                                     lwM2MClient.getFwUpdate().setCurrentTitle(response.getTitle());
-                                    lwM2MClient.getFwUpdate().setCurrentId(new OtaPackageId(new UUID(response.getOtaPackageIdMSB(), response.getOtaPackageIdLSB())).getId());
+                                    log.warn("11) OtaPackageIdMSB: [{}] OtaPackageIdLSB: [{}]", response.getOtaPackageIdMSB(), response.getOtaPackageIdLSB());
+                                    lwM2MClient.getFwUpdate().setCurrentId(new UUID(response.getOtaPackageIdMSB(), response.getOtaPackageIdLSB()));
                                     if (rpcRequest == null) {
                                         lwM2MClient.getFwUpdate().sendReadObserveInfo(lwM2mTransportRequest);
                                     }
--- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/DefaultLwM2mTransportService.java
View file @f0f08fd
+++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/DefaultLwM2mTransportService.java
View file @f0f08fd
@@ -28,7 +28,6 @@ import org.eclipse.leshan.server.californium.registration.CaliforniumRegistratio
 import org.eclipse.leshan.server.model.LwM2mModelProvider;
 import org.eclipse.leshan.server.security.EditableSecurityStore;
 import org.springframework.stereotype.Component;
-import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.queue.util.TbLwM2mTransportComponent;
 import org.thingsboard.server.transport.lwm2m.config.LwM2MTransportServerConfig;
@@ -58,14 +57,14 @@ import java.security.spec.InvalidParameterSpecException;
 import java.security.spec.KeySpec;
 import java.security.spec.PKCS8EncodedKeySpec;
 import java.util.Arrays;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
 import static org.eclipse.californium.scandium.dtls.cipher.CipherSuite.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256;
 import static org.eclipse.californium.scandium.dtls.cipher.CipherSuite.TLS_ECDHE_ECDSA_WITH_AES_128_CCM_8;
 import static org.eclipse.californium.scandium.dtls.cipher.CipherSuite.TLS_PSK_WITH_AES_128_CBC_SHA256;
 import static org.eclipse.californium.scandium.dtls.cipher.CipherSuite.TLS_PSK_WITH_AES_128_CCM_8;
 import static org.thingsboard.server.transport.lwm2m.server.LwM2mNetworkConfig.getCoapConfig;
+import static org.thingsboard.server.transport.lwm2m.server.LwM2mTransportUtil.FW_COAP_RESOURCE;
+import static org.thingsboard.server.transport.lwm2m.server.LwM2mTransportUtil.SW_COAP_RESOURCE;
 @Slf4j
 @Component
@@ -81,7 +80,7 @@ public class DefaultLwM2mTransportService implements LwM2MTransportService {
     private final LwM2mTransportContext context;
     private final LwM2MTransportServerConfig config;
     private final LwM2mTransportServerHelper helper;
-    private final LwM2mTransportMsgHandler handler;
+    private final DefaultLwM2MTransportMsgHandler handler;
     private final CaliforniumRegistrationStore registrationStore;
     private final EditableSecurityStore securityStore;
     private final LwM2mClientContext lwM2mClientContext;
@@ -96,6 +95,19 @@ public class DefaultLwM2mTransportService implements LwM2MTransportService {
             new LWM2MGenerationPSkRPkECC();
         }
         this.server = getLhServer();
+        /**
+         * Add a resource to the server.
+         * CoapResource ->
+         * path = FW_PACKAGE or SW_PACKAGE
+         * nameFile = "BC68JAR01A09_TO_BC68JAR01A10.bin"
+         * "coap://host:port/{path}/{token}/{nameFile}"
+         */
+
+
+        LwM2mTransportCoapResource fwCoapResource = new LwM2mTransportCoapResource(handler, FW_COAP_RESOURCE);
+        LwM2mTransportCoapResource swCoapResource = new LwM2mTransportCoapResource(handler, SW_COAP_RESOURCE);
+        this.server.coap().getServer().add(fwCoapResource);
+        this.server.coap().getServer().add(swCoapResource);
         this.startLhServer();
         this.context.setServer(server);
     }
--- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportCoapResource.java 0 → 100644
View file @f0f08fd
+++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportCoapResource.java 0 → 100644
View file @f0f08fd
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.transport.lwm2m.server;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.californium.core.coap.CoAP;
+import org.eclipse.californium.core.coap.Request;
+import org.eclipse.californium.core.coap.Response;
+import org.eclipse.californium.core.network.Exchange;
+import org.eclipse.californium.core.observe.ObserveRelation;
+import org.eclipse.californium.core.server.resources.CoapExchange;
+import org.eclipse.californium.core.server.resources.Resource;
+import org.eclipse.californium.core.server.resources.ResourceObserver;
+
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+@Slf4j
+public class LwM2mTransportCoapResource extends AbstractLwm2mTransportResource {
+    private final ConcurrentMap<String, ObserveRelation> tokenToObserveRelationMap = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, AtomicInteger> tokenToObserveNotificationSeqMap = new ConcurrentHashMap<>();
+
+    public LwM2mTransportCoapResource(LwM2mTransportMsgHandler handler, String name) {
+        super(handler, name);
+        this.setObservable(true); // enable observing
+        this.addObserver(new CoapResourceObserver());
+    }
+
+
+    @Override
+    public void checkObserveRelation(Exchange exchange, Response response) {
+        String token = getTokenFromRequest(exchange.getRequest());
+        final ObserveRelation relation = exchange.getRelation();
+        if (relation == null || relation.isCanceled()) {
+            return; // because request did not try to establish a relation
+        }
+        if (CoAP.ResponseCode.isSuccess(response.getCode())) {
+
+            if (!relation.isEstablished()) {
+                relation.setEstablished();
+                addObserveRelation(relation);
+            }
+            AtomicInteger notificationCounter = tokenToObserveNotificationSeqMap.computeIfAbsent(token, s -> new AtomicInteger(0));
+            response.getOptions().setObserve(notificationCounter.getAndIncrement());
+        } // ObserveLayer takes care of the else case
+    }
+
+
+    @Override
+    protected void processHandleGet(CoapExchange exchange) {
+        log.warn("1) processHandleGet [{}]", exchange);
+//        exchange.respond(CoAP.ResponseCode.BAD_REQUEST);
+//        int ver = 10;
+        int ver = 9;
+        UUID currentId;
+        if (ver == 10) {
+            long mSB = 4951557297924280811L;
+            long lSb = -8451242882176289074L;
+            currentId = new UUID(mSB, lSb);
+        } else {
+            long mSB = 9085827945869414891L;
+            long lSb = -9086716326447629319L;
+            currentId = new UUID(mSB, lSb);
+        }
+        String resource = exchange.getRequestOptions().getUriPath().get(0);
+        String token = exchange.getRequestOptions().getUriPath().get(1);
+        exchange.respond(CoAP.ResponseCode.CONTENT, this.getFwData(currentId));
+
+    }
+
+    @Override
+    protected void processHandlePost(CoapExchange exchange) {
+        log.warn("2) processHandleGet [{}]", exchange);
+    }
+
+    /**
+     * Override the default behavior so that requests to sub resources (typically /{name}/{token}) are handled by
+     * /name resource.
+     */
+    @Override
+    public Resource getChild(String name) {
+        return this;
+    }
+
+
+    private String getTokenFromRequest(Request request) {
+        return (request.getSourceContext() != null ? request.getSourceContext().getPeerAddress().getAddress().getHostAddress() : "null")
+                + ":" + (request.getSourceContext() != null ? request.getSourceContext().getPeerAddress().getPort() : -1) + ":" + request.getTokenString();
+    }
+
+    public class CoapResourceObserver implements ResourceObserver {
+
+        @Override
+        public void changedName(String old) {
+
+        }
+
+        @Override
+        public void changedPath(String old) {
+
+        }
+
+        @Override
+        public void addedChild(Resource child) {
+
+        }
+
+        @Override
+        public void removedChild(Resource child) {
+
+        }
+
+        @Override
+        public void addedObserveRelation(ObserveRelation relation) {
+
+        }
+
+        @Override
+        public void removedObserveRelation(ObserveRelation relation) {
+
+        }
+    }
+
+    private byte[] getFwData(UUID currentId) {
+        int chunkSize = 0;
+        int chunk = 0;
+        return ((DefaultLwM2MTransportMsgHandler) handler).otaPackageDataCache.get(currentId.toString(), chunkSize, chunk);
+    }
+
+}
--- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportUtil.java
View file @f0f08fd
+++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportUtil.java
View file @f0f08fd
@@ -43,11 +43,11 @@ import org.eclipse.leshan.server.registration.Registration;
 import org.nustaq.serialization.FSTConfiguration;
 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.data.ota.OtaPackageKey;
 import org.thingsboard.server.common.data.ota.OtaPackageType;
 import org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus;
 import org.thingsboard.server.common.data.ota.OtaPackageUtil;
-import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.transport.TransportServiceCallback;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClientProfile;
@@ -77,14 +77,14 @@ import static org.eclipse.leshan.core.model.ResourceModel.Type.OBJLNK;
 import static org.eclipse.leshan.core.model.ResourceModel.Type.OPAQUE;
 import static org.eclipse.leshan.core.model.ResourceModel.Type.STRING;
 import static org.eclipse.leshan.core.model.ResourceModel.Type.TIME;
+import static org.thingsboard.server.common.data.lwm2m.LwM2mConstants.LWM2M_SEPARATOR_KEY;
+import static org.thingsboard.server.common.data.lwm2m.LwM2mConstants.LWM2M_SEPARATOR_PATH;
 import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.DOWNLOADED;
 import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.DOWNLOADING;
 import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.FAILED;
 import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.UPDATED;
 import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.UPDATING;
 import static org.thingsboard.server.common.data.ota.OtaPackageUpdateStatus.VERIFIED;
-import static org.thingsboard.server.common.data.lwm2m.LwM2mConstants.LWM2M_SEPARATOR_KEY;
-import static org.thingsboard.server.common.data.lwm2m.LwM2mConstants.LWM2M_SEPARATOR_PATH;
 @Slf4j
 public class LwM2mTransportUtil {
@@ -140,6 +140,7 @@ public class LwM2mTransportUtil {
     public static final String METHOD_KEY = "methodName";
     // Firmware
+    public static final String FW_COAP_RESOURCE = "fw";
     public static final String FW_UPDATE = "Firmware update";
     public static final Integer FW_ID = 5;
     // Package W
@@ -156,6 +157,7 @@ public class LwM2mTransportUtil {
     public static final String FW_UPDATE_ID = "/5/0/2";
     // Software
+    public static final String SW_COAP_RESOURCE = "sw";
     public static final String SW_UPDATE = "Software update";
     public static final Integer SW_ID = 9;
     // Package W
--- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClientContextImpl.java
View file @f0f08fd
+++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClientContextImpl.java
View file @f0f08fd
@@ -37,6 +37,7 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Predicate;
 import static org.eclipse.leshan.core.SecurityMode.NO_SEC;
 import static org.thingsboard.server.transport.lwm2m.server.LwM2mTransportUtil.convertPathFromObjectIdToIdVer;
@@ -68,12 +69,15 @@ public class LwM2mClientContextImpl implements LwM2mClientContext {
     @Override
     public LwM2mClient getOrRegister(Registration registration) {
+        LwM2mClient client = null;
         if (registration == null) {
-            return null;
+            return client;
+        }
+        if (lwM2mClientsByRegistrationId.size()>0) {
+            client = this.lwM2mClientsByRegistrationId.get(registration.getId());
         }
-        LwM2mClient client = lwM2mClientsByRegistrationId.get(registration.getId());
         if (client == null) {
-            client = lwM2mClientsByEndpoint.get(registration.getEndpoint());
+            client = this.lwM2mClientsByEndpoint.get(registration.getEndpoint());
             if (client == null) {
                 client = registerOrUpdate(registration);
             }
@@ -83,12 +87,29 @@ public class LwM2mClientContextImpl implements LwM2mClientContext {
     @Override
     public LwM2mClient getClient(TransportProtos.SessionInfoProto sessionInfo) {
-        LwM2mClient lwM2mClient =  lwM2mClientsByEndpoint.values().stream().filter(c ->
+        LwM2mClient lwM2mClient = null;
+        Predicate<LwM2mClient> isClientFilter = c ->
                 (new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB()))
-                        .equals((new UUID(c.getSession().getSessionIdMSB(), c.getSession().getSessionIdLSB())))
-
-        ).findAny().get();
-        if (lwM2mClient == null) {
+                        .equals((new UUID(c.getSession().getSessionIdMSB(), c.getSession().getSessionIdLSB())));
+//        if (this.lwM2mClientsByEndpoint.size()>0) {
+//            lwM2mClient = this.lwM2mClientsByEndpoint.values().stream().filter(c ->
+//                    (new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB()))
+//                            .equals((new UUID(c.getSession().getSessionIdMSB(), c.getSession().getSessionIdLSB())))
+//            ).findAny().get();
+//        }
+        if (this.lwM2mClientsByEndpoint.size()>0) {
+            lwM2mClient = this.lwM2mClientsByEndpoint.values().stream().filter(isClientFilter).findAny().get();
+        }
+//        if (lwM2mClient == null && this.lwM2mClientsByRegistrationId.size() > 0) {
+//            lwM2mClient = this.lwM2mClientsByRegistrationId.values().stream().filter(c ->
+//                    (new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB()))
+//                            .equals((new UUID(c.getSession().getSessionIdMSB(), c.getSession().getSessionIdLSB())))
+//            ).findAny().get();
+//        }
+        if (lwM2mClient == null && this.lwM2mClientsByRegistrationId.size() > 0) {
+            lwM2mClient = this.lwM2mClientsByRegistrationId.values().stream().filter(isClientFilter).findAny().get();
+        }
+        if (lwM2mClient == null){
             log.warn("Device TimeOut? lwM2mClient is null.");
             log.warn("SessionInfo input [{}], lwM2mClientsByEndpoint size: [{}]", sessionInfo, lwM2mClientsByEndpoint.values().size());
             log.error("", new RuntimeException());
--- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java
View file @f0f08fd
+++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java
View file @f0f08fd
@@ -259,8 +259,10 @@ public class GatewaySessionHandler {
                                     transportService.process(TransportProtos.TransportToDeviceActorMsg.newBuilder()
                                             .setSessionInfo(deviceSessionInfo)
                                             .setSessionEvent(DefaultTransportService.getSessionEventMsg(TransportProtos.SessionEvent.OPEN))
-                                            .setSubscribeToAttributes(TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().build())
-                                            .setSubscribeToRPC(TransportProtos.SubscribeToRPCMsg.newBuilder().build())
+                                            .setSubscribeToAttributes(TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder()
+                                                    .setSessionType(TransportProtos.SessionType.ASYNC).build())
+                                            .setSubscribeToRPC(TransportProtos.SubscribeToRPCMsg.newBuilder()
+                                                    .setSessionType(TransportProtos.SessionType.ASYNC).build())
                                             .build(), null);
                                 }
                                 futureToSet.set(devices.get(deviceName));
--- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
View file @f0f08fd
+++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
View file @f0f08fd
@@ -708,6 +708,7 @@ public class DefaultTransportService implements TransportService {
             log.debug("Stopping scheduler to avoid resending response if request has been ack.");
             currentSession.getScheduledFuture().cancel(false);
         }
+        log.warn("54) session [{}] deregisterSession Stopping scheduler to avoid resending response if request has been ack.", toSessionId(sessionInfo));
         sessions.remove(toSessionId(sessionInfo));
     }