Merge "Streamline outcome response for subscription creation"

5 files changed, 90 insertions, 21 deletions

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ResponseTimeoutTask.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ResponseTimeoutTask.java
index a81f8fd73..c178700ee 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ResponseTimeoutTask.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ResponseTimeoutTask.java
@@ -49,9 +49,7 @@ public class ResponseTimeoutTask implements Runnable {
     private void generateAndSendResponse() {
         final String subscriptionEventId = subscriptionClientId + subscriptionName;
         if (forwardedSubscriptionEventCache.containsKey(subscriptionEventId)) {
-            final Set<String> dmiNames = forwardedSubscriptionEventCache.get(subscriptionEventId);
-            subscriptionEventResponseOutcome.sendResponse(subscriptionClientId, subscriptionName,
-                    dmiNames.isEmpty());
+            subscriptionEventResponseOutcome.sendResponse(subscriptionClientId, subscriptionName);
             forwardedSubscriptionEventCache.remove(subscriptionEventId);
         }
     }
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java
index 9e363f3cd..1d87a057a 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java
@@ -21,6 +21,7 @@
 package org.onap.cps.ncmp.api.impl.events.avcsubscription;
 
 import com.hazelcast.map.IMap;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
@@ -37,8 +38,11 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.common.header.Headers;
 import org.onap.cps.ncmp.api.impl.config.embeddedcache.ForwardedSubscriptionEventCacheConfig;
 import org.onap.cps.ncmp.api.impl.events.EventsPublisher;
+import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence;
+import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus;
 import org.onap.cps.ncmp.api.impl.utils.DmiServiceNameOrganizer;
 import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle;
+import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent;
 import org.onap.cps.ncmp.api.inventory.InventoryPersistence;
 import org.onap.cps.ncmp.event.model.SubscriptionEvent;
 import org.onap.cps.spi.exceptions.OperationNotYetSupportedException;
@@ -55,6 +59,8 @@ public class SubscriptionEventForwarder {
     private final EventsPublisher<SubscriptionEvent> eventsPublisher;
     private final IMap<String, Set<String>> forwardedSubscriptionEventCache;
     private final SubscriptionEventResponseOutcome subscriptionEventResponseOutcome;
+    private final SubscriptionEventMapper subscriptionEventMapper;
+    private final SubscriptionPersistence subscriptionPersistence;
     private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
     @Value("${app.ncmp.avc.subscription-forward-topic-prefix}")
     private String dmiAvcSubscriptionTopicPrefix;
@@ -83,11 +89,29 @@ public class SubscriptionEventForwarder {
         final Map<String, Map<String, Map<String, String>>> dmiPropertiesPerCmHandleIdPerServiceName
                 = DmiServiceNameOrganizer.getDmiPropertiesPerCmHandleIdPerServiceName(yangModelCmHandles);
 
+        findDmisAndRespond(subscriptionEvent, eventHeaders, cmHandleTargetsAsStrings,
+                dmiPropertiesPerCmHandleIdPerServiceName);
+    }
+
+    private void findDmisAndRespond(final SubscriptionEvent subscriptionEvent, final Headers eventHeaders,
+                           final List<String> cmHandleTargetsAsStrings,
+                           final Map<String, Map<String, Map<String, String>>>
+                                            dmiPropertiesPerCmHandleIdPerServiceName) {
+        final List<String> cmHandlesThatExistsInDb = dmiPropertiesPerCmHandleIdPerServiceName.entrySet().stream()
+                .map(Map.Entry::getValue).map(Map::keySet).flatMap(Set::stream).collect(Collectors.toList());
+
+        final List<String> targetCmHandlesDoesNotExistInDb = new ArrayList<>(cmHandleTargetsAsStrings);
+        targetCmHandlesDoesNotExistInDb.removeAll(cmHandlesThatExistsInDb);
+
         final Set<String> dmisToRespond = new HashSet<>(dmiPropertiesPerCmHandleIdPerServiceName.keySet());
+
+        if (dmisToRespond.isEmpty() || !targetCmHandlesDoesNotExistInDb.isEmpty()) {
+            updatesCmHandlesToRejectedAndPersistSubscriptionEvent(subscriptionEvent, targetCmHandlesDoesNotExistInDb);
+        }
         if (dmisToRespond.isEmpty()) {
             final String clientID = subscriptionEvent.getEvent().getSubscription().getClientID();
             final String subscriptionName = subscriptionEvent.getEvent().getSubscription().getName();
-            subscriptionEventResponseOutcome.sendResponse(clientID, subscriptionName, true);
+            subscriptionEventResponseOutcome.sendResponse(clientID, subscriptionName);
         } else {
             startResponseTimeout(subscriptionEvent, dmisToRespond);
             forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, subscriptionEvent, eventHeaders);
@@ -130,4 +154,24 @@ public class SubscriptionEventForwarder {
             + "-"
             + dmiName;
     }
+
+    private void updatesCmHandlesToRejectedAndPersistSubscriptionEvent(
+            final SubscriptionEvent subscriptionEvent,
+            final List<String> targetCmHandlesDoesNotExistInDb) {
+        final YangModelSubscriptionEvent yangModelSubscriptionEvent =
+                subscriptionEventMapper.toYangModelSubscriptionEvent(subscriptionEvent);
+        yangModelSubscriptionEvent.getPredicates()
+                .setTargetCmHandles(findRejectedCmHandles(targetCmHandlesDoesNotExistInDb,
+                        yangModelSubscriptionEvent));
+        subscriptionPersistence.saveSubscriptionEvent(yangModelSubscriptionEvent);
+    }
+
+    private static List<YangModelSubscriptionEvent.TargetCmHandle> findRejectedCmHandles(
+            final List<String> targetCmHandlesDoesNotExistInDb,
+            final YangModelSubscriptionEvent yangModelSubscriptionEvent) {
+        return yangModelSubscriptionEvent.getPredicates().getTargetCmHandles().stream()
+                    .filter(targetCmHandle -> targetCmHandlesDoesNotExistInDb.contains(targetCmHandle.getCmHandleId()))
+                    .map(target -> new YangModelSubscriptionEvent.TargetCmHandle(target.getCmHandleId(),
+                                    SubscriptionStatus.REJECTED)).collect(Collectors.toList());
+    }
 }
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseConsumer.java
index a1860a613..20df706c0 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseConsumer.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseConsumer.java
@@ -21,6 +21,8 @@
 package org.onap.cps.ncmp.api.impl.events.avcsubscription;
 
 import com.hazelcast.map.IMap;
+import java.util.Collection;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import lombok.RequiredArgsConstructor;
@@ -28,8 +30,11 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.onap.cps.ncmp.api.impl.config.embeddedcache.ForwardedSubscriptionEventCacheConfig;
 import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence;
+import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus;
+import org.onap.cps.ncmp.api.impl.utils.DataNodeHelper;
 import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent;
 import org.onap.cps.ncmp.api.models.SubscriptionEventResponse;
+import org.onap.cps.spi.model.DataNode;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
@@ -64,28 +69,35 @@ public class SubscriptionEventResponseConsumer {
         log.info("subscription event response of clientId: {} is received.", clientId);
         final String subscriptionName = subscriptionEventResponse.getSubscriptionName();
         final String subscriptionEventId = clientId + subscriptionName;
-        boolean isFullOutcomeResponse = false;
+        boolean createOutcomeResponse = false;
         if (forwardedSubscriptionEventCache.containsKey(subscriptionEventId)) {
             final Set<String> dmiNames = forwardedSubscriptionEventCache.get(subscriptionEventId);
 
             dmiNames.remove(subscriptionEventResponse.getDmiName());
             forwardedSubscriptionEventCache.put(subscriptionEventId, dmiNames,
                     ForwardedSubscriptionEventCacheConfig.SUBSCRIPTION_FORWARD_STARTED_TTL_SECS, TimeUnit.SECONDS);
-            isFullOutcomeResponse = forwardedSubscriptionEventCache.get(subscriptionEventId).isEmpty();
-
-            if (isFullOutcomeResponse) {
-                forwardedSubscriptionEventCache.remove(subscriptionEventId);
-            }
+            createOutcomeResponse = forwardedSubscriptionEventCache.get(subscriptionEventId).isEmpty();
         }
         if (subscriptionModelLoaderEnabled) {
             updateSubscriptionEvent(subscriptionEventResponse);
         }
-        if (isFullOutcomeResponse && notificationFeatureEnabled) {
-            subscriptionEventResponseOutcome.sendResponse(clientId, subscriptionName,
-                    isFullOutcomeResponse);
+        if (createOutcomeResponse
+                && notificationFeatureEnabled
+                && hasNoPendingCmHandles(clientId, subscriptionName)) {
+            subscriptionEventResponseOutcome.sendResponse(clientId, subscriptionName);
+            forwardedSubscriptionEventCache.remove(subscriptionEventId);
         }
     }
 
+    private boolean hasNoPendingCmHandles(final String clientId, final String subscriptionName) {
+        final Collection<DataNode> dataNodeSubscription = subscriptionPersistence.getCmHandlesForSubscriptionEvent(
+                clientId, subscriptionName);
+        final Map<String, SubscriptionStatus> cmHandleIdToStatusMap =
+                DataNodeHelper.getCmHandleIdToStatusMapFromDataNodes(
+                dataNodeSubscription);
+        return !cmHandleIdToStatusMap.values().contains(SubscriptionStatus.PENDING);
+    }
+
     private void updateSubscriptionEvent(final SubscriptionEventResponse subscriptionEventResponse) {
         final YangModelSubscriptionEvent yangModelSubscriptionEvent =
                 subscriptionEventResponseMapper
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseOutcome.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseOutcome.java
index 1bfc4ab28..8fdff1794 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseOutcome.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseOutcome.java
@@ -57,28 +57,32 @@ public class SubscriptionEventResponseOutcome {
      *
      * @param subscriptionClientId client id of the subscription.
      * @param subscriptionName name of the subscription.
-     * @param isFullOutcomeResponse the flag to decide on complete or partial response to be generated.
      */
-    public void sendResponse(final String subscriptionClientId, final String subscriptionName,
-                             final boolean isFullOutcomeResponse) {
+    public void sendResponse(final String subscriptionClientId, final String subscriptionName) {
         final SubscriptionEventOutcome subscriptionEventOutcome = generateResponse(
-                subscriptionClientId, subscriptionName, isFullOutcomeResponse);
+                subscriptionClientId, subscriptionName);
         final Headers headers = new RecordHeaders();
         final String subscriptionEventId = subscriptionClientId + subscriptionName;
         outcomeEventsPublisher.publishEvent(subscriptionOutcomeEventTopic,
                 subscriptionEventId, headers, subscriptionEventOutcome);
     }
 
-    private SubscriptionEventOutcome generateResponse(final String subscriptionClientId, final String subscriptionName,
-                                                                 final boolean isFullOutcomeResponse) {
-        final Collection<DataNode> dataNodes = subscriptionPersistence.getDataNodesForSubscriptionEvent();
+    private SubscriptionEventOutcome generateResponse(final String subscriptionClientId,
+                                                      final String subscriptionName) {
+        final Collection<DataNode> dataNodes =
+                subscriptionPersistence.getCmHandlesForSubscriptionEvent(subscriptionClientId, subscriptionName);
         final List<Map<String, Serializable>> dataNodeLeaves = DataNodeHelper.getDataNodeLeaves(dataNodes);
         final List<Collection<Serializable>> cmHandleIdToStatus =
                 DataNodeHelper.getCmHandleIdToStatus(dataNodeLeaves);
+        final Map<String, SubscriptionStatus> cmHandleIdToStatusMap =
+                DataNodeHelper.getCmHandleIdToStatusMap(cmHandleIdToStatus);
         return formSubscriptionOutcomeMessage(cmHandleIdToStatus, subscriptionClientId, subscriptionName,
-                        isFullOutcomeResponse);
+                isFullOutcomeResponse(cmHandleIdToStatusMap));
     }
 
+    private boolean isFullOutcomeResponse(final Map<String, SubscriptionStatus> cmHandleIdToStatusMap) {
+        return !cmHandleIdToStatusMap.values().contains(SubscriptionStatus.PENDING);
+    }
 
     private SubscriptionEventOutcome formSubscriptionOutcomeMessage(
             final List<Collection<Serializable>> cmHandleIdToStatus, final String subscriptionClientId,
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/DataNodeHelper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/DataNodeHelper.java
index 8d44592ae..f42a378fc 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/DataNodeHelper.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/DataNodeHelper.java
@@ -93,4 +93,15 @@ public class DataNodeHelper {
         }
         return resultMap;
     }
+
+    /**
+     * Extracts the mapping of cm handle id to status from data node collection.
+     *
+     * @param dataNodes as a collection
+     * @return cm handle id to status mapping
+     */
+    public static Map<String, SubscriptionStatus> getCmHandleIdToStatusMapFromDataNodes(
+            final Collection<DataNode> dataNodes) {
+        return getCmHandleIdToStatusMap(getCmHandleIdToStatus(getDataNodeLeaves(dataNodes)));
+    }
 }