topic: ${DMI_DEVICE_HEARTBEAT_TOPIC:dmi-device-heartbeat}
cps:
data-updated:
- change-event-notifications-enabled: ${CPS_CHANGE_EVENT_NOTIFICATIONS_ENABLED:true}
+ change-event-notifications-enabled: ${CPS_CHANGE_EVENT_NOTIFICATIONS_ENABLED:false}
topic: ${CPS_CHANGE_EVENT_TOPIC:cps-data-updated-events}
-
-
notification:
enabled: true
async:
@Value("${app.cps.data-updated.topic:cps-data-updated-events}")
private String topicName;
- @Value("${app.cps.data-updated.change-event-notifications-enabled:true}")
+ @Value("${app.cps.data-updated.change-event-notifications-enabled:false}")
private boolean cpsChangeEventNotificationsEnabled;
@Value("${notification.enabled:false}")
package org.onap.cps.impl;
+import static org.onap.cps.api.parameters.FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS;
+
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.onap.cps.api.exceptions.DataNodeNotFoundException;
import org.onap.cps.api.model.Anchor;
import org.onap.cps.api.model.DataNode;
-import org.onap.cps.api.parameters.FetchDescendantsOption;
import org.onap.cps.cpspath.parser.CpsPathUtil;
import org.onap.cps.spi.CpsDataPersistenceService;
import org.onap.cps.utils.ContentType;
final Anchor anchor = cpsAnchorService.getAnchor(ADMIN_DATASPACE, ANCHOR_NAME);
final Collection<DataNode> dataNodes =
- buildDataNodesWithParentNodeXpath(anchor, xpath, notificationSubscriptionAsJson, ContentType.JSON);
- cpsDataPersistenceService.addListElements(ADMIN_DATASPACE, ANCHOR_NAME, xpath,
- dataNodes);
+ buildDataNodesWithParentNodeXpath(anchor, xpath, notificationSubscriptionAsJson);
+ cpsDataPersistenceService.addListElements(ADMIN_DATASPACE, ANCHOR_NAME, xpath, dataNodes);
}
@Override
@Override
public List<Map<String, Object>> getNotificationSubscription(final String xpath) {
final Collection<DataNode> dataNodes =
- cpsDataPersistenceService.getDataNodes(ADMIN_DATASPACE, ANCHOR_NAME, xpath,
- FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS);
+ cpsDataPersistenceService.getDataNodes(ADMIN_DATASPACE, ANCHOR_NAME, xpath, INCLUDE_ALL_DESCENDANTS);
final List<Map<String, Object>> dataMaps = new ArrayList<>(dataNodes.size());
final Anchor anchor = cpsAnchorService.getAnchor(ADMIN_DATASPACE, ANCHOR_NAME);
for (final DataNode dataNode: dataNodes) {
@Override
public boolean isNotificationEnabled(final String dataspaceName, final String anchorName) {
return (isNotificationEnabledForAnchor(dataspaceName, anchorName)
- || notificationEnabledForAllAnchors(dataspaceName));
+ || notificationEnabledForAllAnchors(dataspaceName));
}
private boolean isNotificationEnabledForAnchor(final String dataspaceName, final String anchorName) {
private boolean isNotificationEnabledForXpath(final String xpath) {
try {
- cpsDataPersistenceService.getDataNodes(ADMIN_DATASPACE, ANCHOR_NAME, xpath,
- FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS);
+ cpsDataPersistenceService.getDataNodes(ADMIN_DATASPACE, ANCHOR_NAME, xpath, INCLUDE_ALL_DESCENDANTS);
} catch (final DataNodeNotFoundException e) {
return false;
}
private boolean notificationEnabledForAllAnchors(final String dataspaceName) {
final String dataspaceSubscriptionXpath = String.format(DATASPACE_SUBSCRIPTION_XPATH_FORMAT, dataspaceName);
- return (isNotificationEnabledForXpath(dataspaceSubscriptionXpath)
- && noIndividualAnchorEnabledInDataspace(dataspaceName));
+ return isNotificationEnabledForXpath(dataspaceSubscriptionXpath)
+ && noIndividualAnchorEnabledInDataspace(dataspaceName);
}
private boolean noIndividualAnchorEnabledInDataspace(final String dataspaceName) {
return !isNotificationEnabledForXpath(xpathForAnchors);
}
-
- private Collection<DataNode> buildDataNodesWithParentNodeXpath(final Anchor anchor, final String parentNodeXpath,
- final String nodeData,
- final ContentType contentType) {
-
+ private Collection<DataNode> buildDataNodesWithParentNodeXpath(final Anchor anchor,
+ final String parentNodeXpath,
+ final String nodeData) {
final String normalizedParentNodeXpath = CpsPathUtil.getNormalizedXpath(parentNodeXpath);
final ContainerNode containerNode =
- yangParser.parseData(contentType, nodeData, anchor, normalizedParentNodeXpath);
- final Collection<DataNode> dataNodes = new DataNodeBuilder()
+ yangParser.parseData(ContentType.JSON, nodeData, anchor, normalizedParentNodeXpath);
+ return new DataNodeBuilder()
.withParentNodeXpath(normalizedParentNodeXpath)
.withContainerNode(containerNode)
.buildCollection();
- return dataNodes;
}
-}
\ No newline at end of file
+}
package org.onap.cps.events
-import static org.onap.cps.events.model.Data.Operation.CREATE
-import static org.onap.cps.events.model.Data.Operation.DELETE
-import static org.onap.cps.events.model.Data.Operation.UPDATE
-
-import org.onap.cps.api.CpsNotificationService
import com.fasterxml.jackson.databind.ObjectMapper
import io.cloudevents.CloudEvent
import io.cloudevents.core.CloudEventUtils
import io.cloudevents.jackson.PojoCloudEventDataMapper
-import org.onap.cps.events.model.CpsDataUpdatedEvent
+import org.onap.cps.api.CpsNotificationService
import org.onap.cps.api.model.Anchor
+import org.onap.cps.events.model.CpsDataUpdatedEvent
import org.onap.cps.utils.JsonObjectMapper
import org.springframework.test.context.ContextConfiguration
import spock.lang.Specification
import java.time.OffsetDateTime
+import static org.onap.cps.events.model.Data.Operation.CREATE
+import static org.onap.cps.events.model.Data.Operation.DELETE
+import static org.onap.cps.events.model.Data.Operation.UPDATE
+
@ContextConfiguration(classes = [ObjectMapper, JsonObjectMapper])
class CpsDataUpdateEventsServiceSpec extends Specification {
def mockEventsPublisher = Mock(EventsPublisher)
def setup() {
mockCpsNotificationService.isNotificationEnabled('dataspace01', 'anchor01') >> true
+ objectUnderTest.topicName = 'cps-core-event'
}
- def 'Create and Publish cps update event where events are #scenario'() {
+ def 'Create and Publish cps update event where events are #scenario.'() {
given: 'an anchor, operation and observed timestamp'
def anchor = new Anchor('anchor01', 'dataspace01', 'schema01');
def operation = operationInRequest
and: 'cpsChangeEventNotificationsEnabled is also true'
objectUnderTest.cpsChangeEventNotificationsEnabled = true
when: 'service is called to publish data update event'
- objectUnderTest.topicName = "cps-core-event"
objectUnderTest.publishCpsDataUpdateEvent(anchor, xpath, operation, observedTimestamp)
then: 'the event contains the required attributes'
1 * mockEventsPublisher.publishCloudEvent('cps-core-event', 'dataspace01:anchor01', _) >> {
'non root node xpath and delete operation' | '/test/path' | DELETE || UPDATE
}
- def 'publish cps update event when #scenario'() {
- given: 'an anchor, operation and observed timestamp'
- def anchor = new Anchor('anchor01', 'dataspace01', 'schema01');
- def operation = CREATE
- def observedTimestamp = OffsetDateTime.now()
- and: 'notificationsEnabled is #notificationsEnabled'
- objectUnderTest.notificationsEnabled = notificationsEnabled
- and: 'cpsChangeEventNotificationsEnabled is #cpsChangeEventNotificationsEnabled'
- objectUnderTest.cpsChangeEventNotificationsEnabled = cpsChangeEventNotificationsEnabled
- when: 'service is called to publish data update event'
- objectUnderTest.topicName = "cps-core-event"
- objectUnderTest.publishCpsDataUpdateEvent(anchor, '/', operation, observedTimestamp)
- then: 'the event contains the required attributes'
- expectedCallToPublisher * mockEventsPublisher.publishCloudEvent('cps-core-event', 'dataspace01:anchor01', _)
- where: 'below scenarios are present'
- scenario | notificationsEnabled | cpsChangeEventNotificationsEnabled || expectedCallToPublisher
- 'both notifications enabled' | true | true || 1
- 'both notifications disabled' | false | false || 0
- 'only CPS change event notification enabled' | false | true || 0
- 'only overall notification enabled' | true | false || 0
-
- }
-
- def 'publish cps update event when no timestamp provided'() {
+ def 'Publish cps update event when no timestamp provided.'() {
given: 'an anchor, operation and null timestamp'
def anchor = new Anchor('anchor01', 'dataspace01', 'schema01');
- def operation = CREATE
def observedTimestamp = null
and: 'notificationsEnabled is true'
objectUnderTest.notificationsEnabled = true
and: 'cpsChangeEventNotificationsEnabled is true'
objectUnderTest.cpsChangeEventNotificationsEnabled = true
when: 'service is called to publish data update event'
- objectUnderTest.topicName = "cps-core-event"
- objectUnderTest.publishCpsDataUpdateEvent(anchor, '/', operation, observedTimestamp)
+ objectUnderTest.publishCpsDataUpdateEvent(anchor, '/', CREATE, observedTimestamp)
then: 'the event is published'
1 * mockEventsPublisher.publishCloudEvent('cps-core-event', 'dataspace01:anchor01', _)
}
+
+ def 'Enabling and disabling publish cps update events.'() {
+ given: 'a different anchor'
+ def anchor = new Anchor('anchor02', 'some dataspace', 'some schema');
+ and: 'notificationsEnabled is #notificationsEnabled'
+ objectUnderTest.notificationsEnabled = notificationsEnabled
+ and: 'cpsChangeEventNotificationsEnabled is #cpsChangeEventNotificationsEnabled'
+ objectUnderTest.cpsChangeEventNotificationsEnabled = cpsChangeEventNotificationsEnabled
+ and: 'notification service enabled is: #cpsNotificationServiceisNotificationEnabled'
+ mockCpsNotificationService.isNotificationEnabled(_, 'anchor02') >> cpsNotificationServiceisNotificationEnabled
+ when: 'service is called to publish data update event'
+ objectUnderTest.publishCpsDataUpdateEvent(anchor, '/', CREATE, null)
+ then: 'the event is only published when all related flags are true'
+ expectedCallsToPublisher * mockEventsPublisher.publishCloudEvent(*_)
+ where: 'the following flags are used'
+ notificationsEnabled | cpsChangeEventNotificationsEnabled | cpsNotificationServiceisNotificationEnabled || expectedCallsToPublisher
+ false | true | true || 0
+ true | false | true || 0
+ true | true | false || 0
+ true | true | true || 1
+ }
+
}
def anchorName = 'cps-notification-subscriptions'
def schemaSetName = 'cps-notification-subscriptions'
def anchor = new Anchor(anchorName, dataspaceName, schemaSetName)
+ def someDataNode = new DataNodeBuilder().withXpath('/xpath-1').build()
def mockCpsDataPersistenceService = Mock(CpsDataPersistenceService)
def mockCpsAnchorService = Mock(CpsAnchorService)
def mockYangTextSchemaSourceSetCache = Mock(YangTextSchemaSourceSetCache)
def mockTimedYangTextSchemaSourceSetBuilder = Mock(TimedYangTextSchemaSourceSetBuilder)
def yangParser = new YangParser(new YangParserHelper(), mockYangTextSchemaSourceSetCache, mockTimedYangTextSchemaSourceSetBuilder)
- def mockPrefixResolver = Mock(PrefixResolver)
+ def mockPrefixResolver = Mock(PrefixResolver)
+
def objectUnderTest = new CpsNotificationServiceImpl(mockCpsAnchorService, mockCpsDataPersistenceService, yangParser, mockPrefixResolver)
def 'add notification subscription for list of dataspaces'() {
def 'is notification enabled for given anchor'() {
given: 'data nodes available for given anchor'
- mockCpsDataPersistenceService.getDataNodes(dataspaceName, anchorName, "/dataspaces", FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >>
- [new DataNodeBuilder().withXpath('/xpath-1').build()]
+ mockCpsDataPersistenceService.getDataNodes(dataspaceName, anchorName, "/dataspaces", FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >> [someDataNode]
when: 'is notification enabled is called'
boolean isNotificationEnabled = objectUnderTest.isNotificationEnabled(dataspaceName, anchorName)
then: 'the notification is enabled'
assert !isNotificationEnabled
}
+ def 'is notification enabled for given anchor because all anchors are enabled'() {
+ given: 'data nodes not available for given anchor'
+ mockCpsDataPersistenceService.getDataNodes(dataspaceName, anchorName, "/dataspaces/dataspace[@name='ds01']/anchors/anchor[@name='anchor-01']", FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >>
+ { throw new DataNodeNotFoundException(dataspaceName, anchorName) }
+ and: 'data nodes not available for any specific anchor'
+ mockCpsDataPersistenceService.getDataNodes(dataspaceName, anchorName, "/dataspaces/dataspace[@name='ds01']/anchors", FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >>
+ { throw new DataNodeNotFoundException(dataspaceName, anchorName) }
+ when: 'is notification enabled is called'
+ boolean isNotificationEnabled = objectUnderTest.isNotificationEnabled('ds01', 'anchor-01')
+ then: 'the notification is enabled'
+ assert isNotificationEnabled
+ }
+
def 'is notification enabled for all anchors in a dataspace'() {
given: 'data nodes available for given dataspace'
mockCpsDataPersistenceService.getDataNodes(dataspaceName, anchorName, "/dataspaces/dataspace[@name='ds01']", FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >>
- [new DataNodeBuilder().withXpath('/xpath-1').build()]
+ [someDataNode]
and: 'data nodes not available for any specific anchor'
mockCpsDataPersistenceService.getDataNodes(dataspaceName, anchorName, "/dataspaces/dataspace[@name='ds01']/anchors", FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >>
- { throw new DataNodeNotFoundException(dataspaceName, anchorName) }
+ { throw new DataNodeNotFoundException(dataspaceName, anchorName) }
when: 'is notification enabled is called'
boolean isNotificationEnabled = objectUnderTest.notificationEnabledForAllAnchors('ds01')
then: 'the notification is enabled'
assert isNotificationEnabled
}
- def 'is notification disabled for all anchors in a dataspace'() {
+ def 'is notification disabled for a dataspace'() {
+ given: 'No data nodes available for given dataspace'
+ mockCpsDataPersistenceService.getDataNodes(dataspaceName, anchorName, "/dataspaces/dataspace[@name='ds01']", FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >>
+ { throw new DataNodeNotFoundException(dataspaceName, anchorName) }
+ when: 'is notification enabled is called'
+ boolean isNotificationEnabled = objectUnderTest.notificationEnabledForAllAnchors('ds01')
+ then: 'the notification is disabled'
+ assert !isNotificationEnabled
+ }
+
+ def 'is notification disabled for some anchors in a dataspace'() {
given: 'data nodes available for given dataspace'
mockCpsDataPersistenceService.getDataNodes(dataspaceName, anchorName, "/dataspaces/dataspace[@name='ds01']", FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >>
- [new DataNodeBuilder().withXpath('/xpath-1').build()]
+ [someDataNode]
and: 'data nodes also available for any specific anchor'
mockCpsDataPersistenceService.getDataNodes(dataspaceName, anchorName, "/dataspaces/dataspace[@name='ds01']/anchors", FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) >>
- [new DataNodeBuilder().withXpath('/xpath-1').build()]
+ [someDataNode]
when: 'is notification enabled is called'
boolean isNotificationEnabled = objectUnderTest.notificationEnabledForAllAnchors('ds01')
then: 'the notification is disabled'
Code Review / cps.git / commitdiff