Async request response NCMP -> Client

-Added consumer for DMI events and producer for forwarding to client
-Added schemas for events
-Updated tests
-Added new module for ncmp events
-Used mapstruct for event mapping

Issue-ID: CPS-830
Change-Id: I096d08af9d69092cf8651e11eaa00ce441fc3605
Signed-off-by: sourabh_sourabh <sourabh.sourabh@est.tech>
Signed-off-by: JosephKeenan <joseph.keenan@est.tech>
Signed-off-by: ToineSiebelink <toine.siebelink@est.tech>
Signed-off-by: JosephKeenan <joseph.keenan@est.tech>

6 files changed, 191 insertions, 71 deletions

diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/NetworkCmProxyController.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/NetworkCmProxyController.java
index cedc94672c..11517bcc9e 100755
--- a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/NetworkCmProxyController.java
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/controller/NetworkCmProxyController.java
@@ -46,6 +46,7 @@ import org.onap.cps.ncmp.api.impl.exception.InvalidTopicException;
 import org.onap.cps.ncmp.api.models.CmHandleQueryApiParameters;
 import org.onap.cps.ncmp.api.models.NcmpServiceCmHandle;
 import org.onap.cps.ncmp.rest.api.NetworkCmProxyApi;
+import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor;
 import org.onap.cps.ncmp.rest.mapper.RestOutputCmHandleStateMapper;
 import org.onap.cps.ncmp.rest.model.CmHandleProperties;
 import org.onap.cps.ncmp.rest.model.CmHandleProperty;
@@ -61,6 +62,7 @@ import org.onap.cps.ncmp.rest.model.RestOutputCmHandle;
 import org.onap.cps.ncmp.rest.model.RestOutputCmHandlePublicProperties;
 import org.onap.cps.utils.CpsValidator;
 import org.onap.cps.utils.JsonObjectMapper;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -75,12 +77,14 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
     private static final String NO_BODY = null;
     private static final String NO_REQUEST_ID = null;
     private static final String NO_TOPIC = null;
-    public static final String ASYNC_REQUEST_ID = "requestId";
-
     private final NetworkCmProxyDataService networkCmProxyDataService;
     private final JsonObjectMapper jsonObjectMapper;
     private final NcmpRestInputMapper ncmpRestInputMapper;
     private final RestOutputCmHandleStateMapper restOutputCmHandleStateMapper;
+    private final CpsNcmpTaskExecutor cpsNcmpTaskExecutor;
+
+    @Value("${notification.async.executor.time-out-value-in-ms:2000}")
+    private int timeOutInMilliSeconds;
 
     /**
      * Get resource data from operational datastore.
@@ -96,19 +100,21 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
                                                                         final @NotNull @Valid String resourceIdentifier,
                                                                         final @Valid String optionsParamInQuery,
                                                                         final @Valid String topicParamInQuery) {
-        final ResponseEntity<Map<String, Object>> asyncResponse = populateAsyncResponse(topicParamInQuery);
-        final Map<String, Object> asyncResponseData = asyncResponse.getBody();
+        if (isValidTopic(topicParamInQuery)) {
+            final String requestId = UUID.randomUUID().toString();
+            cpsNcmpTaskExecutor.executeTask(() ->
+                networkCmProxyDataService.getResourceDataOperationalForCmHandle(
+                    cmHandle, resourceIdentifier, optionsParamInQuery, topicParamInQuery,
+                        requestId
+                ), timeOutInMilliSeconds
+            );
+            return acknowledgeAsyncRequest(requestId);
+        }
 
-        final Object responseObject = networkCmProxyDataService.getResourceDataOperationalForCmHandle(cmHandle,
-                resourceIdentifier,
-                optionsParamInQuery,
-                asyncResponseData == null ? NO_TOPIC : topicParamInQuery,
-                asyncResponseData == null ? NO_REQUEST_ID : asyncResponseData.get(ASYNC_REQUEST_ID).toString());
+        final Object responseObject = networkCmProxyDataService.getResourceDataOperationalForCmHandle(
+            cmHandle, resourceIdentifier, optionsParamInQuery, NO_TOPIC, NO_REQUEST_ID);
 
-        if (asyncResponseData == null) {
-            return ResponseEntity.ok(responseObject);
-        }
-        return ResponseEntity.ok(asyncResponse);
+        return ResponseEntity.ok(responseObject);
     }
 
     /**
@@ -125,19 +131,21 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
                                                                     final @NotNull @Valid String resourceIdentifier,
                                                                     final @Valid String optionsParamInQuery,
                                                                     final @Valid String topicParamInQuery) {
-        final ResponseEntity<Map<String, Object>> asyncResponse = populateAsyncResponse(topicParamInQuery);
-        final Map<String, Object> asyncResponseData = asyncResponse.getBody();
+        if (isValidTopic(topicParamInQuery)) {
+            final String resourceDataRequestId = UUID.randomUUID().toString();
+            cpsNcmpTaskExecutor.executeTask(() ->
+                networkCmProxyDataService.getResourceDataPassThroughRunningForCmHandle(
+                    cmHandle, resourceIdentifier, optionsParamInQuery, topicParamInQuery,
+                        resourceDataRequestId
+                ), timeOutInMilliSeconds
+            );
+            return acknowledgeAsyncRequest(resourceDataRequestId);
+        }
 
-        final Object responseObject = networkCmProxyDataService.getResourceDataPassThroughRunningForCmHandle(cmHandle,
-                resourceIdentifier,
-                optionsParamInQuery,
-                asyncResponseData == null ? NO_TOPIC : topicParamInQuery,
-                asyncResponseData == null ? NO_REQUEST_ID : asyncResponseData.get(ASYNC_REQUEST_ID).toString());
+        final Object responseObject = networkCmProxyDataService.getResourceDataPassThroughRunningForCmHandle(
+            cmHandle, resourceIdentifier, optionsParamInQuery, NO_TOPIC, NO_REQUEST_ID);
 
-        if (asyncResponseData == null) {
-            return ResponseEntity.ok(responseObject);
-        }
-        return ResponseEntity.ok(asyncResponse);
+        return ResponseEntity.ok(responseObject);
     }
 
     @Override
@@ -319,18 +327,7 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
         return restOutputCmHandle;
     }
 
-    private ResponseEntity<Map<String, Object>> populateAsyncResponse(final String topicParamInQuery) {
-        final boolean processAsynchronously = hasTopicParameter(topicParamInQuery);
-        final Map<String, Object> responseData;
-        if (processAsynchronously) {
-            responseData = getAsyncResponseData();
-        } else {
-            responseData = null;
-        }
-        return ResponseEntity.ok().body(responseData);
-    }
-
-    private static boolean hasTopicParameter(final String topicName) {
+    private static boolean isValidTopic(final String topicName) {
         if (topicName == null) {
             return false;
         }
@@ -340,11 +337,11 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
         throw new InvalidTopicException("Topic name " + topicName + " is invalid", "invalid topic");
     }
 
-    private Map<String, Object> getAsyncResponseData() {
-        final Map<String, Object> asyncResponseData = new HashMap<>(1);
-        final String resourceDataRequestId = UUID.randomUUID().toString();
-        asyncResponseData.put(ASYNC_REQUEST_ID, resourceDataRequestId);
-        return asyncResponseData;
+    private ResponseEntity<Object> acknowledgeAsyncRequest(final String requestId) {
+        final Map<String, Object> acknowledgeData = new HashMap<>(1);
+        acknowledgeData.put("requestId", requestId);
+        return ResponseEntity.ok(acknowledgeData);
     }
 
 }
+
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/exceptions/CpsTaskExecutionException.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/exceptions/CpsTaskExecutionException.java
new file mode 100644
index 0000000000..3e8929d2e3
--- /dev/null
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/exceptions/CpsTaskExecutionException.java
@@ -0,0 +1,44 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 Nordix Foundation
+ *  ================================================================================
+ *  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.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.rest.exceptions;
+
+import lombok.Getter;
+
+public class CpsTaskExecutionException extends RuntimeException {
+
+    private static final long serialVersionUID = 1481520410918497454L;
+
+    @Getter
+    final String details;
+
+    /**
+     * Constructor.
+     *
+     * @param message the error message
+     * @param details the error details
+     * @param cause   the cause of the exception
+     */
+    public CpsTaskExecutionException(final String message, final String details, final Throwable cause) {
+        super(message, cause);
+        this.details = details;
+    }
+
+}
\ No newline at end of file
diff --git a/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutor.java b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutor.java
new file mode 100644
index 0000000000..93aa2858ca
--- /dev/null
+++ b/cps-ncmp-rest/src/main/java/org/onap/cps/ncmp/rest/executor/CpsNcmpTaskExecutor.java
@@ -0,0 +1,60 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2022 Nordix Foundation
+ *  ================================================================================
+ *  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.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.rest.executor;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+public class CpsNcmpTaskExecutor {
+
+    /**
+     * Execute task asynchronously and publish response to supplied topic.
+     *
+     * @param taskSupplier functional method is get() task need to executed asynchronously
+     * @param timeOutInMillis the time out value in milliseconds
+     */
+    public void executeTask(final Supplier<Object> taskSupplier, final int timeOutInMillis) {
+        CompletableFuture.supplyAsync(taskSupplier::get)
+            .orTimeout(timeOutInMillis, MILLISECONDS)
+            .whenCompleteAsync(
+                (responseAsJson, throwable) -> {
+                    handleTaskCompletion(throwable);
+                }
+            );
+    }
+
+    private void handleTaskCompletion(final Throwable throwable) {
+        if (throwable == null) {
+            log.info("Async task completed successfully.");
+        } else {
+            log.error("Async task failed. caused by : {}", throwable.getMessage());
+        }
+    }
+}
+
+
+
diff --git a/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/NetworkCmProxyControllerSpec.groovy b/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/NetworkCmProxyControllerSpec.groovy
index 6cf1506681..3315304258 100644
--- a/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/NetworkCmProxyControllerSpec.groovy
+++ b/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/controller/NetworkCmProxyControllerSpec.groovy
@@ -28,6 +28,7 @@ import org.onap.cps.ncmp.api.inventory.CmHandleState
 import org.onap.cps.ncmp.api.inventory.CompositeState
 import org.onap.cps.ncmp.api.models.NcmpServiceCmHandle
 import org.onap.cps.ncmp.rest.mapper.RestOutputCmHandleStateMapper
+import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor
 import spock.lang.Shared
 
 import java.time.OffsetDateTime
@@ -83,8 +84,8 @@ class NetworkCmProxyControllerSpec extends Specification {
     @SpringBean
     RestOutputCmHandleStateMapper restOutputCmHandleStateMapper = Mappers.getMapper(RestOutputCmHandleStateMapper)
 
-    def formattedDateAndTime = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ")
-        .format(OffsetDateTime.of(2022, 12, 31, 20, 30, 40, 1, ZoneOffset.UTC))
+    @SpringBean
+    CpsNcmpTaskExecutor spiedCpsTaskExecutor = Spy()
 
     @Value('${rest.api.ncmp-base-path}/v1')
     def ncmpBasePathV1
@@ -95,6 +96,9 @@ class NetworkCmProxyControllerSpec extends Specification {
     def NO_TOPIC = null
     def NO_REQUEST_ID = null
 
+    def formattedDateAndTime = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ")
+        .format(OffsetDateTime.of(2022, 12, 31, 20, 30, 40, 1, ZoneOffset.UTC))
+
     def 'Get Resource Data from pass-through operational.'() {
         given: 'resource data url'
             def getUrl = "$ncmpBasePathV1/ch/testCmHandle/data/ds/ncmp-datastore:passthrough-operational" +
@@ -120,34 +124,40 @@ class NetworkCmProxyControllerSpec extends Specification {
                     "?resourceIdentifier=parent/child&options=(a=1,b=2)${topicQueryParam}"
         when: 'get data resource request is performed'
             def response = mvc.perform(
-                    get(getUrl)
-                    .contentType(MediaType.APPLICATION_JSON)
-            ).andReturn().response
-        then: 'the NCMP data service is called with operational data for cm handle'
-            expectedNumberOfMethodExecutions
-                    * mockNetworkCmProxyDataService."${expectedMethodName}"('testCmHandle',
-                    'parent/child',
-                    '(a=1,b=2)',
-                    expectedTopicName,
-                    _)
-        then: 'response status is expected'
-            response.status == expectedHttpStatus
+                    get(getUrl).contentType(MediaType.APPLICATION_JSON)).andReturn().response
+        then: 'task executor is called appropriate number of times'
+            expectedNumberOfExecutorExecutions * spiedCpsTaskExecutor.executeTask(_, 2000)
+        and: 'response status is expected'
+            response.status == HttpStatus.OK.value()
+        where: 'the following parameters are used'
+            scenario                               | datastoreInUrl            | topicQueryParam        || expectedTopicName | expectedNumberOfExecutorExecutions
+            'url with valid topic'                 | 'passthrough-operational' | '&topic=my-topic-name' || 'my-topic-name'   | 1
+            'no topic in url'                      | 'passthrough-operational' | ''                     || NO_TOPIC          | 0
+            'null topic in url'                    | 'passthrough-operational' | '&topic=null'          || 'null'            | 1
+            'url with valid topic'                 | 'passthrough-running'     | '&topic=my-topic-name' || 'my-topic-name'   | 1
+            'no topic in url'                      | 'passthrough-running'     | ''                     || NO_TOPIC          | 0
+            'null topic in url'                    | 'passthrough-running'     | '&topic=null'          || 'null'            | 1
+    }
+
+    def 'Fail to get Resource Data from #datastoreInUrl when #scenario.'() {
+        given: 'resource data url'
+            def getUrl = "$ncmpBasePathV1/ch/testCmHandle/data/ds/ncmp-datastore:${datastoreInUrl}" +
+                "?resourceIdentifier=parent/child&options=(a=1,b=2)${topicQueryParam}"
+        when: 'get data resource request is performed'
+            def response = mvc.perform(
+                get(getUrl).contentType(MediaType.APPLICATION_JSON)).andReturn().response
+        then: 'abad request is returned'
+            response.status == HttpStatus.BAD_REQUEST.value()
         where: 'the following parameters are used'
-            scenario                               | datastoreInUrl            | topicQueryParam        || expectedTopicName | expectedMethodName                             | expectedNumberOfMethodExecutions | expectedHttpStatus
-            'url with valid topic'                 | 'passthrough-operational' | '&topic=my-topic-name' || 'my-topic-name'   | 'getResourceDataOperationalForCmHandle'        | 1                                | HttpStatus.OK.value()
-            'no topic in url'                      | 'passthrough-operational' | ''                     || NO_TOPIC          | 'getResourceDataOperationalForCmHandle'        | 1                                | HttpStatus.OK.value()
-            'null topic in url'                    | 'passthrough-operational' | '&topic=null'          || 'null'            | 'getResourceDataOperationalForCmHandle'        | 1                                | HttpStatus.OK.value()
-            'empty topic in url'                   | 'passthrough-operational' | '&topic=\"\"'          || null              | 'getResourceDataOperationalForCmHandle'        | 0                                | HttpStatus.BAD_REQUEST.value()
-            'missing topic in url'                 | 'passthrough-operational' | '&topic='              || null              | 'getResourceDataOperationalForCmHandle'        | 0                                | HttpStatus.BAD_REQUEST.value()
-            'blank topic value in url'             | 'passthrough-operational' | '&topic=\" \"'         || null              | 'getResourceDataOperationalForCmHandle'        | 0                                | HttpStatus.BAD_REQUEST.value()
-            'invalid non-empty topic value in url' | 'passthrough-operational' | '&topic=1_5_*_#'       || null              | 'getResourceDataOperationalForCmHandle'        | 0                                | HttpStatus.BAD_REQUEST.value()
-            'url with valid topic'                 | 'passthrough-running'     | '&topic=my-topic-name' || 'my-topic-name'   | 'getResourceDataPassThroughRunningForCmHandle' | 1                                | HttpStatus.OK.value()
-            'no topic in url'                      | 'passthrough-running'     | ''                     || NO_TOPIC          | 'getResourceDataPassThroughRunningForCmHandle' | 1                                | HttpStatus.OK.value()
-            'null topic in url'                    | 'passthrough-running'     | '&topic=null'          || 'null'            | 'getResourceDataPassThroughRunningForCmHandle' | 1                                | HttpStatus.OK.value()
-            'empty topic in url'                   | 'passthrough-running'     | '&topic=\"\"'          || null              | 'getResourceDataPassThroughRunningForCmHandle' | 0                                | HttpStatus.BAD_REQUEST.value()
-            'missing topic in url'                 | 'passthrough-running'     | '&topic='              || null              | 'getResourceDataPassThroughRunningForCmHandle' | 0                                | HttpStatus.BAD_REQUEST.value()
-            'blank topic value in url'             | 'passthrough-running'     | '&topic=\" \"'         || null              | 'getResourceDataPassThroughRunningForCmHandle' | 0                                | HttpStatus.BAD_REQUEST.value()
-            'invalid non-empty topic value in url' | 'passthrough-running'     | '&topic=1_5_*_#'       || null              | 'getResourceDataPassThroughRunningForCmHandle' | 0                                | HttpStatus.BAD_REQUEST.value()
+            scenario                               | datastoreInUrl            | topicQueryParam
+            'empty topic in url'                   | 'passthrough-operational' | '&topic=\"\"'
+            'missing topic in url'                 | 'passthrough-operational' | '&topic='
+            'blank topic value in url'             | 'passthrough-operational' | '&topic=\" \"'
+            'invalid non-empty topic value in url' | 'passthrough-operational' | '&topic=1_5_*_#'
+            'empty topic in url'                   | 'passthrough-running'     | '&topic=\"\"'
+            'missing topic in url'                 | 'passthrough-running'     | '&topic='
+            'blank topic value in url'             | 'passthrough-running'     | '&topic=\" \"'
+            'invalid non-empty topic value in url' | 'passthrough-running'     | '&topic=1_5_*_#'
     }
 
     def 'Get Resource Data from pass-through running with #scenario value in resource identifier param.'() {
diff --git a/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/exceptions/NetworkCmProxyRestExceptionHandlerSpec.groovy b/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/exceptions/NetworkCmProxyRestExceptionHandlerSpec.groovy
index 1258e6e1c4..fd3203b5b1 100644
--- a/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/exceptions/NetworkCmProxyRestExceptionHandlerSpec.groovy
+++ b/cps-ncmp-rest/src/test/groovy/org/onap/cps/ncmp/rest/exceptions/NetworkCmProxyRestExceptionHandlerSpec.groovy
@@ -30,6 +30,7 @@ import org.onap.cps.ncmp.api.impl.exception.HttpClientRequestException
 import org.onap.cps.ncmp.api.impl.exception.ServerNcmpException
 import org.onap.cps.ncmp.rest.controller.NcmpRestInputMapper
 import org.onap.cps.ncmp.rest.mapper.RestOutputCmHandleStateMapper
+import org.onap.cps.ncmp.rest.executor.CpsNcmpTaskExecutor
 import org.onap.cps.spi.exceptions.CpsException
 import org.onap.cps.spi.exceptions.DataNodeNotFoundException
 import org.onap.cps.spi.exceptions.DataValidationException
@@ -62,7 +63,7 @@ class NetworkCmProxyRestExceptionHandlerSpec extends Specification {
     NetworkCmProxyDataService mockNetworkCmProxyDataService = Mock()
 
     @SpringBean
-    JsonObjectMapper jsonObjectMapper = Stub()
+    JsonObjectMapper stubbedJsonObjectMapper = Stub()
 
     @SpringBean
     NcmpRestInputMapper ncmpRestInputMapper = Mappers.getMapper(NcmpRestInputMapper)
@@ -70,6 +71,9 @@ class NetworkCmProxyRestExceptionHandlerSpec extends Specification {
     @SpringBean
     RestOutputCmHandleStateMapper restOutputCmHandleStateMapper = Mappers.getMapper(RestOutputCmHandleStateMapper)
 
+    @SpringBean
+    CpsNcmpTaskExecutor stubbedCpsTaskExecutor = Stub()
+
     @Value('${rest.api.ncmp-base-path}')
     def basePathNcmp
 
diff --git a/cps-ncmp-rest/src/test/resources/application.yml b/cps-ncmp-rest/src/test/resources/application.yml
index f2ca8c759b..0241696c5b 100644
--- a/cps-ncmp-rest/src/test/resources/application.yml
+++ b/cps-ncmp-rest/src/test/resources/application.yml
@@ -21,3 +21,8 @@ rest:
     api:
         ncmp-base-path: /ncmp
         ncmp-inventory-base-path: /ncmpInventory
+
+notification:
+    async:
+        executor:
+            time-out-value-in-ms: 2000
\ No newline at end of file