From 6e71abe10bb9476095880664d945fb56c1d3a110 Mon Sep 17 00:00:00 2001 From: ShvaykaD Date: Tue, 27 Dec 2022 11:46:54 +0200 Subject: [PATCH 1/4] improvements for report device activity during RPC processing --- .../transport/coap/client/DefaultCoapClientContext.java | 2 +- .../server/rpc/RpcDownlinkRequestCallbackProxy.java | 2 +- .../server/transport/mqtt/MqttTransportHandler.java | 5 +---- .../server/common/transport/TransportService.java | 2 ++ .../transport/service/DefaultTransportService.java | 9 ++++++++- 5 files changed, 13 insertions(+), 7 deletions(-) diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java index f8336025d4..9933fd50a6 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java @@ -598,7 +598,7 @@ public class DefaultCoapClientContext implements CoapClientContext { response.addMessageObserver(new TbCoapMessageObserver(requestId, id -> { TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(id); if (rpcRequestMsg != null) { - transportService.process(state.getSession(), rpcRequestMsg, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); + transportService.process(state.getSession(), rpcRequestMsg, RpcStatus.DELIVERED, true, TransportServiceCallback.EMPTY); } }, id -> { TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(id); diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/RpcDownlinkRequestCallbackProxy.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/RpcDownlinkRequestCallbackProxy.java index c26ef2c327..5a16efeaf7 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/RpcDownlinkRequestCallbackProxy.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/RpcDownlinkRequestCallbackProxy.java @@ -65,7 +65,7 @@ public abstract class RpcDownlinkRequestCallbackProxy implements DownlinkR @Override public void onSuccess(R request, T response) { - transportService.process(client.getSession(), this.request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); + transportService.process(client.getSession(), this.request, RpcStatus.DELIVERED, true, TransportServiceCallback.EMPTY); sendRpcReplyOnSuccess(response); if (callback != null) { callback.onSuccess(request, response); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index 7515a1fcd2..eccb7e5cb5 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -21,7 +21,6 @@ import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.codec.mqtt.MqttConnAckMessage; -import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader; import io.netty.handler.codec.mqtt.MqttConnectMessage; import io.netty.handler.codec.mqtt.MqttConnectReturnCode; import io.netty.handler.codec.mqtt.MqttFixedHeader; @@ -95,11 +94,9 @@ import java.util.regex.Matcher; import java.util.regex.Pattern; import static com.amazonaws.util.StringUtils.UTF8; -import static io.netty.handler.codec.mqtt.MqttMessageType.CONNACK; import static io.netty.handler.codec.mqtt.MqttMessageType.CONNECT; import static io.netty.handler.codec.mqtt.MqttMessageType.PINGRESP; import static io.netty.handler.codec.mqtt.MqttMessageType.SUBACK; -import static io.netty.handler.codec.mqtt.MqttMessageType.UNSUBACK; import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; import static org.thingsboard.server.common.transport.service.DefaultTransportService.SESSION_EVENT_MSG_CLOSED; @@ -305,7 +302,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement int msgId = ((MqttPubAckMessage) msg).variableHeader().messageId(); TransportProtos.ToDeviceRpcRequestMsg rpcRequest = rpcAwaitingAck.remove(msgId); if (rpcRequest != null) { - transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY); + transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.DELIVERED, true, TransportServiceCallback.EMPTY); } break; default: diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java index f1d3f7e26c..ce9834834d 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java @@ -117,6 +117,8 @@ public interface TransportService { void process(SessionInfoProto sessionInfo, ToServerRpcRequestMsg msg, TransportServiceCallback callback); + void process(SessionInfoProto sessionInfo, ToDeviceRpcRequestMsg msg, RpcStatus rpcStatus, boolean reportActivity, TransportServiceCallback callback); + void process(SessionInfoProto sessionInfo, ToDeviceRpcRequestMsg msg, RpcStatus rpcStatus, TransportServiceCallback callback); void process(SessionInfoProto sessionInfo, SubscriptionInfoProto msg, TransportServiceCallback callback); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index bf03122b18..b5de6d6976 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -647,6 +647,11 @@ public class DefaultTransportService implements TransportService { @Override public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToDeviceRpcRequestMsg msg, RpcStatus rpcStatus, TransportServiceCallback callback) { + process(sessionInfo, msg, rpcStatus, false, callback); + } + + @Override + public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToDeviceRpcRequestMsg msg, RpcStatus rpcStatus, boolean reportActivity, TransportServiceCallback callback) { TransportProtos.ToDeviceRpcResponseStatusMsg responseMsg = TransportProtos.ToDeviceRpcResponseStatusMsg.newBuilder() .setRequestId(msg.getRequestId()) .setRequestIdLSB(msg.getRequestIdLSB()) @@ -655,7 +660,9 @@ public class DefaultTransportService implements TransportService { .build(); if (checkLimits(sessionInfo, responseMsg, callback)) { - reportActivityInternal(sessionInfo); + if (reportActivity) { + reportActivityInternal(sessionInfo); + } sendToDeviceActor(sessionInfo, TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setRpcResponseStatusMsg(responseMsg).build(), new ApiStatsProxyCallback<>(getTenantId(sessionInfo), getCustomerId(sessionInfo), 1, TransportServiceCallback.EMPTY)); } From 7e346894fb74f24f0f5e5a4a992a6931730f084b Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Tue, 27 Dec 2022 12:02:43 +0200 Subject: [PATCH 2/4] Remove quotes if needed --- .../rule/engine/rest/TbHttpClient.java | 15 +++++++++++++-- .../rest/TbRestApiCallNodeConfiguration.java | 2 ++ 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbHttpClient.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbHttpClient.java index b7414c5e0d..20d99e7107 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbHttpClient.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbHttpClient.java @@ -35,7 +35,6 @@ import org.springframework.http.HttpMethod; import org.springframework.http.ResponseEntity; import org.springframework.http.client.HttpComponentsAsyncClientHttpRequestFactory; import org.springframework.http.client.Netty4ClientHttpRequestFactory; -import org.thingsboard.server.common.data.StringUtils; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; import org.springframework.web.client.AsyncRestTemplate; @@ -49,6 +48,7 @@ import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.rule.engine.credentials.BasicCredentials; import org.thingsboard.rule.engine.credentials.ClientCredentials; import org.thingsboard.rule.engine.credentials.CredentialsType; +import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -192,7 +192,11 @@ public class TbHttpClient { config.isIgnoreRequestBody()) { entity = new HttpEntity<>(headers); } else { - entity = new HttpEntity<>(msg.getData(), headers); + String data = msg.getData(); + if (config.isTrimDoubleQuotes()) { + data = trimDoubleQuotes(data); + } + entity = new HttpEntity<>(data, headers); } URI uri = buildEncodedUri(endpointUrl); @@ -334,4 +338,11 @@ public class TbHttpClient { } } + private static String trimDoubleQuotes(String s) { + if (s.startsWith("\"") && s.endsWith("\"")) { + return s.substring(1, s.length() - 1); + } + return s; + } + } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNodeConfiguration.java index 76e0c0d9c4..4c06ccc9a1 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNodeConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNodeConfiguration.java @@ -38,6 +38,7 @@ public class TbRestApiCallNodeConfiguration implements NodeConfiguration Date: Wed, 1 Mar 2023 17:49:26 +0200 Subject: [PATCH 3/4] Delete unused variables --- .../rule/engine/rest/TbRestApiCallNodeConfiguration.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNodeConfiguration.java index 5d4651a983..3fb28d474b 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNodeConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbRestApiCallNodeConfiguration.java @@ -37,9 +37,7 @@ public class TbRestApiCallNodeConfiguration implements NodeConfiguration Date: Thu, 2 Mar 2023 11:40:46 +0200 Subject: [PATCH 4/4] Change logic similar to existing on PE --- .../rule/engine/rest/TbHttpClient.java | 25 ++++++++++--------- 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbHttpClient.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbHttpClient.java index 1473201d1b..6f9541f5b1 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbHttpClient.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbHttpClient.java @@ -192,11 +192,7 @@ public class TbHttpClient { config.isIgnoreRequestBody()) { entity = new HttpEntity<>(headers); } else { - String data = msg.getData(); - if (config.isTrimDoubleQuotes()) { - data = trimDoubleQuotes(data); - } - entity = new HttpEntity<>(data, headers); + entity = new HttpEntity<>(getData(msg), headers); } URI uri = buildEncodedUri(endpointUrl); @@ -247,6 +243,18 @@ public class TbHttpClient { return uri; } + private String getData(TbMsg msg) { + String data = msg.getData(); + + if (config.isTrimDoubleQuotes()) { + final String dataBefore = data; + data = data.replaceAll("^\"|\"$", "");; + log.trace("Trimming double quotes. Before trim: [{}], after trim: [{}]", dataBefore, data); + } + + return data; + } + private TbMsg processResponse(TbContext ctx, TbMsg origMsg, ResponseEntity response) { TbMsgMetaData metaData = origMsg.getMetaData(); metaData.putValue(STATUS, response.getStatusCode().name()); @@ -338,11 +346,4 @@ public class TbHttpClient { } } - private static String trimDoubleQuotes(String s) { - if (s.startsWith("\"") && s.endsWith("\"")) { - return s.substring(1, s.length() - 1); - } - return s; - } - }