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.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.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 = """
- /datastores/datastore[@name='%s']/cm-handles/cm-handle[@id='%s']/filters/filter[@xpath='%s']""";
+ private static final String SUBSCRIPTION_ANCHOR_NAME = "cm-data-subscriptions";
+ private static final String CPS_PATH_QUERY_FOR_CM_SUBSCRIPTION_WITH_DATASTORE_AND_CMHANDLE = """
+ /datastores/datastore[@name='%s']/cm-handles/cm-handle[@id='%s']/filters
+ """.trim();
+ private static final String CPS_PATH_QUERY_FOR_CM_SUBSCRIPTION_WITH_DATASTORE_CMHANDLE_AND_XPATH =
+ CPS_PATH_QUERY_FOR_CM_SUBSCRIPTION_WITH_DATASTORE_AND_CMHANDLE + "/filter[@xpath='%s']";
+
+ private static final String CPS_PATH_QUERY_FOR_CM_SUBSCRIPTION_WITH_ID = """
+ //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,
+ CPS_PATH_QUERY_FOR_CM_SUBSCRIPTION_WITH_ID.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,
- escapeQuotesByDoublingThem(xpath));
+ CPS_PATH_QUERY_FOR_CM_SUBSCRIPTION_WITH_DATASTORE_CMHANDLE_AND_XPATH.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("subscribers");
+ 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) {
+ final Collection<String> subscriptionIds = getOngoingCmNotificationSubscriptionIds(datastoreType,
+ cmHandleId, xpath);
+ if (subscriptionIds.isEmpty()) {
+ addFirstSubscriptionForDatastoreCmHandleAndXpath(datastoreType, cmHandleId, xpath, subscriptionId);
+ } else if (!subscriptionIds.contains(subscriptionId)) {
+ subscriptionIds.add(subscriptionId);
+ saveSubscriptionDetails(datastoreType, cmHandleId, xpath, subscriptionIds);
+ }
+ }
+
+ @Override
+ public void removeCmNotificationSubscription(final DatastoreType datastoreType, final String cmHandleId,
+ final String xpath, final String subscriptionId) {
+ final Collection<String> subscriptionIds = getOngoingCmNotificationSubscriptionIds(datastoreType,
+ cmHandleId, xpath);
+ if (subscriptionIds.remove(subscriptionId)) {
+ if (isOngoingCmNotificationSubscription(datastoreType, cmHandleId, xpath)) {
+ saveSubscriptionDetails(datastoreType, cmHandleId, xpath, subscriptionIds);
+ log.info("There are subscribers left for the following cps path {} :",
+ CPS_PATH_QUERY_FOR_CM_SUBSCRIPTION_WITH_DATASTORE_CMHANDLE_AND_XPATH.formatted(
+ datastoreType.getDatastoreName(), cmHandleId, escapeQuotesByDoublingThem(xpath)));
+ } else {
+ log.info("No subscribers left for the following cps path {} :",
+ CPS_PATH_QUERY_FOR_CM_SUBSCRIPTION_WITH_DATASTORE_CMHANDLE_AND_XPATH.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,
+ CPS_PATH_QUERY_FOR_CM_SUBSCRIPTION_WITH_DATASTORE_CMHANDLE_AND_XPATH.formatted(
+ datastoreType.getDatastoreName(), cmHandleId, escapeQuotesByDoublingThem(xpath)),
+ OffsetDateTime.now());
+ }
+
+ private boolean isFirstSubscriptionForCmHandle(final DatastoreType datastoreType, final String cmHandleId) {
+ return cpsQueryService.queryDataNodes(NCMP_DATASPACE_NAME, CM_SUBSCRIPTIONS_ANCHOR_NAME,
+ CPS_PATH_QUERY_FOR_CM_SUBSCRIPTION_WITH_DATASTORE_AND_CMHANDLE.formatted(
+ datastoreType.getDatastoreName(), cmHandleId),
+ OMIT_DESCENDANTS).isEmpty();
+ }
+
+ private void addFirstSubscriptionForDatastoreCmHandleAndXpath(final DatastoreType datastoreType,
+ final String cmHandleId,
+ final String xpath,
+ final String subscriptionId) {
+ final Collection<String> newSubscriptionList = Collections.singletonList(subscriptionId);
+ final String subscriptionDetailsAsJson = getSubscriptionDetailsAsJson(xpath, newSubscriptionList);
+ if (isFirstSubscriptionForCmHandle(datastoreType, cmHandleId)) {
+ final String parentXpath = "/datastores/datastore[@name='%s']/cm-handles"
+ .formatted(datastoreType.getDatastoreName());
+ final String subscriptionAsJson = String.format("{\"cm-handle\":[{\"id\":\"%s\",\"filters\":%s}]}",
+ cmHandleId, subscriptionDetailsAsJson);
+ cpsDataService.saveData(NCMP_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME, parentXpath, subscriptionAsJson,
+ OffsetDateTime.now(), ContentType.JSON);
+ } else {
+ cpsDataService.saveListElements(NCMP_DATASPACE_NAME, CM_SUBSCRIPTIONS_ANCHOR_NAME,
+ CPS_PATH_QUERY_FOR_CM_SUBSCRIPTION_WITH_DATASTORE_AND_CMHANDLE.formatted(
+ datastoreType.getDatastoreName(), cmHandleId),
+ subscriptionDetailsAsJson, OffsetDateTime.now());
+ }
+ }
+
+ private void saveSubscriptionDetails(final DatastoreType datastoreType, final String cmHandleId,
+ final String xpath,
+ final Collection<String> subscriptionIds) {
+ final String subscriptionDetailsAsJson = getSubscriptionDetailsAsJson(xpath, subscriptionIds);
+ cpsDataService.updateNodeLeaves(NCMP_DATASPACE_NAME, CM_SUBSCRIPTIONS_ANCHOR_NAME,
+ CPS_PATH_QUERY_FOR_CM_SUBSCRIPTION_WITH_DATASTORE_AND_CMHANDLE.formatted(
+ datastoreType.getDatastoreName(), cmHandleId), subscriptionDetailsAsJson, OffsetDateTime.now());
+ }
+
+ private String getSubscriptionDetailsAsJson(final String xpath, final Collection<String> subscriptionIds) {
+ final Map<String, Serializable> subscriptionDetailsAsMap =
+ Map.of("xpath", xpath, "subscriptionIds", (Serializable) subscriptionIds);
+ return "{\"filter\":[" + jsonObjectMapper.asJsonString(subscriptionDetailsAsMap) + "]}";
}
private static String escapeQuotesByDoublingThem(final String inputXpath) {
Code Review / cps.git / blobdiff