Merge branch 'feature/lwm2m-refactoring-downlink' of https://github.com/YevhenBo…

…ndarenko/thingsboard into feature/lwm2m-refactoring-downlink

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.transport.lwm2m.server.downlink;
+
+import lombok.RequiredArgsConstructor;
+
+import java.util.concurrent.CountDownLatch;
+
+@RequiredArgsConstructor
+public class TbLwM2MLatchCallback<R, T> implements DownlinkRequestCallback<R, T> {
+
+    private final CountDownLatch countDownLatch;
+    private final DownlinkRequestCallback<R, T> callback;
+
+    @Override
+    public void onSuccess(R request, T response) {
+        callback.onSuccess(request, response);
+        countDownLatch.countDown();
+    }
+
+    @Override
+    public void onValidationError(String params, String msg) {
+        callback.onValidationError(params, msg);
+        countDownLatch.countDown();
+    }
+
+    @Override
+    public void onError(String params, Exception e) {
+        callback.onError(params, e);
+        countDownLatch.countDown();
+    }
+}

@@ -18,8 +18,8 @@ package org.thingsboard.server.transport.lwm2m.server.downlink;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.leshan.core.request.ReadRequest;
 import org.eclipse.leshan.core.response.ReadResponse;
-import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient;
+import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
 
 @Slf4j
 public class TbLwM2MReadCallback extends TbLwM2MTargetedCallback<ReadRequest, ReadResponse> {

@@ -28,7 +28,10 @@ import org.eclipse.leshan.core.node.LwM2mObjectInstance;
 import org.eclipse.leshan.core.node.LwM2mPath;
 import org.eclipse.leshan.core.node.LwM2mResource;
 import org.eclipse.leshan.core.observation.Observation;
+import org.eclipse.leshan.core.request.ObserveRequest;
+import org.eclipse.leshan.core.request.ReadRequest;
 import org.eclipse.leshan.core.request.WriteRequest;
+import org.eclipse.leshan.core.response.ObserveResponse;
 import org.eclipse.leshan.core.response.ReadResponse;
 import org.eclipse.leshan.server.registration.Registration;
 import org.springframework.context.annotation.Lazy;
@@ -68,7 +71,9 @@ import org.thingsboard.server.transport.lwm2m.server.client.LwM2mFwSwUpdate;
 import org.thingsboard.server.transport.lwm2m.server.client.ParametersAnalyzeResult;
 import org.thingsboard.server.transport.lwm2m.server.client.ResourceValue;
 import org.thingsboard.server.transport.lwm2m.server.client.ResultsAddKeyValueProto;
+import org.thingsboard.server.transport.lwm2m.server.downlink.DownlinkRequestCallback;
 import org.thingsboard.server.transport.lwm2m.server.downlink.LwM2mDownlinkMsgHandler;
+import org.thingsboard.server.transport.lwm2m.server.downlink.TbLwM2MLatchCallback;
 import org.thingsboard.server.transport.lwm2m.server.downlink.TbLwM2MCancelObserveCallback;
 import org.thingsboard.server.transport.lwm2m.server.downlink.TbLwM2MCancelObserveRequest;
 import org.thingsboard.server.transport.lwm2m.server.downlink.TbLwM2MDiscoverCallback;
@@ -97,6 +102,7 @@ import java.util.Random;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -451,16 +457,29 @@ public class DefaultLwM2MUplinkMsgHandler implements LwM2mUplinkMsgHandler {
         Set<String> targetIds = new HashSet<>(profile.getObserveAttr().getAttribute());
         targetIds.addAll(profile.getObserveAttr().getTelemetry());
         targetIds = targetIds.stream().filter(target -> isSupportedTargetId(supportedObjects, target)).collect(Collectors.toSet());
-        lwM2MClient.getPendingReadRequests().addAll(targetIds);
-        targetIds.forEach(versionedId -> sendReadRequest(lwM2MClient, versionedId));
+
+        CountDownLatch latch = new CountDownLatch(targetIds.size());
+        targetIds.forEach(versionedId -> sendReadRequest(lwM2MClient, versionedId,
+                new TbLwM2MLatchCallback<>(latch, new TbLwM2MReadCallback(this, lwM2MClient, versionedId))));
+        try {
+            latch.await();
+        } catch (InterruptedException e) {
+            log.error("Failed to await Read requests!");
+        }
     }
 
     private void sendObserveRequests(LwM2mClient lwM2MClient, Lwm2mDeviceProfileTransportConfiguration profile, Set<String> supportedObjects) {
         Set<String> targetIds = profile.getObserveAttr().getObserve();
         targetIds = targetIds.stream().filter(target -> isSupportedTargetId(supportedObjects, target)).collect(Collectors.toSet());
-//        TODO: why do we need to put observe into pending read requests?
-//        lwM2MClient.getPendingReadRequests().addAll(targetIds);
-        targetIds.forEach(targetId -> sendObserveRequest(lwM2MClient, targetId));
+
+        CountDownLatch latch = new CountDownLatch(targetIds.size());
+        targetIds.forEach(targetId -> sendObserveRequest(lwM2MClient, targetId,
+                new TbLwM2MLatchCallback<>(latch, new TbLwM2MObserveCallback(this, lwM2MClient, targetId))));
+        try {
+            latch.await();
+        } catch (InterruptedException e) {
+            log.error("Failed to await Observe requests!");
+        }
     }
 
     private void sendWriteAttributeRequests(LwM2mClient lwM2MClient, Lwm2mDeviceProfileTransportConfiguration profile, Set<String> supportedObjects) {
@@ -485,13 +504,21 @@ public class DefaultLwM2MUplinkMsgHandler implements LwM2mUplinkMsgHandler {
     }
 
     private void sendReadRequest(LwM2mClient lwM2MClient, String versionedId) {
+        sendReadRequest(lwM2MClient, versionedId, new TbLwM2MReadCallback(this, lwM2MClient, versionedId));
+    }
+
+    private void sendReadRequest(LwM2mClient lwM2MClient, String versionedId, DownlinkRequestCallback<ReadRequest, ReadResponse> callback) {
         TbLwM2MReadRequest request = TbLwM2MReadRequest.builder().versionedId(versionedId).timeout(this.config.getTimeout()).build();
-        defaultLwM2MDownlinkMsgHandler.sendReadRequest(lwM2MClient, request, new TbLwM2MReadCallback(this, lwM2MClient, versionedId));
+        defaultLwM2MDownlinkMsgHandler.sendReadRequest(lwM2MClient, request, callback);
     }
 
     private void sendObserveRequest(LwM2mClient lwM2MClient, String versionedId) {
+        sendObserveRequest(lwM2MClient, versionedId, new TbLwM2MObserveCallback(this, lwM2MClient, versionedId));
+    }
+
+    private void sendObserveRequest(LwM2mClient lwM2MClient, String versionedId, DownlinkRequestCallback<ObserveRequest, ObserveResponse> callback) {
         TbLwM2MObserveRequest request = TbLwM2MObserveRequest.builder().versionedId(versionedId).timeout(this.config.getTimeout()).build();
-        defaultLwM2MDownlinkMsgHandler.sendObserveRequest(lwM2MClient, request, new TbLwM2MObserveCallback(this, lwM2MClient, versionedId));
+        defaultLwM2MDownlinkMsgHandler.sendObserveRequest(lwM2MClient, request, callback);
     }
 
     private void sendWriteAttributesRequest(LwM2mClient lwM2MClient, String targetId, ObjectAttributes params) {