Merge "CM SUBSCRIPTION: Remove accepted and rejected cache entries"

5 files changed, 95 insertions, 23 deletions

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpOutEventProducer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpOutEventProducer.java
index 76ee08e64c..e7af69e500 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpOutEventProducer.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpOutEventProducer.java
@@ -34,7 +34,6 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.onap.cps.events.EventsPublisher;
 import org.onap.cps.ncmp.api.impl.events.cmsubscription.mapper.CmNotificationSubscriptionNcmpOutEventMapper;
-import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.DmiCmNotificationSubscriptionDetails;
 import org.onap.cps.ncmp.events.cmsubscription_merge1_0_0.ncmp_to_client.CmNotificationSubscriptionNcmpOutEvent;
 import org.onap.cps.utils.JsonObjectMapper;
 import org.springframework.beans.factory.annotation.Value;
@@ -55,8 +54,8 @@ public class CmNotificationSubscriptionNcmpOutEventProducer {
 
     private final EventsPublisher<CloudEvent> eventsPublisher;
     private final JsonObjectMapper jsonObjectMapper;
-    private final Map<String, Map<String, DmiCmNotificationSubscriptionDetails>> cmNotificationSubscriptionCache;
     private final CmNotificationSubscriptionNcmpOutEventMapper cmNotificationSubscriptionNcmpOutEventMapper;
+    private final DmiCmNotificationSubscriptionCacheHandler dmiCmNotificationSubscriptionCacheHandler;
     private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
     private static final Map<String, ScheduledFuture<?>> scheduledTasksPerSubscriptionId = new ConcurrentHashMap<>();
 
@@ -72,8 +71,9 @@ public class CmNotificationSubscriptionNcmpOutEventProducer {
      *                                               or published now
      */
     public void publishCmNotificationSubscriptionNcmpOutEvent(final String subscriptionId, final String eventType,
-            final CmNotificationSubscriptionNcmpOutEvent cmNotificationSubscriptionNcmpOutEvent,
-            final boolean isScheduledEvent) {
+                                                              final CmNotificationSubscriptionNcmpOutEvent
+                                                                      cmNotificationSubscriptionNcmpOutEvent,
+                                                              final boolean isScheduledEvent) {
 
         if (isScheduledEvent && !scheduledTasksPerSubscriptionId.containsKey(subscriptionId)) {
             final ScheduledFuture<?> scheduledFuture =
@@ -86,16 +86,15 @@ public class CmNotificationSubscriptionNcmpOutEventProducer {
                     cmNotificationSubscriptionNcmpOutEvent);
             log.info("Published CmNotificationSubscriptionEvent on demand for subscriptionId : {}", subscriptionId);
         }
-
     }
 
     private ScheduledFuture<?> scheduleAndPublishCmNotificationSubscriptionNcmpOutEvent(final String subscriptionId,
-            final String eventType) {
+                                                                                        final String eventType) {
         final CmNotificationSubscriptionNcmpOutEventPublishingTask
                 cmNotificationSubscriptionNcmpOutEventPublishingTask =
                 new CmNotificationSubscriptionNcmpOutEventPublishingTask(cmNotificationSubscriptionNcmpOutEventTopic,
-                        subscriptionId, eventType, eventsPublisher, jsonObjectMapper, cmNotificationSubscriptionCache,
-                        cmNotificationSubscriptionNcmpOutEventMapper);
+                        subscriptionId, eventType, eventsPublisher, jsonObjectMapper,
+                        cmNotificationSubscriptionNcmpOutEventMapper, dmiCmNotificationSubscriptionCacheHandler);
         return scheduledExecutorService.schedule(cmNotificationSubscriptionNcmpOutEventPublishingTask,
                 cmNotificationSubscriptionDmiOutEventTimeoutInMs, TimeUnit.MILLISECONDS);
     }
@@ -112,12 +111,15 @@ public class CmNotificationSubscriptionNcmpOutEventProducer {
 
 
     private void publishCmNotificationSubscriptionNcmpOutEventNow(final String subscriptionId, final String eventType,
-            final CmNotificationSubscriptionNcmpOutEvent cmNotificationSubscriptionNcmpOutEvent) {
+                                                                  final CmNotificationSubscriptionNcmpOutEvent
+                                                                          cmNotificationSubscriptionNcmpOutEvent) {
         final CloudEvent cmNotificationSubscriptionNcmpOutEventAsCloudEvent =
                 buildAndGetCmNotificationNcmpOutEventAsCloudEvent(jsonObjectMapper, subscriptionId, eventType,
                         cmNotificationSubscriptionNcmpOutEvent);
         eventsPublisher.publishCloudEvent(cmNotificationSubscriptionNcmpOutEventTopic, subscriptionId,
                 cmNotificationSubscriptionNcmpOutEventAsCloudEvent);
+        dmiCmNotificationSubscriptionCacheHandler
+                .removeAcceptedAndRejectedDmiCmNotificationSubscriptionEntries(subscriptionId);
     }
 
     protected static CloudEvent buildAndGetCmNotificationNcmpOutEventAsCloudEvent(
@@ -125,9 +127,9 @@ public class CmNotificationSubscriptionNcmpOutEventProducer {
             final CmNotificationSubscriptionNcmpOutEvent cmNotificationSubscriptionNcmpOutEvent) {
 
         return CloudEventBuilder.v1().withId(UUID.randomUUID().toString()).withType(eventType)
-                       .withSource(URI.create("NCMP")).withDataSchema(URI.create("org.onap.ncmp.cm.subscription:1.0.0"))
-                       .withExtension("correlationid", subscriptionId)
-                       .withData(jsonObjectMapper.asJsonBytes(cmNotificationSubscriptionNcmpOutEvent)).build();
+                .withSource(URI.create("NCMP")).withDataSchema(URI.create("org.onap.ncmp.cm.subscription:1.0.0"))
+                .withExtension("correlationid", subscriptionId)
+                .withData(jsonObjectMapper.asJsonBytes(cmNotificationSubscriptionNcmpOutEvent)).build();
     }
 
 }
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpOutEventPublishingTask.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpOutEventPublishingTask.java
index f7ea4a465c..b2c0a5cfb3 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpOutEventPublishingTask.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpOutEventPublishingTask.java
@@ -42,8 +42,8 @@ public class CmNotificationSubscriptionNcmpOutEventPublishingTask implements Run
     private final String eventType;
     private final EventsPublisher<CloudEvent> eventsPublisher;
     private final JsonObjectMapper jsonObjectMapper;
-    private final Map<String, Map<String, DmiCmNotificationSubscriptionDetails>> cmNotificationSubscriptionCache;
     private final CmNotificationSubscriptionNcmpOutEventMapper cmNotificationSubscriptionNcmpOutEventMapper;
+    private final DmiCmNotificationSubscriptionCacheHandler dmiCmNotificationSubscriptionCacheHandler;
 
     /**
      * Delegating the responsibility of publishing CmNotificationSubscriptionNcmpOutEvent as a separate task which will
@@ -52,13 +52,15 @@ public class CmNotificationSubscriptionNcmpOutEventPublishingTask implements Run
     @Override
     public void run() {
         final Map<String, DmiCmNotificationSubscriptionDetails> dmiCmNotificationSubscriptionDetailsMap =
-                cmNotificationSubscriptionCache.get(subscriptionId);
+                dmiCmNotificationSubscriptionCacheHandler.get(subscriptionId);
         final CmNotificationSubscriptionNcmpOutEvent cmNotificationSubscriptionNcmpOutEvent =
                 cmNotificationSubscriptionNcmpOutEventMapper.toCmNotificationSubscriptionNcmpOutEvent(subscriptionId,
                         dmiCmNotificationSubscriptionDetailsMap);
         eventsPublisher.publishCloudEvent(topicName, subscriptionId,
                 buildAndGetCmNotificationNcmpOutEventAsCloudEvent(jsonObjectMapper, subscriptionId, eventType,
                         cmNotificationSubscriptionNcmpOutEvent));
+        dmiCmNotificationSubscriptionCacheHandler
+                .removeAcceptedAndRejectedDmiCmNotificationSubscriptionEntries(subscriptionId);
     }
 
 }
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/DmiCmNotificationSubscriptionCacheHandler.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/DmiCmNotificationSubscriptionCacheHandler.java
index 4b3a085147..34ffb5e19d 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/DmiCmNotificationSubscriptionCacheHandler.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/DmiCmNotificationSubscriptionCacheHandler.java
@@ -29,6 +29,7 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.CmNotificationSubscriptionStatus;
 import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.DmiCmNotificationSubscriptionDetails;
@@ -51,7 +52,7 @@ public class DmiCmNotificationSubscriptionCacheHandler {
     /**
      * Adds new subscription to the subscription cache.
      *
-     * @param subscriptionId    subscription Id
+     * @param subscriptionId    subscription id
      * @param predicates        subscription request predicates
      */
     public void add(final String subscriptionId, final List<Predicate> predicates) {
@@ -59,6 +60,33 @@ public class DmiCmNotificationSubscriptionCacheHandler {
     }
 
     /**
+     * Get cm notification subscription cache entry via subscription id.
+     *
+     * @param subscriptionId    subscription id
+     * @return map of dmi cm notification subscriptions per dmi
+     */
+    public Map<String, DmiCmNotificationSubscriptionDetails> get(final String subscriptionId) {
+        return cmNotificationSubscriptionCache.get(subscriptionId);
+    }
+
+
+    /**
+     * Remove cache entries with CmNotificationSubscriptionStatus ACCEPTED/REJECTED via subscription id.
+     *
+     * @param subscriptionId subscription id as key in CM notification Subscription cache.
+     */
+    public void removeAcceptedAndRejectedDmiCmNotificationSubscriptionEntries(final String subscriptionId) {
+        final Map<String, DmiCmNotificationSubscriptionDetails> dmiCmNotificationSubscriptionsPerDmi =
+                cmNotificationSubscriptionCache.get(subscriptionId);
+        final Map<String, DmiCmNotificationSubscriptionDetails> updatedDmiCmNotificationSubscriptionsPerDmi =
+                dmiCmNotificationSubscriptionsPerDmi.entrySet().stream().filter(
+                                dmiCmNotificationSubscription ->
+                                        !isAcceptedOrRejected(dmiCmNotificationSubscription.getValue()))
+                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        cmNotificationSubscriptionCache.put(subscriptionId, updatedDmiCmNotificationSubscriptionsPerDmi);
+    }
+
+    /**
      *  Creates map of subscription details per DMI.
      *
      * @param predicates    CM Subscription Create Request Predicates
@@ -95,9 +123,9 @@ public class DmiCmNotificationSubscriptionCacheHandler {
      *
      */
     public void updateDmiCmNotificationSubscriptionStatusPerDmi(
-        final String subscriptionId, final String dmiServiceName, final CmNotificationSubscriptionStatus status) {
+            final String subscriptionId, final String dmiServiceName, final CmNotificationSubscriptionStatus status) {
         cmNotificationSubscriptionCache.get(subscriptionId).get(dmiServiceName)
-            .setCmNotificationSubscriptionStatus(status);
+                .setCmNotificationSubscriptionStatus(status);
     }
 
     /**
@@ -109,10 +137,10 @@ public class DmiCmNotificationSubscriptionCacheHandler {
      */
     public void persistIntoDatabasePerDmi(final String subscriptionId, final String dmiServiceName) {
         final List<DmiCmNotificationSubscriptionPredicate> dmiCmNotificationSubscriptionPredicateList =
-            cmNotificationSubscriptionCache.get(subscriptionId).get(dmiServiceName)
-            .getDmiCmNotificationSubscriptionPredicates();
+                cmNotificationSubscriptionCache.get(subscriptionId).get(dmiServiceName)
+                        .getDmiCmNotificationSubscriptionPredicates();
         for (final DmiCmNotificationSubscriptionPredicate dmiCmNotificationSubscriptionPredicate:
-            dmiCmNotificationSubscriptionPredicateList) {
+                dmiCmNotificationSubscriptionPredicateList) {
             final DatastoreType datastoreType = dmiCmNotificationSubscriptionPredicate.getDatastoreType();
             final Set<String> cmHandles = dmiCmNotificationSubscriptionPredicate.getTargetCmHandleIds();
             final Set<String> xpaths = dmiCmNotificationSubscriptionPredicate.getXpaths();
@@ -153,4 +181,10 @@ public class DmiCmNotificationSubscriptionCacheHandler {
         }
         return targetCmHandlesByDmiServiceNames;
     }
+
+    private boolean isAcceptedOrRejected(
+            final DmiCmNotificationSubscriptionDetails dmiCmNotificationSubscription) {
+        return dmiCmNotificationSubscription.getCmNotificationSubscriptionStatus().toString().equals("ACCEPTED")
+                || dmiCmNotificationSubscription.getCmNotificationSubscriptionStatus().toString().equals("REJECTED");
+    }
 }
\ No newline at end of file
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpOutEventProducerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpOutEventProducerSpec.groovy
index 970d7e67b0..0f1bdc65b7 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpOutEventProducerSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmNotificationSubscriptionNcmpOutEventProducerSpec.groovy
@@ -4,7 +4,6 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import io.cloudevents.CloudEvent
 import org.onap.cps.events.EventsPublisher
 import org.onap.cps.ncmp.api.impl.events.cmsubscription.mapper.CmNotificationSubscriptionNcmpOutEventMapper
-import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.DmiCmNotificationSubscriptionDetails
 import org.onap.cps.ncmp.api.impl.events.mapper.CloudEventMapper
 import org.onap.cps.ncmp.events.cmsubscription_merge1_0_0.ncmp_to_client.CmNotificationSubscriptionNcmpOutEvent
 import org.onap.cps.ncmp.events.cmsubscription_merge1_0_0.ncmp_to_client.Data
@@ -15,10 +14,11 @@ class CmNotificationSubscriptionNcmpOutEventProducerSpec extends Specification {
 
     def mockEventsPublisher = Mock(EventsPublisher)
     def jsonObjectMapper = new JsonObjectMapper(new ObjectMapper())
-    def mockCmNotificationSubscriptionCache = Mock(Map<String, Map<String, DmiCmNotificationSubscriptionDetails>>)
     def mockCmNotificationSubscriptionNcmpOutEventMapper = Mock(CmNotificationSubscriptionNcmpOutEventMapper)
+    def mockDmiCmNotificationSubscriptionCacheHandler = Mock(DmiCmNotificationSubscriptionCacheHandler)
 
-    def objectUnderTest = new CmNotificationSubscriptionNcmpOutEventProducer(mockEventsPublisher, jsonObjectMapper, mockCmNotificationSubscriptionCache, mockCmNotificationSubscriptionNcmpOutEventMapper)
+    def objectUnderTest = new CmNotificationSubscriptionNcmpOutEventProducer(mockEventsPublisher, jsonObjectMapper,
+        mockCmNotificationSubscriptionNcmpOutEventMapper, mockDmiCmNotificationSubscriptionCacheHandler)
 
     def 'Create and #scenario Cm Notification Subscription NCMP out event'() {
         given: 'a cm subscription response for the client'
@@ -80,6 +80,8 @@ class CmNotificationSubscriptionNcmpOutEventProducerSpec extends Specification {
                         assert CloudEventMapper.toTargetEvent(cmNotificationSubscriptionNcmpOutEventAsCloudEvent, CmNotificationSubscriptionNcmpOutEvent) == cmNotificationSubscriptionNcmpOutEvent
                     }
             }
+        then: 'the cache handler is called once to remove accepted and rejected entries in cache'
+            1 * mockDmiCmNotificationSubscriptionCacheHandler.removeAcceptedAndRejectedDmiCmNotificationSubscriptionEntries(subscriptionId)
     }
 
 
diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/DmiCmNotificationSubscriptionCacheHandlerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/DmiCmNotificationSubscriptionCacheHandlerSpec.groovy
index 10e060fee6..43568be501 100644
--- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/DmiCmNotificationSubscriptionCacheHandlerSpec.groovy
+++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/cmsubscription/DmiCmNotificationSubscriptionCacheHandlerSpec.groovy
@@ -25,6 +25,7 @@ import io.cloudevents.CloudEvent
 import io.cloudevents.core.builder.CloudEventBuilder
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.CmNotificationSubscriptionStatus
+import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.DmiCmNotificationSubscriptionDetails
 import org.onap.cps.ncmp.api.impl.events.cmsubscription.service.CmNotificationSubscriptionPersistenceService
 import org.onap.cps.ncmp.api.impl.inventory.InventoryPersistence
 import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle
@@ -75,6 +76,37 @@ class DmiCmNotificationSubscriptionCacheHandlerSpec extends MessagingBaseSpec {
             assert testCache.containsKey(subscriptionId)
     }
 
+    def 'Get cache entry via subscription id'() {
+        given: 'the cache contains value for some-id'
+            testCache.put('some-id',[:])
+        when: 'the get method is called'
+            def result = objectUnderTest.get('some-id')
+        then: 'correct value is returned as expected'
+            assert result == [:]
+    }
+
+    def 'Remove accepted and rejected entries from cache via subscription id'() {
+        given: 'a map as the value for cache entry for some-id'
+            def testMap = [:]
+            testMap.put("dmi-1",
+                new DmiCmNotificationSubscriptionDetails([],CmNotificationSubscriptionStatus.ACCEPTED))
+            testMap.put("dmi-2",
+                new DmiCmNotificationSubscriptionDetails([],CmNotificationSubscriptionStatus.REJECTED))
+            testMap.put("dmi-3",
+                new DmiCmNotificationSubscriptionDetails([],CmNotificationSubscriptionStatus.PENDING))
+            testCache.put("test-id", testMap)
+            assert testCache.get("test-id").size() == 3
+        when: 'the method to remove accepted and rejected entries for test-id is called'
+            objectUnderTest.removeAcceptedAndRejectedDmiCmNotificationSubscriptionEntries("test-id")
+        then: 'all entries with status accepted/rejected are no longer present for test-id'
+            testCache.get("test-id").each { key, testResultMap ->
+                assert testResultMap.cmNotificationSubscriptionStatus != CmNotificationSubscriptionStatus.ACCEPTED
+                    || testResultMap.cmNotificationSubscriptionStatus != CmNotificationSubscriptionStatus.REJECTED
+            }
+        and: 'the size of the map for cache entry test-id is as expected'
+            assert testCache.get("test-id").size() == 1
+    }
+
     def 'Create map for DMI cm notification subscription per DMI service name'() {
         given: 'list of predicates from the create subscription event'
             def predicates = cmNotificationSubscriptionNcmpInEvent.getData().getPredicates()