From f932a90b00b7f8fffc61285b389dc8d45ac68449 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Mon, 26 Jul 2021 15:57:27 +0300 Subject: [PATCH] Improvements to Persistent RPC call delivery confirmation --- .../device/DeviceActorMessageProcessor.java | 19 +++++++++++--- common/queue/src/main/proto/queue.proto | 2 +- .../coap/client/DefaultCoapClientContext.java | 20 +++++++++------ .../transport/http/DeviceApiController.java | 2 +- .../rpc/DefaultLwM2MRpcRequestHandler.java | 14 +++++------ .../server/rpc/LwM2MRpcResponseBody.java | 1 - .../rpc/RpcDownlinkRequestCallbackProxy.java | 25 +++++++++---------- .../transport/mqtt/MqttTransportHandler.java | 16 +++++++----- .../mqtt/session/GatewayDeviceSessionCtx.java | 11 +++++--- .../snmp/session/DeviceSessionContext.java | 2 +- .../common/transport/TransportService.java | 2 +- .../service/DefaultTransportService.java | 12 ++------- 12 files changed, 71 insertions(+), 55 deletions(-) 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 5963350297..a10ae17b88 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 @@ -15,6 +15,7 @@ */ package org.thingsboard.server.actors.device; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; @@ -35,6 +36,7 @@ import org.thingsboard.server.actors.TbActorCtx; import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.edge.EdgeEvent; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.edge.EdgeEventType; @@ -512,11 +514,20 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { ToDeviceRpcRequestMetadata requestMd = toDeviceRpcPendingMap.remove(responseMsg.getRequestId()); boolean success = requestMd != null; if (success) { - systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(), - responseMsg.getPayload(), null)); + boolean hasError = StringUtils.isNotEmpty(responseMsg.getError()); + String payload = hasError ? responseMsg.getError() : responseMsg.getPayload(); + systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor( + new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(), + payload, hasError ? RpcError.INTERNAL : null)); if (requestMd.getMsg().getMsg().isPersisted()) { - RpcStatus status = responseMsg.getFailed() ? RpcStatus.FAILED : RpcStatus.SUCCESSFUL; - systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), status, JacksonUtil.toJsonNode(responseMsg.getPayload())); + RpcStatus status = hasError ? RpcStatus.FAILED : RpcStatus.SUCCESSFUL; + JsonNode response; + try { + response = JacksonUtil.toJsonNode(payload); + } catch (IllegalArgumentException e) { + response = JacksonUtil.newObjectNode().put("error", payload); + } + systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), status, response); } } 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 73f8dc0aae..b496eaaf22 100644 --- a/common/queue/src/main/proto/queue.proto +++ b/common/queue/src/main/proto/queue.proto @@ -339,7 +339,7 @@ message ToDeviceRpcRequestMsg { message ToDeviceRpcResponseMsg { int32 requestId = 1; string payload = 2; - bool failed = 3; + string error = 3; } message UplinkNotificationMsg { 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 757ec45961..265980c0a9 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 @@ -28,6 +28,7 @@ import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.DeviceTransportType; +import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.device.data.PowerMode; import org.thingsboard.server.common.data.device.data.PowerSavingConfiguration; import org.thingsboard.server.common.data.device.profile.CoapDeviceProfileTransportConfiguration; @@ -507,6 +508,7 @@ public class DefaultCoapClientContext implements CoapClientContext { return; } boolean sent = false; + String error = null; boolean conRequest = AbstractSyncSessionCallback.isConRequest(state.getRpc()); try { Response response = state.getAdaptor().convertToPublish(conRequest, msg, state.getConfiguration().getRpcRequestDynamicMessageBuilder()); @@ -515,15 +517,12 @@ public class DefaultCoapClientContext implements CoapClientContext { if (msg.getPersisted() && conRequest) { transportContext.getRpcAwaitingAck().put(requestId, msg); transportContext.getScheduler().schedule(() -> { - TransportProtos.ToDeviceRpcRequestMsg awaitingAckMsg = transportContext.getRpcAwaitingAck().remove(requestId); - if (awaitingAckMsg != null) { - transportService.process(state.getSession(), msg, true, TransportServiceCallback.EMPTY); - } + transportContext.getRpcAwaitingAck().remove(requestId); }, Math.max(0, msg.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS); response.addMessageObserver(new TbCoapMessageObserver(requestId, id -> { TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(id); if (rpcRequestMsg != null) { - transportService.process(state.getSession(), rpcRequestMsg, false, TransportServiceCallback.EMPTY); + transportService.process(state.getSession(), rpcRequestMsg, TransportServiceCallback.EMPTY); } }, null)); } @@ -536,9 +535,16 @@ public class DefaultCoapClientContext implements CoapClientContext { log.trace("Failed to reply due to error", e); cancelObserveRelation(state.getRpc()); cancelRpcSubscription(state); + error = "Failed to convert device RPC command to CoAP msg"; + } catch (Exception e) { + error = "Internal error: " + e.getMessage(); } finally { - if (msg.getPersisted() && !conRequest) { - transportService.process(state.getSession(), msg, sent, TransportServiceCallback.EMPTY); + if (StringUtils.isNotEmpty(error)) { + transportService.process(state.getSession(), + TransportProtos.ToDeviceRpcResponseMsg.newBuilder() + .setRequestId(msg.getRequestId()).setError(error).build(), TransportServiceCallback.EMPTY); + } else if (msg.getPersisted() && !conRequest && sent) { + transportService.process(state.getSession(), msg, TransportServiceCallback.EMPTY); } } } diff --git a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java index 9c258ef00f..aab76e350c 100644 --- a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java +++ b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java @@ -409,7 +409,7 @@ public class DeviceApiController implements TbTransportService { public void onToDeviceRpcRequest(UUID sessionId, ToDeviceRpcRequestMsg msg) { log.trace("[{}] Received RPC command to device", sessionId); responseWriter.setResult(new ResponseEntity<>(JsonConverter.toJson(msg, true).toString(), HttpStatus.OK)); - transportService.process(sessionInfo, msg, false, TransportServiceCallback.EMPTY); + transportService.process(sessionInfo, msg, TransportServiceCallback.EMPTY); } @Override diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java index 5b212665c3..1a8d0f8ce3 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java @@ -89,12 +89,12 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler { if (!this.rpcSubscriptions.containsKey(requestUUID)) { LwM2mOperationType operationType = LwM2mOperationType.fromType(rpcRequest.getMethodName()); if (operationType == null) { - this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.METHOD_NOT_ALLOWED.getName(), "Unsupported operation type: " + rpcRequest.getMethodName()); + this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.METHOD_NOT_ALLOWED, "Unsupported operation type: " + rpcRequest.getMethodName()); return; } LwM2mClient client = clientContext.getClientBySessionInfo(sessionInfo); if (client.getRegistration() == null) { - this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.INTERNAL_SERVER_ERROR.getName(), "Registration is empty"); + this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.INTERNAL_SERVER_ERROR, "Registration is empty"); return; } try { @@ -145,7 +145,7 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler { } } else { this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), - ResponseCode.INTERNAL_SERVER_ERROR.getName(), "This device does not support Composite Operation"); + ResponseCode.INTERNAL_SERVER_ERROR, "This device does not support Composite Operation"); } } else { switch (operationType) { @@ -165,7 +165,7 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler { } } } catch (IllegalArgumentException e) { - this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.BAD_REQUEST.getName(), e.getMessage()); + this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.BAD_REQUEST, e.getMessage()); } } } @@ -312,9 +312,9 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler { } } - private void sendErrorRpcResponse(TransportProtos.SessionInfoProto sessionInfo, int requestId, String result, String error) { - String payload = JacksonUtil.toString(JacksonUtil.newObjectNode().put("result", result).put("error", error)); - TransportProtos.ToDeviceRpcResponseMsg msg = TransportProtos.ToDeviceRpcResponseMsg.newBuilder().setRequestId(requestId).setPayload(payload).setFailed(true).build(); + private void sendErrorRpcResponse(TransportProtos.SessionInfoProto sessionInfo, int requestId, ResponseCode result, String error) { + String payload = JacksonUtil.toString(LwM2MRpcResponseBody.builder().result(result.getName()).error(error).build()); + TransportProtos.ToDeviceRpcResponseMsg msg = TransportProtos.ToDeviceRpcResponseMsg.newBuilder().setRequestId(requestId).setError(payload).build(); transportService.process(sessionInfo, msg, null); } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/LwM2MRpcResponseBody.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/LwM2MRpcResponseBody.java index 16b743101b..8836db71e3 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/LwM2MRpcResponseBody.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/LwM2MRpcResponseBody.java @@ -27,6 +27,5 @@ public class LwM2MRpcResponseBody { private String result; private String value; private String error; - private String info; } 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 f8c43f75e4..c94c9c5805 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 @@ -18,6 +18,7 @@ package org.thingsboard.server.transport.lwm2m.server.rpc; import org.eclipse.leshan.core.ResponseCode; import org.eclipse.leshan.core.request.exception.ClientSleepingException; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.gen.transport.TransportProtos; @@ -43,7 +44,7 @@ public abstract class RpcDownlinkRequestCallbackProxy implements DownlinkR @Override public void onSuccess(R request, T response) { - transportService.process(client.getSession(), this.request, false, TransportServiceCallback.EMPTY); + transportService.process(client.getSession(), this.request, TransportServiceCallback.EMPTY); sendRpcReplyOnSuccess(response); if (callback != null) { callback.onSuccess(request, response); @@ -69,26 +70,24 @@ public abstract class RpcDownlinkRequestCallbackProxy implements DownlinkR } 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); + TransportProtos.ToDeviceRpcResponseMsg.Builder msg = TransportProtos.ToDeviceRpcResponseMsg.newBuilder().setRequestId(request.getRequestId()); + String responseAsString = JacksonUtil.toString(response); + if (StringUtils.isEmpty(response.getError())) { + msg.setPayload(responseAsString); + } else { + msg.setError(responseAsString); + } + transportService.process(client.getSession(), msg.build(), null); } abstract protected void sendRpcReplyOnSuccess(T response); protected void sendRpcReplyOnValidationError(String msg) { - reply(LwM2MRpcResponseBody.builder().result(ResponseCode.BAD_REQUEST.getName()).error(msg).build(), true); + reply(LwM2MRpcResponseBody.builder().result(ResponseCode.BAD_REQUEST.getName()).error(msg).build()); } protected void sendRpcReplyOnError(Exception e) { - reply(LwM2MRpcResponseBody.builder().result(ResponseCode.INTERNAL_SERVER_ERROR.getName()).error(e.getMessage()).build(), true); + reply(LwM2MRpcResponseBody.builder().result(ResponseCode.INTERNAL_SERVER_ERROR.getName()).error(e.getMessage()).build()); } } 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 6afdadffff..630aec946d 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 @@ -249,7 +249,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, false, TransportServiceCallback.EMPTY); + transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, TransportServiceCallback.EMPTY); } break; default: @@ -829,18 +829,22 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement if (rpcRequest.getPersisted() && isAckExpected(payload)) { rpcAwaitingAck.put(msgId, rpcRequest); context.getScheduler().schedule(() -> { - TransportProtos.ToDeviceRpcRequestMsg awaitingAckMsg = rpcAwaitingAck.remove(msgId); - if (awaitingAckMsg != null) { - transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, true, TransportServiceCallback.EMPTY); - } + rpcAwaitingAck.remove(msgId); }, Math.max(0, rpcRequest.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS); } var cf = publish(payload, deviceSessionCtx); if (rpcRequest.getPersisted() && !isAckExpected(payload)) { - cf.addListener(result -> transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, result.cause() != null, TransportServiceCallback.EMPTY)); + cf.addListener(result -> { + if (result.cause() == null) { + transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, TransportServiceCallback.EMPTY); + } + }); } }); } catch (Exception e) { + transportService.process(deviceSessionCtx.getSessionInfo(), + TransportProtos.ToDeviceRpcResponseMsg.newBuilder() + .setRequestId(rpcRequest.getRequestId()).setError("Failed to convert device RPC command to MQTT msg").build(), TransportServiceCallback.EMPTY); log.trace("[{}] Failed to convert device RPC command to MQTT msg", sessionId, e); } } diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java index fb41093c35..3fed5e51ea 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java @@ -101,13 +101,18 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext imple payload -> { ChannelFuture channelFuture = parent.writeAndFlush(payload); if (request.getPersisted()) { - channelFuture.addListener(future -> - transportService.process(getSessionInfo(), request, future.cause() != null, TransportServiceCallback.EMPTY) - ); + channelFuture.addListener(future -> { + if (future.cause() == null) { + transportService.process(getSessionInfo(), request, TransportServiceCallback.EMPTY); + } + }); } } ); } catch (Exception e) { + transportService.process(getSessionInfo(), + TransportProtos.ToDeviceRpcResponseMsg.newBuilder() + .setRequestId(request.getRequestId()).setError("Failed to convert device RPC command to MQTT msg").build(), TransportServiceCallback.EMPTY); log.trace("[{}] Failed to convert device attributes response to MQTT msg", sessionId, e); } } diff --git a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java index 1fba12782a..1927aadc56 100644 --- a/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java +++ b/common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java @@ -142,7 +142,7 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S public void onToDeviceRpcRequest(UUID sessionId, ToDeviceRpcRequestMsg toDeviceRequest) { log.trace("[{}] Received RPC command to device", sessionId); snmpTransportContext.getSnmpTransportService().onToDeviceRpcRequest(this, toDeviceRequest); - snmpTransportContext.getTransportService().process(getSessionInfo(), toDeviceRequest, false, TransportServiceCallback.EMPTY); + snmpTransportContext.getTransportService().process(getSessionInfo(), toDeviceRequest, TransportServiceCallback.EMPTY); } @Override 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 5227f671bc..5f2fa4f197 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 @@ -110,7 +110,7 @@ public interface TransportService { void process(SessionInfoProto sessionInfo, ToServerRpcRequestMsg msg, TransportServiceCallback callback); - void process(SessionInfoProto sessionInfo, ToDeviceRpcRequestMsg msg, boolean isFailedRpc, TransportServiceCallback callback); + void process(SessionInfoProto sessionInfo, ToDeviceRpcRequestMsg msg, 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 eda114e498..e3390eb9bb 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 @@ -580,17 +580,9 @@ public class DefaultTransportService implements TransportService { } @Override - public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToDeviceRpcRequestMsg msg, boolean isFailedRpc, TransportServiceCallback callback) { + public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToDeviceRpcRequestMsg msg, TransportServiceCallback callback) { if (msg.getPersisted()) { - RpcStatus status; - - if (isFailedRpc) { - status = RpcStatus.FAILED; - } else if (msg.getOneway()) { - status = RpcStatus.SUCCESSFUL; - } else { - status = RpcStatus.DELIVERED; - } + RpcStatus status = msg.getOneway() ? RpcStatus.SUCCESSFUL : RpcStatus.DELIVERED; TransportProtos.ToDevicePersistedRpcResponseMsg responseMsg = TransportProtos.ToDevicePersistedRpcResponseMsg.newBuilder() .setRequestId(msg.getRequestId())