diff options
4 files changed, 51 insertions, 7 deletions
diff --git a/cps-application/src/main/resources/application.yml b/cps-application/src/main/resources/application.yml index bba8f09eb0..802da9e87c 100644 --- a/cps-application/src/main/resources/application.yml +++ b/cps-application/src/main/resources/application.yml @@ -99,7 +99,7 @@ app: topic: ${NCMP_ASYNC_M2M_TOPIC:ncmp-async-m2m} avc: subscription-topic: ${NCMP_CM_AVC_SUBSCRIPTION:cm-avc-subscription} - subscription-forward-topic: ${NCMP_FORWARD_CM_AVC_SUBSCRIPTION:ncmp-dmi-cm-avc-subscription} + subscription-forward-topic-prefix: ${NCMP_FORWARD_CM_AVC_SUBSCRIPTION:ncmp-dmi-cm-avc-subscription-} subscription-response-topic: ${NCMP_RESPONSE_CM_AVC_SUBSCRIPTION:dmi-ncmp-cm-avc-subscription} subscription-outcome-topic: ${NCMP_OUTCOME_CM_AVC_SUBSCRIPTION:cm-avc-subscription-response} cm-events-topic: ${NCMP_CM_EVENTS_TOPIC:cm-events} diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java index 4afa051d30..4654b148c6 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarder.java @@ -56,7 +56,8 @@ public class SubscriptionEventForwarder { private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(); - private static final String DMI_AVC_SUBSCRIPTION_TOPIC_PREFIX = "ncmp-dmi-cm-avc-subscription-"; + @Value("${app.ncmp.avc.subscription-forward-topic-prefix}") + private String dmiAvcSubscriptionTopicPrefix; @Value("${ncmp.timers.subscription-forwarding.dmi-response-timeout-ms:30000}") private int dmiResponseTimeoutInMs; @@ -82,8 +83,15 @@ public class SubscriptionEventForwarder { = DmiServiceNameOrganizer.getDmiPropertiesPerCmHandleIdPerServiceName(yangModelCmHandles); final Set<String> dmisToRespond = new HashSet<>(dmiPropertiesPerCmHandleIdPerServiceName.keySet()); - startResponseTimeout(subscriptionEvent, dmisToRespond); - forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, subscriptionEvent); + if (dmisToRespond.isEmpty()) { + log.info("placeholder to create full outcome response for subscriptionEventId: {}.", + subscriptionEvent.getEvent().getSubscription().getClientID() + + subscriptionEvent.getEvent().getSubscription().getName()); + //TODO outcome response with no cmhandles + } else { + startResponseTimeout(subscriptionEvent, dmisToRespond); + forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, subscriptionEvent); + } } private void forwardEventToDmis(final Map<String, Map<String, Map<String, String>>> dmiNameCmHandleMap, @@ -91,8 +99,8 @@ public class SubscriptionEventForwarder { dmiNameCmHandleMap.forEach((dmiName, cmHandlePropertiesMap) -> { subscriptionEvent.getEvent().getPredicates().setTargets(Collections.singletonList(cmHandlePropertiesMap)); final String eventKey = createEventKey(subscriptionEvent, dmiName); - eventsPublisher.publishEvent( - DMI_AVC_SUBSCRIPTION_TOPIC_PREFIX + dmiName, eventKey, subscriptionEvent); + final String dmiAvcSubscriptionTopic = dmiAvcSubscriptionTopicPrefix + dmiName; + eventsPublisher.publishEvent(dmiAvcSubscriptionTopic, eventKey, subscriptionEvent); }); } diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarderSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarderSpec.groovy index 457eb6fa99..a3dec29ede 100644 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarderSpec.groovy +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventForwarderSpec.groovy @@ -113,6 +113,41 @@ class SubscriptionEventForwarderSpec extends MessagingBaseSpec { 'wildcard' | ['CMHandle*'] } + def 'Forward valid CM create subscription where targets are not associated to any existing CMHandles'() { + given: 'an event' + def jsonData = TestUtils.getResourceFileContent('avcSubscriptionCreationEvent.json') + def testEventSent = jsonObjectMapper.convertJsonString(jsonData, SubscriptionEvent.class) + and: 'the InventoryPersistence returns no private properties for the supplied CM Handles' + 1 * mockInventoryPersistence.getYangModelCmHandles(["CMHandle1", "CMHandle2", "CMHandle3"]) >> [] + and: 'the thread creation delay is reduced to 2 seconds for testing' + objectUnderTest.dmiResponseTimeoutInMs = 2000 + and: 'a Blocking Variable is used for the Asynchronous call with a timeout of 5 seconds' + def block = new BlockingVariable<Object>(5) + when: 'the valid event is forwarded' + objectUnderTest.forwardCreateSubscriptionEvent(testEventSent) + then: 'the event is not added to the forwarded subscription event cache' + 0 * mockForwardedSubscriptionEventCache.put("SCO-9989752cm-subscription-001", ["DMIName1", "DMIName2"] as Set) + and: 'the event is forwarded twice with the CMHandle private properties and provides a valid listenable future' + 0 * mockSubscriptionEventPublisher.publishEvent("ncmp-dmi-cm-avc-subscription-DMIName1", "SCO-9989752-cm-subscription-001-DMIName1", + subscriptionEvent -> { + Map targets = subscriptionEvent.getEvent().getPredicates().getTargets().get(0) + targets["CMHandle1"] == ["shape":"circle"] + targets["CMHandle2"] == ["shape":"square"] + } + ) + 0 * mockSubscriptionEventPublisher.publishEvent("ncmp-dmi-cm-avc-subscription-DMIName2", "SCO-9989752-cm-subscription-001-DMIName2", + subscriptionEvent -> { + Map targets = subscriptionEvent.getEvent().getPredicates().getTargets().get(0) + targets["CMHandle3"] == ["shape":"triangle"] + } + ) + and: 'a separate thread has been created where the map is polled' + 0 * mockForwardedSubscriptionEventCache.containsKey("SCO-9989752cm-subscription-001") >> true + 0 * mockForwardedSubscriptionEventCache.get(_) + and: 'the subscription id is removed from the event cache map returning the asynchronous blocking variable' + 0 * mockForwardedSubscriptionEventCache.remove("SCO-9989752cm-subscription-001") >> {block.set(_)} + } + static def createYangModelCmHandleWithDmiProperty(id, dmiId,propertyName, propertyValue) { return new YangModelCmHandle(id:"CMHandle" + id, dmiDataServiceName: "DMIName" + dmiId, dmiProperties: [new YangModelCmHandle.Property(propertyName,propertyValue)]) } diff --git a/cps-ncmp-service/src/test/resources/application.yml b/cps-ncmp-service/src/test/resources/application.yml index 66194ad9e5..1016f2b033 100644 --- a/cps-ncmp-service/src/test/resources/application.yml +++ b/cps-ncmp-service/src/test/resources/application.yml @@ -1,5 +1,5 @@ # ============LICENSE_START======================================================= -# Copyright (C) 2021-2022 Nordix Foundation +# Copyright (C) 2021-2023 Nordix Foundation # ================================================================================ # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -21,6 +21,7 @@ app: avc: subscription-topic: cm-avc-subscription cm-events-topic: cm-events + subscription-forward-topic-prefix: ${NCMP_FORWARD_CM_AVC_SUBSCRIPTION:ncmp-dmi-cm-avc-subscription-} ncmp: dmi: |