From 8022cfb15537e658bb77cb6e1ab6dc93d89283e2 Mon Sep 17 00:00:00 2001 From: nickAS21 Date: Wed, 3 Feb 2021 19:14:10 +0200 Subject: [PATCH] Lwm2m: back: fix bug delay operation --- .../lwm2m/server/LwM2MTransportRequest.java | 74 +++----- .../server/LwM2MTransportServiceImpl.java | 172 +++++++++--------- .../lwm2m/server/LwM2mServerListener.java | 2 +- .../lwm2m/server/client/LwM2MClient.java | 79 ++------ .../secure/LwM2mInMemorySecurityStore.java | 6 +- 5 files changed, 128 insertions(+), 205 deletions(-) diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2MTransportRequest.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2MTransportRequest.java index 845ec201f7..e02746eb0a 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2MTransportRequest.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2MTransportRequest.java @@ -96,7 +96,7 @@ public class LwM2MTransportRequest { this.converter = LwM2mValueConverterImpl.getInstance(); executorResponse = Executors.newFixedThreadPool(this.context.getCtxServer().getRequestPoolSize(), new NamedThreadFactory(String.format("LwM2M %s channel response", RESPONSE_CHANNEL))); - executorResponseError = Executors.newFixedThreadPool(this.context.getCtxServer().getRequestErrorPoolSize(), + executorResponseError = Executors.newFixedThreadPool(this.context.getCtxServer().getRequestErrorPoolSize(), new NamedThreadFactory(String.format("LwM2M %s channel response Error", RESPONSE_CHANNEL))); } @@ -112,16 +112,15 @@ public class LwM2MTransportRequest { /** * Device management and service enablement, including Read, Write, Execute, Discover, Create, Delete and Write-Attributes * - * @param lwServer - * @param registration - * @param target - * @param typeOper - * @param contentFormatParam - * @param lwM2MClient - * @param observation + * @param lwServer - + * @param registration - + * @param target - + * @param typeOper - + * @param contentFormatParam - + * @param observation - */ - public void sendAllRequest(LeshanServer lwServer, Registration registration, String target, String typeOper, String contentFormatParam, - LwM2MClient lwM2MClient, Observation observation, Object params, long timeoutInMs, boolean isDelayedUpdate) { + public void sendAllRequest(LeshanServer lwServer, Registration registration, String target, String typeOper, + String contentFormatParam, Observation observation, Object params, long timeoutInMs) { LwM2mPath resultIds = new LwM2mPath(target); if (registration != null && resultIds.getObjectId() >= 0) { DownlinkRequest request = null; @@ -221,13 +220,7 @@ public class LwM2MTransportRequest { } if (request != null) { - this.sendRequest(lwServer, registration, request, lwM2MClient, timeoutInMs, isDelayedUpdate); - } else if (isDelayedUpdate) { - String msg = String.format(LOG_LW2M_ERROR + ": sendRequest: Resource path - %s msg No SendRequest to Client", target); - service.sentLogsToThingsboard(msg, registration); - log.error("[{}] - [{}] No SendRequest", target); -// this.handleResponseError(registration, target, lwM2MClient, true); - + this.sendRequest(lwServer, registration, request, timeoutInMs); } } } @@ -237,45 +230,37 @@ public class LwM2MTransportRequest { * @param lwServer - * @param registration - * @param request - - * @param lwM2MClient - * @param timeoutInMs - */ - private void sendRequest(LeshanServer lwServer, Registration registration, DownlinkRequest request, LwM2MClient lwM2MClient, long timeoutInMs, boolean isDelayedUpdate) { + private void sendRequest(LeshanServer lwServer, Registration registration, DownlinkRequest request, long timeoutInMs) { + LwM2MClient lwM2MClient = this.service.lwM2mInMemorySecurityStore.getLwM2MClientWithReg(registration, null); lwServer.send(registration, request, timeoutInMs, (ResponseCallback) response -> { + if (!lwM2MClient.isInit()) lwM2MClient.initValue(this.service, request.getPath().toString()); if (isSuccess(((Response) response.getCoapResponse()).getCode())) { - this.handleResponse(registration, request.getPath().toString(), response, request, lwM2MClient, isDelayedUpdate); + this.handleResponse(registration, request.getPath().toString(), response, request); if (request instanceof WriteRequest && ((WriteRequest) request).isReplaceRequest()) { - String delayedUpdateStr = ""; - if (isDelayedUpdate) { - delayedUpdateStr = " (delayedUpdate) "; - } - String msg = String.format(LOG_LW2M_INFO + ": sendRequest Replace%s: CoapCde - %s Lwm2m code - %d name - %s Resource path - %s value - %s SendRequest to Client", - delayedUpdateStr, ((Response) response.getCoapResponse()).getCode(), response.getCode().getCode(), response.getCode().getName(), request.getPath().toString(), + String msg = String.format(LOG_LW2M_INFO + ": sendRequest Replace: CoapCde - %s Lwm2m code - %d name - %s Resource path - %s value - %s SendRequest to Client", + ((Response) response.getCoapResponse()).getCode(), response.getCode().getCode(), response.getCode().getName(), request.getPath().toString(), ((LwM2mSingleResource) ((WriteRequest) request).getNode()).getValue().toString()); service.sentLogsToThingsboard(msg, registration); - log.info("[{}] - [{}] [{}] [{}] Update SendRequest[{}]", ((Response) response.getCoapResponse()).getCode(), response.getCode(), request.getPath().toString(), - ((LwM2mSingleResource) ((WriteRequest) request).getNode()).getValue(), delayedUpdateStr); + log.info("[{}] [{}] - [{}] [{}] Update SendRequest[{}]", registration.getEndpoint(), ((Response) response.getCoapResponse()).getCode(), response.getCode(), request.getPath().toString(), + ((LwM2mSingleResource) ((WriteRequest) request).getNode()).getValue()); } } else { String msg = String.format(LOG_LW2M_ERROR + ": sendRequest: CoapCode - %s Lwm2m code - %d name - %s Resource path - %s SendRequest to Client", ((Response) response.getCoapResponse()).getCode(), response.getCode().getCode(), response.getCode().getName(), request.getPath().toString()); service.sentLogsToThingsboard(msg, registration); log.error("[{}] - [{}] [{}] error SendRequest", ((Response) response.getCoapResponse()).getCode(), response.getCode(), request.getPath().toString()); -// if (request instanceof WriteRequest && ((WriteRequest) request).isReplaceRequest() && isDelayedUpdate) { -// this.handleResponseError(registration, request.getPath().toString(), lwM2MClient, isDelayedUpdate); -// } } - }, e -> { + if (!lwM2MClient.isInit()) lwM2MClient.initValue(this.service, request.getPath().toString()); String msg = String.format(LOG_LW2M_ERROR + ": sendRequest: Resource path - %s msg error - %s SendRequest to Client", request.getPath().toString(), e.toString()); service.sentLogsToThingsboard(msg, registration); log.error("[{}] - [{}] error SendRequest", request.getPath().toString(), e.toString()); -// if (request instanceof WriteRequest && ((WriteRequest) request).isReplaceRequest() && isDelayedUpdate) { -// this.handleResponseError(registration, request.getPath().toString(), lwM2MClient, isDelayedUpdate); -// } }); + } private WriteRequest getWriteRequestSingleResource(ContentFormat contentFormat, Integer objectId, Integer instanceId, Integer resourceId, Object value, ResourceModel.Type type, Registration registration) { @@ -308,34 +293,23 @@ public class LwM2MTransportRequest { } } - private void handleResponse(Registration registration, final String path, LwM2mResponse response, DownlinkRequest request, LwM2MClient lwM2MClient, boolean isDelayedUpdate) { + private void handleResponse(Registration registration, final String path, LwM2mResponse response, DownlinkRequest request) { executorResponse.submit(() -> { try { - sendResponse(registration, path, response, request, lwM2MClient, isDelayedUpdate); + sendResponse(registration, path, response, request); } catch (Exception e) { log.error("[{}] endpoint [{}] path [{}] Exception Unable to after send response.", registration.getEndpoint(), path, e); } }); } -// -// private void handleResponseError(Registration registration, final String path, LwM2MClient lwM2MClient, boolean isDelayedUpdate) { -// executorResponseError.submit(() -> { -// try { -// if (isDelayedUpdate) lwM2MClient.onSuccessOrErrorDelayedRequests(path); -// } catch (RuntimeException t) { -// log.error("[{}] endpoint [{}] path [{}] RuntimeException Unable to after send response.", registration.getEndpoint(), path, t); -// } -// }); -// } /** * processing a response from a client * @param registration - * @param path - * @param response - - * @param lwM2MClient - */ - private void sendResponse(Registration registration, String path, LwM2mResponse response, DownlinkRequest request, LwM2MClient lwM2MClient, boolean isDelayedUpdate) { + private void sendResponse(Registration registration, String path, LwM2mResponse response, DownlinkRequest request) { if (response instanceof ObserveResponse || response instanceof ReadResponse) { service.onObservationResponse(registration, path, (ReadResponse) response); } else if (response instanceof CancelObservationResponse) { @@ -350,7 +324,7 @@ public class LwM2MTransportRequest { log.info("[{}] Path [{}] WriteAttributesResponse 8_Send", path, response); } else if (response instanceof WriteResponse) { log.info("[{}] Path [{}] WriteAttributesResponse 9_Send", path, response); - service.onWriteResponseOk(registration, path, (WriteRequest) request, isDelayedUpdate); + service.onWriteResponseOk(registration, path, (WriteRequest) request); } } } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2MTransportServiceImpl.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2MTransportServiceImpl.java index 90a0131069..656433f2a6 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2MTransportServiceImpl.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2MTransportServiceImpl.java @@ -121,6 +121,7 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { this.context.getScheduler().scheduleAtFixedRate(this::checkInactivityAndReportActivity, new Random().nextInt((int) context.getCtxServer().getSessionReportTimeout()), context.getCtxServer().getSessionReportTimeout(), TimeUnit.MILLISECONDS); this.executorRegistered = Executors.newFixedThreadPool(this.context.getCtxServer().getRegisteredPoolSize(), new NamedThreadFactory(String.format("LwM2M %s channel registered", SERVICE_CHANNEL))); +// this.executorRegistered = Executors.newWorkStealingPool(this.context.getCtxServer().getRegisteredPoolSize()); this.executorUpdateRegistered = Executors.newFixedThreadPool(this.context.getCtxServer().getUpdateRegisteredPoolSize(), new NamedThreadFactory(String.format("LwM2M %s channel update registered", SERVICE_CHANNEL))); this.executorUnRegistered = Executors.newFixedThreadPool(this.context.getCtxServer().getUnRegisteredPoolSize(), @@ -147,14 +148,10 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { executorRegistered.submit(() -> { try { log.warn("[{}] [{{}] Client: create after Registration", registration.getEndpoint(), registration.getId()); - LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.updateInSessionsLwM2MClient(lwServer, registration); + LwM2MClient lwM2MClient = this.lwM2mInMemorySecurityStore.updateInSessionsLwM2MClient(lwServer, registration); if (lwM2MClient != null) { lwM2MClient.setLwM2MTransportServiceImpl(this); - lwM2MClient.setSessionUuid(UUID.randomUUID()); this.sentLogsToThingsboard(LOG_LW2M_INFO + ": Client Registered", registration); - LwM2MClientProfile lwM2MClientProfile = lwM2mInMemorySecurityStore.getProfile(registration.getId()); - this.putDelayedUpdateResourcesThingsboard(lwM2MClient); - this.setLwM2mFromClientValue(lwServer, registration, lwM2MClient, lwM2MClientProfile); SessionInfoProto sessionInfo = this.getValidateSessionInfo(registration); if (sessionInfo != null) { lwM2MClient.setDeviceUuid(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB())); @@ -165,9 +162,8 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { transportService.process(sessionInfo, DefaultTransportService.getSessionEventMsg(SessionEvent.OPEN), null); transportService.process(sessionInfo, TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().build(), null); this.sentLogsToThingsboard(LOG_LW2M_INFO + ": Client create after Registration", registration); -// if (LwM2MTransportHandler.getClientUpdateValueAfterConnect(lwM2MClientProfile)) { -// this.putDelayedUpdateResourcesThingsboard(lwM2MClient); -// } + this.initLwM2mFromClientValue(lwServer, registration, lwM2MClient); + } else { log.error("Client: [{}] onRegistered [{}] name [{}] sessionInfo ", registration.getId(), registration.getEndpoint(), null); } @@ -301,8 +297,7 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { if (!path.isEmpty() && (this.validatePathInAttrProfile(profile, path) || this.validatePathInTelemetryProfile(profile, path))) { if (resourceModel != null && resourceModel.operations.isWritable()) { lwM2MTransportRequest.sendAllRequest(lwM2MClient.getLwServer(), lwM2MClient.getRegistration(), path, POST_TYPE_OPER_WRITE_REPLACE, - ContentFormat.TLV.getName(), lwM2MClient, null, value, this.context.getCtxServer().getTimeout(), - false); + ContentFormat.TLV.getName(), null, value, this.context.getCtxServer().getTimeout()); } else { log.error("Resource path - [{}] value - [{}] is not Writable and cannot be updated", path, value); String logMsg = String.format(LOG_LW2M_ERROR + ": attributeUpdate: Resource path - %s value - %s is not Writable and cannot be updated", path, value); @@ -356,8 +351,7 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { @Override public void doTrigger(LeshanServer lwServer, Registration registration, String path) { lwM2MTransportRequest.sendAllRequest(lwServer, registration, path, POST_TYPE_OPER_EXECUTE, - ContentFormat.TLV.getName(), null, null, null, this.context.getCtxServer().getTimeout(), - false); + ContentFormat.TLV.getName(), null, null, this.context.getCtxServer().getTimeout()); } /** @@ -461,23 +455,26 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { * @param registration - Registration LwM2M Client * @param lwM2MClient - object with All parameters off client */ - private void setLwM2mFromClientValue(LeshanServer lwServer, Registration registration, LwM2MClient lwM2MClient, LwM2MClientProfile lwM2MClientProfile) { + private void initLwM2mFromClientValue(LeshanServer lwServer, Registration registration, LwM2MClient lwM2MClient) { + LwM2MClientProfile lwM2MClientProfile = lwM2mInMemorySecurityStore.getProfile(registration.getId()); Set clientObjects = this.getAllOjectsInClient(registration); - if (clientObjects != null) { + if (clientObjects != null && !LwM2MTransportHandler.getClientOnlyObserveAfterConnect(lwM2MClientProfile)) { // #2 - if (!LwM2MTransportHandler.getClientOnlyObserveAfterConnect(lwM2MClientProfile) && !LwM2MTransportHandler.getClientUpdateValueAfterConnect(lwM2MClientProfile)) { - this.onSentReadAttrTelemetryToClient(lwServer, registration); + if (!LwM2MTransportHandler.getClientUpdateValueAfterConnect(lwM2MClientProfile)) { + this.initReadAttrTelemetryObserveToClient(lwServer, registration, lwM2MClient, GET_TYPE_OPER_READ); + } // #3 else { + lwM2MClient.getPendingRequests().addAll(clientObjects); clientObjects.forEach(path -> { lwM2MTransportRequest.sendAllRequest(lwServer, registration, path, GET_TYPE_OPER_READ, ContentFormat.TLV.getName(), - lwM2MClient, null, null, this.context.getCtxServer().getTimeout(), false); + null, null, this.context.getCtxServer().getTimeout()); }); } } // #1 - this.onSentObserveToClient(lwServer, registration); + this.initReadAttrTelemetryObserveToClient(lwServer, registration, lwM2MClient, GET_TYPE_OPER_OBSERVE); } /** @@ -519,7 +516,6 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { private void updateResourcesValue(Registration registration, LwM2mResource lwM2mResource, String path) { LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.getLwM2MClientWithReg(registration, null); lwM2MClient.updateResourceValue(path, lwM2mResource); - log.warn("upDateResize: [{}] [{}]", lwM2MClient.getEndPoint(), path); Set paths = new HashSet<>(); paths.add(path); this.updateAttrTelemetry(registration, false, paths); @@ -582,30 +578,45 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { } /** - * Start observe + * Start observe/read: Attr/Telemetry * #1 - Analyze: - * #1.1 path in observe profile == client resource - * - * @param lwServer - LeshanServer - * @param registration - Registration LwM2M Client + * #1.1 path in resource profile == client resource + * @param lwServer - + * @param registration - */ - private void onSentObserveToClient(LeshanServer lwServer, Registration registration) { - if (lwServer.getObservationService().getObservations(registration).size() > 0) { - this.setCancelObservations(lwServer, registration); - } - LwM2MClientProfile lwM2MClientProfile = lwM2mInMemorySecurityStore.getProfile(registration.getId()); - Set clientInstances = this.getAllInstancesInClient(registration); - lwM2MClientProfile.getPostObserveProfile().forEach(p -> { - // #1.1 - String target = p.getAsString().toString(); - String[] resPath = target.split("/"); - String instance = "/" + resPath[1] + "/" + resPath[2]; - if (clientInstances.contains(instance)) { - lwM2MTransportRequest.sendAllRequest(lwServer, registration, target, GET_TYPE_OPER_OBSERVE, - null, null, null, null, this.context.getCtxServer().getTimeout(), - false); + private void initReadAttrTelemetryObserveToClient(LeshanServer lwServer, Registration registration, LwM2MClient lwM2MClient, String typeOper) { + try { + LwM2MClientProfile lwM2MClientProfile = lwM2mInMemorySecurityStore.getProfile(registration.getId()); + Set clientInstances = this.getAllInstancesInClient(registration); + Set result; + if (GET_TYPE_OPER_READ.equals(typeOper)) { + result = new ObjectMapper().readValue(lwM2MClientProfile.getPostAttributeProfile().getAsJsonArray().toString().getBytes(), Set.class); + result.addAll(new ObjectMapper().readValue(lwM2MClientProfile.getPostTelemetryProfile().getAsJsonArray().toString().getBytes(), Set.class)); } - }); + else { + result = new ObjectMapper().readValue(lwM2MClientProfile.getPostObserveProfile().getAsJsonArray().toString().getBytes(), Set.class); + } + Set pathSent = ConcurrentHashMap.newKeySet(); + result.forEach(p -> { + // #1.1 + String target = p; + String[] resPath = target.split("/"); + String instance = "/" + resPath[1] + "/" + resPath[2]; + if (clientInstances.contains(instance)) { + pathSent.add(target); + } + }); + lwM2MClient.getPendingRequests().addAll(pathSent); + pathSent.forEach(target -> { + lwM2MTransportRequest.sendAllRequest(lwServer, registration, target, typeOper, ContentFormat.TLV.getName(), + null, null, this.context.getCtxServer().getTimeout()); + }); + if (GET_TYPE_OPER_OBSERVE.equals(typeOper)) { + lwM2MClient.initValue(this, null); + } + } catch (IOException e) { + e.printStackTrace(); + } } /** @@ -657,28 +668,6 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { return (clientInstances.size() > 0) ? clientInstances : null; } - private void onSentReadAttrTelemetryToClient(LeshanServer lwServer, Registration registration) { - LwM2MClientProfile lwM2MClientProfile = lwM2mInMemorySecurityStore.getProfile(registration.getId()); - Set clientInstances = this.getAllInstancesInClient(registration); - Set attr = ConcurrentHashMap.newKeySet(); - try { - attr = new ObjectMapper().readValue(lwM2MClientProfile.getPostAttributeProfile().getAsJsonArray().toString().getBytes(), Set.class); - attr.addAll(new ObjectMapper().readValue(lwM2MClientProfile.getPostTelemetryProfile().getAsJsonArray().toString().getBytes(), Set.class)); - } catch (IOException e) { - e.printStackTrace(); - } - attr.forEach(p -> { - // #1.1 - String target = p; - String[] resPath = target.split("/"); - String instance = "/" + resPath[1] + "/" + resPath[2]; - if (clientInstances.contains(instance)) { - lwM2MTransportRequest.sendAllRequest(lwServer, registration, target, GET_TYPE_OPER_READ, ContentFormat.TLV.getName(), - null, null, null, this.context.getCtxServer().getTimeout(), false); - } - }); - } - /** * get AttrName/TelemetryName with value from Client * @@ -767,7 +756,7 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { * @param path - * @param request - */ - public void onWriteResponseOk(Registration registration, String path, WriteRequest request, boolean isDelayedUpdate) { + public void onWriteResponseOk(Registration registration, String path, WriteRequest request) { this.updateResourcesValue(registration, ((LwM2mResource) request.getNode()), path); } @@ -841,8 +830,7 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.getLwM2MClientWithReg(null, registrationId); LeshanServer lwServer = lwM2MClient.getLwServer(); Registration registration = lwM2mInMemorySecurityStore.getByRegistration(registrationId); - log.warn("[{}] # 4.1", registration.getEndpoint()); - this.updateResourceValueObserve(lwServer, registration, sentAttrToThingsboard.getPathPostParametersAdd(), GET_TYPE_OPER_READ); + this.readResourceValueObserve(lwServer, registration, sentAttrToThingsboard.getPathPostParametersAdd(), GET_TYPE_OPER_READ); // sent attr/telemetry to tingsboard for new path this.updateAttrTelemetry(registration, false, sentAttrToThingsboard.getPathPostParametersAdd()); }); @@ -870,8 +858,7 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.getLwM2MClient(null, registrationId); LeshanServer lwServer = lwM2MClient.getLwServer(); Registration registration = lwM2mInMemorySecurityStore.getByRegistration(registrationId); - log.warn("[{}] # 5.1", registration.getEndpoint()); - this.updateResourceValueObserve(lwServer, registration, postObserveAnalyzer.getPathPostParametersAdd(), GET_TYPE_OPER_OBSERVE); + this.readResourceValueObserve(lwServer, registration, postObserveAnalyzer.getPathPostParametersAdd(), GET_TYPE_OPER_OBSERVE); // 5.3 del // sent Request cancel observe to Client this.cancelObserveIsValue(lwServer, registration, postObserveAnalyzer.getPathPostParametersDel()); @@ -914,18 +901,16 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { * @param registration - Registration LwM2M Client * @param targets - path Resources == [ "/2/0/0", "/2/0/1"] */ - private void updateResourceValueObserve(LeshanServer lwServer, Registration registration, Set targets, String typeOper) { + private void readResourceValueObserve(LeshanServer lwServer, Registration registration, Set targets, String typeOper) { targets.forEach(target -> { LwM2mPath pathIds = new LwM2mPath(target); if (pathIds.isResource()) { if (GET_TYPE_OPER_READ.equals(typeOper)) { lwM2MTransportRequest.sendAllRequest(lwServer, registration, target, typeOper, - ContentFormat.TLV.getName(), null, null, null, this.context.getCtxServer().getTimeout(), - false); + ContentFormat.TLV.getName(), null, null, this.context.getCtxServer().getTimeout()); } else if (GET_TYPE_OPER_OBSERVE.equals(typeOper)) { lwM2MTransportRequest.sendAllRequest(lwServer, registration, target, typeOper, - null, null, null, null, this.context.getCtxServer().getTimeout(), - false); + null, null, null, this.context.getCtxServer().getTimeout()); } } }); @@ -952,10 +937,11 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { } private void putDelayedUpdateResourcesClient(LwM2MClient lwM2MClient, Object valueOld, Object valueNew, String path) { - if (valueNew != null && !valueNew.toString().equals(valueOld.toString())) { + if (valueNew != null && (valueOld == null || !valueNew.toString().equals(valueOld.toString()))) { lwM2MTransportRequest.sendAllRequest(lwM2MClient.getLwServer(), lwM2MClient.getRegistration(), path, POST_TYPE_OPER_WRITE_REPLACE, - ContentFormat.TLV.getName(), lwM2MClient, null, valueNew, this.context.getCtxServer().getTimeout(), - true); + ContentFormat.TLV.getName(), null, valueNew, this.context.getCtxServer().getTimeout()); + } else { + log.error("05 delayError"); } } @@ -1007,22 +993,26 @@ public class LwM2MTransportServiceImpl implements LwM2MTransportService { * @param sessionInfo - */ public void onGetAttributesResponse(TransportProtos.GetAttributeResponseMsg attributesResponse, TransportProtos.SessionInfoProto sessionInfo) { - LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.getLwM2MClient(sessionInfo); - attributesResponse.getSharedAttributeListList().forEach(attr -> { - String path = this.getPathAttributeUpdate(sessionInfo, attr.getKv().getKey()); - // #1.1 - if (lwM2MClient.getDelayedRequests().containsKey(path) && attr.getTs() > lwM2MClient.getDelayedRequests().get(path).getTs()) { - lwM2MClient.getDelayedRequests().put(path, attr); - } else { - lwM2MClient.getDelayedRequests().put(path, attr); - } - }); - // #2.1 - lwM2MClient.getDelayedRequests().forEach((k, v) -> { - ArrayList listV = new ArrayList<>(); - listV.add(v.getKv()); - this.putDelayedUpdateResourcesClient(lwM2MClient, this.getResourceValueToString(lwM2MClient, k), getJsonObject(listV).get(v.getKv().getKey()), k); - }); + try { + LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.getLwM2MClient(sessionInfo); + attributesResponse.getSharedAttributeListList().forEach(attr -> { + String path = this.getPathAttributeUpdate(sessionInfo, attr.getKv().getKey()); + // #1.1 + if (lwM2MClient.getDelayedRequests().containsKey(path) && attr.getTs() > lwM2MClient.getDelayedRequests().get(path).getTs()) { + lwM2MClient.getDelayedRequests().put(path, attr); + } else { + lwM2MClient.getDelayedRequests().put(path, attr); + } + }); + // #2.1 + lwM2MClient.getDelayedRequests().forEach((k, v) -> { + ArrayList listV = new ArrayList<>(); + listV.add(v.getKv()); + this.putDelayedUpdateResourcesClient(lwM2MClient, this.getResourceValueToString(lwM2MClient, k), getJsonObject(listV).get(v.getKv().getKey()), k); + }); + } catch (Exception e) { + log.error(String.valueOf(e)); + } } /** diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mServerListener.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mServerListener.java index 937d2c7dcd..5ff826b25b 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mServerListener.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mServerListener.java @@ -108,7 +108,7 @@ public class LwM2mServerListener { @Override public void newObservation(Observation observation, Registration registration) { - log.info("Received newObservation from [{}] endpoint [{}] ", observation.getPath(), registration.getEndpoint()); +// log.info("Received newObservation from [{}] endpoint [{}] ", observation.getPath(), registration.getEndpoint()); } }; diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2MClient.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2MClient.java index 38885e1f49..f2f352ab18 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2MClient.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2MClient.java @@ -28,6 +28,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceCreden import org.thingsboard.server.transport.lwm2m.server.LwM2MTransportServiceImpl; import org.thingsboard.server.transport.lwm2m.utils.LwM2mValueConverterImpl; +import java.util.ArrayList; import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; @@ -49,73 +50,30 @@ public class LwM2MClient implements Cloneable { private ValidateDeviceCredentialsResponseMsg credentialsResponse; private Map attributes; private Map resources; - // private Set pendingRequests; private Map delayedRequests; -// private Set delayedRequestsId; -// private Map responses; + private ArrayList pendingRequests; + private boolean init; private final LwM2mValueConverterImpl converter; public Object clone() throws CloneNotSupportedException { return super.clone(); } - public LwM2MClient(String endPoint, String identity, SecurityInfo securityInfo, ValidateDeviceCredentialsResponseMsg credentialsResponse, UUID profileUuid) { + public LwM2MClient(String endPoint, String identity, SecurityInfo securityInfo, ValidateDeviceCredentialsResponseMsg credentialsResponse, UUID profileUuid, UUID sessionUuid) { this.endPoint = endPoint; this.identity = identity; this.securityInfo = securityInfo; this.credentialsResponse = credentialsResponse; this.attributes = new ConcurrentHashMap<>(); -// this.pendingRequests = ConcurrentHashMap.newKeySet(); this.delayedRequests = new ConcurrentHashMap<>(); + this.pendingRequests = new ArrayList<>(); this.resources = new ConcurrentHashMap<>(); -// this.delayedRequestsId = ConcurrentHashMap.newKeySet(); this.profileUuid = profileUuid; - /** - * Key , response instance -> resources: value...> - */ -// this.responses = new ConcurrentHashMap<>(); + this.sessionUuid = sessionUuid; this.converter = LwM2mValueConverterImpl.getInstance(); + this.init = false; } -// /** -// * Fill with data -> Model client -// * -// * @param path - -// * @param response - -// */ -// public void onSuccessHandler(String path, LwM2mResponse response) { -// this.responses.put(path, response); -// this.pendingRequests.remove(path); -// if (this.pendingRequests.size() == 0) { -// this.initValue(); -// this.lwM2MTransportServiceImpl.putDelayedUpdateResourcesThingsboard(this); -// } -// } -// -// private void initValue() { -// this.responses.forEach((key, lwM2mResponse) -> { -// LwM2mPath pathIds = new LwM2mPath(key); -// if (pathIds.isObjectInstance()) { -// ((LwM2mObjectInstance) ((ReadResponse) lwM2mResponse).getContent()).getResources().forEach((k, v) -> { -// String pathRez = pathIds.toString() + "/" + k; -// this.updateResourceValue(pathRez, v); -// }); -// } -// else if (pathIds.isResource()) { -// this.updateResourceValue(pathIds.toString(), ((LwM2mResource) ((ReadResponse) lwM2mResponse).getContent())); -// } -// }); -// if (this.responses.size() == 0) this.responses = new ConcurrentHashMap<>(); -// } - -// public void updateObjectInstanceResourceValue(String pathInst, LwM2mObjectInstance instance) { -// LwM2mPath pathIds = new LwM2mPath(pathInst); -// instance.getResources().forEach((k, v) -> { -// String pathRez = pathIds.toString() + "/" + k; -// this.updateResourceValue(pathRez, v); -// }); -// } - public void updateResourceValue(String pathRez, LwM2mResource rez) { if (rez instanceof LwM2mMultipleResource) { this.resources.put(pathRez, new ResourceValue(rez.getValues(), null, true)); @@ -124,17 +82,18 @@ public class LwM2MClient implements Cloneable { } } -// /** -// * if path != null -// * -// * @param path -// */ -// public void onSuccessOrErrorDelayedRequests(String path) { -// if (path != null) this.delayedRequests.remove(path); -// if (this.delayedRequests.size() == 0 && this.getDelayedRequestsId().size() == 0) { -// this.lwM2MTransportServiceImpl.updatesAndSentModelParameter(this); -// } -// } + public void initValue(LwM2MTransportServiceImpl lwM2MTransportService, String path) { + if (path != null) { + this.pendingRequests.remove(path); + } + if (this.pendingRequests.size() == 0) { + this.init = true; + lwM2MTransportService.putDelayedUpdateResourcesThingsboard(this); + } + } + public LwM2MClient copy() { + return new LwM2MClient(this.endPoint, this.identity, this.securityInfo, this.credentialsResponse, this.profileUuid, this.sessionUuid); + } } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/secure/LwM2mInMemorySecurityStore.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/secure/LwM2mInMemorySecurityStore.java index b49193434d..b75b7d6102 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/secure/LwM2mInMemorySecurityStore.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/secure/LwM2mInMemorySecurityStore.java @@ -165,7 +165,7 @@ public class LwM2mInMemorySecurityStore extends InMemorySecurityStore { if (this.sessions.get(registration.getEndpoint()) == null) { this.addLwM2MClientToSession(registration.getEndpoint()); } - LwM2MClient lwM2MClient = this.sessions.get(registration.getEndpoint()); + LwM2MClient lwM2MClient = this.sessions.get(registration.getEndpoint()).copy(); lwM2MClient.setLwServer(lwServer); lwM2MClient.setRegistration(registration); lwM2MClient.setAttributes(registration.getAdditionalRegistrationAttributes()); @@ -203,9 +203,9 @@ public class LwM2mInMemorySecurityStore extends InMemorySecurityStore { UUID profileUuid = (store.getDeviceProfile() != null && addUpdateProfileParameters(store.getDeviceProfile())) ? store.getDeviceProfile().getUuidId() : null; if (store.getSecurityInfo() != null && profileUuid != null) { String endpoint = store.getSecurityInfo().getEndpoint(); - sessions.put(endpoint, new LwM2MClient(endpoint, store.getSecurityInfo().getIdentity(), store.getSecurityInfo(), store.getMsg(), profileUuid)); + sessions.put(endpoint, new LwM2MClient(endpoint, store.getSecurityInfo().getIdentity(), store.getSecurityInfo(), store.getMsg(), profileUuid, UUID.randomUUID())); } else if (store.getSecurityMode() == NO_SEC.code && profileUuid != null) { - sessions.put(identity, new LwM2MClient(identity, null, null, store.getMsg(), profileUuid)); + sessions.put(identity, new LwM2MClient(identity, null, null, store.getMsg(), profileUuid, UUID.randomUUID())); } else { log.error("Registration failed: FORBIDDEN/profileUuid/device [{}] , endpointId: [{}]", profileUuid, identity); /**