Merge pull request #1724 from elbstack/1686-No-response-to-client-side-RPC-via-CoAP

1686 no response to client side rpc via co ap

@@ -192,6 +192,7 @@ public class CoapTransportResource extends CoapResource {
                                         new CoapOkCallback(exchange));
                                         new CoapOkCallback(exchange));
                                 break;
                                 break;
                             case TO_SERVER_RPC_REQUEST:
                             case TO_SERVER_RPC_REQUEST:
+                                transportService.registerSyncSession(sessionInfo, new CoapSessionListener(sessionId, exchange), transportContext.getTimeout());
                                 transportService.process(sessionInfo,
                                 transportService.process(sessionInfo,
                                         transportContext.getAdaptor().convertToServerRpcRequest(sessionId, request),
                                         transportContext.getAdaptor().convertToServerRpcRequest(sessionId, request),
                                         new CoapNoOpCallback(exchange));
                                         new CoapNoOpCallback(exchange));

@@ -178,15 +178,24 @@ public abstract class AbstractTransportService implements TransportService {
 
 
     @Override
     @Override
     public void registerSyncSession(TransportProtos.SessionInfoProto sessionInfo, SessionMsgListener listener, long timeout) {
     public void registerSyncSession(TransportProtos.SessionInfoProto sessionInfo, SessionMsgListener listener, long timeout) {
-        sessions.putIfAbsent(toId(sessionInfo), new SessionMetaData(sessionInfo, TransportProtos.SessionType.SYNC, listener));
-        schedulerExecutor.schedule(() -> {
+        SessionMetaData currentSession = new SessionMetaData(sessionInfo, TransportProtos.SessionType.SYNC, listener);
+        sessions.putIfAbsent(toId(sessionInfo), currentSession);
+
+        ScheduledFuture executorFuture = schedulerExecutor.schedule(() -> {
             listener.onRemoteSessionCloseCommand(TransportProtos.SessionCloseNotificationProto.getDefaultInstance());
             listener.onRemoteSessionCloseCommand(TransportProtos.SessionCloseNotificationProto.getDefaultInstance());
             deregisterSession(sessionInfo);
             deregisterSession(sessionInfo);
         }, timeout, TimeUnit.MILLISECONDS);
         }, timeout, TimeUnit.MILLISECONDS);
+
+        currentSession.setScheduledFuture(executorFuture);
     }
     }
 
 
     @Override
     @Override
     public void deregisterSession(TransportProtos.SessionInfoProto sessionInfo) {
     public void deregisterSession(TransportProtos.SessionInfoProto sessionInfo) {
+        SessionMetaData currentSession = sessions.get(toId(sessionInfo));
+        if (currentSession.hasScheduledFuture()) {
+            log.debug("Stopping scheduler to avoid resending response if request has been ack.");
+            currentSession.getScheduledFuture().cancel(false);
+        }
         sessions.remove(toId(sessionInfo));
         sessions.remove(toId(sessionInfo));
     }
     }
 
 

@@ -19,6 +19,8 @@ import lombok.Data;
 import org.thingsboard.server.common.transport.SessionMsgListener;
 import org.thingsboard.server.common.transport.SessionMsgListener;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos;
 
 
+import java.util.concurrent.ScheduledFuture;
+
 /**
 /**
  * Created by ashvayka on 15.10.18.
  * Created by ashvayka on 15.10.18.
  */
  */
@@ -29,19 +31,33 @@ class SessionMetaData {
     private final TransportProtos.SessionType sessionType;
     private final TransportProtos.SessionType sessionType;
     private final SessionMsgListener listener;
     private final SessionMsgListener listener;
 
 
+    private ScheduledFuture scheduledFuture;
+
     private volatile long lastActivityTime;
     private volatile long lastActivityTime;
     private volatile boolean subscribedToAttributes;
     private volatile boolean subscribedToAttributes;
     private volatile boolean subscribedToRPC;
     private volatile boolean subscribedToRPC;
 
 
-    SessionMetaData(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.SessionType sessionType, SessionMsgListener listener) {
+    SessionMetaData(
+            TransportProtos.SessionInfoProto sessionInfo,
+            TransportProtos.SessionType sessionType,
+            SessionMsgListener listener
+    ) {
         this.sessionInfo = sessionInfo;
         this.sessionInfo = sessionInfo;
         this.sessionType = sessionType;
         this.sessionType = sessionType;
         this.listener = listener;
         this.listener = listener;
         this.lastActivityTime = System.currentTimeMillis();
         this.lastActivityTime = System.currentTimeMillis();
+        this.scheduledFuture = null;
     }
     }
 
 
     void updateLastActivityTime() {
     void updateLastActivityTime() {
         this.lastActivityTime = System.currentTimeMillis();
         this.lastActivityTime = System.currentTimeMillis();
     }
     }
 
 
+    void setScheduledFuture(ScheduledFuture scheduledFuture) { this.scheduledFuture = scheduledFuture; }
+
+    public ScheduledFuture getScheduledFuture() {
+        return scheduledFuture;
+    }
+
+    public boolean hasScheduledFuture() { return null != this.scheduledFuture; }
 }
 }