Merge branch 'master' of github.com:thingsboard/thingsboard

@@ -233,7 +233,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
         rpc.setExpirationTime(request.getExpirationTime());
         rpc.setRequest(JacksonUtil.valueToTree(request));
         rpc.setStatus(status);
-        systemContext.getTbRpcService().save(tenantId, rpc);
         return systemContext.getTbRpcService().save(tenantId, rpc);
     }
 

@@ -31,6 +31,7 @@ import org.thingsboard.server.common.data.TransportPayloadType;
 import org.thingsboard.server.common.msg.session.FeatureType;
 import org.thingsboard.server.transport.coap.AbstractCoapIntegrationTest;
 
+import java.nio.charset.StandardCharsets;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 

@@ -15,11 +15,14 @@
  */
 package org.thingsboard.server.transport.coap;
 
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.californium.core.CoapResource;
 import org.eclipse.californium.core.coap.CoAP;
+import org.eclipse.californium.core.coap.MessageObserver;
 import org.eclipse.californium.core.coap.Response;
 import org.eclipse.californium.core.server.resources.CoapExchange;
+import org.eclipse.californium.elements.EndpointContext;
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.transport.TransportContext;
 import org.thingsboard.server.common.transport.TransportService;
@@ -29,8 +32,12 @@ import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsRes
 import org.thingsboard.server.gen.transport.TransportProtos;
 
 import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.function.BiConsumer;
 
+import static org.eclipse.californium.core.coap.Message.MAX_MID;
+import static org.eclipse.californium.core.coap.Message.NONE;
+
 @Slf4j
 public abstract class AbstractCoapTransportResource extends CoapResource {
 
@@ -75,77 +82,8 @@ public abstract class AbstractCoapTransportResource extends CoapResource {
                 .setEvent(event).build();
     }
 
-    public static class CoapDeviceAuthCallback implements TransportServiceCallback<ValidateDeviceCredentialsResponse> {
-        private final TransportContext transportContext;
-        private final CoapExchange exchange;
-        private final BiConsumer<TransportProtos.SessionInfoProto, DeviceProfile> onSuccess;
-
-        public CoapDeviceAuthCallback(TransportContext transportContext, CoapExchange exchange, BiConsumer<TransportProtos.SessionInfoProto, DeviceProfile> onSuccess) {
-            this.transportContext = transportContext;
-            this.exchange = exchange;
-            this.onSuccess = onSuccess;
-        }
-
-        @Override
-        public void onSuccess(ValidateDeviceCredentialsResponse msg) {
-            DeviceProfile deviceProfile = msg.getDeviceProfile();
-            if (msg.hasDeviceInfo() && deviceProfile != null) {
-                TransportProtos.SessionInfoProto sessionInfoProto = SessionInfoCreator.create(msg, transportContext, UUID.randomUUID());
-                onSuccess.accept(sessionInfoProto, deviceProfile);
-            } else {
-                exchange.respond(CoAP.ResponseCode.UNAUTHORIZED);
-            }
-        }
-
-        @Override
-        public void onError(Throwable e) {
-            log.warn("Failed to process request", e);
-            exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR);
-        }
-    }
-
-    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();
-        }
+    protected int getNextMsgId() {
+        return ThreadLocalRandom.current().nextInt(NONE, MAX_MID + 1);
     }
 
-    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);
-        }
-    }
 }

@@ -22,10 +22,14 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Component;
 import org.thingsboard.server.common.transport.TransportContext;
+import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.transport.coap.adaptors.JsonCoapAdaptor;
 import org.thingsboard.server.transport.coap.adaptors.ProtoCoapAdaptor;
 import org.thingsboard.server.transport.coap.efento.adaptor.EfentoCoapAdaptor;
 
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
 
 /**
  * Created by ashvayka on 18.10.18.
@@ -33,22 +37,21 @@ import org.thingsboard.server.transport.coap.efento.adaptor.EfentoCoapAdaptor;
 @Slf4j
 @ConditionalOnExpression("'${service.type:null}'=='tb-transport' || ('${service.type:null}'=='monolith' && '${transport.api_enabled:true}'=='true' && '${transport.coap.enabled}'=='true')")
 @Component
+@Getter
 public class CoapTransportContext extends TransportContext {
 
-    @Getter
     @Value("${transport.sessions.report_timeout}")
     private long sessionReportTimeout;
 
-    @Getter
     @Autowired
     private JsonCoapAdaptor jsonCoapAdaptor;
 
-    @Getter
     @Autowired
     private ProtoCoapAdaptor protoCoapAdaptor;
 
-    @Getter
     @Autowired
     private EfentoCoapAdaptor efentoCoapAdaptor;
 
+    private final ConcurrentMap<Integer, TransportProtos.ToDeviceRpcRequestMsg> rpcAwaitingAck = new ConcurrentHashMap<>();
+
 }

@@ -19,6 +19,7 @@ import com.google.gson.JsonParseException;
 import com.google.protobuf.Descriptors;
 import com.google.protobuf.DynamicMessage;
 import lombok.Data;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.californium.core.coap.CoAP;
 import org.eclipse.californium.core.coap.Request;
@@ -53,6 +54,9 @@ import org.thingsboard.server.common.transport.adaptor.AdaptorException;
 import org.thingsboard.server.common.transport.adaptor.JsonConverter;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.transport.coap.adaptors.CoapTransportAdaptor;
+import org.thingsboard.server.transport.coap.callback.CoapDeviceAuthCallback;
+import org.thingsboard.server.transport.coap.callback.CoapNoOpCallback;
+import org.thingsboard.server.transport.coap.callback.CoapOkCallback;
 
 import java.util.List;
 import java.util.Map;
@@ -81,9 +85,8 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
     private final Set<UUID> rpcSubscriptions = ConcurrentHashMap.newKeySet();
     private final Set<UUID> attributeSubscriptions = ConcurrentHashMap.newKeySet();
 
-    private ConcurrentMap<String, TbCoapDtlsSessionInfo> dtlsSessionIdMap;
-    private long timeout;
-    private long sessionReportTimeout;
+    private final ConcurrentMap<String, TbCoapDtlsSessionInfo> dtlsSessionIdMap;
+    private final long timeout;
 
     public CoapTransportResource(CoapTransportContext ctx, CoapServerService coapServerService, String name) {
         super(ctx, name);
@@ -91,7 +94,7 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
         this.addObserver(new CoapResourceObserver());
         this.dtlsSessionIdMap = coapServerService.getDtlsSessionsMap();
         this.timeout = coapServerService.getTimeout();
-        this.sessionReportTimeout = ctx.getSessionReportTimeout();
+        long sessionReportTimeout = ctx.getSessionReportTimeout();
         ctx.getScheduler().scheduleAtFixedRate(() -> {
             Set<CoapObserveSessionInfo> coapObserveSessionInfos = sessionInfoToObserveRelationMap.keySet();
             Set<TransportProtos.SessionInfoProto> observeSessions = coapObserveSessionInfos
@@ -110,7 +113,6 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
             return; // because request did not try to establish a relation
         }
         if (CoAP.ResponseCode.isSuccess(response.getCode())) {
-
             if (!relation.isEstablished()) {
                 relation.setEstablished();
                 addObserveRelation(relation);
@@ -280,8 +282,8 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
                     CoapObserveSessionInfo currentCoapObserveAttrSessionInfo = tokenToCoapSessionInfoMap.get(getTokenFromRequest(request));
                     if (currentCoapObserveAttrSessionInfo == null) {
                         attributeSubscriptions.add(sessionId);
-                        registerAsyncCoapSession(exchange, sessionInfo, coapTransportAdaptor,
-                                transportConfigurationContainer.getRpcRequestDynamicMessageBuilder(), getTokenFromRequest(request));
+                        registerAsyncCoapSession(exchange, coapTransportAdaptor, transportConfigurationContainer.getRpcRequestDynamicMessageBuilder(),
+                                sessionInfo, getTokenFromRequest(request));
                         transportService.process(sessionInfo,
                                 TransportProtos.SubscribeToAttributeUpdatesMsg.getDefaultInstance(), new CoapNoOpCallback(exchange));
                         transportService.process(sessionInfo,
@@ -305,11 +307,11 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
                     CoapObserveSessionInfo currentCoapObserveRpcSessionInfo = tokenToCoapSessionInfoMap.get(getTokenFromRequest(request));
                     if (currentCoapObserveRpcSessionInfo == null) {
                         rpcSubscriptions.add(sessionId);
-                        registerAsyncCoapSession(exchange, sessionInfo, coapTransportAdaptor,
-                                transportConfigurationContainer.getRpcRequestDynamicMessageBuilder(), getTokenFromRequest(request));
+                        registerAsyncCoapSession(exchange, coapTransportAdaptor, transportConfigurationContainer.getRpcRequestDynamicMessageBuilder()
+                                , sessionInfo, getTokenFromRequest(request));
                         transportService.process(sessionInfo,
                                 TransportProtos.SubscribeToRPCMsg.getDefaultInstance(),
-                                new CoapOkCallback(exchange, CoAP.ResponseCode.VALID, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)
+                                 new CoapOkCallback(exchange, CoAP.ResponseCode.VALID, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)
                         );
                     }
                     break;
@@ -359,14 +361,16 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
         return tokenToCoapSessionInfoMap.remove(token);
     }
 
-    private void registerAsyncCoapSession(CoapExchange exchange, TransportProtos.SessionInfoProto sessionInfo, CoapTransportAdaptor coapTransportAdaptor, DynamicMessage.Builder rpcRequestDynamicMessageBuilder, String token) {
+    private void registerAsyncCoapSession(CoapExchange exchange, CoapTransportAdaptor coapTransportAdaptor,
+                                          DynamicMessage.Builder rpcRequestDynamicMessageBuilder, TransportProtos.SessionInfoProto sessionInfo, String token) {
         tokenToCoapSessionInfoMap.putIfAbsent(token, new CoapObserveSessionInfo(sessionInfo));
         transportService.registerAsyncSession(sessionInfo, getCoapSessionListener(exchange, coapTransportAdaptor, rpcRequestDynamicMessageBuilder, sessionInfo));
         transportService.process(sessionInfo, getSessionEventMsg(TransportProtos.SessionEvent.OPEN), null);
     }
 
-    private CoapSessionListener getCoapSessionListener(CoapExchange exchange, CoapTransportAdaptor coapTransportAdaptor, DynamicMessage.Builder rpcRequestDynamicMessageBuilder, TransportProtos.SessionInfoProto sessionInfo) {
-        return new CoapSessionListener(this, exchange, coapTransportAdaptor, rpcRequestDynamicMessageBuilder, sessionInfo);
+    private CoapSessionListener getCoapSessionListener(CoapExchange exchange, CoapTransportAdaptor coapTransportAdaptor,
+                                                       DynamicMessage.Builder rpcRequestDynamicMessageBuilder, TransportProtos.SessionInfoProto sessionInfo) {
+        return new CoapSessionListener(exchange, coapTransportAdaptor, rpcRequestDynamicMessageBuilder, sessionInfo);
     }
 
     private String getTokenFromRequest(Request request) {
@@ -448,22 +452,14 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
         }
     }
 
-    private static class CoapSessionListener implements SessionMsgListener {
+    @RequiredArgsConstructor
+    private class CoapSessionListener implements SessionMsgListener {
 
-        private final CoapTransportResource coapTransportResource;
         private final CoapExchange exchange;
         private final CoapTransportAdaptor coapTransportAdaptor;
         private final DynamicMessage.Builder rpcRequestDynamicMessageBuilder;
         private final TransportProtos.SessionInfoProto sessionInfo;
 
-        CoapSessionListener(CoapTransportResource coapTransportResource, CoapExchange exchange, CoapTransportAdaptor coapTransportAdaptor, DynamicMessage.Builder rpcRequestDynamicMessageBuilder, TransportProtos.SessionInfoProto sessionInfo) {
-            this.coapTransportResource = coapTransportResource;
-            this.exchange = exchange;
-            this.coapTransportAdaptor = coapTransportAdaptor;
-            this.rpcRequestDynamicMessageBuilder = rpcRequestDynamicMessageBuilder;
-            this.sessionInfo = sessionInfo;
-        }
-
         @Override
         public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg msg) {
             try {
@@ -496,18 +492,30 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
         @Override
         public void onToDeviceRpcRequest(UUID sessionId, TransportProtos.ToDeviceRpcRequestMsg msg) {
             log.trace("[{}] Received RPC command to device", sessionId);
-            boolean successful = true;
             try {
-                exchange.respond(coapTransportAdaptor.convertToPublish(isConRequest(), msg, rpcRequestDynamicMessageBuilder));
+                Response response = coapTransportAdaptor.convertToPublish(isConRequest(), msg, rpcRequestDynamicMessageBuilder);
+                int requestId = getNextMsgId();
+                response.setMID(requestId);
+                if (msg.getPersisted()) {
+                    transportContext.getRpcAwaitingAck().put(requestId, msg);
+                    transportContext.getScheduler().schedule(() -> {
+                        TransportProtos.ToDeviceRpcRequestMsg awaitingAckMsg = transportContext.getRpcAwaitingAck().remove(requestId);
+                        if (awaitingAckMsg != null) {
+                            transportService.process(sessionInfo, msg, true, TransportServiceCallback.EMPTY);
+                        }
+                    }, Math.max(0, msg.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+                }
+                response.addMessageObserver(new TbCoapMessageObserver(requestId, id -> {
+                    TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(id);
+                    if (rpcRequestMsg != null) {
+                        transportService.process(sessionInfo, rpcRequestMsg, false, TransportServiceCallback.EMPTY);
+                    }
+                }));
+                exchange.respond(response);
             } catch (AdaptorException e) {
                 log.trace("Failed to reply due to error", e);
                 closeObserveRelationAndNotify(sessionId, CoAP.ResponseCode.INTERNAL_SERVER_ERROR);
-                successful = false;
-            } finally {
-                coapTransportResource.transportService.process(sessionInfo, msg, !successful, TransportServiceCallback.EMPTY);
-                if (!successful) {
-                    closeAndDeregister();
-                }
+                closeAndDeregister();
             }
         }
 
@@ -526,8 +534,8 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
         }
 
         private void closeObserveRelationAndNotify(UUID sessionId, CoAP.ResponseCode responseCode) {
-            Map<CoapObserveSessionInfo, ObserveRelation> sessionToObserveRelationMap = coapTransportResource.getCoapSessionInfoToObserveRelationMap();
-            if (coapTransportResource.getObserverCount() > 0 && !CollectionUtils.isEmpty(sessionToObserveRelationMap)) {
+            Map<CoapObserveSessionInfo, ObserveRelation> sessionToObserveRelationMap = CoapTransportResource.this.getCoapSessionInfoToObserveRelationMap();
+            if (CoapTransportResource.this.getObserverCount() > 0 && !CollectionUtils.isEmpty(sessionToObserveRelationMap)) {
                 Optional<CoapObserveSessionInfo> observeSessionToClose = sessionToObserveRelationMap.keySet().stream().filter(coapObserveSessionInfo -> {
                     TransportProtos.SessionInfoProto sessionToDelete = coapObserveSessionInfo.getSessionInfoProto();
                     UUID observeSessionId = new UUID(sessionToDelete.getSessionIdMSB(), sessionToDelete.getSessionIdLSB());
@@ -536,16 +544,16 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
                 if (observeSessionToClose.isPresent()) {
                     CoapObserveSessionInfo coapObserveSessionInfo = observeSessionToClose.get();
                     ObserveRelation observeRelation = sessionToObserveRelationMap.get(coapObserveSessionInfo);
-                    coapTransportResource.clearAndNotifyObserveRelation(observeRelation, responseCode);
+                    CoapTransportResource.this.clearAndNotifyObserveRelation(observeRelation, responseCode);
                 }
             }
         }
 
         private void closeAndDeregister() {
             Request request = exchange.advanced().getRequest();
-            String token = coapTransportResource.getTokenFromRequest(request);
-            CoapObserveSessionInfo deleted = coapTransportResource.lookupAsyncSessionInfo(token);
-            coapTransportResource.closeAndDeregister(deleted.getSessionInfoProto());
+            String token = CoapTransportResource.this.getTokenFromRequest(request);
+            CoapObserveSessionInfo deleted = CoapTransportResource.this.lookupAsyncSessionInfo(token);
+            CoapTransportResource.this.closeAndDeregister(deleted.getSessionInfoProto());
         }
 
     }

@@ -20,22 +20,19 @@ 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 org.thingsboard.common.util.ThingsBoardExecutors;
 import org.thingsboard.server.common.data.DeviceTransportType;
 import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.server.common.data.ota.OtaPackageType;
 import org.thingsboard.server.common.data.security.DeviceTokenCredentials;
 import org.thingsboard.server.common.transport.TransportServiceCallback;
 import org.thingsboard.server.gen.transport.TransportProtos;
+import org.thingsboard.server.transport.coap.callback.CoapDeviceAuthCallback;
 
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
-import java.util.concurrent.ExecutorService;
 
 @Slf4j
 public class OtaPackageTransportResource extends AbstractCoapTransportResource {

+/**
+ * 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.coap;
+
+import lombok.RequiredArgsConstructor;
+import org.eclipse.californium.core.coap.MessageObserver;
+import org.eclipse.californium.core.coap.Response;
+import org.eclipse.californium.elements.EndpointContext;
+
+import java.util.function.Consumer;
+
+@RequiredArgsConstructor
+public class TbCoapMessageObserver implements MessageObserver {
+
+    private final int msgId;
+    private final Consumer<Integer> onAcknowledge;
+
+    @Override
+    public void onRetransmission() {
+
+    }
+
+    @Override
+    public void onResponse(Response response) {
+
+    }
+
+    @Override
+    public void onAcknowledgement() {
+        onAcknowledge.accept(msgId);
+    }
+
+    @Override
+    public void onReject() {
+
+    }
+
+    @Override
+    public void onTimeout() {
+
+    }
+
+    @Override
+    public void onCancel() {
+
+    }
+
+    @Override
+    public void onReadyToSend() {
+
+    }
+
+    @Override
+    public void onConnecting() {
+
+    }
+
+    @Override
+    public void onDtlsRetransmission(int flight) {
+
+    }
+
+    @Override
+    public void onSent(boolean retransmission) {
+
+    }
+
+    @Override
+    public void onSendError(Throwable error) {
+
+    }
+
+    @Override
+    public void onContextEstablished(EndpointContext endpointContext) {
+
+    }
+
+    @Override
+    public void onComplete() {
+
+    }
+}

@@ -150,7 +150,7 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor {
     private Response getObserveNotification(boolean confirmable, JsonElement json) {
         Response response = new Response(CoAP.ResponseCode._UNKNOWN_SUCCESS_CODE);
         response.setPayload(json.toString());
-        response.setAcknowledged(confirmable);
+        response.setConfirmable(confirmable);
         return response;
     }
 

@@ -122,7 +122,7 @@ public class ProtoCoapAdaptor implements CoapTransportAdaptor {
     @Override
     public Response convertToPublish(boolean isConfirmable, TransportProtos.ToServerRpcResponseMsg msg) throws AdaptorException {
         Response response = new Response(CoAP.ResponseCode.CONTENT);
-        response.setAcknowledged(isConfirmable);
+        response.setConfirmable(isConfirmable);
         response.setPayload(msg.toByteArray());
         return response;
     }

+/**
+ * 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.coap.callback;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.californium.core.coap.CoAP;
+import org.eclipse.californium.core.server.resources.CoapExchange;
+import org.thingsboard.server.common.data.DeviceProfile;
+import org.thingsboard.server.common.transport.TransportContext;
+import org.thingsboard.server.common.transport.TransportServiceCallback;
+import org.thingsboard.server.common.transport.auth.SessionInfoCreator;
+import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
+import org.thingsboard.server.gen.transport.TransportProtos;
+import org.thingsboard.server.transport.coap.AbstractCoapTransportResource;
+
+import java.util.UUID;
+import java.util.function.BiConsumer;
+
+@Slf4j
+public class CoapDeviceAuthCallback implements TransportServiceCallback<ValidateDeviceCredentialsResponse> {
+    private final TransportContext transportContext;
+    private final CoapExchange exchange;
+    private final BiConsumer<TransportProtos.SessionInfoProto, DeviceProfile> onSuccess;
+
+    public CoapDeviceAuthCallback(TransportContext transportContext, CoapExchange exchange, BiConsumer<TransportProtos.SessionInfoProto, DeviceProfile> onSuccess) {
+        this.transportContext = transportContext;
+        this.exchange = exchange;
+        this.onSuccess = onSuccess;
+    }
+
+    @Override
+    public void onSuccess(ValidateDeviceCredentialsResponse msg) {
+        DeviceProfile deviceProfile = msg.getDeviceProfile();
+        if (msg.hasDeviceInfo() && deviceProfile != null) {
+            TransportProtos.SessionInfoProto sessionInfoProto = SessionInfoCreator.create(msg, transportContext, UUID.randomUUID());
+            onSuccess.accept(sessionInfoProto, deviceProfile);
+        } else {
+            exchange.respond(CoAP.ResponseCode.UNAUTHORIZED);
+        }
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        log.warn("Failed to process request", e);
+        exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR);
+    }
+}

+/**
+ * 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.coap.callback;
+
+import org.eclipse.californium.core.coap.CoAP;
+import org.eclipse.californium.core.server.resources.CoapExchange;
+import org.thingsboard.server.common.transport.TransportServiceCallback;
+
+public class CoapNoOpCallback implements TransportServiceCallback<Void> {
+    private final CoapExchange exchange;
+
+    public 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);
+    }
+}

+/**
+ * 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.coap.callback;
+
+import org.eclipse.californium.core.coap.CoAP;
+import org.eclipse.californium.core.coap.Response;
+import org.eclipse.californium.core.server.resources.CoapExchange;
+import org.thingsboard.server.common.transport.TransportServiceCallback;
+
+public class CoapOkCallback implements TransportServiceCallback<Void> {
+
+    protected final CoapExchange exchange;
+    protected final CoAP.ResponseCode onSuccessResponse;
+    protected 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.setConfirmable(isConRequest());
+        exchange.respond(response);
+    }
+
+    @Override
+    public void onError(Throwable e) {
+        exchange.respond(onFailureResponse);
+    }
+
+    protected boolean isConRequest() {
+        return exchange.advanced().getRequest().isConfirmable();
+    }
+}

@@ -35,6 +35,8 @@ import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.coap.MeasurementTypeProtos;
 import org.thingsboard.server.gen.transport.coap.MeasurementsProtos;
 import org.thingsboard.server.transport.coap.AbstractCoapTransportResource;
+import org.thingsboard.server.transport.coap.callback.CoapDeviceAuthCallback;
+import org.thingsboard.server.transport.coap.callback.CoapOkCallback;
 import org.thingsboard.server.transport.coap.CoapTransportContext;
 import org.thingsboard.server.transport.coap.efento.utils.CoapEfentoUtils;
 

@@ -17,7 +17,6 @@ package org.thingsboard.server.transport.mqtt;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.gson.JsonParseException;
-import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.handler.codec.mqtt.MqttConnAckMessage;
@@ -40,6 +39,7 @@ import io.netty.util.CharsetUtil;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
+import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.thingsboard.server.common.data.DataConstants;
@@ -129,6 +129,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
 
     private final ConcurrentHashMap<String, String> otaPackSessions;
     private final ConcurrentHashMap<String, Integer> chunkSizes;
+    private final ConcurrentMap<Integer, TransportProtos.ToDeviceRpcRequestMsg> rpcAwaitingAck;
 
     MqttTransportHandler(MqttTransportContext context, SslHandler sslHandler) {
         this.sessionId = UUID.randomUUID();
@@ -140,6 +141,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
         this.deviceSessionCtx = new DeviceSessionCtx(sessionId, mqttQoSMap, context);
         this.otaPackSessions = new ConcurrentHashMap<>();
         this.chunkSizes = new ConcurrentHashMap<>();
+        this.rpcAwaitingAck = new ConcurrentHashMap<>();
     }
 
     @Override
@@ -243,6 +245,13 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
                     processDisconnect(ctx);
                 }
                 break;
+            case PUBACK:
+                int msgId = ((MqttPubAckMessage) msg).variableHeader().messageId();
+                TransportProtos.ToDeviceRpcRequestMsg rpcRequest = rpcAwaitingAck.remove(msgId);
+                if (rpcRequest != null) {
+                    transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, false, TransportServiceCallback.EMPTY);
+                }
+                break;
             default:
                 break;
         }
@@ -815,16 +824,22 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
     public void onToDeviceRpcRequest(UUID sessionId, TransportProtos.ToDeviceRpcRequestMsg rpcRequest) {
         log.trace("[{}] Received RPC command to device", sessionId);
         try {
-            deviceSessionCtx.getPayloadAdaptor().convertToPublish(deviceSessionCtx, rpcRequest)
-                    .ifPresent(payload -> {
-                        ChannelFuture channelFuture = deviceSessionCtx.getChannel().writeAndFlush(payload);
-                        if (rpcRequest.getPersisted()) {
-                            channelFuture.addListener(future ->
-                                    transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest,
-                                            future.cause() != null, TransportServiceCallback.EMPTY)
-                            );
-                        }
-                    });
+            deviceSessionCtx.getPayloadAdaptor().convertToPublish(deviceSessionCtx, rpcRequest).ifPresent(payload -> {
+                RequestInfo requestInfo = publish(payload, deviceSessionCtx);
+                int msgId = requestInfo.getMsgId();
+
+                if (isAckExpected(payload)) {
+                    if (rpcRequest.getPersisted()) {
+                        rpcAwaitingAck.put(msgId, rpcRequest);
+                        context.getScheduler().schedule(() -> {
+                            TransportProtos.ToDeviceRpcRequestMsg awaitingAckMsg = rpcAwaitingAck.remove(msgId);
+                            if (awaitingAckMsg != null) {
+                                transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, true, TransportServiceCallback.EMPTY);
+                            }
+                        }, Math.max(0, rpcRequest.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+                    }
+                }
+            });
         } catch (Exception e) {
             log.trace("[{}] Failed to convert device RPC command to MQTT msg", sessionId, e);
         }
@@ -840,6 +855,19 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
         }
     }
 
+    private RequestInfo publish(MqttMessage message, DeviceSessionCtx deviceSessionCtx) {
+        deviceSessionCtx.getChannel().writeAndFlush(message);
+
+        int msgId = ((MqttPublishMessage) message).variableHeader().packetId();
+        RequestInfo requestInfo = new RequestInfo(msgId, System.currentTimeMillis(), deviceSessionCtx.getSessionInfo());
+
+        return requestInfo;
+    }
+
+    private boolean isAckExpected(MqttMessage message) {
+        return message.fixedHeader().qosLevel().value() > 0;
+    }
+
     @Override
     public void onDeviceProfileUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceProfile deviceProfile) {
         deviceSessionCtx.onDeviceProfileUpdate(sessionInfo, deviceProfile);
@@ -849,4 +877,11 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
     public void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device, Optional<DeviceProfile> deviceProfileOpt) {
         deviceSessionCtx.onDeviceUpdate(sessionInfo, device, deviceProfileOpt);
     }
+
+    @Data
+    public static class RequestInfo {
+        private final int msgId;
+        private final long requestTime;
+        private final TransportProtos.SessionInfoProto sessionInfo;
+    }
 }