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;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+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.spi.exceptions.OperationNotYetSupportedException;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.client_to_ncmp.SubscriptionEvent;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.dmi_to_ncmp.Data;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.dmi_to_ncmp.SubscriptionEventResponse;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.CmHandle;
+import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
public class SubscriptionEventForwarder {
private final InventoryPersistence inventoryPersistence;
- private final EventsPublisher<SubscriptionEvent> eventsPublisher;
- private static final String DMI_AVC_SUBSCRIPTION_TOPIC_PREFIX = "ncmp-dmi-cm-avc-subscription-";
+ 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}")
+ private String dmiAvcSubscriptionTopicPrefix;
+
+ @Value("${ncmp.timers.subscription-forwarding.dmi-response-timeout-ms:30000}")
+ private int dmiResponseTimeoutInMs;
/**
* Forward subscription event.
*
* @param subscriptionEvent the event to be forwarded
*/
- public void forwardCreateSubscriptionEvent(final SubscriptionEvent subscriptionEvent) {
- final List<Object> cmHandleTargets = subscriptionEvent.getEvent().getPredicates().getTargets();
+ public void forwardCreateSubscriptionEvent(final SubscriptionEvent subscriptionEvent, final String eventType) {
+ final List<String> cmHandleTargets = subscriptionEvent.getData().getPredicates().getTargets();
if (cmHandleTargets == null || cmHandleTargets.isEmpty()
- || cmHandleTargets.stream().anyMatch(id -> ((String) id).contains("*"))) {
- throw new OperationNotYetSupportedException(
+ || cmHandleTargets.stream().anyMatch(id -> (id).contains("*"))) {
+ throw new UnsupportedOperationException(
"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);
- dmiPropertiesPerCmHandleIdPerServiceName.forEach((dmiServiceName, cmHandlePropertiesMap) -> {
- subscriptionEvent.getEvent().getPredicates().setTargets(Collections
- .singletonList(cmHandlePropertiesMap));
- final String eventKey = createEventKey(subscriptionEvent, dmiServiceName);
- eventsPublisher.publishEvent(DMI_AVC_SUBSCRIPTION_TOPIC_PREFIX + dmiServiceName, eventKey,
- subscriptionEvent);
+ findDmisAndRespond(subscriptionEvent, eventType, cmHandleTargets, dmiPropertiesPerCmHandleIdPerServiceName);
+ }
+
+ private void findDmisAndRespond(final SubscriptionEvent subscriptionEvent, final String eventType,
+ final List<String> cmHandleTargetsAsStrings,
+ final Map<String, Map<String, Map<String, String>>>
+ dmiPropertiesPerCmHandleIdPerServiceName) {
+ final SubscriptionEventResponse emptySubscriptionEventResponse =
+ new SubscriptionEventResponse().withData(new Data());
+ emptySubscriptionEventResponse.getData().setSubscriptionName(
+ subscriptionEvent.getData().getSubscription().getName());
+ emptySubscriptionEventResponse.getData().setClientId(
+ subscriptionEvent.getData().getSubscription().getClientID());
+ final List<String> cmHandlesThatExistsInDb = dmiPropertiesPerCmHandleIdPerServiceName.entrySet().stream()
+ .map(Map.Entry::getValue).map(Map::keySet).flatMap(Set::stream).collect(Collectors.toList());
+
+ final List<String> targetCmHandlesDoesNotExistInDb = new ArrayList<>(cmHandleTargetsAsStrings);
+ targetCmHandlesDoesNotExistInDb.removeAll(cmHandlesThatExistsInDb);
+
+ final Set<String> dmisToRespond = new HashSet<>(dmiPropertiesPerCmHandleIdPerServiceName.keySet());
+
+ if (dmisToRespond.isEmpty() || !targetCmHandlesDoesNotExistInDb.isEmpty()) {
+ updatesCmHandlesToRejectedAndPersistSubscriptionEvent(subscriptionEvent, targetCmHandlesDoesNotExistInDb);
+ }
+ if (dmisToRespond.isEmpty()) {
+ subscriptionEventResponseOutcome.sendResponse(emptySubscriptionEventResponse,
+ "subscriptionCreatedStatus");
+ } else {
+ startResponseTimeout(emptySubscriptionEventResponse, dmisToRespond);
+ final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent ncmpSubscriptionEvent =
+ clientSubscriptionEventMapper.toNcmpSubscriptionEvent(subscriptionEvent);
+ forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, ncmpSubscriptionEvent, eventType);
+ }
+ }
+
+ private void startResponseTimeout(final SubscriptionEventResponse emptySubscriptionEventResponse,
+ final Set<String> dmisToRespond) {
+ final String subscriptionClientId = emptySubscriptionEventResponse.getData().getClientId();
+ final String subscriptionName = emptySubscriptionEventResponse.getData().getSubscriptionName();
+ final String subscriptionEventId = subscriptionClientId + subscriptionName;
+
+ forwardedSubscriptionEventCache.put(subscriptionEventId, dmisToRespond,
+ ForwardedSubscriptionEventCacheConfig.SUBSCRIPTION_FORWARD_STARTED_TTL_SECS, TimeUnit.SECONDS);
+ final ResponseTimeoutTask responseTimeoutTask =
+ new ResponseTimeoutTask(forwardedSubscriptionEventCache, subscriptionEventResponseOutcome,
+ emptySubscriptionEventResponse);
+ try {
+ executorService.schedule(responseTimeoutTask, dmiResponseTimeoutInMs, TimeUnit.MILLISECONDS);
+ } catch (final RuntimeException ex) {
+ log.info("Caught exception in ScheduledExecutorService for ResponseTimeoutTask. StackTrace: {}",
+ ex.toString());
+ }
+ }
+
+ private void forwardEventToDmis(final Map<String, Map<String, Map<String, String>>> dmiNameCmHandleMap,
+ final org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_dmi.SubscriptionEvent
+ ncmpSubscriptionEvent, final String eventType) {
+ dmiNameCmHandleMap.forEach((dmiName, cmHandlePropertiesMap) -> {
+ 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;
+
+ final CloudEvent ncmpSubscriptionCloudEvent =
+ SubscriptionEventCloudMapper.toCloudEvent(ncmpSubscriptionEvent, eventKey, eventType);
+ 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;
}
+ private void updatesCmHandlesToRejectedAndPersistSubscriptionEvent(
+ final SubscriptionEvent subscriptionEvent,
+ final List<String> targetCmHandlesDoesNotExistInDb) {
+ final YangModelSubscriptionEvent yangModelSubscriptionEvent =
+ subscriptionEventMapper.toYangModelSubscriptionEvent(subscriptionEvent);
+ yangModelSubscriptionEvent.getPredicates()
+ .setTargetCmHandles(findRejectedCmHandles(targetCmHandlesDoesNotExistInDb,
+ yangModelSubscriptionEvent));
+ subscriptionPersistence.saveSubscriptionEvent(yangModelSubscriptionEvent);
+ }
+
+ private static List<YangModelSubscriptionEvent.TargetCmHandle> findRejectedCmHandles(
+ final List<String> targetCmHandlesDoesNotExistInDb,
+ final YangModelSubscriptionEvent yangModelSubscriptionEvent) {
+ return yangModelSubscriptionEvent.getPredicates().getTargetCmHandles().stream()
+ .filter(targetCmHandle -> targetCmHandlesDoesNotExistInDb.contains(targetCmHandle.getCmHandleId()))
+ .map(target -> new YangModelSubscriptionEvent.TargetCmHandle(target.getCmHandleId(),
+ SubscriptionStatus.REJECTED, "Targets not found"))
+ .collect(Collectors.toList());
+ }
}
Code Review / cps.git / blobdiff