Shvaika Dmytro
2 days ago
committed by
GitHub
GitHub
8 changed files with 241 additions and 17 deletions
@ -0,0 +1,75 @@ |
|||
/** |
|||
* 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.transport.mqtt.mqttv3.rpc; |
|||
|
|||
import io.netty.handler.codec.mqtt.MqttQoS; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.junit.Before; |
|||
import org.junit.Test; |
|||
import org.springframework.test.context.TestPropertySource; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.server.common.data.TransportPayloadType; |
|||
import org.thingsboard.server.common.data.rpc.RpcStatus; |
|||
import org.thingsboard.server.dao.service.DaoSqlTest; |
|||
import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; |
|||
|
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.junit.Assert.assertEquals; |
|||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; |
|||
|
|||
@Slf4j |
|||
@DaoSqlTest |
|||
@TestPropertySource(properties = { |
|||
"actors.rpc.close_session_on_rpc_delivery_timeout=true", |
|||
"transport.mqtt.timeout=100", |
|||
}) |
|||
public class MqttGatewayServerSideRpcDeliveryTimeoutIntegrationTest extends AbstractMqttServerSideRpcIntegrationTest { |
|||
|
|||
@Before |
|||
public void beforeTest() throws Exception { |
|||
MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() |
|||
.deviceName("RPC timeout test device") |
|||
.gatewayName("RPC timeout test gateway") |
|||
.transportPayloadType(TransportPayloadType.JSON) |
|||
.build(); |
|||
processBeforeTest(configProperties); |
|||
} |
|||
|
|||
@Test |
|||
public void testGatewayPersistentRpcTimeoutWhenNoPuback() throws Exception { |
|||
// manualAcks = true → the gateway client never PUBACKs the QoS1 downlink.
|
|||
GatewayRpcSession session = connectGatewayAndSubscribeForRpc("Gateway Device Persistent RPC Timeout Json", true); |
|||
|
|||
long expirationTime = System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1); |
|||
String rpcRequest = "{\"method\":\"toggle_gpio\",\"params\":{\"pin\":1},\"persistent\":true,\"retries\":0,\"expirationTime\":" + expirationTime + "}"; |
|||
String response = doPostAsync("/api/rpc/twoway/" + session.device().getId().getId().toString(), rpcRequest, String.class, status().isOk()); |
|||
String rpcId = JacksonUtil.toJsonNode(response).get("rpcId").asText(); |
|||
|
|||
session.callback().getSubscribeLatch().await(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS); |
|||
// Confirm the downlink arrived at QoS1 — a QoS0 downlink would produce no PUBACK path at all,
|
|||
// so the timeout would fire for the wrong reason and mask a regression.
|
|||
assertEquals(MqttQoS.AT_LEAST_ONCE.value(), session.callback().getMessageArrivedQoS()); |
|||
|
|||
// No PUBACK is ever sent. The gateway awaiting-ack scheduler emits TIMEOUT after
|
|||
// transport.mqtt.timeout (100 ms); the RPC status reverts to QUEUED (re-queued via the
|
|||
// close_session_on_rpc_delivery_timeout path).
|
|||
awaitRpcStatus(rpcId, RpcStatus.QUEUED, 50, 100); |
|||
|
|||
session.client().disconnect(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,35 @@ |
|||
/** |
|||
* 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.transport.mqtt.util; |
|||
|
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcRequestMsg; |
|||
|
|||
public final class MqttRpcStatusUtil { |
|||
|
|||
private MqttRpcStatusUtil() { |
|||
} |
|||
|
|||
/** |
|||
* Whether the transport should track and emit a delivery status (DELIVERED / TIMEOUT) for the given RPC. |
|||
* Non-persistent one-way RPCs self-complete on send in the device actor and are never added to its pending |
|||
* map, so a delivery status for them lands on nothing and only produces a benign |
|||
* "RPC has already been removed from pending map" warning. Every other RPC is tracked. |
|||
*/ |
|||
public static boolean requireDeliveryTracking(ToDeviceRpcRequestMsg rpc) { |
|||
return !(rpc.getOneway() && !rpc.getPersisted()); |
|||
} |
|||
|
|||
} |
|||
Loading…
Reference in new issue