From 58d5a44c1dea06e51bb4ed02ef5151fc8c0d6ede Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 25 Jun 2026 17:08:58 +0300 Subject: [PATCH] feat(rpc): async RPC persistence with striped post-flush rule-engine notification --- .../server/service/rpc/TbRpcService.java | 49 +++++++++++-- .../server/service/rpc/TbRpcServiceTest.java | 70 +++++++++++++++++++ 2 files changed, 115 insertions(+), 4 deletions(-) create mode 100644 application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java diff --git a/application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java b/application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java index 912e8242c6..60ace279ba 100644 --- a/application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/rpc/TbRpcService.java @@ -16,10 +16,16 @@ package org.thingsboard.server.service.rpc; import com.fasterxml.jackson.databind.JsonNode; +import com.google.common.util.concurrent.ListenableFuture; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; +import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.RpcId; @@ -34,6 +40,11 @@ import org.thingsboard.server.common.msg.TbMsgMetaData; import org.thingsboard.server.dao.rpc.RpcService; import org.thingsboard.server.queue.util.TbCoreComponent; +import java.util.UUID; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + @TbCoreComponent @Service @RequiredArgsConstructor @@ -42,10 +53,40 @@ public class TbRpcService { private final RpcService rpcService; private final TbClusterService tbClusterService; - public Rpc save(TenantId tenantId, Rpc rpc) { - Rpc saved = rpcService.save(rpc); - pushRpcMsgToRuleEngine(tenantId, saved); - return saved; + @Value("${sql.rpc.callback_threads:3}") + private int callbackThreads; + + private ExecutorService[] callbackExecutors; + + @PostConstruct + private void init() { + callbackExecutors = new ExecutorService[callbackThreads]; + for (int i = 0; i < callbackThreads; i++) { + callbackExecutors[i] = Executors.newSingleThreadExecutor( + ThingsBoardThreadFactory.forName("rpc-persist-callback-" + i)); + } + } + + @PreDestroy + private void destroy() { + if (callbackExecutors != null) { + for (ExecutorService executor : callbackExecutors) { + executor.shutdownNow(); + } + } + } + + public void save(TenantId tenantId, Rpc rpc) { + ListenableFuture future = rpcService.saveAsync(rpc); + DonAsynchron.withCallback(future, + v -> pushRpcMsgToRuleEngine(tenantId, rpc), + t -> log.error("[{}][{}][{}] Failed to persist RPC with status [{}]", + tenantId, rpc.getDeviceId(), rpc.getId(), rpc.getStatus(), t), + executorFor(rpc.getUuidId())); + } + + private Executor executorFor(UUID rpcId) { + return callbackExecutors[(rpcId.hashCode() & 0x7FFFFFFF) % callbackThreads]; } public void save(TenantId tenantId, RpcId rpcId, RpcStatus newStatus, JsonNode response) { diff --git a/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java new file mode 100644 index 0000000000..d0736ada26 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/service/rpc/TbRpcServiceTest.java @@ -0,0 +1,70 @@ +/** + * Copyright © 2016-2026 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.service.rpc; + +import com.google.common.util.concurrent.Futures; +import org.junit.Before; +import org.junit.Test; +import org.springframework.test.util.ReflectionTestUtils; +import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.cluster.TbClusterService; +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.rpc.Rpc; +import org.thingsboard.server.common.data.rpc.RpcStatus; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.dao.rpc.RpcService; + +import java.util.UUID; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class TbRpcServiceTest { + + private final RpcService rpcService = mock(RpcService.class); + private final TbClusterService clusterService = mock(TbClusterService.class); + private TbRpcService tbRpcService; + + @Before + public void setUp() { + tbRpcService = new TbRpcService(rpcService, clusterService); + ReflectionTestUtils.setField(tbRpcService, "callbackThreads", 1); + ReflectionTestUtils.invokeMethod(tbRpcService, "init"); + } + + @Test + public void savePushesToRuleEngineAfterFlush() { + Rpc rpc = new Rpc(new RpcId(UUID.randomUUID())); + rpc.setTenantId(TenantId.SYS_TENANT_ID); + rpc.setDeviceId(new DeviceId(UUID.randomUUID())); + rpc.setStatus(RpcStatus.QUEUED); + rpc.setRequest(JacksonUtil.toJsonNode("{}")); + + when(rpcService.saveAsync(rpc)).thenReturn(Futures.immediateFuture(null)); + + tbRpcService.save(rpc.getTenantId(), rpc); + + verify(clusterService, timeout(5000)) + .pushMsgToRuleEngine(eq(rpc.getTenantId()), eq(rpc.getDeviceId()), any(TbMsg.class), isNull()); + } +}