Merge remote-tracking branch 'origin/bug/rpc-duplicates-on-timeout'

application/src/main/java/org/thingsboard/server/controller/RpcController.java → application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java

@@ -16,7 +16,6 @@
 package org.thingsboard.server.controller;
 
 import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.util.concurrent.FutureCallback;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -26,13 +25,12 @@ import org.springframework.http.ResponseEntity;
 import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.util.StringUtils;
 import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.ResponseBody;
-import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.context.request.async.DeferredResult;
+import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.rule.engine.api.RpcError;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.audit.ActionType;
@@ -59,20 +57,15 @@ import org.thingsboard.server.service.security.permission.Operation;
 import org.thingsboard.server.service.telemetry.exception.ToErrorResponseEntity;
 
 import javax.annotation.Nullable;
-import java.io.IOException;
 import java.util.Optional;
 import java.util.UUID;
 
 /**
  * Created by ashvayka on 22.03.18.
  */
-@RestController
 @TbCoreComponent
-@RequestMapping(TbUrlConstants.RPC_URL_PREFIX)
 @Slf4j
-public class RpcController extends BaseController {
+public abstract class AbstractRpcController extends BaseController {
-
-    protected final ObjectMapper jsonMapper = new ObjectMapper();
 
     @Autowired
     private TbCoreDeviceRpcService deviceRpcService;
@@ -81,75 +74,15 @@ public class RpcController extends BaseController {
     private AccessValidator accessValidator;
 
     @Value("${server.rest.server_side_rpc.min_timeout:5000}")
-    private long minTimeout;
+    protected long minTimeout;
 
     @Value("${server.rest.server_side_rpc.default_timeout:10000}")
-    private long defaultTimeout;
+    protected long defaultTimeout;
-
-    @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')")
-    @RequestMapping(value = "/oneway/{deviceId}", method = RequestMethod.POST)
-    @ResponseBody
-    public DeferredResult<ResponseEntity> handleOneWayDeviceRPCRequest(@PathVariable("deviceId") String deviceIdStr, @RequestBody String requestBody) throws ThingsboardException {
-        return handleDeviceRPCRequest(true, new DeviceId(UUID.fromString(deviceIdStr)), requestBody);
-    }
-
-    @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')")
-    @RequestMapping(value = "/twoway/{deviceId}", method = RequestMethod.POST)
-    @ResponseBody
-    public DeferredResult<ResponseEntity> handleTwoWayDeviceRPCRequest(@PathVariable("deviceId") String deviceIdStr, @RequestBody String requestBody) throws ThingsboardException {
-        return handleDeviceRPCRequest(false, new DeviceId(UUID.fromString(deviceIdStr)), requestBody);
-    }
-
-    @PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')")
-    @RequestMapping(value = "/persistent/{rpcId}", method = RequestMethod.GET)
-    @ResponseBody
-    public Rpc getPersistedRpc(@PathVariable("rpcId") String strRpc) throws ThingsboardException {
-        checkParameter("RpcId", strRpc);
-        try {
-            RpcId rpcId = new RpcId(UUID.fromString(strRpc));
-            return checkRpcId(rpcId, Operation.READ);
-        } catch (Exception e) {
-            throw handleException(e);
-        }
-    }
-
-    @PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')")
-    @RequestMapping(value = "/persistent/device/{deviceId}", method = RequestMethod.GET)
-    @ResponseBody
-    public PageData<Rpc> getPersistedRpcByDevice(@PathVariable("deviceId") String strDeviceId,
-                                                 @RequestParam int pageSize,
-                                                 @RequestParam int page,
-                                                 @RequestParam RpcStatus rpcStatus,
-                                                 @RequestParam(required = false) String textSearch,
-                                                 @RequestParam(required = false) String sortProperty,
-                                                 @RequestParam(required = false) String sortOrder) throws ThingsboardException {
-        checkParameter("DeviceId", strDeviceId);
-        try {
-            TenantId tenantId = getCurrentUser().getTenantId();
-            PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder);
-            DeviceId deviceId = new DeviceId(UUID.fromString(strDeviceId));
-            return checkNotNull(rpcService.findAllByDeviceIdAndStatus(tenantId, deviceId, rpcStatus, pageLink));
-        } catch (Exception e) {
-            throw handleException(e);
-        }
-    }
-
-    @PreAuthorize("hasAnyAuthority('TENANT_ADMIN')")
-    @RequestMapping(value = "/persistent/{rpcId}", method = RequestMethod.DELETE)
-    @ResponseBody
-    public void deleteResource(@PathVariable("rpcId") String strRpc) throws ThingsboardException {
-        checkParameter("RpcId", strRpc);
-        try {
-            rpcService.deleteRpc(getTenantId(), new RpcId(UUID.fromString(strRpc)));
-        } catch (Exception e) {
-            throw handleException(e);
-        }
-    }
 
-    private DeferredResult<ResponseEntity> handleDeviceRPCRequest(boolean oneWay, DeviceId deviceId, String requestBody) throws ThingsboardException {
+    protected DeferredResult<ResponseEntity> handleDeviceRPCRequest(boolean oneWay, DeviceId deviceId, String requestBody, HttpStatus timeoutStatus, HttpStatus noActiveConnectionStatus) throws ThingsboardException {
         try {
-            JsonNode rpcRequestBody = jsonMapper.readTree(requestBody);
+            JsonNode rpcRequestBody = JacksonUtil.toJsonNode(requestBody);
-            ToDeviceRpcRequestBody body = new ToDeviceRpcRequestBody(rpcRequestBody.get("method").asText(), jsonMapper.writeValueAsString(rpcRequestBody.get("params")));
+            ToDeviceRpcRequestBody body = new ToDeviceRpcRequestBody(rpcRequestBody.get("method").asText(), JacksonUtil.toString(rpcRequestBody.get("params")));
             SecurityUser currentUser = getCurrentUser();
             TenantId tenantId = currentUser.getTenantId();
             final DeferredResult<ResponseEntity> response = new DeferredResult<>();
@@ -157,7 +90,7 @@ public class RpcController extends BaseController {
             long expTime = System.currentTimeMillis() + Math.max(minTimeout, timeout);
             UUID rpcRequestUUID = rpcRequestBody.has("requestUUID") ? UUID.fromString(rpcRequestBody.get("requestUUID").asText()) : UUID.randomUUID();
             boolean persisted = rpcRequestBody.has(DataConstants.PERSISTENT) && rpcRequestBody.get(DataConstants.PERSISTENT).asBoolean();
-            accessValidator.validate(currentUser, Operation.RPC_CALL, deviceId, new HttpValidationCallback(response, new FutureCallback<DeferredResult<ResponseEntity>>() {
+            accessValidator.validate(currentUser, Operation.RPC_CALL, deviceId, new HttpValidationCallback(response, new FutureCallback<>() {
                 @Override
                 public void onSuccess(@Nullable DeferredResult<ResponseEntity> result) {
                     ToDeviceRpcRequest rpcRequest = new ToDeviceRpcRequest(rpcRequestUUID,
@@ -168,7 +101,7 @@ public class RpcController extends BaseController {
                             body,
                             persisted
                     );
-                    deviceRpcService.processRestApiRpcRequest(rpcRequest, fromDeviceRpcResponse -> reply(new LocalRequestMetaData(rpcRequest, currentUser, result), fromDeviceRpcResponse), currentUser);
+                    deviceRpcService.processRestApiRpcRequest(rpcRequest, fromDeviceRpcResponse -> reply(new LocalRequestMetaData(rpcRequest, currentUser, result), fromDeviceRpcResponse, timeoutStatus, noActiveConnectionStatus), currentUser);
                 }
 
                 @Override
@@ -184,12 +117,12 @@ public class RpcController extends BaseController {
                 }
             }));
             return response;
-        } catch (IOException ioe) {
+        } catch (IllegalArgumentException ioe) {
             throw new ThingsboardException("Invalid request body", ioe, ThingsboardErrorCode.BAD_REQUEST_PARAMS);
         }
     }
 
-    public void reply(LocalRequestMetaData rpcRequest, FromDeviceRpcResponse response) {
+    public void reply(LocalRequestMetaData rpcRequest, FromDeviceRpcResponse response, HttpStatus timeoutStatus, HttpStatus noActiveConnectionStatus) {
         Optional<RpcError> rpcError = response.getError();
         DeferredResult<ResponseEntity> responseWriter = rpcRequest.getResponseWriter();
         if (rpcError.isPresent()) {
@@ -197,13 +130,13 @@ public class RpcController extends BaseController {
             RpcError error = rpcError.get();
             switch (error) {
                 case TIMEOUT:
-                    responseWriter.setResult(new ResponseEntity<>(HttpStatus.REQUEST_TIMEOUT));
+                    responseWriter.setResult(new ResponseEntity<>(timeoutStatus));
                     break;
                 case NO_ACTIVE_CONNECTION:
-                    responseWriter.setResult(new ResponseEntity<>(HttpStatus.CONFLICT));
+                    responseWriter.setResult(new ResponseEntity<>(noActiveConnectionStatus));
                     break;
                 default:
-                    responseWriter.setResult(new ResponseEntity<>(HttpStatus.REQUEST_TIMEOUT));
+                    responseWriter.setResult(new ResponseEntity<>(timeoutStatus));
                     break;
             }
         } else {
@@ -212,8 +145,8 @@ public class RpcController extends BaseController {
                 String data = responseData.get();
                 try {
                     logRpcCall(rpcRequest, rpcError, null);
-                    responseWriter.setResult(new ResponseEntity<>(jsonMapper.readTree(data), HttpStatus.OK));
+                    responseWriter.setResult(new ResponseEntity<>(JacksonUtil.toJsonNode(data), HttpStatus.OK));
-                } catch (IOException e) {
+                } catch (IllegalArgumentException e) {
                     log.debug("Failed to decode device response: {}", data, e);
                     logRpcCall(rpcRequest, rpcError, e);
                     responseWriter.setResult(new ResponseEntity<>(HttpStatus.NOT_ACCEPTABLE));

application/src/main/java/org/thingsboard/server/controller/RpcV1Controller.java

@@ -0,0 +1,55 @@
+/**
+ * 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.controller;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.security.access.prepost.PreAuthorize;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.ResponseBody;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.context.request.async.DeferredResult;
+import org.thingsboard.server.common.data.exception.ThingsboardException;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.queue.util.TbCoreComponent;
+
+import java.util.UUID;
+
+@RestController
+@TbCoreComponent
+@RequestMapping(TbUrlConstants.RPC_V1_URL_PREFIX)
+@Slf4j
+public class RpcV1Controller extends AbstractRpcController {
+
+    @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')")
+    @RequestMapping(value = "/oneway/{deviceId}", method = RequestMethod.POST)
+    @ResponseBody
+    public DeferredResult<ResponseEntity> handleOneWayDeviceRPCRequest(@PathVariable("deviceId") String deviceIdStr, @RequestBody String requestBody) throws ThingsboardException {
+        return handleDeviceRPCRequest(true, new DeviceId(UUID.fromString(deviceIdStr)), requestBody, HttpStatus.REQUEST_TIMEOUT, HttpStatus.CONFLICT);
+    }
+
+    @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')")
+    @RequestMapping(value = "/twoway/{deviceId}", method = RequestMethod.POST)
+    @ResponseBody
+    public DeferredResult<ResponseEntity> handleTwoWayDeviceRPCRequest(@PathVariable("deviceId") String deviceIdStr, @RequestBody String requestBody) throws ThingsboardException {
+        return handleDeviceRPCRequest(false, new DeviceId(UUID.fromString(deviceIdStr)), requestBody, HttpStatus.REQUEST_TIMEOUT, HttpStatus.CONFLICT);
+    }
+
+}

application/src/main/java/org/thingsboard/server/controller/RpcV2Controller.java

@@ -0,0 +1,108 @@
+/**
+ * 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.controller;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.security.access.prepost.PreAuthorize;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.context.request.async.DeferredResult;
+import org.thingsboard.server.common.data.exception.ThingsboardException;
+import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.id.RpcId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.page.PageData;
+import org.thingsboard.server.common.data.page.PageLink;
+import org.thingsboard.server.common.data.rpc.Rpc;
+import org.thingsboard.server.common.data.rpc.RpcStatus;
+import org.thingsboard.server.queue.util.TbCoreComponent;
+import org.thingsboard.server.service.security.permission.Operation;
+
+import java.util.UUID;
+
+@RestController
+@TbCoreComponent
+@RequestMapping(TbUrlConstants.RPC_V2_URL_PREFIX)
+@Slf4j
+public class RpcV2Controller extends AbstractRpcController {
+
+    @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')")
+    @RequestMapping(value = "/oneway/{deviceId}", method = RequestMethod.POST)
+    @ResponseBody
+    public DeferredResult<ResponseEntity> handleOneWayDeviceRPCRequest(@PathVariable("deviceId") String deviceIdStr, @RequestBody String requestBody) throws ThingsboardException {
+        return handleDeviceRPCRequest(true, new DeviceId(UUID.fromString(deviceIdStr)), requestBody, HttpStatus.GATEWAY_TIMEOUT, HttpStatus.GATEWAY_TIMEOUT);
+    }
+
+    @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')")
+    @RequestMapping(value = "/twoway/{deviceId}", method = RequestMethod.POST)
+    @ResponseBody
+    public DeferredResult<ResponseEntity> handleTwoWayDeviceRPCRequest(@PathVariable("deviceId") String deviceIdStr, @RequestBody String requestBody) throws ThingsboardException {
+        return handleDeviceRPCRequest(false, new DeviceId(UUID.fromString(deviceIdStr)), requestBody, HttpStatus.GATEWAY_TIMEOUT, HttpStatus.GATEWAY_TIMEOUT);
+    }
+
+    @PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')")
+    @RequestMapping(value = "/persistent/{rpcId}", method = RequestMethod.GET)
+    @ResponseBody
+    public Rpc getPersistedRpc(@PathVariable("rpcId") String strRpc) throws ThingsboardException {
+        checkParameter("RpcId", strRpc);
+        try {
+            RpcId rpcId = new RpcId(UUID.fromString(strRpc));
+            return checkRpcId(rpcId, Operation.READ);
+        } catch (Exception e) {
+            throw handleException(e);
+        }
+    }
+
+    @PreAuthorize("hasAnyAuthority('TENANT_ADMIN', 'CUSTOMER_USER')")
+    @RequestMapping(value = "/persistent/device/{deviceId}", method = RequestMethod.GET)
+    @ResponseBody
+    public PageData<Rpc> getPersistedRpcByDevice(@PathVariable("deviceId") String strDeviceId,
+                                                 @RequestParam int pageSize,
+                                                 @RequestParam int page,
+                                                 @RequestParam RpcStatus rpcStatus,
+                                                 @RequestParam(required = false) String textSearch,
+                                                 @RequestParam(required = false) String sortProperty,
+                                                 @RequestParam(required = false) String sortOrder) throws ThingsboardException {
+        checkParameter("DeviceId", strDeviceId);
+        try {
+            TenantId tenantId = getCurrentUser().getTenantId();
+            PageLink pageLink = createPageLink(pageSize, page, textSearch, sortProperty, sortOrder);
+            DeviceId deviceId = new DeviceId(UUID.fromString(strDeviceId));
+            return checkNotNull(rpcService.findAllByDeviceIdAndStatus(tenantId, deviceId, rpcStatus, pageLink));
+        } catch (Exception e) {
+            throw handleException(e);
+        }
+    }
+
+    @PreAuthorize("hasAnyAuthority('TENANT_ADMIN')")
+    @RequestMapping(value = "/persistent/{rpcId}", method = RequestMethod.DELETE)
+    @ResponseBody
+    public void deleteResource(@PathVariable("rpcId") String strRpc) throws ThingsboardException {
+        checkParameter("RpcId", strRpc);
+        try {
+            rpcService.deleteRpc(getTenantId(), new RpcId(UUID.fromString(strRpc)));
+        } catch (Exception e) {
+            throw handleException(e);
+        }
+    }
+}

application/src/main/java/org/thingsboard/server/controller/TbUrlConstants.java

@@ -20,5 +20,6 @@ package org.thingsboard.server.controller;
  */
 public class TbUrlConstants {
     public static final String TELEMETRY_URL_PREFIX = "/api/plugins/telemetry";
-    public static final String RPC_URL_PREFIX = "/api/plugins/rpc";
+    public static final String RPC_V1_URL_PREFIX = "/api/plugins/rpc";
+    public static final String RPC_V2_URL_PREFIX = "/api/rpc";
 }

application/src/test/java/org/thingsboard/server/transport/TransportSqlTestSuite.java

@@ -27,13 +27,13 @@ import java.util.Arrays;
 @RunWith(ClasspathSuite.class)
 @ClasspathSuite.ClassnameFilters({
         "org.thingsboard.server.transport.*.rpc.sql.*Test",
-        "org.thingsboard.server.transport.*.telemetry.timeseries.sql.*Test",
+//        "org.thingsboard.server.transport.*.telemetry.timeseries.sql.*Test",
-        "org.thingsboard.server.transport.*.telemetry.attributes.sql.*Test",
+//        "org.thingsboard.server.transport.*.telemetry.attributes.sql.*Test",
-        "org.thingsboard.server.transport.*.attributes.updates.sql.*Test",
+//        "org.thingsboard.server.transport.*.attributes.updates.sql.*Test",
-        "org.thingsboard.server.transport.*.attributes.request.sql.*Test",
+//        "org.thingsboard.server.transport.*.attributes.request.sql.*Test",
-        "org.thingsboard.server.transport.*.claim.sql.*Test",
+//        "org.thingsboard.server.transport.*.claim.sql.*Test",
-        "org.thingsboard.server.transport.*.provision.sql.*Test",
+//        "org.thingsboard.server.transport.*.provision.sql.*Test",
-        "org.thingsboard.server.transport.lwm2m.*Test"
+//        "org.thingsboard.server.transport.lwm2m.*Test"
 })
 public class TransportSqlTestSuite {
 

application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcDefaultIntegrationTest.java

@@ -43,7 +43,7 @@ public abstract class AbstractCoapServerSideRpcDefaultIntegrationTest extends Ab
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"24\",\"value\": 1},\"timeout\": 6000}";
         String deviceId = savedDevice.getId().getId().toString();
 
-        doPostAsync("/api/plugins/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().is(409),
+        doPostAsync("/api/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().is(504),
                 asyncContextTimeoutToUseRpcPlugin);
     }
 
@@ -52,7 +52,7 @@ public abstract class AbstractCoapServerSideRpcDefaultIntegrationTest extends Ab
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"25\",\"value\": 1}}";
         String nonExistentDeviceId = Uuids.timeBased().toString();
 
-        String result = doPostAsync("/api/plugins/rpc/oneway/" + nonExistentDeviceId, setGpioRequest, String.class,
+        String result = doPostAsync("/api/rpc/oneway/" + nonExistentDeviceId, setGpioRequest, String.class,
                 status().isNotFound());
         Assert.assertEquals(AccessValidator.DEVICE_WITH_REQUESTED_ID_NOT_FOUND, result);
     }
@@ -62,7 +62,7 @@ public abstract class AbstractCoapServerSideRpcDefaultIntegrationTest extends Ab
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"27\",\"value\": 1},\"timeout\": 6000}";
         String deviceId = savedDevice.getId().getId().toString();
 
-        doPostAsync("/api/plugins/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().is(409),
+        doPostAsync("/api/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().is(504),
                 asyncContextTimeoutToUseRpcPlugin);
     }
 
@@ -71,7 +71,7 @@ public abstract class AbstractCoapServerSideRpcDefaultIntegrationTest extends Ab
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"28\",\"value\": 1}}";
         String nonExistentDeviceId = Uuids.timeBased().toString();
 
-        String result = doPostAsync("/api/plugins/rpc/twoway/" + nonExistentDeviceId, setGpioRequest, String.class,
+        String result = doPostAsync("/api/rpc/twoway/" + nonExistentDeviceId, setGpioRequest, String.class,
                 status().isNotFound());
         Assert.assertEquals(AccessValidator.DEVICE_WITH_REQUESTED_ID_NOT_FOUND, result);
     }

application/src/test/java/org/thingsboard/server/transport/coap/rpc/AbstractCoapServerSideRpcIntegrationTest.java

@@ -71,7 +71,7 @@ public abstract class AbstractCoapServerSideRpcIntegrationTest extends AbstractC
 
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"23\",\"value\": 1}}";
         String deviceId = savedDevice.getId().getId().toString();
-        String result = doPostAsync("/api/plugins/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().isOk());
+        String result = doPostAsync("/api/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().isOk());
 
         latch.await(3, TimeUnit.SECONDS);
 
@@ -99,14 +99,14 @@ public abstract class AbstractCoapServerSideRpcIntegrationTest extends AbstractC
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"26\",\"value\": 1}}";
         String deviceId = savedDevice.getId().getId().toString();
 
-        String actualResult = doPostAsync("/api/plugins/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().isOk());
+        String actualResult = doPostAsync("/api/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().isOk());
         latch.await(3, TimeUnit.SECONDS);
 
         validateTwoWayStateChangedNotification(callback, 1, expectedResponseResult, actualResult);
 
         latch = new CountDownLatch(1);
 
-        actualResult = doPostAsync("/api/plugins/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().isOk());
+        actualResult = doPostAsync("/api/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().isOk());
         latch.await(3, TimeUnit.SECONDS);
 
         validateTwoWayStateChangedNotification(callback, 2, expectedResponseResult, actualResult);

application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcDefaultIntegrationTest.java

@@ -46,7 +46,7 @@ public abstract class AbstractMqttServerSideRpcDefaultIntegrationTest extends Ab
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"24\",\"value\": 1},\"timeout\": 6000}";
         String deviceId = savedDevice.getId().getId().toString();
 
-        doPostAsync("/api/plugins/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().is(409),
+        doPostAsync("/api/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().is(504),
                 asyncContextTimeoutToUseRpcPlugin);
     }
 
@@ -55,7 +55,7 @@ public abstract class AbstractMqttServerSideRpcDefaultIntegrationTest extends Ab
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"25\",\"value\": 1}}";
         String nonExistentDeviceId = Uuids.timeBased().toString();
 
-        String result = doPostAsync("/api/plugins/rpc/oneway/" + nonExistentDeviceId, setGpioRequest, String.class,
+        String result = doPostAsync("/api/rpc/oneway/" + nonExistentDeviceId, setGpioRequest, String.class,
                 status().isNotFound());
         Assert.assertEquals(AccessValidator.DEVICE_WITH_REQUESTED_ID_NOT_FOUND, result);
     }
@@ -65,7 +65,7 @@ public abstract class AbstractMqttServerSideRpcDefaultIntegrationTest extends Ab
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"27\",\"value\": 1},\"timeout\": 6000}";
         String deviceId = savedDevice.getId().getId().toString();
 
-        doPostAsync("/api/plugins/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().is(409),
+        doPostAsync("/api/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().is(504),
                 asyncContextTimeoutToUseRpcPlugin);
     }
 
@@ -74,7 +74,7 @@ public abstract class AbstractMqttServerSideRpcDefaultIntegrationTest extends Ab
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"28\",\"value\": 1}}";
         String nonExistentDeviceId = Uuids.timeBased().toString();
 
-        String result = doPostAsync("/api/plugins/rpc/twoway/" + nonExistentDeviceId, setGpioRequest, String.class,
+        String result = doPostAsync("/api/rpc/twoway/" + nonExistentDeviceId, setGpioRequest, String.class,
                 status().isNotFound());
         Assert.assertEquals(AccessValidator.DEVICE_WITH_REQUESTED_ID_NOT_FOUND, result);
     }

application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java

@@ -69,7 +69,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
 
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"23\",\"value\": 1}}";
         String deviceId = savedDevice.getId().getId().toString();
-        String result = doPostAsync("/api/plugins/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().isOk());
+        String result = doPostAsync("/api/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().isOk());
         Assert.assertTrue(StringUtils.isEmpty(result));
         latch.await(3, TimeUnit.SECONDS);
         assertEquals(MqttQoS.AT_MOST_ONCE.value(), callback.getQoS());
@@ -95,7 +95,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"26\",\"value\": 1}}";
         String deviceId = savedDevice.getId().getId().toString();
 
-        String result = doPostAsync("/api/plugins/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().isOk());
+        String result = doPostAsync("/api/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().isOk());
         String expected = "{\"value1\":\"A\",\"value2\":\"B\"}";
         latch.await(3, TimeUnit.SECONDS);
         Assert.assertEquals(expected, result);
@@ -130,7 +130,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
 
         String setGpioRequest = "{\"method\": \"toggle_gpio\", \"params\": {\"pin\":1}}";
         String deviceId = savedDevice.getId().getId().toString();
-        String result = doPostAsync("/api/plugins/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().isOk());
+        String result = doPostAsync("/api/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().isOk());
         Assert.assertTrue(StringUtils.isEmpty(result));
         latch.await(3, TimeUnit.SECONDS);
         assertEquals(MqttQoS.AT_MOST_ONCE.value(), callback.getQoS());
@@ -156,7 +156,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
 
         String setGpioRequest = "{\"method\": \"toggle_gpio\", \"params\": {\"pin\":1}}";
         String deviceId = savedDevice.getId().getId().toString();
-        String result = doPostAsync("/api/plugins/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().isOk());
+        String result = doPostAsync("/api/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().isOk());
         latch.await(3, TimeUnit.SECONDS);
         String expected = "{\"success\":true}";
         assertEquals(expected, result);

application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcProtoIntegrationTest.java

@@ -131,7 +131,7 @@ public abstract class AbstractMqttServerSideRpcProtoIntegrationTest extends Abst
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"26\",\"value\": 1}}";
         String deviceId = savedDevice.getId().getId().toString();
 
-        String result = doPostAsync("/api/plugins/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().isOk());
+        String result = doPostAsync("/api/rpc/twoway/" + deviceId, setGpioRequest, String.class, status().isOk());
         String expected = "{\"payload\":\"{\\\"value1\\\":\\\"A\\\",\\\"value2\\\":\\\"B\\\"}\"}";
         latch.await(3, TimeUnit.SECONDS);
         Assert.assertEquals(expected, result);

common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java

@@ -77,96 +77,89 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler {
     private final LwM2mUplinkMsgHandler uplinkHandler;
     private final LwM2mDownlinkMsgHandler downlinkHandler;
     private final LwM2MTelemetryLogService logService;
-    private final Map<UUID, Long> rpcSubscriptions = new ConcurrentHashMap<>();
 
     @Override
     public void onToDeviceRpcRequest(TransportProtos.ToDeviceRpcRequestMsg rpcRequest, TransportProtos.SessionInfoProto sessionInfo) {
-        this.cleanupOldSessions();
-        UUID requestUUID = new UUID(rpcRequest.getRequestIdMSB(), rpcRequest.getRequestIdLSB());
         log.debug("Received params: {}", rpcRequest.getParams());
-        // We use this map to protect from browser issue that the same command is sent twice.
+        LwM2mOperationType operationType = LwM2mOperationType.fromType(rpcRequest.getMethodName());
-        // TODO: This is probably not the best place and should be moved to DeviceActor
+        if (operationType == null) {
-        if (!this.rpcSubscriptions.containsKey(requestUUID)) {
+            this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.METHOD_NOT_ALLOWED, "Unsupported operation type: " + rpcRequest.getMethodName());
-            LwM2mOperationType operationType = LwM2mOperationType.fromType(rpcRequest.getMethodName());
+            return;
-            if (operationType == null) {
+        }
-                this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.METHOD_NOT_ALLOWED, "Unsupported operation type: " + rpcRequest.getMethodName());
+        LwM2mClient client = clientContext.getClientBySessionInfo(sessionInfo);
-                return;
+        if (client.getRegistration() == null) {
-            }
+            this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.INTERNAL_SERVER_ERROR, "Registration is empty");
-            LwM2mClient client = clientContext.getClientBySessionInfo(sessionInfo);
+            return;
-            if (client.getRegistration() == null) {
+        }
-                this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.INTERNAL_SERVER_ERROR, "Registration is empty");
+        try {
-                return;
+            if (operationType.isHasObjectId()) {
-            }
+                String objectId = getIdFromParameters(client, rpcRequest);
-            try {
+                switch (operationType) {
-                if (operationType.isHasObjectId()) {
+                    case READ:
-                    String objectId = getIdFromParameters(client, rpcRequest);
+                        sendReadRequest(client, rpcRequest, objectId);
+                        break;
+                    case OBSERVE:
+                        sendObserveRequest(client, rpcRequest, objectId);
+                        break;
+                    case DISCOVER:
+                        sendDiscoverRequest(client, rpcRequest, objectId);
+                        break;
+                    case EXECUTE:
+                        sendExecuteRequest(client, rpcRequest, objectId);
+                        break;
+                    case WRITE_ATTRIBUTES:
+                        sendWriteAttributesRequest(client, rpcRequest, objectId);
+                        break;
+                    case OBSERVE_CANCEL:
+                        sendCancelObserveRequest(client, rpcRequest, objectId);
+                        break;
+                    case DELETE:
+                        sendDeleteRequest(client, rpcRequest, objectId);
+                        break;
+                    case WRITE_UPDATE:
+                        sendWriteUpdateRequest(client, rpcRequest, objectId);
+                        break;
+                    case WRITE_REPLACE:
+                        sendWriteReplaceRequest(client, rpcRequest, objectId);
+                        break;
+                    default:
+                        throw new IllegalArgumentException("Unsupported operation: " + operationType.name());
+                }
+            } else if (operationType.isComposite()) {
+                if (clientContext.isComposite(client)) {
                     switch (operationType) {
-                        case READ:
+                        case READ_COMPOSITE:
-                            sendReadRequest(client, rpcRequest, objectId);
+                            sendReadCompositeRequest(client, rpcRequest);
-                            break;
-                        case OBSERVE:
-                            sendObserveRequest(client, rpcRequest, objectId);
-                            break;
-                        case DISCOVER:
-                            sendDiscoverRequest(client, rpcRequest, objectId);
-                            break;
-                        case EXECUTE:
-                            sendExecuteRequest(client, rpcRequest, objectId);
-                            break;
-                        case WRITE_ATTRIBUTES:
-                            sendWriteAttributesRequest(client, rpcRequest, objectId);
-                            break;
-                        case OBSERVE_CANCEL:
-                            sendCancelObserveRequest(client, rpcRequest, objectId);
-                            break;
-                        case DELETE:
-                            sendDeleteRequest(client, rpcRequest, objectId);
-                            break;
-                        case WRITE_UPDATE:
-                            sendWriteUpdateRequest(client, rpcRequest, objectId);
                             break;
-                        case WRITE_REPLACE:
+                        case WRITE_COMPOSITE:
-                            sendWriteReplaceRequest(client, rpcRequest, objectId);
+                            sendWriteCompositeRequest(client, rpcRequest);
                             break;
                         default:
                             throw new IllegalArgumentException("Unsupported operation: " + operationType.name());
                     }
-                } else if (operationType.isComposite()) {
-                    if (clientContext.isComposite(client)) {
-                        switch (operationType) {
-                            case READ_COMPOSITE:
-                                sendReadCompositeRequest(client, rpcRequest);
-                                break;
-                            case WRITE_COMPOSITE:
-                                sendWriteCompositeRequest(client, rpcRequest);
-                                break;
-                            default:
-                                throw new IllegalArgumentException("Unsupported operation: " + operationType.name());
-                        }
-                    } else {
-                        this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(),
-                                ResponseCode.INTERNAL_SERVER_ERROR, "This device does not support Composite Operation");
-                    }
                 } else {
-                    switch (operationType) {
+                    this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(),
-                        case OBSERVE_CANCEL_ALL:
+                            ResponseCode.INTERNAL_SERVER_ERROR, "This device does not support Composite Operation");
-                            sendCancelAllObserveRequest(client, rpcRequest);
+                }
-                            break;
+            } else {
-                        case OBSERVE_READ_ALL:
+                switch (operationType) {
-                            sendObserveAllRequest(client, rpcRequest);
+                    case OBSERVE_CANCEL_ALL:
-                            break;
+                        sendCancelAllObserveRequest(client, rpcRequest);
-                        case DISCOVER_ALL:
+                        break;
-                            sendDiscoverAllRequest(client, rpcRequest);
+                    case OBSERVE_READ_ALL:
-                            break;
+                        sendObserveAllRequest(client, rpcRequest);
-                        case FW_UPDATE:
+                        break;
-                            //TODO: implement and add break statement
+                    case DISCOVER_ALL:
-                        default:
+                        sendDiscoverAllRequest(client, rpcRequest);
-                            throw new IllegalArgumentException("Unsupported operation: " + operationType.name());
+                        break;
-                    }
+                    case FW_UPDATE:
+                        //TODO: implement and add break statement
+                    default:
+                        throw new IllegalArgumentException("Unsupported operation: " + operationType.name());
                 }
-            } catch (IllegalArgumentException e) {
-                this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.BAD_REQUEST, e.getMessage());
             }
+        } catch (IllegalArgumentException e) {
+            this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.BAD_REQUEST, e.getMessage());
         }
     }
 
@@ -318,17 +311,6 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler {
         transportService.process(sessionInfo, msg, null);
     }
 
-    private void cleanupOldSessions() {
-        log.debug("Before rpcSubscriptions.size(): [{}]", rpcSubscriptions.size());
-        if (rpcSubscriptions.size() > 0) {
-            long currentTime = System.currentTimeMillis();
-            Set<UUID> rpcSubscriptionsToRemove = rpcSubscriptions.entrySet().stream().filter(kv -> currentTime > kv.getValue()).map(Map.Entry::getKey).collect(Collectors.toSet());
-            log.debug("RpcSubscriptionsToRemove: [{}]", rpcSubscriptionsToRemove);
-            rpcSubscriptionsToRemove.forEach(rpcSubscriptions::remove);
-        }
-        log.debug("After rpcSubscriptions.size(): [{}]", rpcSubscriptions.size());
-    }
-
     @Override
     public void onToDeviceRpcResponse(TransportProtos.ToDeviceRpcResponseMsg toDeviceResponse, TransportProtos.SessionInfoProto sessionInfo) {
         log.debug("OnToDeviceRpcResponse: [{}], sessionUUID: [{}]", toDeviceResponse, new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB()));

msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java

@@ -267,7 +267,7 @@ public class MqttClientTest extends AbstractContainerTest {
         ListenableFuture<ResponseEntity> future = service.submit(() -> {
             try {
                 return restClient.getRestTemplate()
-                        .postForEntity(HTTPS_URL + "/api/plugins/rpc/twoway/{deviceId}",
+                        .postForEntity(HTTPS_URL + "/api/rpc/twoway/{deviceId}",
                                 mapper.readTree(serverRpcPayload.toString()), String.class,
                                 device.getId());
             } catch (IOException e) {

msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttGatewayClientTest.java

@@ -263,7 +263,7 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
         ListenableFuture<ResponseEntity> future = service.submit(() -> {
             try {
                 return restClient.getRestTemplate()
-                        .postForEntity(HTTPS_URL + "/api/plugins/rpc/twoway/{deviceId}",
+                        .postForEntity(HTTPS_URL + "/api/rpc/twoway/{deviceId}",
                                 mapper.readTree(serverRpcPayload.toString()), String.class,
                                 createdDevice.getId());
             } catch (IOException e) {

rest-client/src/main/java/org/thingsboard/rest/client/RestClient.java

@@ -1811,12 +1811,12 @@ public class RestClient implements ClientHttpRequestInterceptor, Closeable {
     }
 
     public void handleOneWayDeviceRPCRequest(DeviceId deviceId, JsonNode requestBody) {
-        restTemplate.postForLocation(baseURL + "/api/plugins/rpc/oneway/{deviceId}", requestBody, deviceId.getId());
+        restTemplate.postForLocation(baseURL + "/api/rpc/oneway/{deviceId}", requestBody, deviceId.getId());
     }
 
     public JsonNode handleTwoWayDeviceRPCRequest(DeviceId deviceId, JsonNode requestBody) {
         return restTemplate.exchange(
-                baseURL + "/api/plugins/rpc/twoway/{deviceId}",
+                baseURL + "/api/rpc/twoway/{deviceId}",
                 HttpMethod.POST,
                 new HttpEntity<>(requestBody),
                 new ParameterizedTypeReference<JsonNode>() {

ui-ngx/src/app/core/api/widget-subscription.ts

@@ -663,7 +663,7 @@ export class WidgetSubscription implements IWidgetSubscription {
     if (!this.rpcEnabled) {
       return throwError(new Error('Rpc disabled!'));
     } else {
-      if (this.rpcRejection && this.rpcRejection.status !== 408) {
+      if (this.rpcRejection && this.rpcRejection.status !== 504) {
         this.rpcRejection = null;
         this.rpcErrorText = null;
         this.callbacks.onRpcErrorCleared(this);
@@ -715,12 +715,10 @@ export class WidgetSubscription implements IWidgetSubscription {
           }
           this.executingRpcRequest = this.executingSubjects.length > 0;
           this.callbacks.rpcStateChanged(this);
-          if (!this.executingRpcRequest || rejection.status === 408) {
+          if (!this.executingRpcRequest || rejection.status === 504) {
             this.rpcRejection = rejection;
-            if (rejection.status === 408) {
+            if (rejection.status === 504) {
               this.rpcErrorText = 'Request Timeout.';
-            } else if (rejection.status === 409) {
-              this.rpcErrorText = 'Device is offline.';
             } else {
               this.rpcErrorText =  'Error : ' + rejection.status + ' - ' + rejection.statusText;
               const error = this.extractRejectionErrorText(rejection);

ui-ngx/src/app/core/http/device.service.ts

@@ -130,11 +130,11 @@ export class DeviceService {
   }
 
   public sendOneWayRpcCommand(deviceId: string, requestBody: any, config?: RequestConfig): Observable<any> {
-    return this.http.post<Device>(`/api/plugins/rpc/oneway/${deviceId}`, requestBody, defaultHttpOptionsFromConfig(config));
+    return this.http.post<Device>(`/api/rpc/oneway/${deviceId}`, requestBody, defaultHttpOptionsFromConfig(config));
   }
 
   public sendTwoWayRpcCommand(deviceId: string, requestBody: any, config?: RequestConfig): Observable<any> {
-    return this.http.post<Device>(`/api/plugins/rpc/twoway/${deviceId}`, requestBody, defaultHttpOptionsFromConfig(config));
+    return this.http.post<Device>(`/api/rpc/twoway/${deviceId}`, requestBody, defaultHttpOptionsFromConfig(config));
   }
 
   public findByQuery(query: DeviceSearchQuery,

ui-ngx/src/app/core/interceptors/global-http-interceptor.ts

@@ -47,7 +47,7 @@ export class GlobalHttpInterceptor implements HttpInterceptor {
 
   private internalUrlPrefixes = [
     '/api/auth/token',
-    '/api/plugins/rpc'
+    '/api/rpc'
   ];
 
   private activeRequests = 0;
@@ -142,7 +142,7 @@ export class GlobalHttpInterceptor implements HttpInterceptor {
       }
     } else if (errorResponse.status === 0 || errorResponse.status === -1) {
         this.showError('Unable to connect');
-    } else if (!req.url.startsWith('/api/plugins/rpc')) {
+    } else if (!(req.url.startsWith('/api/rpc') || req.url.startsWith('/api/plugins/rpc'))) {
       if (errorResponse.status === 404) {
         if (!ignoreErrors) {
           this.showError(req.method + ': ' + req.url + '<br/>' +