dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improved persistent RPC for unack requests

Browse Source
This commit is contained in:
Andrii Shvaika 2021-07-14 14:23:54 +03:00
parent e914425b22
commit 0c60e18ea6
2 changed files with 35 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java
View File

@ -492,36 +492,37 @@ public class CoapTransportResource extends AbstractCoapTransportResource {
@Override @Override
public void onToDeviceRpcRequest(UUID sessionId, TransportProtos.ToDeviceRpcRequestMsg msg) { public void onToDeviceRpcRequest(UUID sessionId, TransportProtos.ToDeviceRpcRequestMsg msg) {
log.trace("[{}] Received RPC command to device", sessionId); log.trace("[{}] Received RPC command to device", sessionId);
boolean sent = false;
try { try {
Response response = coapTransportAdaptor.convertToPublish(isConRequest(), msg, rpcRequestDynamicMessageBuilder); Response response = coapTransportAdaptor.convertToPublish(isConRequest(), msg, rpcRequestDynamicMessageBuilder);
int requestId = getNextMsgId(); int requestId = getNextMsgId();
response.setMID(requestId); response.setMID(requestId);
if (msg.getPersisted()) { if (msg.getPersisted() && isConRequest()) {
if (isConRequest()) { transportContext.getRpcAwaitingAck().put(requestId, msg);
transportContext.getRpcAwaitingAck().put(requestId, msg); transportContext.getScheduler().schedule(() -> {
transportContext.getScheduler().schedule(() -> { TransportProtos.ToDeviceRpcRequestMsg awaitingAckMsg = transportContext.getRpcAwaitingAck().remove(requestId);
TransportProtos.ToDeviceRpcRequestMsg awaitingAckMsg = transportContext.getRpcAwaitingAck().remove(requestId); if (awaitingAckMsg != null) {
if (awaitingAckMsg != null) { transportService.process(sessionInfo, msg, true, TransportServiceCallback.EMPTY);
transportService.process(sessionInfo, msg, true, TransportServiceCallback.EMPTY); }
} }, Math.max(0, msg.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
}, Math.max(0, msg.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS); response.addMessageObserver(new TbCoapMessageObserver(requestId, id -> {
response.addMessageObserver(new TbCoapMessageObserver(requestId, id -> { TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(id);
TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(id); if (rpcRequestMsg != null) {
if (rpcRequestMsg != null) { transportService.process(sessionInfo, rpcRequestMsg, false, TransportServiceCallback.EMPTY);
transportService.process(sessionInfo, rpcRequestMsg, false, TransportServiceCallback.EMPTY); }
} }));
}));
} else {
transportService.process(sessionInfo, msg, false, TransportServiceCallback.EMPTY);
}
} }
exchange.respond(response); exchange.respond(response);
sent = true;
} catch (AdaptorException e) { } catch (AdaptorException e) {
log.trace("Failed to reply due to error", e); log.trace("Failed to reply due to error", e);
closeObserveRelationAndNotify(sessionId, CoAP.ResponseCode.INTERNAL_SERVER_ERROR); closeObserveRelationAndNotify(sessionId, CoAP.ResponseCode.INTERNAL_SERVER_ERROR);
closeAndDeregister(); closeAndDeregister();
} finally {
if (msg.getPersisted() && !isConRequest()) {
transportService.process(sessionInfo, msg, sent, TransportServiceCallback.EMPTY);
}
} }
} }

30
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
View File

@ -17,6 +17,7 @@ package org.thingsboard.server.transport.mqtt;
import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.JsonNode;
import com.google.gson.JsonParseException; import com.google.gson.JsonParseException;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttConnAckMessage; import io.netty.handler.codec.mqtt.MqttConnAckMessage;
@ -825,20 +826,19 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
try { try {
deviceSessionCtx.getPayloadAdaptor().convertToPublish(deviceSessionCtx, rpcRequest).ifPresent(payload -> { deviceSessionCtx.getPayloadAdaptor().convertToPublish(deviceSessionCtx, rpcRequest).ifPresent(payload -> {
int msgId = ((MqttPublishMessage) payload).variableHeader().packetId(); int msgId = ((MqttPublishMessage) payload).variableHeader().packetId();
if (rpcRequest.getPersisted()) { if (rpcRequest.getPersisted() && isAckExpected(payload)) {
if (isAckExpected(payload)) { rpcAwaitingAck.put(msgId, rpcRequest);
rpcAwaitingAck.put(msgId, rpcRequest); context.getScheduler().schedule(() -> {
context.getScheduler().schedule(() -> { TransportProtos.ToDeviceRpcRequestMsg awaitingAckMsg = rpcAwaitingAck.remove(msgId);
TransportProtos.ToDeviceRpcRequestMsg awaitingAckMsg = rpcAwaitingAck.remove(msgId); if (awaitingAckMsg != null) {
if (awaitingAckMsg != null) { transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, true, TransportServiceCallback.EMPTY);
transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, true, TransportServiceCallback.EMPTY); }
} }, Math.max(0, rpcRequest.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
}, Math.max(0, rpcRequest.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS); }
} else { var cf = publish(payload, deviceSessionCtx);
transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, false, TransportServiceCallback.EMPTY); if (rpcRequest.getPersisted() && !isAckExpected(payload)) {
} cf.addListener(result -> transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, result.cause() != null, TransportServiceCallback.EMPTY));
} }
publish(payload, deviceSessionCtx);
}); });
} catch (Exception e) { } catch (Exception e) {
log.trace("[{}] Failed to convert device RPC command to MQTT msg", sessionId, e); log.trace("[{}] Failed to convert device RPC command to MQTT msg", sessionId, e);
@ -855,8 +855,8 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
} }
} }
private void publish(MqttMessage message, DeviceSessionCtx deviceSessionCtx) { private ChannelFuture publish(MqttMessage message, DeviceSessionCtx deviceSessionCtx) {
deviceSessionCtx.getChannel().writeAndFlush(message); return deviceSessionCtx.getChannel().writeAndFlush(message);
} }
private boolean isAckExpected(MqttMessage message) { private boolean isAckExpected(MqttMessage message) {