From de1a56a4889f211894f5b62348e3ef18e675ac34 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Thu, 4 Nov 2021 15:52:33 +0200 Subject: [PATCH] CoAP ack/confirmable mess fixing --- .../transport/coap/CoapTransportResource.java | 4 +--- .../coap/adaptors/CoapTransportAdaptor.java | 8 ++++---- .../coap/adaptors/JsonCoapAdaptor.java | 18 +++++++----------- .../coap/adaptors/ProtoCoapAdaptor.java | 18 +++++++----------- .../coap/callback/CoapOkCallback.java | 4 +--- .../GetAttributesSyncSessionCallback.java | 2 +- .../ToServerRpcSyncSessionCallback.java | 2 +- .../coap/client/DefaultCoapClientContext.java | 9 +++++---- 8 files changed, 27 insertions(+), 38 deletions(-) diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java index cf8c40a51c..78b25c2d42 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java @@ -223,9 +223,7 @@ public class CoapTransportResource extends AbstractCoapTransportResource { return; } transportService.process(DeviceTransportType.COAP, TransportProtos.ValidateDeviceTokenRequestMsg.newBuilder().setToken(credentials.get().getCredentialsId()).build(), - new CoapDeviceAuthCallback(exchange, (deviceCredentials, deviceProfile) -> { - processRequest(exchange, type, request, deviceCredentials, deviceProfile); - })); + new CoapDeviceAuthCallback(exchange, (deviceCredentials, deviceProfile) -> processRequest(exchange, type, request, deviceCredentials, deviceProfile))); } private void processRequest(CoapExchange exchange, SessionMsgType type, Request request, ValidateDeviceCredentialsResponse deviceCredentials, DeviceProfile deviceProfile) { diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/CoapTransportAdaptor.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/CoapTransportAdaptor.java index b38ac3c322..0c7875aebd 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/CoapTransportAdaptor.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/CoapTransportAdaptor.java @@ -39,13 +39,13 @@ public interface CoapTransportAdaptor { TransportProtos.ClaimDeviceMsg convertToClaimDevice(UUID sessionId, Request inbound, TransportProtos.SessionInfoProto sessionInfo) throws AdaptorException; - Response convertToPublish(boolean isConfirmable, TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException; + Response convertToPublish(TransportProtos.GetAttributeResponseMsg responseMsg) throws AdaptorException; - Response convertToPublish(boolean isConfirmable, TransportProtos.AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException; + Response convertToPublish(TransportProtos.AttributeUpdateNotificationMsg notificationMsg) throws AdaptorException; - Response convertToPublish(boolean isConfirmable, TransportProtos.ToDeviceRpcRequestMsg rpcRequest, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) throws AdaptorException; + Response convertToPublish(TransportProtos.ToDeviceRpcRequestMsg rpcRequest, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) throws AdaptorException; - Response convertToPublish(boolean isConfirmable, TransportProtos.ToServerRpcResponseMsg msg) throws AdaptorException; + Response convertToPublish(TransportProtos.ToServerRpcResponseMsg msg) throws AdaptorException; ProvisionDeviceRequestMsg convertToProvisionRequestMsg(UUID sessionId, Request inbound) throws AdaptorException; diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java index e24e14faab..d0babc51cf 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/JsonCoapAdaptor.java @@ -93,21 +93,20 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor { } @Override - public Response convertToPublish(boolean isConfirmable, TransportProtos.AttributeUpdateNotificationMsg msg) throws AdaptorException { - return getObserveNotification(isConfirmable, JsonConverter.toJson(msg)); + public Response convertToPublish(TransportProtos.AttributeUpdateNotificationMsg msg) throws AdaptorException { + return getObserveNotification(JsonConverter.toJson(msg)); } @Override - public Response convertToPublish(boolean isConfirmable, TransportProtos.ToDeviceRpcRequestMsg msg, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) throws AdaptorException { - return getObserveNotification(isConfirmable, JsonConverter.toJson(msg, true)); + public Response convertToPublish(TransportProtos.ToDeviceRpcRequestMsg msg, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) throws AdaptorException { + return getObserveNotification(JsonConverter.toJson(msg, true)); } @Override - public Response convertToPublish(boolean isConfirmable, TransportProtos.ToServerRpcResponseMsg msg) throws AdaptorException { + public Response convertToPublish(TransportProtos.ToServerRpcResponseMsg msg) throws AdaptorException { Response response = new Response(CoAP.ResponseCode.CONTENT); JsonElement result = JsonConverter.toJson(msg); response.setPayload(result.toString()); - response.setConfirmable(isConfirmable); return response; } @@ -122,11 +121,10 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor { } @Override - public Response convertToPublish(boolean isConfirmable, TransportProtos.GetAttributeResponseMsg msg) throws AdaptorException { + public Response convertToPublish(TransportProtos.GetAttributeResponseMsg msg) throws AdaptorException { if (msg.getSharedStateMsg()) { if (StringUtils.isEmpty(msg.getError())) { Response response = new Response(CoAP.ResponseCode.CONTENT); - response.setConfirmable(isConfirmable); TransportProtos.AttributeUpdateNotificationMsg notificationMsg = TransportProtos.AttributeUpdateNotificationMsg.newBuilder().addAllSharedUpdated(msg.getSharedAttributeListList()).build(); JsonObject result = JsonConverter.toJson(notificationMsg); response.setPayload(result.toString()); @@ -139,7 +137,6 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor { return new Response(CoAP.ResponseCode.NOT_FOUND); } else { Response response = new Response(CoAP.ResponseCode.CONTENT); - response.setConfirmable(isConfirmable); JsonObject result = JsonConverter.toJson(msg); response.setPayload(result.toString()); return response; @@ -147,10 +144,9 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor { } } - private Response getObserveNotification(boolean confirmable, JsonElement json) { + private Response getObserveNotification(JsonElement json) { Response response = new Response(CoAP.ResponseCode.CONTENT); response.setPayload(json.toString()); - response.setConfirmable(confirmable); return response; } diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/ProtoCoapAdaptor.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/ProtoCoapAdaptor.java index 1ad29524d2..e4a125008e 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/ProtoCoapAdaptor.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/adaptors/ProtoCoapAdaptor.java @@ -113,29 +113,27 @@ public class ProtoCoapAdaptor implements CoapTransportAdaptor { } @Override - public Response convertToPublish(boolean isConfirmable, TransportProtos.AttributeUpdateNotificationMsg msg) throws AdaptorException { - return getObserveNotification(isConfirmable, msg.toByteArray()); + public Response convertToPublish(TransportProtos.AttributeUpdateNotificationMsg msg) throws AdaptorException { + return getObserveNotification(msg.toByteArray()); } @Override - public Response convertToPublish(boolean isConfirmable, TransportProtos.ToDeviceRpcRequestMsg rpcRequest, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) throws AdaptorException { - return getObserveNotification(isConfirmable, ProtoConverter.convertToRpcRequest(rpcRequest, rpcRequestDynamicMessageBuilder)); + public Response convertToPublish(TransportProtos.ToDeviceRpcRequestMsg rpcRequest, DynamicMessage.Builder rpcRequestDynamicMessageBuilder) throws AdaptorException { + return getObserveNotification(ProtoConverter.convertToRpcRequest(rpcRequest, rpcRequestDynamicMessageBuilder)); } @Override - public Response convertToPublish(boolean isConfirmable, TransportProtos.ToServerRpcResponseMsg msg) throws AdaptorException { + public Response convertToPublish(TransportProtos.ToServerRpcResponseMsg msg) throws AdaptorException { Response response = new Response(CoAP.ResponseCode.CONTENT); - response.setConfirmable(isConfirmable); response.setPayload(msg.toByteArray()); return response; } @Override - public Response convertToPublish(boolean isConfirmable, TransportProtos.GetAttributeResponseMsg msg) throws AdaptorException { + public Response convertToPublish(TransportProtos.GetAttributeResponseMsg msg) throws AdaptorException { if (msg.getSharedStateMsg()) { if (StringUtils.isEmpty(msg.getError())) { Response response = new Response(CoAP.ResponseCode.CONTENT); - response.setConfirmable(isConfirmable); TransportProtos.AttributeUpdateNotificationMsg notificationMsg = TransportProtos.AttributeUpdateNotificationMsg.newBuilder().addAllSharedUpdated(msg.getSharedAttributeListList()).build(); response.setPayload(notificationMsg.toByteArray()); return response; @@ -147,17 +145,15 @@ public class ProtoCoapAdaptor implements CoapTransportAdaptor { return new Response(CoAP.ResponseCode.NOT_FOUND); } else { Response response = new Response(CoAP.ResponseCode.CONTENT); - response.setConfirmable(isConfirmable); response.setPayload(msg.toByteArray()); return response; } } } - private Response getObserveNotification(boolean confirmable, byte[] notification) { + private Response getObserveNotification(byte[] notification) { Response response = new Response(CoAP.ResponseCode.CONTENT); response.setPayload(notification); - response.setConfirmable(confirmable); return response; } diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/CoapOkCallback.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/CoapOkCallback.java index 40e1b894ee..491eea221e 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/CoapOkCallback.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/CoapOkCallback.java @@ -34,9 +34,7 @@ public class CoapOkCallback implements TransportServiceCallback { @Override public void onSuccess(Void msg) { - Response response = new Response(onSuccessResponse); - response.setConfirmable(isConRequest()); - exchange.respond(response); + exchange.respond(new Response(onSuccessResponse)); } @Override diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/GetAttributesSyncSessionCallback.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/GetAttributesSyncSessionCallback.java index f6e5b52ff7..f7adceec7f 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/GetAttributesSyncSessionCallback.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/GetAttributesSyncSessionCallback.java @@ -34,7 +34,7 @@ public class GetAttributesSyncSessionCallback extends AbstractSyncSessionCallbac @Override public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg msg) { try { - respond(state.getAdaptor().convertToPublish(request.isConfirmable(), msg)); + respond(state.getAdaptor().convertToPublish(msg)); } catch (AdaptorException e) { log.trace("[{}] Failed to reply due to error", state.getDeviceId(), e); exchange.respond(new Response(CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/ToServerRpcSyncSessionCallback.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/ToServerRpcSyncSessionCallback.java index fb4b855f39..ebf5547d24 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/ToServerRpcSyncSessionCallback.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/ToServerRpcSyncSessionCallback.java @@ -33,7 +33,7 @@ public class ToServerRpcSyncSessionCallback extends AbstractSyncSessionCallback @Override public void onToServerRpcResponse(TransportProtos.ToServerRpcResponseMsg toServerResponse) { try { - respond(state.getAdaptor().convertToPublish(request.isConfirmable(), toServerResponse)); + respond(state.getAdaptor().convertToPublish(toServerResponse)); } catch (AdaptorException e) { log.trace("Failed to reply due to error", e); exchange.respond(CoAP.ResponseCode.INTERNAL_SERVER_ERROR); 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 082a94dbe6..2a1911c232 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 @@ -435,8 +435,7 @@ public class DefaultCoapClientContext implements CoapClientContext { TbCoapObservationState attrs = state.getAttrs(); if (attrs != null) { try { - boolean conRequest = AbstractSyncSessionCallback.isConRequest(state.getAttrs()); - Response response = state.getAdaptor().convertToPublish(conRequest, msg); + Response response = state.getAdaptor().convertToPublish(msg); respond(attrs.getExchange(), response, state.getContentFormat()); } catch (AdaptorException e) { log.trace("Failed to reply due to error", e); @@ -466,7 +465,8 @@ public class DefaultCoapClientContext implements CoapClientContext { try { boolean conRequest = AbstractSyncSessionCallback.isConRequest(state.getAttrs()); int requestId = getNextMsgId(); - Response response = state.getAdaptor().convertToPublish(conRequest, msg); + Response response = state.getAdaptor().convertToPublish(msg); + response.setConfirmable(conRequest); response.setMID(requestId); if (conRequest) { response.addMessageObserver(new TbCoapMessageObserver(requestId, id -> awake(state), id -> asleep(state))); @@ -527,7 +527,8 @@ public class DefaultCoapClientContext implements CoapClientContext { String error = null; boolean conRequest = AbstractSyncSessionCallback.isConRequest(state.getRpc()); try { - Response response = state.getAdaptor().convertToPublish(conRequest, msg, state.getConfiguration().getRpcRequestDynamicMessageBuilder()); + Response response = state.getAdaptor().convertToPublish(msg, state.getConfiguration().getRpcRequestDynamicMessageBuilder()); + response.setConfirmable(conRequest); int requestId = getNextMsgId(); response.setMID(requestId); if (conRequest) {