Refactoring persistence classes

[cps.git] / cps-ncmp-service / src / main / java / org / onap / cps / ncmp / api / impl / events / cmsubscription / CmSubscriptionDmiOutEventConsumer.java

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmSubscriptionDmiOutEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmSubscriptionDmiOutEventConsumer.java
index 1ac4044..d2b596e 100644 (file)
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmSubscriptionDmiOutEventConsumer.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/cmsubscription/CmSubscriptionDmiOutEventConsumer.java
@@ -24,8 +24,6 @@ import static org.onap.cps.ncmp.api.impl.events.mapper.CloudEventMapper.toTarget
 
 import com.hazelcast.map.IMap;
 import io.cloudevents.CloudEvent;
-import java.util.Collection;
-import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import lombok.RequiredArgsConstructor;
@@ -33,11 +31,9 @@ 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.CmSubscriptionEvent;
 import org.onap.cps.ncmp.events.cmsubscription1_0_0.dmi_to_ncmp.CmSubscriptionDmiOutEvent;
-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;
@@ -66,7 +62,7 @@ public class CmSubscriptionDmiOutEventConsumer {
      */
     @KafkaListener(topics = "${app.ncmp.avc.subscription-response-topic}",
             containerFactory = "cloudEventConcurrentKafkaListenerContainerFactory")
-    public void consumeSubscriptionEventResponse(
+    public void consumeDmiOutEvent(
             final ConsumerRecord<String, CloudEvent> cmSubscriptionDmiOutConsumerRecord) {
         final CloudEvent cloudEvent = cmSubscriptionDmiOutConsumerRecord.value();
         final String eventType = cmSubscriptionDmiOutConsumerRecord.value().getType();
@@ -76,7 +72,7 @@ public class CmSubscriptionDmiOutEventConsumer {
         log.info("subscription event response of clientId: {} is received.", clientId);
         final String subscriptionName = cmSubscriptionDmiOutEvent.getData().getSubscriptionName();
         final String subscriptionEventId = clientId + subscriptionName;
-        boolean createOutcomeResponse = false;
+        boolean createOutcomeResponse = true;
         if (forwardedSubscriptionEventCache.containsKey(subscriptionEventId)) {
             final Set<String> dmiNames = forwardedSubscriptionEventCache.get(subscriptionEventId);
             dmiNames.remove(cmSubscriptionDmiOutEvent.getData().getDmiName());
@@ -88,25 +84,15 @@ public class CmSubscriptionDmiOutEventConsumer {
             updateSubscriptionEvent(cmSubscriptionDmiOutEvent);
         }
         if (createOutcomeResponse
-                && notificationFeatureEnabled
-                && hasNoPendingCmHandles(clientId, subscriptionName)) {
-            cmSubscriptionNcmpOutEventPublisher.sendResponse(cmSubscriptionDmiOutEvent, eventType);
-            forwardedSubscriptionEventCache.remove(subscriptionEventId);
-        }
-    }
+                && notificationFeatureEnabled) {
 
-    private boolean hasNoPendingCmHandles(final String clientId, final String subscriptionName) {
-        final Collection<DataNode> dataNodeSubscription = subscriptionPersistence.getCmHandlesForSubscriptionEvent(
-                clientId, subscriptionName);
-        final Map<String, Map<String, String>> cmHandleIdToStatusAndDetailsAsMapOriginal =
-                DataNodeHelper.cmHandleIdToStatusAndDetailsAsMapFromDataNode(dataNodeSubscription);
-        for (final Map<String, String> statusAndDetailsMap : cmHandleIdToStatusAndDetailsAsMapOriginal.values()) {
-            final String status = statusAndDetailsMap.get("status");
-            if (SubscriptionStatus.PENDING.toString().equals(status)) {
-                return false;
-            }
+            final CmSubscriptionEvent cmSubscriptionEvent = new CmSubscriptionEvent();
+            cmSubscriptionEvent.setClientId(cmSubscriptionDmiOutEvent.getData().getClientId());
+            cmSubscriptionEvent.setSubscriptionName(cmSubscriptionDmiOutEvent.getData().getSubscriptionName());
+
+            cmSubscriptionNcmpOutEventPublisher.sendResponse(cmSubscriptionEvent, eventType);
+            forwardedSubscriptionEventCache.remove(subscriptionEventId);
         }
-        return true;
     }
 
     private void updateSubscriptionEvent(final CmSubscriptionDmiOutEvent cmSubscriptionDmiOutEvent) {