Merge "Handle null NcmpOut event"

2 files changed, 21 insertions, 16 deletions

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/CmSubscriptionHandlerImpl.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/CmSubscriptionHandlerImpl.java
index c2c71dbaae..1cdc7ed3e0 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/CmSubscriptionHandlerImpl.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/CmSubscriptionHandlerImpl.java
@@ -154,7 +154,7 @@ public class CmSubscriptionHandlerImpl implements CmSubscriptionHandler {
                             dmiSubscriptionDetails.getDmiCmSubscriptionPredicates());
 
             if (dmiCmSubscriptionPredicates.isEmpty()) {
-                acceptAndPublishNcmpOutEventPerDmi(subscriptionId, dmiPluginName);
+                acceptAndPersistCmSubscriptionPerDmi(subscriptionId, dmiPluginName);
             } else {
                 publishDmiInEventPerDmi(subscriptionId, dmiPluginName, dmiCmSubscriptionPredicates);
             }
@@ -168,7 +168,7 @@ public class CmSubscriptionHandlerImpl implements CmSubscriptionHandler {
                 "subscriptionCreateRequest", dmiInEvent);
     }
 
-    private void acceptAndPublishNcmpOutEventPerDmi(final String subscriptionId, final String dmiPluginName) {
+    private void acceptAndPersistCmSubscriptionPerDmi(final String subscriptionId, final String dmiPluginName) {
         dmiCacheHandler.updateDmiSubscriptionStatus(subscriptionId, dmiPluginName,
                 CmSubscriptionStatus.ACCEPTED);
         dmiCacheHandler.persistIntoDatabasePerDmi(subscriptionId, dmiPluginName);
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/NcmpOutEventProducer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/NcmpOutEventProducer.java
index 01d720937f..3371d59f7a 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/NcmpOutEventProducer.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/NcmpOutEventProducer.java
@@ -57,7 +57,8 @@ public class NcmpOutEventProducer {
     private final NcmpOutEventMapper ncmpOutEventMapper;
     private final DmiCacheHandler dmiCacheHandler;
     private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
-    private static final Map<String, ScheduledFuture<?>> scheduledTasksPerSubscriptionId = new ConcurrentHashMap<>();
+    private static final Map<String, ScheduledFuture<?>> scheduledTasksPerSubscriptionIdAndEventType =
+            new ConcurrentHashMap<>();
 
     /**
      * Publish the event to the client who requested the subscription with key as subscription id and event is Cloud
@@ -73,14 +74,20 @@ public class NcmpOutEventProducer {
     public void publishNcmpOutEvent(final String subscriptionId, final String eventType,
             final NcmpOutEvent ncmpOutEvent, final boolean isScheduledEvent) {
 
-        if (isScheduledEvent && !scheduledTasksPerSubscriptionId.containsKey(subscriptionId)) {
+        final String taskKey = subscriptionId.concat(eventType);
+
+        if (isScheduledEvent && !scheduledTasksPerSubscriptionIdAndEventType.containsKey(taskKey)) {
             final ScheduledFuture<?> scheduledFuture = scheduleAndPublishNcmpOutEvent(subscriptionId, eventType);
-            scheduledTasksPerSubscriptionId.putIfAbsent(subscriptionId, scheduledFuture);
-            log.debug("Scheduled the CmNotificationSubscriptionEvent for subscriptionId : {}", subscriptionId);
+            scheduledTasksPerSubscriptionIdAndEventType.putIfAbsent(taskKey, scheduledFuture);
+            log.debug("Scheduled the Cm Subscription Event for subscriptionId : {} and eventType : {}", subscriptionId,
+                    eventType);
         } else {
-            cancelScheduledTaskForSubscriptionId(subscriptionId);
-            publishNcmpOutEventNow(subscriptionId, eventType, ncmpOutEvent);
-            log.info("Published CmNotificationSubscriptionEvent on demand for subscriptionId : {}", subscriptionId);
+            cancelScheduledTask(taskKey);
+            if (ncmpOutEvent != null) {
+                publishNcmpOutEventNow(subscriptionId, eventType, ncmpOutEvent);
+                log.debug("Published Cm Subscription Event on demand for subscriptionId : {} and eventType : {}",
+                        subscriptionId, eventType);
+            }
         }
     }
 
@@ -92,12 +99,12 @@ public class NcmpOutEventProducer {
                 TimeUnit.MILLISECONDS);
     }
 
-    private void cancelScheduledTaskForSubscriptionId(final String subscriptionId) {
+    private void cancelScheduledTask(final String taskKey) {
 
-        final ScheduledFuture<?> scheduledFuture = scheduledTasksPerSubscriptionId.get(subscriptionId);
+        final ScheduledFuture<?> scheduledFuture = scheduledTasksPerSubscriptionIdAndEventType.get(taskKey);
         if (scheduledFuture != null) {
             scheduledFuture.cancel(true);
-            scheduledTasksPerSubscriptionId.remove(subscriptionId);
+            scheduledTasksPerSubscriptionIdAndEventType.remove(taskKey);
         }
 
     }
@@ -106,10 +113,8 @@ public class NcmpOutEventProducer {
     private void publishNcmpOutEventNow(final String subscriptionId, final String eventType,
             final NcmpOutEvent ncmpOutEvent) {
         final CloudEvent ncmpOutEventAsCloudEvent =
-                buildAndGetNcmpOutEventAsCloudEvent(jsonObjectMapper, subscriptionId, eventType,
-                        ncmpOutEvent);
-        eventsPublisher.publishCloudEvent(ncmpOutEventTopic, subscriptionId,
-                ncmpOutEventAsCloudEvent);
+                buildAndGetNcmpOutEventAsCloudEvent(jsonObjectMapper, subscriptionId, eventType, ncmpOutEvent);
+        eventsPublisher.publishCloudEvent(ncmpOutEventTopic, subscriptionId, ncmpOutEventAsCloudEvent);
         dmiCacheHandler.removeAcceptedAndRejectedDmiSubscriptionEntries(subscriptionId);
     }