Subscription Creation: NCMP to DMI CloudEvent transformation

- Add mapper to convert client event into ncmp event
- Add sample json object of ncmp version
- Change subscription event consumer to consume CloudEvents
- Change subscription event forwarder to publish CloudEvents
- Change test producer config to support CloudEvents
- Change sample subscription event json to comply with new schema
- Add more test for missing branches
- Change packages of the mappers into relevant directory

Issue-ID: CPS-1737
Change-Id: I8c9e6e7bf713a8fb530a0586dfb2bce796a462f5
Signed-off-by: halil.cakal <halil.cakal@est.tech>

5 files changed, 174 insertions, 54 deletions

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ClientSubscriptionEventMapper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ClientSubscriptionEventMapper.java
new file mode 100644
index 000000000..59b1d09c7
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/ClientSubscriptionEventMapper.java
@@ -0,0 +1,34 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2023 Nordix Foundation
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.events.avcsubscription;
+
+import org.mapstruct.Mapper;
+import org.mapstruct.Mapping;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent;
+
+@Mapper(componentModel = "spring")
+public interface ClientSubscriptionEventMapper {
+
+    @Mapping(target = "data.predicates.targets", ignore = true)
+    org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent toNcmpSubscriptionEvent(
+            SubscriptionEvent subscriptionEvent);
+
+}
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java
index 88b41d007..f511965c7 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventConsumer.java
@@ -20,13 +20,14 @@
 
 package org.onap.cps.ncmp.api.impl.events.avcsubscription;
 
+import io.cloudevents.CloudEvent;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionPersistence;
+import org.onap.cps.ncmp.api.impl.utils.SubscriptionEventCloudMapper;
 import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent;
-import org.onap.cps.ncmp.event.model.InnerSubscriptionEvent;
-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.spi.exceptions.OperationNotYetSupportedException;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.annotation.KafkaListener;
@@ -54,28 +55,25 @@ public class SubscriptionEventConsumer {
      * @param subscriptionEventConsumerRecord the event to be consumed
      */
     @KafkaListener(topics = "${app.ncmp.avc.subscription-topic}",
-            properties = {"spring.json.value.default.type=org.onap.cps.ncmp.event.model.SubscriptionEvent"})
-    public void consumeSubscriptionEvent(
-            final ConsumerRecord<String, SubscriptionEvent> subscriptionEventConsumerRecord) {
-        final SubscriptionEvent subscriptionEvent = subscriptionEventConsumerRecord.value();
-        final InnerSubscriptionEvent event = subscriptionEvent.getEvent();
-        final String eventDatastore = event.getPredicates().getDatastore();
+            containerFactory = "cloudEventConcurrentKafkaListenerContainerFactory")
+    public void consumeSubscriptionEvent(final ConsumerRecord<String, CloudEvent> subscriptionEventConsumerRecord) {
+        final CloudEvent cloudEvent = subscriptionEventConsumerRecord.value();
+        final SubscriptionEvent subscriptionEvent = SubscriptionEventCloudMapper.toSubscriptionEvent(cloudEvent);
+        final String eventDatastore = subscriptionEvent.getData().getPredicates().getDatastore();
         if (!(eventDatastore.equals("passthrough-running") || eventDatastore.equals("passthrough-operational"))) {
             throw new OperationNotYetSupportedException(
                 "passthrough datastores are currently only supported for event subscriptions");
         }
-        if ("CM".equals(event.getDataType().getDataCategory())) {
-            log.debug("Consuming event {} ...", subscriptionEvent);
+        if ("CM".equals(subscriptionEvent.getData().getDataType().getDataCategory())) {
             if (subscriptionModelLoaderEnabled) {
                 persistSubscriptionEvent(subscriptionEvent);
             }
-            if ("CREATE".equals(subscriptionEvent.getEventType().value())) {
+            if ("CREATE".equals(cloudEvent.getType())) {
                 log.info("Subscription for ClientID {} with name {} ...",
-                        event.getSubscription().getClientID(),
-                        event.getSubscription().getName());
+                        subscriptionEvent.getData().getSubscription().getClientID(),
+                        subscriptionEvent.getData().getSubscription().getName());
                 if (notificationFeatureEnabled) {
-                    subscriptionEventForwarder.forwardCreateSubscriptionEvent(subscriptionEvent,
-                            subscriptionEventConsumerRecord.headers());
+                    subscriptionEventForwarder.forwardCreateSubscriptionEvent(subscriptionEvent);
                 }
             }
         } else {
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;
     }
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java
index dcbdcf339..bf9ceb1c3 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventMapper.java
@@ -27,17 +27,16 @@ import org.mapstruct.Mapping;
 import org.mapstruct.Named;
 import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus;
 import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent;
-import org.onap.cps.ncmp.event.model.SubscriptionEvent;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent;
 
 @Mapper(componentModel = "spring")
 public interface SubscriptionEventMapper {
 
-    @Mapping(source = "event.subscription.clientID", target = "clientId")
-    @Mapping(source = "event.subscription.name", target = "subscriptionName")
-    @Mapping(source = "event.subscription.isTagged", target = "tagged")
-    @Mapping(source = "event.predicates.targets", target = "predicates.targetCmHandles",
+    @Mapping(source = "data.subscription.clientID", target = "clientId")
+    @Mapping(source = "data.subscription.name", target = "subscriptionName")
+    @Mapping(source = "data.predicates.targets", target = "predicates.targetCmHandles",
             qualifiedByName = "mapTargetsToCmHandleTargets")
-    @Mapping(source = "event.predicates.datastore", target = "predicates.datastore")
+    @Mapping(source = "data.predicates.datastore", target = "predicates.datastore")
     YangModelSubscriptionEvent toYangModelSubscriptionEvent(SubscriptionEvent subscriptionEvent);
 
     /**
@@ -47,8 +46,8 @@ public interface SubscriptionEventMapper {
      * @return TargetCmHandle list
      */
     @Named("mapTargetsToCmHandleTargets")
-    default List<YangModelSubscriptionEvent.TargetCmHandle> mapTargetsToCmHandleTargets(List<Object> targets) {
-        return targets.stream().map(target -> new YangModelSubscriptionEvent.TargetCmHandle(target.toString(),
+    default List<YangModelSubscriptionEvent.TargetCmHandle> mapTargetsToCmHandleTargets(List<String> targets) {
+        return targets.stream().map(target -> new YangModelSubscriptionEvent.TargetCmHandle(target,
                         SubscriptionStatus.PENDING))
                 .collect(Collectors.toList());
     }
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/SubscriptionEventCloudMapper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/SubscriptionEventCloudMapper.java
new file mode 100644
index 000000000..a7de47904
--- /dev/null
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/SubscriptionEventCloudMapper.java
@@ -0,0 +1,79 @@
+/*
+ *  ============LICENSE_START=======================================================
+ *  Copyright (C) 2023 Nordix Foundation
+ *  ================================================================================
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *  ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.ncmp.api.impl.utils;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.CloudEventUtils;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import io.cloudevents.core.data.PojoCloudEventData;
+import io.cloudevents.jackson.PojoCloudEventDataMapper;
+import java.net.URI;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+@Slf4j
+public class SubscriptionEventCloudMapper {
+
+    private static final ObjectMapper objectMapper = new ObjectMapper();
+
+    /**
+     * Maps CloudEvent object to SubscriptionEvent.
+     *
+     * @param cloudEvent object.
+     * @return SubscriptionEvent deserialized.
+     */
+    public static SubscriptionEvent toSubscriptionEvent(final CloudEvent cloudEvent) {
+        final PojoCloudEventData<SubscriptionEvent> deserializedCloudEvent = CloudEventUtils
+                .mapData(cloudEvent, PojoCloudEventDataMapper.from(objectMapper, SubscriptionEvent.class));
+        if (deserializedCloudEvent == null) {
+            log.debug("No data found in the consumed event");
+            return null;
+        } else {
+            final SubscriptionEvent subscriptionEvent = deserializedCloudEvent.getValue();
+            log.debug("Consuming event {}", subscriptionEvent);
+            return subscriptionEvent;
+        }
+    }
+
+    /**
+     * Maps SubscriptionEvent to a CloudEvent.
+     *
+     * @param ncmpSubscriptionEvent object.
+     * @param eventKey as String.
+     * @return CloudEvent builded.
+     */
+    public static CloudEvent toCloudEvent(
+            final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent ncmpSubscriptionEvent,
+            final String eventKey) {
+        try {
+            return CloudEventBuilder.v1()
+                    .withData(objectMapper.writeValueAsBytes(ncmpSubscriptionEvent))
+                    .withId(eventKey).withType("CREATE").withSource(
+                            URI.create(ncmpSubscriptionEvent.getData().getSubscription().getClientID())).build();
+        } catch (final Exception ex) {
+            throw new RuntimeException("The Cloud Event could not be constructed.", ex);
+        }
+    }
+}