diff options
Diffstat (limited to 'cps-ncmp-service/src')
6 files changed, 47 insertions, 32 deletions
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumer.java index 0207fb90e3..9e90eabbc4 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumer.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumer.java @@ -21,8 +21,6 @@ package org.onap.cps.ncmp.impl.cmnotificationsubscription.cmavc; import io.cloudevents.CloudEvent; -import io.cloudevents.core.builder.CloudEventBuilder; -import java.util.UUID; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -56,11 +54,8 @@ public class CmAvcEventConsumer { containerFactory = "cloudEventConcurrentKafkaListenerContainerFactory") public void consumeAndForward( final ConsumerRecord<String, CloudEvent> cmAvcEventAsConsumerRecord) { - log.debug("Consuming AVC event {} ...", cmAvcEventAsConsumerRecord.value()); - final String newEventId = UUID.randomUUID().toString(); - final CloudEvent outgoingAvcEvent = - CloudEventBuilder.from(cmAvcEventAsConsumerRecord.value()).withId(newEventId) - .build(); - eventsPublisher.publishCloudEvent(cmEventsTopicName, newEventId, outgoingAvcEvent); + final CloudEvent outgoingAvcEvent = cmAvcEventAsConsumerRecord.value(); + log.debug("Consuming AVC event {} ...", outgoingAvcEvent); + eventsPublisher.publishCloudEvent(cmEventsTopicName, outgoingAvcEvent.getId(), outgoingAvcEvent); } } diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/CmSubscriptionHandlerImpl.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/CmSubscriptionHandlerImpl.java index c2c71dbaae..1cdc7ed3e0 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/CmSubscriptionHandlerImpl.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/CmSubscriptionHandlerImpl.java @@ -154,7 +154,7 @@ public class CmSubscriptionHandlerImpl implements CmSubscriptionHandler { dmiSubscriptionDetails.getDmiCmSubscriptionPredicates()); if (dmiCmSubscriptionPredicates.isEmpty()) { - acceptAndPublishNcmpOutEventPerDmi(subscriptionId, dmiPluginName); + acceptAndPersistCmSubscriptionPerDmi(subscriptionId, dmiPluginName); } else { publishDmiInEventPerDmi(subscriptionId, dmiPluginName, dmiCmSubscriptionPredicates); } @@ -168,7 +168,7 @@ public class CmSubscriptionHandlerImpl implements CmSubscriptionHandler { "subscriptionCreateRequest", dmiInEvent); } - private void acceptAndPublishNcmpOutEventPerDmi(final String subscriptionId, final String dmiPluginName) { + private void acceptAndPersistCmSubscriptionPerDmi(final String subscriptionId, final String dmiPluginName) { dmiCacheHandler.updateDmiSubscriptionStatus(subscriptionId, dmiPluginName, CmSubscriptionStatus.ACCEPTED); dmiCacheHandler.persistIntoDatabasePerDmi(subscriptionId, dmiPluginName); diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/NcmpOutEventProducer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/NcmpOutEventProducer.java index 01d720937f..3371d59f7a 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/NcmpOutEventProducer.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/NcmpOutEventProducer.java @@ -57,7 +57,8 @@ public class NcmpOutEventProducer { private final NcmpOutEventMapper ncmpOutEventMapper; private final DmiCacheHandler dmiCacheHandler; private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(); - private static final Map<String, ScheduledFuture<?>> scheduledTasksPerSubscriptionId = new ConcurrentHashMap<>(); + private static final Map<String, ScheduledFuture<?>> scheduledTasksPerSubscriptionIdAndEventType = + new ConcurrentHashMap<>(); /** * Publish the event to the client who requested the subscription with key as subscription id and event is Cloud @@ -73,14 +74,20 @@ public class NcmpOutEventProducer { public void publishNcmpOutEvent(final String subscriptionId, final String eventType, final NcmpOutEvent ncmpOutEvent, final boolean isScheduledEvent) { - if (isScheduledEvent && !scheduledTasksPerSubscriptionId.containsKey(subscriptionId)) { + final String taskKey = subscriptionId.concat(eventType); + + if (isScheduledEvent && !scheduledTasksPerSubscriptionIdAndEventType.containsKey(taskKey)) { final ScheduledFuture<?> scheduledFuture = scheduleAndPublishNcmpOutEvent(subscriptionId, eventType); - scheduledTasksPerSubscriptionId.putIfAbsent(subscriptionId, scheduledFuture); - log.debug("Scheduled the CmNotificationSubscriptionEvent for subscriptionId : {}", subscriptionId); + scheduledTasksPerSubscriptionIdAndEventType.putIfAbsent(taskKey, scheduledFuture); + log.debug("Scheduled the Cm Subscription Event for subscriptionId : {} and eventType : {}", subscriptionId, + eventType); } else { - cancelScheduledTaskForSubscriptionId(subscriptionId); - publishNcmpOutEventNow(subscriptionId, eventType, ncmpOutEvent); - log.info("Published CmNotificationSubscriptionEvent on demand for subscriptionId : {}", subscriptionId); + cancelScheduledTask(taskKey); + if (ncmpOutEvent != null) { + publishNcmpOutEventNow(subscriptionId, eventType, ncmpOutEvent); + log.debug("Published Cm Subscription Event on demand for subscriptionId : {} and eventType : {}", + subscriptionId, eventType); + } } } @@ -92,12 +99,12 @@ public class NcmpOutEventProducer { TimeUnit.MILLISECONDS); } - private void cancelScheduledTaskForSubscriptionId(final String subscriptionId) { + private void cancelScheduledTask(final String taskKey) { - final ScheduledFuture<?> scheduledFuture = scheduledTasksPerSubscriptionId.get(subscriptionId); + final ScheduledFuture<?> scheduledFuture = scheduledTasksPerSubscriptionIdAndEventType.get(taskKey); if (scheduledFuture != null) { scheduledFuture.cancel(true); - scheduledTasksPerSubscriptionId.remove(subscriptionId); + scheduledTasksPerSubscriptionIdAndEventType.remove(taskKey); } } @@ -106,10 +113,8 @@ public class NcmpOutEventProducer { private void publishNcmpOutEventNow(final String subscriptionId, final String eventType, final NcmpOutEvent ncmpOutEvent) { final CloudEvent ncmpOutEventAsCloudEvent = - buildAndGetNcmpOutEventAsCloudEvent(jsonObjectMapper, subscriptionId, eventType, - ncmpOutEvent); - eventsPublisher.publishCloudEvent(ncmpOutEventTopic, subscriptionId, - ncmpOutEventAsCloudEvent); + buildAndGetNcmpOutEventAsCloudEvent(jsonObjectMapper, subscriptionId, eventType, ncmpOutEvent); + eventsPublisher.publishCloudEvent(ncmpOutEventTopic, subscriptionId, ncmpOutEventAsCloudEvent); dmiCacheHandler.removeAcceptedAndRejectedDmiSubscriptionEntries(subscriptionId); } diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumerSpec.groovy index a8b5250edf..06651be913 100644 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumerSpec.groovy +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/cmnotificationsubscription/cmavc/CmAvcEventConsumerSpec.groovy @@ -37,7 +37,6 @@ import org.springframework.beans.factory.annotation.Autowired import org.springframework.boot.test.context.SpringBootTest import org.springframework.test.annotation.DirtiesContext import org.testcontainers.spock.Testcontainers - import java.time.Duration import static org.onap.cps.ncmp.utils.events.CloudEventMapper.toTargetEvent @@ -86,8 +85,8 @@ class CmAvcEventConsumerSpec extends MessagingBaseSpec { def convertedAvcEvent = toTargetEvent(cloudEvent, AvcEvent.class) and: 'we have correct headers forwarded where correlation id matches' assert KafkaHeaders.getParsedKafkaHeader(record.headers(), 'ce_correlationid') == 'test-cmhandle1' - and: 'event id differs(as per requirement) between consumed and forwarded' - assert KafkaHeaders.getParsedKafkaHeader(record.headers(), 'ce_id') != 'sample-eventid' + and: 'event id is same between consumed and forwarded' + assert KafkaHeaders.getParsedKafkaHeader(record.headers(), 'ce_id') == 'sample-eventid' and: 'the event payload still matches' assert testEventSent == convertedAvcEvent } diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/NcmpOutEventProducerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/NcmpOutEventProducerSpec.groovy index e03682d8c9..afa2e9874e 100644 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/NcmpOutEventProducerSpec.groovy +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/impl/cmnotificationsubscription/ncmp/NcmpOutEventProducerSpec.groovy @@ -83,5 +83,24 @@ class NcmpOutEventProducerSpec extends Specification { 1 * mockDmiCacheHandler.removeAcceptedAndRejectedDmiSubscriptionEntries(subscriptionId) } + def 'No event published when NCMP out event is null'() { + given: 'a cm subscription response for the client' + def subscriptionId = 'test-subscription-id-3' + def eventType = 'subscriptionCreateResponse' + def ncmpOutEvent = null + and: 'also we have target topic for publishing to client' + objectUnderTest.ncmpOutEventTopic = 'client-test-topic' + and: 'a deadline to an event' + objectUnderTest.dmiOutEventTimeoutInMs = 1000 + when: 'the event is scheduled to be published' + objectUnderTest.publishNcmpOutEvent(subscriptionId, eventType, ncmpOutEvent, true) + then: 'we wait for 10ms and then we receive response from DMI' + Thread.sleep(10) + and: 'we receive NO response from DMI so we publish the message on demand' + objectUnderTest.publishNcmpOutEvent(subscriptionId, eventType, ncmpOutEvent, false) + and: 'no event published' + 0 * mockEventsPublisher.publishCloudEvent(*_) + } + } diff --git a/cps-ncmp-service/src/test/resources/sampleAvcInputEvent.json b/cps-ncmp-service/src/test/resources/sampleAvcInputEvent.json index 5b297c86c2..1dc14bd65e 100644 --- a/cps-ncmp-service/src/test/resources/sampleAvcInputEvent.json +++ b/cps-ncmp-service/src/test/resources/sampleAvcInputEvent.json @@ -10,11 +10,8 @@ "operation":"replace", "target":"ran-network:ran-network/NearRTRIC[@id='22']/GNBCUCPFunction[@id='cucpserver2']/NRCellCU[@id='15549']/NRCellRelation[@id='14427']", "value":{ - "attributes":[ - { - "isHoAllowed":true - } - ] + "color": "yellow", + "name": "Apple" } }, { |