package org.onap.cps.ncmp.api.impl.events.cmsubscription.service;
+import static org.onap.cps.spi.FetchDescendantsOption.OMIT_DESCENDANTS;
+
+import java.io.Serializable;
+import java.time.OffsetDateTime;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.onap.cps.api.CpsDataService;
import org.onap.cps.api.CpsQueryService;
+import org.onap.cps.cpspath.parser.CpsPathUtil;
import org.onap.cps.ncmp.api.impl.operations.DatastoreType;
-import org.onap.cps.spi.FetchDescendantsOption;
import org.onap.cps.spi.model.DataNode;
+import org.onap.cps.utils.ContentType;
+import org.onap.cps.utils.JsonObjectMapper;
import org.springframework.stereotype.Service;
@Slf4j
@RequiredArgsConstructor
public class CmNotificationSubscriptionPersistenceServiceImpl implements CmNotificationSubscriptionPersistenceService {
- private static final String IS_ONGOING_CM_SUBSCRIPTION_CPS_PATH_QUERY = """
+ private static final String SUBSCRIPTION_ANCHOR_NAME = "cm-data-subscriptions";
+ private static final String CM_SUBSCRIPTION_CPS_PATH_QUERY = """
/datastores/datastore[@name='%s']/cm-handles/cm-handle[@id='%s']/filters/filter[@xpath='%s']
""".trim();
+ private static final String SUBSCRIPTION_IDS_CPS_PATH_QUERY = """
+ //filter/subscriptionIds[text()='%s']
+ """.trim();
+ private final JsonObjectMapper jsonObjectMapper;
private final CpsQueryService cpsQueryService;
+ private final CpsDataService cpsDataService;
@Override
public boolean isOngoingCmNotificationSubscription(final DatastoreType datastoreType, final String cmHandleId,
- final String xpath) {
+ final String xpath) {
return !getOngoingCmNotificationSubscriptionIds(datastoreType, cmHandleId, xpath).isEmpty();
}
+ @Override
+ public boolean isUniqueSubscriptionId(final String subscriptionId) {
+ return cpsQueryService.queryDataNodes(NCMP_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME,
+ SUBSCRIPTION_IDS_CPS_PATH_QUERY.formatted(subscriptionId),
+ OMIT_DESCENDANTS).isEmpty();
+ }
+
@Override
public Collection<String> getOngoingCmNotificationSubscriptionIds(final DatastoreType datastoreType,
- final String cmHandleId, final String xpath) {
+ final String cmHandleId, final String xpath) {
final String isOngoingCmSubscriptionCpsPathQuery =
- IS_ONGOING_CM_SUBSCRIPTION_CPS_PATH_QUERY.formatted(datastoreType.getDatastoreName(), cmHandleId,
+ CM_SUBSCRIPTION_CPS_PATH_QUERY.formatted(datastoreType.getDatastoreName(), cmHandleId,
escapeQuotesByDoublingThem(xpath));
final Collection<DataNode> existingNodes =
cpsQueryService.queryDataNodes(NCMP_DATASPACE_NAME, CM_SUBSCRIPTIONS_ANCHOR_NAME,
- isOngoingCmSubscriptionCpsPathQuery, FetchDescendantsOption.OMIT_DESCENDANTS);
+ isOngoingCmSubscriptionCpsPathQuery, OMIT_DESCENDANTS);
if (existingNodes.isEmpty()) {
return Collections.emptyList();
}
return (List<String>) existingNodes.iterator().next().getLeaves().get("subscriptionIds");
}
+ @Override
+ public void addCmNotificationSubscription(final DatastoreType datastoreType, final String cmHandleId,
+ final String xpath, final String subscriptionId) {
+ if (isOngoingCmNotificationSubscription(datastoreType, cmHandleId, xpath)
+ && (!getOngoingCmNotificationSubscriptionIds(datastoreType, cmHandleId, xpath)
+ .contains(subscriptionId))) {
+ final DataNode subscriptionAsDataNode = getSubscriptionAsDataNode(datastoreType, cmHandleId, xpath);
+ final Collection<String> subscriptionIds = getOngoingCmNotificationSubscriptionIds(datastoreType,
+ cmHandleId, xpath);
+ subscriptionIds.add(subscriptionId);
+ saveSubscriptionDetails(subscriptionAsDataNode, subscriptionIds);
+ } else {
+ addNewSubscriptionViaDatastore(datastoreType, cmHandleId, xpath, subscriptionId);
+ }
+ }
+
+ @Override
+ public void removeCmNotificationSubscription(final DatastoreType datastoreType, final String cmHandleId,
+ final String xpath, final String subscriptionId) {
+ final DataNode subscriptionAsDataNode = getSubscriptionAsDataNode(datastoreType, cmHandleId, xpath);
+ final Collection<String> subscriptionIds = getOngoingCmNotificationSubscriptionIds(datastoreType,
+ cmHandleId, xpath);
+ subscriptionIds.remove(subscriptionId);
+ saveSubscriptionDetails(subscriptionAsDataNode, subscriptionIds);
+ if (isOngoingCmNotificationSubscription(datastoreType, cmHandleId, xpath)) {
+ log.info("There are subscribers left for the following cps path {} :",
+ CM_SUBSCRIPTION_CPS_PATH_QUERY.formatted(datastoreType.getDatastoreName(), cmHandleId,
+ escapeQuotesByDoublingThem(xpath)));
+ } else {
+ log.info("No subscribers left for the following cps path {} :",
+ CM_SUBSCRIPTION_CPS_PATH_QUERY.formatted(datastoreType.getDatastoreName(), cmHandleId,
+ escapeQuotesByDoublingThem(xpath)));
+ deleteListOfSubscriptionsFor(datastoreType, cmHandleId, xpath);
+ }
+ }
+
+ private void deleteListOfSubscriptionsFor(final DatastoreType datastoreType, final String cmHandleId,
+ final String xpath) {
+ cpsDataService.deleteDataNode(NCMP_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME,
+ CM_SUBSCRIPTION_CPS_PATH_QUERY.formatted(datastoreType.getDatastoreName(), cmHandleId,
+ escapeQuotesByDoublingThem(xpath)),
+ OffsetDateTime.now());
+ }
+
+ private DataNode getSubscriptionAsDataNode(final DatastoreType datastoreType, final String cmHandleId,
+ final String xpath) {
+ return cpsQueryService.queryDataNodes(NCMP_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME,
+ CM_SUBSCRIPTION_CPS_PATH_QUERY.formatted(datastoreType.getDatastoreName(), cmHandleId,
+ escapeQuotesByDoublingThem(xpath)),
+ OMIT_DESCENDANTS).iterator().next();
+ }
+
+ private void addNewSubscriptionViaDatastore(final DatastoreType datastoreType, final String cmHandleId,
+ final String xpath, final String newSubscriptionId) {
+ final String parentXpath = "/datastores/datastore[@name='%s']/cm-handles"
+ .formatted(datastoreType.getDatastoreName());
+ final String subscriptionAsJson = String.format("{\"cm-handle\":[{\"id\":\"%s\",\"filters\":{\"filter\":"
+ + "[{\"xpath\":\"%s\",\"subscriptionIds\":[\"%s\"]}]}}]}", cmHandleId, xpath, newSubscriptionId);
+ cpsDataService.saveData(NCMP_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME, parentXpath, subscriptionAsJson,
+ OffsetDateTime.now(), ContentType.JSON);
+ }
+
+ private void saveSubscriptionDetails(final DataNode subscriptionDetailsAsDataNode,
+ final Collection<String> subscriptionIds) {
+ final Map<String, Serializable> subscriptionDetailsAsMap = new HashMap<>();
+ subscriptionDetailsAsMap.put("xpath", subscriptionDetailsAsDataNode.getLeaves().get("xpath"));
+ subscriptionDetailsAsMap.put("subscriptionIds", (Serializable) subscriptionIds);
+ final String parentXpath = CpsPathUtil.getNormalizedParentXpath(subscriptionDetailsAsDataNode.getXpath());
+ final String subscriptionDetailsAsJson = "{\"filter\":["
+ + jsonObjectMapper.asJsonString(subscriptionDetailsAsMap).replace("'", "\"") + "]}";
+ cpsDataService.updateNodeLeaves(NCMP_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME, parentXpath,
+ subscriptionDetailsAsJson, OffsetDateTime.now());
+ }
+
private static String escapeQuotesByDoublingThem(final String inputXpath) {
return inputXpath.replace("'", "''");
}