package org.onap.cps.ncmp.api.impl.events.avcsubscription;
-import java.io.Serializable;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Iterator;
+import io.cloudevents.CloudEvent;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.common.header.Headers;
-import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.onap.cps.ncmp.api.NcmpEventResponseCode;
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.DataNodeHelper;
-import org.onap.cps.ncmp.api.models.SubscriptionEventResponse;
-import org.onap.cps.ncmp.events.avc.subscription.v1.SubscriptionEventOutcome;
-import org.onap.cps.spi.model.DataNode;
+import org.onap.cps.ncmp.api.impl.utils.SubscriptionOutcomeCloudMapper;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.dmi_to_ncmp.SubscriptionEventResponse;
+import org.onap.cps.ncmp.events.avcsubscription1_0_0.ncmp_to_client.SubscriptionEventOutcome;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
private final SubscriptionPersistence subscriptionPersistence;
- private final EventsPublisher<SubscriptionEventOutcome> outcomeEventsPublisher;
+ private final EventsPublisher<CloudEvent> outcomeEventsPublisher;
private final SubscriptionOutcomeMapper subscriptionOutcomeMapper;
- @Value("${app.ncmp.avc.subscription-outcome-topic:cm-avc-subscription-response}")
+ @Value("${app.ncmp.avc.subscription-outcome-topic:subscription-response}")
private String subscriptionOutcomeEventTopic;
/**
* This is for construction of outcome message to be published for client apps.
*
- * @param subscriptionClientId client id of the subscription.
- * @param subscriptionName name of the subscription.
- * @param isFullOutcomeResponse the flag to decide on complete or partial response to be generated.
+ * @param subscriptionEventResponse event produced by Dmi Plugin
*/
- public void sendResponse(final String subscriptionClientId, final String subscriptionName,
- final boolean isFullOutcomeResponse) {
- final SubscriptionEventOutcome subscriptionEventOutcome = generateResponse(
- subscriptionClientId, subscriptionName, isFullOutcomeResponse);
- final Headers headers = new RecordHeaders();
+ public void sendResponse(final SubscriptionEventResponse subscriptionEventResponse, final String eventKey) {
+ final SubscriptionEventOutcome subscriptionEventOutcome =
+ formSubscriptionOutcomeMessage(subscriptionEventResponse);
+ final String subscriptionClientId = subscriptionEventResponse.getData().getClientId();
+ final String subscriptionName = subscriptionEventResponse.getData().getSubscriptionName();
final String subscriptionEventId = subscriptionClientId + subscriptionName;
- outcomeEventsPublisher.publishEvent(subscriptionOutcomeEventTopic,
- subscriptionEventId, headers, subscriptionEventOutcome);
+ final CloudEvent subscriptionOutcomeCloudEvent =
+ SubscriptionOutcomeCloudMapper.toCloudEvent(subscriptionEventOutcome,
+ subscriptionEventId, eventKey);
+ outcomeEventsPublisher.publishCloudEvent(subscriptionOutcomeEventTopic,
+ subscriptionEventId, subscriptionOutcomeCloudEvent);
}
- private SubscriptionEventOutcome generateResponse(final String subscriptionClientId, final String subscriptionName,
- final boolean isFullOutcomeResponse) {
- final Collection<DataNode> dataNodes = subscriptionPersistence.getDataNodesForSubscriptionEvent();
- final List<Map<String, Serializable>> dataNodeLeaves = DataNodeHelper.getDataNodeLeaves(dataNodes);
- final List<Collection<Serializable>> cmHandleIdToStatus =
- DataNodeHelper.getCmHandleIdToStatus(dataNodeLeaves);
- return formSubscriptionOutcomeMessage(cmHandleIdToStatus, subscriptionClientId, subscriptionName,
- isFullOutcomeResponse);
+ private SubscriptionEventOutcome formSubscriptionOutcomeMessage(
+ final SubscriptionEventResponse subscriptionEventResponse) {
+ final Map<String, Map<String, String>> cmHandleIdToStatusAndDetailsAsMap =
+ DataNodeHelper.cmHandleIdToStatusAndDetailsAsMapFromDataNode(
+ subscriptionPersistence.getCmHandlesForSubscriptionEvent(
+ subscriptionEventResponse.getData().getClientId(),
+ subscriptionEventResponse.getData().getSubscriptionName()));
+ final List<org.onap.cps.ncmp.events.avcsubscription1_0_0.dmi_to_ncmp.SubscriptionStatus>
+ subscriptionStatusList = mapCmHandleIdStatusDetailsMapToSubscriptionStatusList(
+ cmHandleIdToStatusAndDetailsAsMap);
+ subscriptionEventResponse.getData().setSubscriptionStatus(subscriptionStatusList);
+ return fromSubscriptionEventResponse(subscriptionEventResponse,
+ decideOnNcmpEventResponseCodeForSubscription(cmHandleIdToStatusAndDetailsAsMap));
}
+ private static List<org.onap.cps.ncmp.events.avcsubscription1_0_0.dmi_to_ncmp.SubscriptionStatus>
+ mapCmHandleIdStatusDetailsMapToSubscriptionStatusList(
+ final Map<String, Map<String, String>> cmHandleIdToStatusAndDetailsAsMap) {
+ return cmHandleIdToStatusAndDetailsAsMap.entrySet()
+ .stream().map(entryset -> {
+ final org.onap.cps.ncmp.events.avcsubscription1_0_0.dmi_to_ncmp.SubscriptionStatus
+ subscriptionStatus = new org.onap.cps.ncmp.events.avcsubscription1_0_0
+ .dmi_to_ncmp.SubscriptionStatus();
+ final String cmHandleId = entryset.getKey();
+ final Map<String, String> statusAndDetailsMap = entryset.getValue();
+ final String status = statusAndDetailsMap.get("status");
+ final String details = statusAndDetailsMap.get("details");
+ subscriptionStatus.setId(cmHandleId);
+ subscriptionStatus.setStatus(
+ org.onap.cps.ncmp.events.avcsubscription1_0_0.dmi_to_ncmp
+ .SubscriptionStatus.Status.fromValue(status));
+ subscriptionStatus.setDetails(details);
+ return subscriptionStatus;
+ }).collect(Collectors.toList());
+ }
- private SubscriptionEventOutcome formSubscriptionOutcomeMessage(
- final List<Collection<Serializable>> cmHandleIdToStatus, final String subscriptionClientId,
- final String subscriptionName, final boolean isFullOutcomeResponse) {
+ private NcmpEventResponseCode decideOnNcmpEventResponseCodeForSubscription(
+ final Map<String, Map<String, String>> cmHandleIdToStatusAndDetailsAsMap) {
- final SubscriptionEventResponse subscriptionEventResponse = toSubscriptionEventResponse(
- cmHandleIdToStatus, subscriptionClientId, subscriptionName);
+ final boolean isAllTargetsPending = isAllTargetCmHandleStatusMatch(cmHandleIdToStatusAndDetailsAsMap,
+ SubscriptionStatus.PENDING);
- final SubscriptionEventOutcome subscriptionEventOutcome =
- subscriptionOutcomeMapper.toSubscriptionEventOutcome(subscriptionEventResponse);
+ final boolean isAllTargetsRejected = isAllTargetCmHandleStatusMatch(cmHandleIdToStatusAndDetailsAsMap,
+ SubscriptionStatus.REJECTED);
+
+ final boolean isAllTargetsAccepted = isAllTargetCmHandleStatusMatch(cmHandleIdToStatusAndDetailsAsMap,
+ SubscriptionStatus.ACCEPTED);
- if (isFullOutcomeResponse) {
- subscriptionEventOutcome.setEventType(SubscriptionEventOutcome.EventType.COMPLETE_OUTCOME);
+ if (isAllTargetsAccepted) {
+ return NcmpEventResponseCode.SUCCESSFULLY_APPLIED_SUBSCRIPTION;
+ } else if (isAllTargetsRejected) {
+ return NcmpEventResponseCode.SUBSCRIPTION_NOT_APPLICABLE;
+ } else if (isAllTargetsPending) {
+ return NcmpEventResponseCode.SUBSCRIPTION_PENDING;
} else {
- subscriptionEventOutcome.setEventType(SubscriptionEventOutcome.EventType.PARTIAL_OUTCOME);
+ return NcmpEventResponseCode.PARTIALLY_APPLIED_SUBSCRIPTION;
}
+ }
- return subscriptionEventOutcome;
+ private boolean isAllTargetCmHandleStatusMatch(
+ final Map<String, Map<String, String>> cmHandleIdToStatusAndDetailsAsMap,
+ final SubscriptionStatus subscriptionStatus) {
+ return cmHandleIdToStatusAndDetailsAsMap.values().stream()
+ .allMatch(entryset -> entryset.containsValue(subscriptionStatus.toString()));
}
- private SubscriptionEventResponse toSubscriptionEventResponse(
- final List<Collection<Serializable>> cmHandleIdToStatus, final String subscriptionClientId,
- final String subscriptionName) {
- final Map<String, SubscriptionStatus> cmHandleIdToStatusMap = new HashMap<>();
- final SubscriptionEventResponse subscriptionEventResponse = new SubscriptionEventResponse();
- subscriptionEventResponse.setClientId(subscriptionClientId);
- subscriptionEventResponse.setSubscriptionName(subscriptionName);
-
- for (final Collection<Serializable> cmHandleToStatusBucket: cmHandleIdToStatus) {
- final Iterator<Serializable> bucketIterator = cmHandleToStatusBucket.iterator();
- while (bucketIterator.hasNext()) {
- final String item = (String) bucketIterator.next();
- if ("PENDING".equals(item)) {
- cmHandleIdToStatusMap.put((String) bucketIterator.next(),
- SubscriptionStatus.PENDING);
- }
- if ("REJECTED".equals(item)) {
- cmHandleIdToStatusMap.put((String) bucketIterator.next(),
- SubscriptionStatus.REJECTED);
- }
- if ("ACCEPTED".equals(item)) {
- cmHandleIdToStatusMap.put((String) bucketIterator.next(),
- SubscriptionStatus.ACCEPTED);
- }
- }
- }
- subscriptionEventResponse.setCmHandleIdToStatus(cmHandleIdToStatusMap);
+ private SubscriptionEventOutcome fromSubscriptionEventResponse(
+ final SubscriptionEventResponse subscriptionEventResponse,
+ final NcmpEventResponseCode ncmpEventResponseCode) {
- return subscriptionEventResponse;
+ final SubscriptionEventOutcome subscriptionEventOutcome =
+ subscriptionOutcomeMapper.toSubscriptionEventOutcome(subscriptionEventResponse);
+ subscriptionEventOutcome.getData().setStatusCode(Integer.parseInt(ncmpEventResponseCode.getStatusCode()));
+ subscriptionEventOutcome.getData().setStatusMessage(ncmpEventResponseCode.getStatusMessage());
+
+ return subscriptionEventOutcome;
}
}
Code Review / cps.git / blobdiff