Use Flux stream processing for CM-handle searches

This greatly reduces memory consumption to fetch CM-handles in NCMP
by fetching in batches in a Flux. Full CM-handle search operations
now consume much less memory than before. The lower memory usage and
database pressure improves overall performance.

Issue-ID: CPS-2712
Signed-off-by: danielhanrahan <daniel.hanrahan@est.tech>
Change-Id: I7f653fadeadbf9612e0847f9451654b01a1a5604

2 files changed, 7 insertions, 7 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 d7b38d1a46..6215427dc1 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
@@ -1,7 +1,7 @@
 /*
  *  ============LICENSE_START=======================================================
  *  Copyright (C) 2021 Pantheon.tech
- *  Modifications Copyright (C) 2021-2025 Nordix Foundation
+ *  Modifications Copyright (C) 2021-2025 OpenInfra Foundation Europe
  *  Modifications Copyright (C) 2021 highstreet technologies GmbH
  *  Modifications Copyright (C) 2021-2022 Bell Canada
  *  ================================================================================
@@ -262,10 +262,9 @@ public class NetworkCmProxyController implements NetworkCmProxyApi {
             final CmHandleQueryParameters cmHandleQueryParameters) {
         final CmHandleQueryApiParameters cmHandleQueryApiParameters =
                 deprecationHelper.mapOldConditionProperties(cmHandleQueryParameters);
-        final Collection<NcmpServiceCmHandle> cmHandles = networkCmProxyInventoryFacade
-                .executeCmHandleSearch(cmHandleQueryApiParameters);
         final List<RestOutputCmHandle> restOutputCmHandles =
-                cmHandles.stream().map(this::toRestOutputCmHandle).collect(Collectors.toList());
+                networkCmProxyInventoryFacade.executeCmHandleSearch(cmHandleQueryApiParameters)
+                        .map(this::toRestOutputCmHandle).collectList().block();
         return ResponseEntity.ok(restOutputCmHandles);
     }
 
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 c3aca5a99b..94c113c053 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
@@ -2,7 +2,7 @@
  *  ============LICENSE_START=======================================================
  *  Copyright (C) 2021 Pantheon.tech
  *  Modifications Copyright (C) 2021 highstreet technologies GmbH
- *  Modifications Copyright (C) 2021-2024 Nordix Foundation
+ *  Modifications Copyright (C) 2021-2025 OpenInfra Foundation Europe
  *  Modifications Copyright (C) 2021-2022 Bell Canada.
  *  ================================================================================
  *  Licensed under the Apache License, Version 2.0 (the "License");
@@ -59,6 +59,7 @@ import org.springframework.http.HttpStatus
 import org.springframework.http.MediaType
 import org.springframework.http.ResponseEntity
 import org.springframework.test.web.servlet.MockMvc
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import spock.lang.Shared
 import spock.lang.Specification
@@ -275,7 +276,7 @@ class NetworkCmProxyControllerSpec extends Specification {
             cmHandle2.alternateId = 'someAlternateId'
             cmHandle2.moduleSetTag = 'someModuleSetTag'
             cmHandle2.dataProducerIdentifier = 'someDataProducerIdentifier'
-            mockNetworkCmProxyInventoryFacade.executeCmHandleSearch(_) >> [cmHandle1, cmHandle2]
+            mockNetworkCmProxyInventoryFacade.executeCmHandleSearch(_) >> Flux.fromIterable([cmHandle1, cmHandle2])
         when: 'the searches api is invoked'
             def response = mvc.perform(post(searchesEndpoint).contentType(MediaType.APPLICATION_JSON).content(jsonString)).andReturn().response
         then: 'response status returns OK'
@@ -352,7 +353,7 @@ class NetworkCmProxyControllerSpec extends Specification {
             cmHandle2.cmHandleId = 'ch-2'
             cmHandle2.publicProperties = [color: 'green']
             cmHandle2.currentTrustLevel = TrustLevel.NONE
-            mockNetworkCmProxyInventoryFacade.executeCmHandleSearch(_) >> [cmHandle1, cmHandle2]
+            mockNetworkCmProxyInventoryFacade.executeCmHandleSearch(_) >> Flux.fromIterable([cmHandle1, cmHandle2])
         when: 'the searches api is invoked'
             def response = mvc.perform(post(searchesEndpoint).contentType(MediaType.APPLICATION_JSON).content(jsonString)).andReturn().response
         then: 'an empty cm handle identifier is returned'