diff options
Diffstat (limited to 'cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java')
-rw-r--r-- | cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java | 72 |
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; } |