Merge pull request #5522 from YevhenBondarenko/master

[3.3.2] Lwm2M Improvements to avoid duplication of RPC call

@@ -117,6 +117,10 @@ public class LwM2mClient implements Serializable {
     @Getter
     @Getter
     private final AtomicInteger retryAttempts;
     private final AtomicInteger retryAttempts;
 
 
+    @Getter
+    @Setter
+    private UUID lastSentRpcId;
+
     public Object clone() throws CloneNotSupportedException {
     public Object clone() throws CloneNotSupportedException {
         return super.clone();
         return super.clone();
     }
     }

@@ -319,6 +319,10 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im
         Registration registration = client.getRegistration();
         Registration registration = client.getRegistration();
         try {
         try {
             logService.log(client, String.format("[%s][%s] Sending request: %s to %s", registration.getId(), registration.getSocketAddress(), request.getClass().getSimpleName(), pathToStringFunction.apply(request)));
             logService.log(client, String.format("[%s][%s] Sending request: %s to %s", registration.getId(), registration.getSocketAddress(), request.getClass().getSimpleName(), pathToStringFunction.apply(request)));
+            if (!callback.onSent(request)) {
+                return;
+            }
+
             context.getServer().send(registration, request, timeoutInMs, response -> {
             context.getServer().send(registration, request, timeoutInMs, response -> {
                 executor.submit(() -> {
                 executor.submit(() -> {
                     try {
                     try {
@@ -330,7 +334,6 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im
                     }
                     }
                 });
                 });
             }, e -> handleDownlinkError(client, request, callback, e));
             }, e -> handleDownlinkError(client, request, callback, e));
-            callback.onSent(request);
         } catch (Exception e) {
         } catch (Exception e) {
             handleDownlinkError(client, request, callback, e);
             handleDownlinkError(client, request, callback, e);
         }
         }
@@ -366,6 +369,7 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im
 
 
     private <R extends DownlinkRequest<T>, T extends LwM2mResponse> void handleDownlinkError(LwM2mClient client, R request, DownlinkRequestCallback<R, T> callback, Exception e) {
     private <R extends DownlinkRequest<T>, T extends LwM2mResponse> void handleDownlinkError(LwM2mClient client, R request, DownlinkRequestCallback<R, T> callback, Exception e) {
         log.trace("[{}] Received downlink error: {}.", client.getEndpoint(), e);
         log.trace("[{}] Received downlink error: {}.", client.getEndpoint(), e);
+        client.updateLastUplinkTime();
         executor.submit(() -> {
         executor.submit(() -> {
             if (e instanceof TimeoutException || e instanceof ClientSleepingException) {
             if (e instanceof TimeoutException || e instanceof ClientSleepingException) {
                 log.trace("[{}] Received {}, client is probably sleeping", client.getEndpoint(), e.getClass().getSimpleName());
                 log.trace("[{}] Received {}, client is probably sleeping", client.getEndpoint(), e.getClass().getSimpleName());

@@ -17,7 +17,9 @@ package org.thingsboard.server.transport.lwm2m.server.downlink;
 
 
 public interface DownlinkRequestCallback<R, T> {
 public interface DownlinkRequestCallback<R, T> {
 
 
-    default void onSent(R request){};
+    default boolean onSent(R request){
+        return true;
+    };
 
 
     void onSuccess(R request, T response);
     void onSuccess(R request, T response);
 
 

@@ -95,6 +95,12 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler {
             this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.INTERNAL_SERVER_ERROR, "Registration is empty");
             this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.INTERNAL_SERVER_ERROR, "Registration is empty");
             return;
             return;
         }
         }
+        UUID rpcId = new UUID(rpcRequest.getRequestIdMSB(), rpcRequest.getRequestIdLSB());
+
+        if (rpcId.equals(client.getLastSentRpcId())) {
+            log.debug("[{}]][{}] Rpc has already sent!", client.getEndpoint(), rpcId);
+            return;
+        }
         try {
         try {
             if (operationType.isHasObjectId()) {
             if (operationType.isHasObjectId()) {
                 String objectId = getIdFromParameters(client, rpcRequest);
                 String objectId = getIdFromParameters(client, rpcRequest);

@@ -15,6 +15,7 @@
  */
  */
 package org.thingsboard.server.transport.lwm2m.server.rpc;
 package org.thingsboard.server.transport.lwm2m.server.rpc;
 
 
+import lombok.extern.slf4j.Slf4j;
 import org.eclipse.leshan.core.ResponseCode;
 import org.eclipse.leshan.core.ResponseCode;
 import org.eclipse.leshan.core.request.exception.ClientSleepingException;
 import org.eclipse.leshan.core.request.exception.ClientSleepingException;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.common.util.JacksonUtil;
@@ -26,8 +27,10 @@ import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient;
 import org.thingsboard.server.transport.lwm2m.server.downlink.DownlinkRequestCallback;
 import org.thingsboard.server.transport.lwm2m.server.downlink.DownlinkRequestCallback;
 
 
+import java.util.UUID;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.TimeoutException;
 
 
+@Slf4j
 public abstract class RpcDownlinkRequestCallbackProxy<R, T> implements DownlinkRequestCallback<R, T> {
 public abstract class RpcDownlinkRequestCallbackProxy<R, T> implements DownlinkRequestCallback<R, T> {
 
 
     private final TransportService transportService;
     private final TransportService transportService;
@@ -44,8 +47,20 @@ public abstract class RpcDownlinkRequestCallbackProxy<R, T> implements DownlinkR
     }
     }
 
 
     @Override
     @Override
-    public void onSent(R request) {
+    public boolean onSent(R request) {
+        client.lock();
+        try {
+            UUID rpcId = new UUID(this.request.getRequestIdMSB(), this.request.getRequestIdLSB());
+            if (rpcId.equals(client.getLastSentRpcId())) {
+                log.debug("[{}]][{}] Rpc has already sent!", client.getEndpoint(), rpcId);
+                return false;
+            }
+            client.setLastSentRpcId(rpcId);
+        } finally {
+            client.unlock();
+        }
         transportService.process(client.getSession(), this.request, RpcStatus.SENT, TransportServiceCallback.EMPTY);
         transportService.process(client.getSession(), this.request, RpcStatus.SENT, TransportServiceCallback.EMPTY);
+        return true;
     }
     }
 
 
     @Override
     @Override
@@ -68,6 +83,7 @@ public abstract class RpcDownlinkRequestCallbackProxy<R, T> implements DownlinkR
     @Override
     @Override
     public void onError(String params, Exception e) {
     public void onError(String params, Exception e) {
         if (e instanceof TimeoutException || e instanceof org.eclipse.leshan.core.request.exception.TimeoutException) {
         if (e instanceof TimeoutException || e instanceof org.eclipse.leshan.core.request.exception.TimeoutException) {
+            client.setLastSentRpcId(null);
             transportService.process(client.getSession(), this.request, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
             transportService.process(client.getSession(), this.request, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
         } else if (!(e instanceof ClientSleepingException)) {
         } else if (!(e instanceof ClientSleepingException)) {
             sendRpcReplyOnError(e);
             sendRpcReplyOnError(e);