1 files changed, 41 insertions, 31 deletions

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 1d87a057a..1fe963a27 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,13 +21,12 @@
 package org.onap.cps.ncmp.api.impl.events.avcsubscription;
 
 import com.hazelcast.map.IMap;
+import io.cloudevents.CloudEvent;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -35,16 +34,17 @@ import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 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.utils.SubscriptionEventCloudMapper;
 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.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.CmHandle;
 import org.onap.cps.spi.exceptions.OperationNotYetSupportedException;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
@@ -56,10 +56,11 @@ import org.springframework.stereotype.Component;
 public class SubscriptionEventForwarder {
 
     private final InventoryPersistence inventoryPersistence;
-    private final EventsPublisher<SubscriptionEvent> eventsPublisher;
+    private final EventsPublisher<CloudEvent> eventsPublisher;
     private final IMap<String, Set<String>> forwardedSubscriptionEventCache;
     private final SubscriptionEventResponseOutcome subscriptionEventResponseOutcome;
     private final SubscriptionEventMapper subscriptionEventMapper;
+    private final ClientSubscriptionEventMapper clientSubscriptionEventMapper;
     private final SubscriptionPersistence subscriptionPersistence;
     private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
     @Value("${app.ncmp.avc.subscription-forward-topic-prefix}")
@@ -73,28 +74,22 @@ public class SubscriptionEventForwarder {
      *
      * @param subscriptionEvent the event to be forwarded
      */
-    public void forwardCreateSubscriptionEvent(final SubscriptionEvent subscriptionEvent,
-                                               final Headers eventHeaders) {
-        final List<Object> cmHandleTargets = subscriptionEvent.getEvent().getPredicates().getTargets();
+    public void forwardCreateSubscriptionEvent(final SubscriptionEvent subscriptionEvent) {
+        final List<String> cmHandleTargets = subscriptionEvent.getData().getPredicates().getTargets();
         if (cmHandleTargets == null || cmHandleTargets.isEmpty()
-                || cmHandleTargets.stream().anyMatch(id -> ((String) id).contains("*"))) {
+                || cmHandleTargets.stream().anyMatch(id -> (id).contains("*"))) {
             throw new OperationNotYetSupportedException(
                     "CMHandle targets are required. \"Wildcard\" operations are not yet supported");
         }
-        final List<String> cmHandleTargetsAsStrings = cmHandleTargets.stream().map(
-                Objects::toString).collect(Collectors.toList());
         final Collection<YangModelCmHandle> yangModelCmHandles =
-                inventoryPersistence.getYangModelCmHandles(cmHandleTargetsAsStrings);
-
+                inventoryPersistence.getYangModelCmHandles(cmHandleTargets);
         final Map<String, Map<String, Map<String, String>>> dmiPropertiesPerCmHandleIdPerServiceName
                 = DmiServiceNameOrganizer.getDmiPropertiesPerCmHandleIdPerServiceName(yangModelCmHandles);
-
-        findDmisAndRespond(subscriptionEvent, eventHeaders, cmHandleTargetsAsStrings,
-                dmiPropertiesPerCmHandleIdPerServiceName);
+        findDmisAndRespond(subscriptionEvent, cmHandleTargets, dmiPropertiesPerCmHandleIdPerServiceName);
     }
 
-    private void findDmisAndRespond(final SubscriptionEvent subscriptionEvent, final Headers eventHeaders,
-                           final List<String> cmHandleTargetsAsStrings,
+    private void findDmisAndRespond(final SubscriptionEvent subscriptionEvent,
+                                    final List<String> cmHandleTargetsAsStrings,
                            final Map<String, Map<String, Map<String, String>>>
                                             dmiPropertiesPerCmHandleIdPerServiceName) {
         final List<String> cmHandlesThatExistsInDb = dmiPropertiesPerCmHandleIdPerServiceName.entrySet().stream()
@@ -109,18 +104,20 @@ public class SubscriptionEventForwarder {
             updatesCmHandlesToRejectedAndPersistSubscriptionEvent(subscriptionEvent, targetCmHandlesDoesNotExistInDb);
         }
         if (dmisToRespond.isEmpty()) {
-            final String clientID = subscriptionEvent.getEvent().getSubscription().getClientID();
-            final String subscriptionName = subscriptionEvent.getEvent().getSubscription().getName();
+            final String clientID = subscriptionEvent.getData().getSubscription().getClientID();
+            final String subscriptionName = subscriptionEvent.getData().getSubscription().getName();
             subscriptionEventResponseOutcome.sendResponse(clientID, subscriptionName);
         } else {
             startResponseTimeout(subscriptionEvent, dmisToRespond);
-            forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, subscriptionEvent, eventHeaders);
+            final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent ncmpSubscriptionEvent =
+                    clientSubscriptionEventMapper.toNcmpSubscriptionEvent(subscriptionEvent);
+            forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, ncmpSubscriptionEvent);
         }
     }
 
     private void startResponseTimeout(final SubscriptionEvent subscriptionEvent, final Set<String> dmisToRespond) {
-        final String subscriptionClientId = subscriptionEvent.getEvent().getSubscription().getClientID();
-        final String subscriptionName = subscriptionEvent.getEvent().getSubscription().getName();
+        final String subscriptionClientId = subscriptionEvent.getData().getSubscription().getClientID();
+        final String subscriptionName = subscriptionEvent.getData().getSubscription().getName();
         final String subscriptionEventId = subscriptionClientId + subscriptionName;
 
         forwardedSubscriptionEventCache.put(subscriptionEventId, dmisToRespond,
@@ -137,20 +134,33 @@ public class SubscriptionEventForwarder {
     }
 
     private void forwardEventToDmis(final Map<String, Map<String, Map<String, String>>> dmiNameCmHandleMap,
-                                    final SubscriptionEvent subscriptionEvent,
-                                    final Headers eventHeaders) {
+                                    final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent
+                                            ncmpSubscriptionEvent) {
         dmiNameCmHandleMap.forEach((dmiName, cmHandlePropertiesMap) -> {
-            subscriptionEvent.getEvent().getPredicates().setTargets(Collections.singletonList(cmHandlePropertiesMap));
-            final String eventKey = createEventKey(subscriptionEvent, dmiName);
+            final List<CmHandle> cmHandleTargets = cmHandlePropertiesMap.entrySet().stream().map(
+                    cmHandleAndProperties -> {
+                        final CmHandle cmHandle = new CmHandle();
+                        cmHandle.setId(cmHandleAndProperties.getKey());
+                        cmHandle.setAdditionalProperties(cmHandleAndProperties.getValue());
+                        return cmHandle;
+                    }).collect(Collectors.toList());
+
+            ncmpSubscriptionEvent.getData().getPredicates().setTargets(cmHandleTargets);
+            final String eventKey = createEventKey(ncmpSubscriptionEvent, dmiName);
             final String dmiAvcSubscriptionTopic = dmiAvcSubscriptionTopicPrefix + dmiName;
-            eventsPublisher.publishEvent(dmiAvcSubscriptionTopic, eventKey, eventHeaders, subscriptionEvent);
+
+            final CloudEvent ncmpSubscriptionCloudEvent =
+                    SubscriptionEventCloudMapper.toCloudEvent(ncmpSubscriptionEvent, eventKey);
+            eventsPublisher.publishCloudEvent(dmiAvcSubscriptionTopic, eventKey, ncmpSubscriptionCloudEvent);
         });
     }
 
-    private String createEventKey(final SubscriptionEvent subscriptionEvent, final String dmiName) {
-        return subscriptionEvent.getEvent().getSubscription().getClientID()
+    private String createEventKey(
+            final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent subscriptionEvent,
+            final String dmiName) {
+        return subscriptionEvent.getData().getSubscription().getClientID()
             + "-"
-            + subscriptionEvent.getEvent().getSubscription().getName()
+            + subscriptionEvent.getData().getSubscription().getName()
             + "-"
             + dmiName;
     }