dshvaika
3 weeks ago
14 changed files with 650 additions and 42 deletions
@ -0,0 +1,154 @@ |
|||
/** |
|||
* 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.mockito.ArgumentCaptor; |
|||
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.msg.TbMsgType; |
|||
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.List; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.CountDownLatch; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.junit.Assert.assertEquals; |
|||
import static org.junit.Assert.assertTrue; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.eq; |
|||
import static org.mockito.ArgumentMatchers.isNull; |
|||
import static org.mockito.Mockito.after; |
|||
import static org.mockito.Mockito.doAnswer; |
|||
import static org.mockito.Mockito.mock; |
|||
import static org.mockito.Mockito.never; |
|||
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, 1); |
|||
} |
|||
|
|||
@Test |
|||
public void updatePersistsViaUpdateAsyncThenPushesToRuleEngine() { |
|||
Rpc rpc = newRpc(); |
|||
when(rpcService.updateAsync(rpc)).thenReturn(Futures.immediateFuture(true)); |
|||
|
|||
tbRpcService.update(rpc.getTenantId(), rpc); |
|||
|
|||
verify(rpcService).updateAsync(rpc); |
|||
verify(clusterService, timeout(5000)) |
|||
.pushMsgToRuleEngine(eq(rpc.getTenantId()), eq(rpc.getDeviceId()), any(TbMsg.class), isNull()); |
|||
} |
|||
|
|||
@Test |
|||
public void createPersistsSynchronouslyThenPushesToRuleEngine() { |
|||
Rpc rpc = newRpc(); |
|||
when(rpcService.save(rpc)).thenReturn(rpc); |
|||
|
|||
tbRpcService.create(rpc.getTenantId(), rpc); |
|||
|
|||
// create must persist synchronously via save(...), never via the async batch path
|
|||
verify(rpcService).save(rpc); |
|||
verify(rpcService, never()).createAsync(any()); |
|||
verify(clusterService, timeout(5000)) |
|||
.pushMsgToRuleEngine(eq(rpc.getTenantId()), eq(rpc.getDeviceId()), any(TbMsg.class), isNull()); |
|||
} |
|||
|
|||
@Test |
|||
public void updateDoesNotNotifyRuleEngineWhenRowMissing() { |
|||
Rpc rpc = newRpc(); |
|||
// updateAsync resolves false: the UPDATE matched no row (RPC was deleted), so the rule engine
|
|||
// must not be notified for a status change that never persisted.
|
|||
when(rpcService.updateAsync(rpc)).thenReturn(Futures.immediateFuture(false)); |
|||
|
|||
tbRpcService.update(rpc.getTenantId(), rpc); |
|||
|
|||
verify(rpcService).updateAsync(rpc); |
|||
verify(clusterService, after(500).never()) |
|||
.pushMsgToRuleEngine(any(TenantId.class), any(DeviceId.class), any(TbMsg.class), isNull()); |
|||
} |
|||
|
|||
@Test |
|||
public void sameRpcIdUpdateNotificationsRunInSubmissionOrderOnStripedExecutor() throws InterruptedException { |
|||
// A single rpcId always maps to one stripe; this checks that two update notifications for the
|
|||
// SAME rpcId are serialized on that stripe (DELIVERED before SUCCESSFUL).
|
|||
tbRpcService = new TbRpcService(rpcService, clusterService, 3); |
|||
|
|||
RpcId rpcId = new RpcId(UUID.randomUUID()); |
|||
DeviceId deviceId = new DeviceId(UUID.randomUUID()); |
|||
Rpc delivered = newRpc(rpcId, deviceId, RpcStatus.DELIVERED); |
|||
Rpc successful = newRpc(rpcId, deviceId, RpcStatus.SUCCESSFUL); |
|||
when(rpcService.updateAsync(delivered)).thenReturn(Futures.immediateFuture(true)); |
|||
when(rpcService.updateAsync(successful)).thenReturn(Futures.immediateFuture(true)); |
|||
|
|||
// Make the DELIVERED notification block while it runs (signal start, then sleep) - holding the
|
|||
// stripe. If the two callbacks for this rpcId were NOT serialized on one stripe, the fast
|
|||
// SUCCESSFUL one would overtake the sleeping DELIVERED one and be recorded first.
|
|||
CountDownLatch deliveredStarted = new CountDownLatch(1); |
|||
doAnswer(invocation -> { |
|||
TbMsg msg = invocation.getArgument(2); |
|||
if (msg.getInternalType() == TbMsgType.RPC_DELIVERED) { |
|||
deliveredStarted.countDown(); |
|||
Thread.sleep(300); |
|||
} |
|||
return null; |
|||
}).when(clusterService).pushMsgToRuleEngine(eq(TenantId.SYS_TENANT_ID), eq(deviceId), any(TbMsg.class), isNull()); |
|||
|
|||
tbRpcService.update(delivered.getTenantId(), delivered); |
|||
// Don't submit SUCCESSFUL until DELIVERED is in flight (and now sleeping) on the stripe.
|
|||
assertTrue(deliveredStarted.await(5, TimeUnit.SECONDS)); |
|||
tbRpcService.update(successful.getTenantId(), successful); |
|||
|
|||
ArgumentCaptor<TbMsg> msgCaptor = ArgumentCaptor.forClass(TbMsg.class); |
|||
verify(clusterService, timeout(5000).times(2)) |
|||
.pushMsgToRuleEngine(eq(TenantId.SYS_TENANT_ID), eq(deviceId), msgCaptor.capture(), isNull()); |
|||
|
|||
List<TbMsg> msgs = msgCaptor.getAllValues(); |
|||
assertEquals(TbMsgType.RPC_DELIVERED, msgs.get(0).getInternalType()); |
|||
assertEquals(TbMsgType.RPC_SUCCESSFUL, msgs.get(1).getInternalType()); |
|||
} |
|||
|
|||
private Rpc newRpc() { |
|||
return newRpc(new RpcId(UUID.randomUUID()), new DeviceId(UUID.randomUUID()), RpcStatus.QUEUED); |
|||
} |
|||
|
|||
private Rpc newRpc(RpcId rpcId, DeviceId deviceId, RpcStatus status) { |
|||
Rpc rpc = new Rpc(rpcId); |
|||
rpc.setTenantId(TenantId.SYS_TENANT_ID); |
|||
rpc.setDeviceId(deviceId); |
|||
rpc.setStatus(status); |
|||
rpc.setRequest(JacksonUtil.toJsonNode("{}")); |
|||
return rpc; |
|||
} |
|||
} |
|||
@ -0,0 +1,32 @@ |
|||
/** |
|||
* 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.common.util; |
|||
|
|||
/** |
|||
* Maps a hash code to a non-negative partition index in {@code [0, partitions)}. |
|||
*/ |
|||
public final class HashPartitioner { |
|||
|
|||
private HashPartitioner() { |
|||
} |
|||
|
|||
public static int resolvePartition(int hashCode, int partitions) { |
|||
if (partitions <= 0) { |
|||
throw new IllegalArgumentException("partitions must be > 0, but was " + partitions); |
|||
} |
|||
return (hashCode & 0x7FFFFFFF) % partitions; |
|||
} |
|||
} |
|||
@ -0,0 +1,99 @@ |
|||
/** |
|||
* 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.dao.sql.rpc; |
|||
|
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import org.springframework.jdbc.core.BatchPreparedStatementSetter; |
|||
import org.springframework.stereotype.Repository; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.server.dao.model.sql.RpcEntity; |
|||
import org.thingsboard.server.dao.sqlts.insert.AbstractInsertRepository; |
|||
|
|||
import java.sql.PreparedStatement; |
|||
import java.sql.SQLException; |
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Repository |
|||
public class RpcInsertRepository extends AbstractInsertRepository { |
|||
|
|||
private static final String INSERT = |
|||
"INSERT INTO rpc (id, created_time, tenant_id, device_id, expiration_time, request, response, additional_info, status) " + |
|||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) " + |
|||
"ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, response = COALESCE(EXCLUDED.response, rpc.response);"; |
|||
|
|||
private static final String UPDATE = |
|||
"UPDATE rpc SET status = ?, response = COALESCE(?, response) WHERE id = ?;"; |
|||
|
|||
@FunctionalInterface |
|||
private interface ColumnBinder { |
|||
void bind(PreparedStatement ps, RpcEntity rpc) throws SQLException; |
|||
} |
|||
|
|||
List<Boolean> saveOrUpdate(List<RpcQueueEntry> entries) { |
|||
Map<Boolean, List<RpcEntity>> byIntent = entries.stream().collect(Collectors.partitioningBy( |
|||
RpcQueueEntry::insert, Collectors.mapping(RpcQueueEntry::entity, Collectors.toList()))); |
|||
List<RpcEntity> inserts = byIntent.get(true); |
|||
List<RpcEntity> updates = byIntent.get(false); |
|||
return transactionTemplate.execute(status -> { |
|||
if (!inserts.isEmpty()) { |
|||
batch(INSERT, inserts, (ps, rpc) -> { |
|||
ps.setObject(1, rpc.getUuid()); |
|||
ps.setLong(2, rpc.getCreatedTime()); |
|||
ps.setObject(3, rpc.getTenantId()); |
|||
ps.setObject(4, rpc.getDeviceId()); |
|||
ps.setLong(5, rpc.getExpirationTime()); |
|||
ps.setString(6, toJsonStr(rpc.getRequest())); |
|||
ps.setString(7, toJsonStr(rpc.getResponse())); |
|||
ps.setString(8, toJsonStr(rpc.getAdditionalInfo())); |
|||
ps.setString(9, rpc.getStatus().name()); |
|||
}); |
|||
} |
|||
int[] updateCounts = updates.isEmpty() ? new int[0] : batch(UPDATE, updates, (ps, rpc) -> { |
|||
ps.setString(1, rpc.getStatus().name()); |
|||
ps.setString(2, toJsonStr(rpc.getResponse())); |
|||
ps.setObject(3, rpc.getUuid()); |
|||
}); |
|||
|
|||
List<Boolean> persisted = new ArrayList<>(entries.size()); |
|||
int updateIdx = 0; |
|||
for (RpcQueueEntry entry : entries) { |
|||
persisted.add(entry.insert() || updateCounts[updateIdx++] > 0); |
|||
} |
|||
return persisted; |
|||
}); |
|||
} |
|||
|
|||
private int[] batch(String sql, List<RpcEntity> entities, ColumnBinder binder) { |
|||
return jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() { |
|||
@Override |
|||
public void setValues(PreparedStatement ps, int i) throws SQLException { |
|||
binder.bind(ps, entities.get(i)); |
|||
} |
|||
|
|||
@Override |
|||
public int getBatchSize() { |
|||
return entities.size(); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private String toJsonStr(JsonNode node) { |
|||
return node == null ? null : replaceNullChars(JacksonUtil.toString(node)); |
|||
} |
|||
} |
|||
@ -0,0 +1,29 @@ |
|||
/** |
|||
* 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.dao.sql.rpc; |
|||
|
|||
import org.thingsboard.server.dao.model.sql.RpcEntity; |
|||
|
|||
record RpcQueueEntry(RpcEntity entity, boolean insert) { |
|||
|
|||
static RpcQueueEntry forInsert(RpcEntity entity) { |
|||
return new RpcQueueEntry(entity, true); |
|||
} |
|||
|
|||
static RpcQueueEntry forUpdate(RpcEntity entity) { |
|||
return new RpcQueueEntry(entity, false); |
|||
} |
|||
} |
|||
Loading…
Reference in new issue