X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;ds=sidebyside;f=ves-agent%2Fsrc%2Fmain%2Fjava%2Forg%2Fcommscope%2Ftr069adapter%2Fvesagent%2Fasync%2FWaitForNotifications.java;fp=ves-agent%2Fsrc%2Fmain%2Fjava%2Forg%2Fcommscope%2Ftr069adapter%2Fvesagent%2Fasync%2FWaitForNotifications.java;h=614ce447c4b10b9ed2a7997fb71de1c47453fe17;hb=641a6c47b4ee74412e9386b3c95693adda3cafc9;hp=0000000000000000000000000000000000000000;hpb=8c7432b8380a355e89df05f070e7d88e599912fd;p=oam%2Ftr069-adapter.git

diff --git a/ves-agent/src/main/java/org/commscope/tr069adapter/vesagent/async/WaitForNotifications.java b/ves-agent/src/main/java/org/commscope/tr069adapter/vesagent/async/WaitForNotifications.java
new file mode 100644
index 0000000..614ce44
--- /dev/null
+++ b/ves-agent/src/main/java/org/commscope/tr069adapter/vesagent/async/WaitForNotifications.java
@@ -0,0 +1,100 @@
+package org.commscope.tr069adapter.vesagent.async;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import org.commscope.tr069adapter.acs.common.DeviceRPCResponse;
+import org.commscope.tr069adapter.acs.common.OperationCode;
+import org.commscope.tr069adapter.acs.common.OperationResponse;
+import org.commscope.tr069adapter.acs.common.dto.CustomOperationCode;
+import org.commscope.tr069adapter.mapper.model.VESNotification;
+import org.commscope.tr069adapter.vesagent.util.VesAgentConstants;
+import org.commscope.tr069adapter.vesagent.util.VesAgentUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+@Component
+public class WaitForNotifications {
+
+  private static final Logger LOG = LoggerFactory.getLogger(WaitForNotifications.class);
+
+  private static Map<String, Future<DeviceRPCResponse>> opFutureMap = new HashMap<>();
+  private static Map<String, DeviceRPCResponse> opResultMap = new HashMap<>();
+  private static Map<String, Semaphore> semaphoreMap = new HashMap<>();
+
+  public void notifyDeviceNotification(VESNotification notification) {
+    String deviceOperationKey = VesAgentUtils.getDeviceOperationKey(
+        notification.getDevnotification().getDeviceDetails().getDeviceId(),
+        CustomOperationCode.CONNECT);
+
+    if (!semaphoreMap.containsKey(deviceOperationKey)) {
+      return;
+    }
+
+    DeviceRPCResponse response = new DeviceRPCResponse();
+    response.setDeviceDetails(notification.getDevnotification().getDeviceDetails());
+
+    OperationResponse operationResponse = new OperationResponse();
+    operationResponse.setStatus(VesAgentConstants.DEVICE_IS_REACHABLE);
+    operationResponse.setOperationCode(CustomOperationCode.CONNECT);
+
+    response.setOperationResponse(operationResponse);
+
+    opResultMap.put(deviceOperationKey, response);
+    Semaphore mutex = semaphoreMap.remove(deviceOperationKey);
+    mutex.release();
+  }
+
+
+  public void notifyResult(DeviceRPCResponse opResult) {
+    String deviceOperationKey =
+        VesAgentUtils.getDeviceOperationKey(opResult.getDeviceDetails().getDeviceId(),
+            opResult.getOperationResponse().getOperationCode());
+
+    if (!semaphoreMap.containsKey(deviceOperationKey)) {
+      return;
+    }
+
+    opResultMap.put(deviceOperationKey, opResult);
+    Semaphore mutex = semaphoreMap.remove(deviceOperationKey);
+    mutex.release();
+  }
+
+  public DeviceRPCResponse getOperationResult(String deviceId, OperationCode opCode) {
+    return opResultMap.remove(VesAgentUtils.getDeviceOperationKey(deviceId, opCode));
+  }
+
+  public boolean waitForResult(String deviceId, OperationCode opCode,
+      Future<DeviceRPCResponse> futureResponse, long timeout) throws InterruptedException {
+    LOG.debug("Waiting for operation result for device:{}, operation: {}", deviceId, opCode);
+
+    String deviceOperationKey = VesAgentUtils.getDeviceOperationKey(deviceId, opCode);
+    opFutureMap.put(deviceOperationKey, futureResponse);
+
+    Semaphore semaphore = new Semaphore(0);
+    semaphoreMap.put(deviceOperationKey, semaphore);
+
+    LOG.debug("Semaphore MAP size = {}", semaphoreMap.size());
+    LOG.debug("opResultMap MAP size = {}", opResultMap.size());
+    LOG.debug("opFutureMap MAP size = {}", opFutureMap.size());
+
+    return semaphore.tryAcquire(timeout, TimeUnit.SECONDS);
+  }
+
+  public void stopOperation(String deviceId, OperationCode opCode) {
+    LOG.debug("Stopping waiting for operation result thread for device:{}, operation: {}", deviceId,
+        opCode);
+
+    Future<DeviceRPCResponse> operationInstance =
+        opFutureMap.remove(VesAgentUtils.getDeviceOperationKey(deviceId, opCode));
+
+    if (null != operationInstance) {
+      LOG.info("Stopping operation result waiting thread for operation : {}", operationInstance);
+      operationInstance.cancel(true);
+    }
+  }
+}