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.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.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 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}")
*
* @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 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);
-
- findDmisAndRespond(subscriptionEvent, eventHeaders, cmHandleTargetsAsStrings,
- dmiPropertiesPerCmHandleIdPerServiceName);
+ findDmisAndRespond(subscriptionEvent, eventType, cmHandleTargets, dmiPropertiesPerCmHandleIdPerServiceName);
}
- private void findDmisAndRespond(final SubscriptionEvent subscriptionEvent, final Headers eventHeaders,
- final List<String> cmHandleTargetsAsStrings,
+ 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());
updatesCmHandlesToRejectedAndPersistSubscriptionEvent(subscriptionEvent, targetCmHandlesDoesNotExistInDb);
}
if (dmisToRespond.isEmpty()) {
- final String clientID = subscriptionEvent.getEvent().getSubscription().getClientID();
- final String subscriptionName = subscriptionEvent.getEvent().getSubscription().getName();
- subscriptionEventResponseOutcome.sendResponse(clientID, subscriptionName);
+ subscriptionEventResponseOutcome.sendResponse(emptySubscriptionEventResponse,
+ "subscriptionCreatedStatus");
} else {
- startResponseTimeout(subscriptionEvent, dmisToRespond);
- forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, subscriptionEvent, eventHeaders);
+ 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 SubscriptionEvent subscriptionEvent, final Set<String> dmisToRespond) {
- final String subscriptionClientId = subscriptionEvent.getEvent().getSubscription().getClientID();
- final String subscriptionName = subscriptionEvent.getEvent().getSubscription().getName();
+ 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,
- subscriptionClientId, subscriptionName);
+ emptySubscriptionEventResponse);
try {
executorService.schedule(responseTimeoutTask, dmiResponseTimeoutInMs, TimeUnit.MILLISECONDS);
} catch (final RuntimeException ex) {
}
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, final String eventType) {
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, 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;
}
return yangModelSubscriptionEvent.getPredicates().getTargetCmHandles().stream()
.filter(targetCmHandle -> targetCmHandlesDoesNotExistInDb.contains(targetCmHandle.getCmHandleId()))
.map(target -> new YangModelSubscriptionEvent.TargetCmHandle(target.getCmHandleId(),
- SubscriptionStatus.REJECTED)).collect(Collectors.toList());
+ SubscriptionStatus.REJECTED, "Targets not found"))
+ .collect(Collectors.toList());
}
}
Code Review / cps.git / blobdiff