/**
* Check if we have an ongoing cm subscription based on the parameters.
*
- * @param datastoreType valid datastore type
- * @param cmHandleId cmhandle id
- * @param xpath valid xpath
+ * @param datastoreType the susbcription target datastore type
+ * @param cmHandleId the id of the cm handle for the susbcription
+ * @param xpath the target xpath
* @return true for ongoing cmsubscription , otherwise false
*/
boolean isOngoingCmNotificationSubscription(final DatastoreType datastoreType, final String cmHandleId,
/**
* Get all ongoing cm notification subscription based on the parameters.
*
- * @param datastoreType valid datastore type
- * @param cmHandleId cmhandle id
- * @param xpath valid xpath
+ * @param datastoreType the susbcription target datastore type
+ * @param cmHandleId the id of the cm handle for the susbcription
+ * @param xpath the target xpath
* @return collection of subscription ids of ongoing cm notification subscription
*/
Collection<String> getOngoingCmNotificationSubscriptionIds(final DatastoreType datastoreType,
final String cmHandleId, final String xpath);
/**
- * Add or update cm notification subscription.
+ * Add cm notification subscription.
*
- * @param datastoreType valid datastore type
- * @param cmHandle cmhandle id
- * @param xpath valid xpath
- * @param newSubscriptionId subscription Id to be added
+ * @param datastoreType the susbcription target datastore type
+ * @param cmHandleId the id of the cm handle for the susbcription
+ * @param xpath the target xpath
+ * @param newSubscriptionId subscription id to be added
*/
- void addOrUpdateCmNotificationSubscription(final DatastoreType datastoreType, final String cmHandle,
- final String xpath, final String newSubscriptionId);
+ void addCmNotificationSubscription(final DatastoreType datastoreType, final String cmHandleId,
+ final String xpath, final String newSubscriptionId);
+
+ /**
+ * Remove cm notification Subscription.
+ *
+ * @param datastoreType the susbcription target datastore type
+ * @param cmHandleId the id of the cm handle for the susbcription
+ * @param xpath the target xpath
+ * @param subscriptionId subscription id to remove
+ */
+ void removeCmNotificationSubscription(final DatastoreType datastoreType, final String cmHandleId,
+ final String xpath, final String subscriptionId);
+
}
+
import java.io.Serializable;
import java.time.OffsetDateTime;
-import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@Override
public boolean isOngoingCmNotificationSubscription(final DatastoreType datastoreType, final String cmHandleId,
- final String xpath) {
+ final String xpath) {
return !getOngoingCmNotificationSubscriptionIds(datastoreType, cmHandleId, xpath).isEmpty();
}
@Override
public Collection<String> getOngoingCmNotificationSubscriptionIds(final DatastoreType datastoreType,
- final String cmHandleId, final String xpath) {
+ final String cmHandleId, final String xpath) {
final String isOngoingCmSubscriptionCpsPathQuery =
CM_SUBSCRIPTION_CPS_PATH_QUERY.formatted(datastoreType.getDatastoreName(), cmHandleId,
}
@Override
- public void addOrUpdateCmNotificationSubscription(final DatastoreType datastoreType, final String cmHandleId,
- final String xpath, final String newSubscriptionId) {
- if (isOngoingCmNotificationSubscription(datastoreType, cmHandleId, xpath)) {
- final DataNode existingFilterNode =
- cpsQueryService.queryDataNodes(NCMP_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME,
- CM_SUBSCRIPTION_CPS_PATH_QUERY.formatted(datastoreType.getDatastoreName(), cmHandleId,
- escapeQuotesByDoublingThem(xpath)),
- OMIT_DESCENDANTS).iterator().next();
- final Collection<String> existingSubscriptionIds = getOngoingCmNotificationSubscriptionIds(datastoreType,
+ 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);
- if (!existingSubscriptionIds.contains(newSubscriptionId)) {
- updateListOfSubscribers(existingSubscriptionIds, newSubscriptionId, existingFilterNode);
- }
+ 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 {
- addNewSubscriptionViaDatastore(datastoreType, cmHandleId, xpath, newSubscriptionId);
+ 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 updatedJson = String.format("{\"cm-handle\":[{\"id\":\"%s\",\"filters\":{\"filter\":"
+ 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, updatedJson,
+ cpsDataService.saveData(NCMP_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME, parentXpath, subscriptionAsJson,
OffsetDateTime.now(), ContentType.JSON);
}
- private void updateListOfSubscribers(final Collection<String> existingSubscriptionIds,
- final String newSubscriptionId, final DataNode existingFilterNode) {
- final String parentXpath = CpsPathUtil.getNormalizedParentXpath(existingFilterNode.getXpath());
- final List<String> updatedSubscribers = new ArrayList<>(existingSubscriptionIds);
- updatedSubscribers.add(newSubscriptionId);
- final Map<String, Serializable> updatedLeaves = new HashMap<>();
- updatedLeaves.put("xpath", existingFilterNode.getLeaves().get("xpath"));
- updatedLeaves.put("subscriptionIds", (Serializable) updatedSubscribers);
- final String updatedJson = "{\"filter\":[" + jsonObjectMapper.asJsonString(updatedLeaves) + "]}";
- cpsDataService.updateNodeLeaves(NCMP_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME, parentXpath, updatedJson,
- OffsetDateTime.now());
+ 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) {
def 'Add new subscriber to an ongoing cm notification subscription'() {
given: 'a valid cm subscription path query'
- def cpsPathQuery = objectUnderTest.CM_SUBSCRIPTION_CPS_PATH_QUERY.formatted('ncmp-datastore:passthrough-running', 'ch-1', '/x/y');
+ def cpsPathQuery = objectUnderTest.CM_SUBSCRIPTION_CPS_PATH_QUERY.formatted('ncmp-datastore:passthrough-running', 'ch-1', '/x/y')
and: 'a dataNode exists for the given cps path query'
mockCpsQueryService.queryDataNodes('NCMP-Admin', 'cm-data-subscriptions',
cpsPathQuery, FetchDescendantsOption.OMIT_DESCENDANTS) >> [new DataNode(xpath: cpsPathQuery, leaves: ['xpath': '/x/y','subscriptionIds': ['sub-1']])]
when: 'the method to add/update cm notification subscription is called'
- objectUnderTest.addOrUpdateCmNotificationSubscription(DatastoreType.PASSTHROUGH_RUNNING, 'ch-1','/x/y', 'newSubId')
+ objectUnderTest.addCmNotificationSubscription(DatastoreType.PASSTHROUGH_RUNNING, 'ch-1','/x/y', 'newSubId')
then: 'data service method to update list of subscribers is called once'
1 * mockCpsDataService.updateNodeLeaves(
'NCMP-Admin',
cpsPathQuery.formatted(datastoreName),
FetchDescendantsOption.OMIT_DESCENDANTS) >> []
when: 'the method to add/update cm notification subscription is called'
- objectUnderTest.addOrUpdateCmNotificationSubscription(datastoreType, 'ch-1','/x/y', 'newSubId')
+ objectUnderTest.addCmNotificationSubscription(datastoreType, 'ch-1','/x/y', 'newSubId')
then: 'data service method to update list of subscribers is called once with the correct parameters'
1 * mockCpsDataService.saveData(
'NCMP-Admin',
'passthrough_running' | DatastoreType.PASSTHROUGH_RUNNING || "ncmp-datastore:passthrough-running"
'passthrough_operational' | DatastoreType.PASSTHROUGH_OPERATIONAL || "ncmp-datastore:passthrough-operational"
}
+
+ def 'Remove subscriber from a list of an ongoing cm notification subscription'() {
+ given: 'a subscription exists when queried'
+ def cpsPathQuery = objectUnderTest.CM_SUBSCRIPTION_CPS_PATH_QUERY.formatted('ncmp-datastore:passthrough-running', 'ch-1', '/x/y')
+ mockCpsQueryService.queryDataNodes('NCMP-Admin', 'cm-data-subscriptions',
+ cpsPathQuery, FetchDescendantsOption.OMIT_DESCENDANTS) >> [new DataNode(xpath: cpsPathQuery, leaves: ['xpath': '/x/y','subscriptionIds': ['sub-1', 'sub-2']])]
+ when: 'the subscriber is removed'
+ objectUnderTest.removeCmNotificationSubscription(DatastoreType.PASSTHROUGH_RUNNING, 'ch-1', '/x/y', 'sub-1')
+ then: 'the list of subscribers is updated'
+ 1 * mockCpsDataService.updateNodeLeaves('NCMP-Admin', 'cm-data-subscriptions',
+ '/datastores/datastore[@name=\'ncmp-datastore:passthrough-running\']/cm-handles/cm-handle[@id=\'ch-1\']/filters',
+ '{"filter":[{"xpath":"/x/y","subscriptionIds":["sub-2"]}]}', _)
+ }
+
+ def 'Removing ongoing subscription with no subscribers'(){
+ given: 'a subscription exists when queried but has no subscribers'
+ def cpsPathQuery = objectUnderTest.CM_SUBSCRIPTION_CPS_PATH_QUERY.formatted('ncmp-datastore:passthrough-running', 'ch-1', '/x/y')
+ mockCpsQueryService.queryDataNodes('NCMP-Admin', 'cm-data-subscriptions',
+ cpsPathQuery, FetchDescendantsOption.OMIT_DESCENDANTS) >> [new DataNode(xpath: cpsPathQuery, leaves: ['xpath': '/x/y','subscriptionIds': []])]
+ when: 'a an ongoing subscription is refreshed'
+ objectUnderTest.removeCmNotificationSubscription(DatastoreType.PASSTHROUGH_RUNNING, 'ch-1', '/x/y', 'sub-1')
+ then: 'the subscription with empty subscriber list is removed'
+ 1 * mockCpsDataService.deleteDataNode('NCMP-Admin', 'cm-data-subscriptions',
+ '/datastores/datastore[@name=\'ncmp-datastore:passthrough-running\']/cm-handles/cm-handle[@id=\'ch-1\']/filters/filter[@xpath=\'/x/y\']',
+ _)
+ }
}
\ No newline at end of file
assert cmNotificationSubscriptionPersistenceService.
getOngoingCmNotificationSubscriptionIds(datastoreType,cmHandleId,xpath).size() == 0
when: 'we add a new cm notification subscription'
- cmNotificationSubscriptionPersistenceService.addOrUpdateCmNotificationSubscription(datastoreType,cmHandleId,xpath,
+ cmNotificationSubscriptionPersistenceService.addCmNotificationSubscription(datastoreType,cmHandleId,xpath,
'subId-1')
then: 'there is an ongoing cm subscription for that CM handle and xpath'
assert cmNotificationSubscriptionPersistenceService.isOngoingCmNotificationSubscription(datastoreType,cmHandleId,xpath)
getOngoingCmNotificationSubscriptionIds(datastoreType,cmHandleId,xpath).size() == 1
}
- def 'Adding a cm notification subscription to an already existing'() {
- given: 'an ongoing cm subscription'
+ def 'Adding a cm notification subscription to the already existing'() {
+ given: 'an ongoing cm subscription with the following details'
def datastoreType = PASSTHROUGH_RUNNING
def cmHandleId = 'ch-1'
def xpath = '/x/y'
- cmNotificationSubscriptionPersistenceService.addOrUpdateCmNotificationSubscription(datastoreType,cmHandleId,xpath,
- 'subId-1')
when: 'a new cm notification subscription is made for the SAME CM handle and xpath'
- cmNotificationSubscriptionPersistenceService.addOrUpdateCmNotificationSubscription(datastoreType,cmHandleId,xpath,
+ cmNotificationSubscriptionPersistenceService.addCmNotificationSubscription(datastoreType,cmHandleId,xpath,
'subId-2')
then: 'it is added to the ongoing list of subscription ids'
def subscriptionIds = cmNotificationSubscriptionPersistenceService.getOngoingCmNotificationSubscriptionIds(datastoreType,cmHandleId,xpath)
and: 'both subscription ids exists for the CM handle and xpath'
assert subscriptionIds.contains("subId-1") && subscriptionIds.contains("subId-2")
}
+
+ def 'Removing cm notification subscriber among other subscribers'() {
+ given: 'an ongoing cm subscription with the following details'
+ def datastoreType = PASSTHROUGH_RUNNING
+ def cmHandleId = 'ch-1'
+ def xpath = '/x/y'
+ and: 'the number of subscribers is as follows'
+ def originalNumberOfSubscribers =
+ cmNotificationSubscriptionPersistenceService.getOngoingCmNotificationSubscriptionIds(datastoreType,cmHandleId,xpath).size()
+ when: 'a subscriber is removed'
+ cmNotificationSubscriptionPersistenceService.removeCmNotificationSubscription(datastoreType,cmHandleId,xpath,'subId-2')
+ then: 'the number of subscribers is reduced by 1'
+ def updatedNumberOfSubscribers =
+ cmNotificationSubscriptionPersistenceService.getOngoingCmNotificationSubscriptionIds(datastoreType,cmHandleId,xpath).size()
+ assert updatedNumberOfSubscribers == originalNumberOfSubscribers-1
+ }
+
+ def 'Removing the LAST cm notification subscriber for a given cm handle, datastore and xpath'() {
+ given: 'an ongoing cm subscription with the following details'
+ def datastoreType = PASSTHROUGH_RUNNING
+ def cmHandleId = 'ch-1'
+ def xpath = '/x/y'
+ and: 'there is only one subscriber'
+ assert cmNotificationSubscriptionPersistenceService
+ .getOngoingCmNotificationSubscriptionIds(datastoreType,cmHandleId,xpath).size() == 1
+ when: 'only subscriber is removed'
+ cmNotificationSubscriptionPersistenceService.removeCmNotificationSubscription(datastoreType,cmHandleId,xpath,'subId-1')
+ then: 'there are no longer any subscriptions for the cm handle, datastore and xpath'
+ assert !cmNotificationSubscriptionPersistenceService.isOngoingCmNotificationSubscription(datastoreType, cmHandleId, xpath)
+ }
+
}
Code Review / cps.git / commitdiff