diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index 42e58ea9e5..5963350297 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -515,7 +515,8 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(), responseMsg.getPayload(), null)); if (requestMd.getMsg().getMsg().isPersisted()) { - systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), RpcStatus.SUCCESSFUL, JacksonUtil.toJsonNode(responseMsg.getPayload())); + RpcStatus status = responseMsg.getFailed() ? RpcStatus.FAILED : RpcStatus.SUCCESSFUL; + systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), status, JacksonUtil.toJsonNode(responseMsg.getPayload())); } } else { log.debug("[{}] Rpc command response [{}] is stale!", deviceId, responseMsg.getRequestId()); diff --git a/common/queue/src/main/proto/queue.proto b/common/queue/src/main/proto/queue.proto index aebb212dae..73f8dc0aae 100644 --- a/common/queue/src/main/proto/queue.proto +++ b/common/queue/src/main/proto/queue.proto @@ -339,6 +339,7 @@ message ToDeviceRpcRequestMsg { message ToDeviceRpcResponseMsg { int32 requestId = 1; string payload = 2; + bool failed = 3; } message UplinkNotificationMsg { diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java index ec688bfffa..e9b0391250 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java @@ -84,7 +84,6 @@ public class LwM2mSessionMsgListener implements GenericFutureListener implements DownlinkRequestCallback { private final TransportService transportService; @@ -39,6 +43,7 @@ public abstract class RpcDownlinkRequestCallbackProxy implements DownlinkR @Override public void onSuccess(R request, T response) { + transportService.process(client.getSession(), this.request, false, TransportServiceCallback.EMPTY); sendRpcReplyOnSuccess(response); if (callback != null) { callback.onSuccess(request, response); @@ -55,16 +60,23 @@ public abstract class RpcDownlinkRequestCallbackProxy implements DownlinkR @Override public void onError(String params, Exception e) { - sendRpcReplyOnError(e); + if (!(e instanceof TimeoutException || e instanceof ClientSleepingException)) { + sendRpcReplyOnError(e); + } if (callback != null) { callback.onError(params, e); } } protected void reply(LwM2MRpcResponseBody response) { + reply(response, false); + } + + protected void reply(LwM2MRpcResponseBody response, boolean failed) { TransportProtos.ToDeviceRpcResponseMsg msg = TransportProtos.ToDeviceRpcResponseMsg.newBuilder() .setPayload(JacksonUtil.toString(response)) .setRequestId(request.getRequestId()) + .setFailed(failed) .build(); transportService.process(client.getSession(), msg, null); } @@ -72,11 +84,11 @@ public abstract class RpcDownlinkRequestCallbackProxy implements DownlinkR abstract protected void sendRpcReplyOnSuccess(T response); protected void sendRpcReplyOnValidationError(String msg) { - reply(LwM2MRpcResponseBody.builder().result(ResponseCode.BAD_REQUEST.getName()).error(msg).build()); + reply(LwM2MRpcResponseBody.builder().result(ResponseCode.BAD_REQUEST.getName()).error(msg).build(), true); } protected void sendRpcReplyOnError(Exception e) { - reply(LwM2MRpcResponseBody.builder().result(ResponseCode.INTERNAL_SERVER_ERROR.getName()).error(e.getMessage()).build()); + reply(LwM2MRpcResponseBody.builder().result(ResponseCode.INTERNAL_SERVER_ERROR.getName()).error(e.getMessage()).build(), true); } }