import static org.onap.cps.ncmp.api.impl.events.mapper.CloudEventMapper.toTargetEvent;
import io.cloudevents.CloudEvent;
+import java.util.List;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.onap.cps.ncmp.api.impl.events.cmsubscription.service.CmNotificationSubscriptionHandlerService;
import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.client_to_ncmp.CmNotificationSubscriptionNcmpInEvent;
+import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.client_to_ncmp.Predicate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
cmNotificationSubscriptionNcmpInEvent.getData().getSubscriptionId());
final String subscriptionId = cmNotificationSubscriptionNcmpInEvent.getData().getSubscriptionId();
+ final List<Predicate> predicates = cmNotificationSubscriptionNcmpInEvent.getData().getPredicates();
if ("subscriptionCreateRequest".equals(cloudEvent.getType())) {
- log.info("Subscription for source {} with subscription id {} ...", cloudEvent.getSource(), subscriptionId);
- cmNotificationSubscriptionHandlerService.processSubscriptionCreateRequest(
- cmNotificationSubscriptionNcmpInEvent);
+ log.info("Subscription create request for source {} with subscription id {} ...",
+ cloudEvent.getSource(), subscriptionId);
+ cmNotificationSubscriptionHandlerService.processSubscriptionCreateRequest(subscriptionId, predicates);
+ }
+ if ("subscriptionDeleteRequest".equals(cloudEvent.getType())) {
+ log.info("Subscription delete request for source {} with subscription id {} ...",
+ cloudEvent.getSource(), subscriptionId);
+ cmNotificationSubscriptionHandlerService.processSubscriptionDeleteRequest(subscriptionId, predicates);
}
}
}
\ No newline at end of file
package org.onap.cps.ncmp.api.impl.events.cmsubscription.service;
-import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.client_to_ncmp.CmNotificationSubscriptionNcmpInEvent;
+import java.util.List;
+import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.client_to_ncmp.Predicate;
public interface CmNotificationSubscriptionHandlerService {
/**
- * Process cm notification subscription request.
+ * Process cm notification subscription create request.
*
- * @param cmNotificationSubscriptionNcmpInEvent CM Notification Subscription event
+ * @param subscriptionId subscription id
+ * @param predicates subscription predicates
*/
- void processSubscriptionCreateRequest(
- final CmNotificationSubscriptionNcmpInEvent cmNotificationSubscriptionNcmpInEvent);
+ void processSubscriptionCreateRequest(final String subscriptionId, final List<Predicate> predicates);
-}
+ /**
+ * Process cm notification subscription delete request.
+ *
+ * @param subscriptionId subscription id
+ * @param predicates subscription predicates
+ */
+ void processSubscriptionDeleteRequest(final String subscriptionId, final List<Predicate> predicates);
+
+}
\ No newline at end of file
import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.CmNotificationSubscriptionStatus;
import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.DmiCmNotificationSubscriptionDetails;
import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.DmiCmNotificationSubscriptionPredicate;
-import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.client_to_ncmp.CmNotificationSubscriptionNcmpInEvent;
import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.client_to_ncmp.Predicate;
import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.ncmp_to_dmi.CmNotificationSubscriptionDmiInEvent;
import org.onap.cps.ncmp.events.cmsubscription_merge1_0_0.ncmp_to_client.CmNotificationSubscriptionNcmpOutEvent;
private final DmiCmNotificationSubscriptionCacheHandler dmiCmNotificationSubscriptionCacheHandler;
@Override
- public void processSubscriptionCreateRequest(
- final CmNotificationSubscriptionNcmpInEvent cmNotificationSubscriptionNcmpInEvent) {
- final String subscriptionId = cmNotificationSubscriptionNcmpInEvent.getData().getSubscriptionId();
- final List<Predicate> predicates = cmNotificationSubscriptionNcmpInEvent.getData().getPredicates();
-
+ public void processSubscriptionCreateRequest(final String subscriptionId, final List<Predicate> predicates) {
if (cmNotificationSubscriptionPersistenceService.isUniqueSubscriptionId(subscriptionId)) {
dmiCmNotificationSubscriptionCacheHandler.add(subscriptionId, predicates);
handleCmNotificationSubscriptionDelta(subscriptionId);
- scheduleCmNotificationSubscriptionNcmpOutEventResponse(subscriptionId);
+ scheduleCmNotificationSubscriptionNcmpOutEventResponse(subscriptionId,
+ "subscriptionCreateResponse");
} else {
rejectAndPublishCmNotificationSubscriptionCreateRequest(subscriptionId, predicates);
}
}
- private void scheduleCmNotificationSubscriptionNcmpOutEventResponse(final String subscriptionId) {
+ @Override
+ public void processSubscriptionDeleteRequest(final String subscriptionId, final List<Predicate> predicates) {
+ dmiCmNotificationSubscriptionCacheHandler.add(subscriptionId, predicates);
+ sendSubscriptionDeleteRequestToDmi(subscriptionId);
+ scheduleCmNotificationSubscriptionNcmpOutEventResponse(subscriptionId, "subscriptionDeleteResponse");
+ }
+
+ private void scheduleCmNotificationSubscriptionNcmpOutEventResponse(final String subscriptionId,
+ final String eventType) {
cmNotificationSubscriptionEventsHandler.publishCmNotificationSubscriptionNcmpOutEvent(subscriptionId,
- "subscriptionCreateResponse", null, true);
+ eventType, null, true);
}
private void rejectAndPublishCmNotificationSubscriptionCreateRequest(final String subscriptionId,
- final List<Predicate> predicates) {
+
final List<Predicate> predicates) {
final Set<String> subscriptionTargetFilters =
predicates.stream().flatMap(predicate -> predicate.getTargetFilter().stream())
.collect(Collectors.toSet());
}
private void publishCmNotificationSubscriptionDmiInEventPerDmi(final String subscriptionId,
- final String dmiPluginName,
- final List<DmiCmNotificationSubscriptionPredicate> dmiCmNotificationSubscriptionPredicates) {
+ final String dmiPluginName,
+ final List<DmiCmNotificationSubscriptionPredicate>
+ dmiCmNotificationSubscriptionPredicates) {
final CmNotificationSubscriptionDmiInEvent cmNotificationSubscriptionDmiInEvent =
cmNotificationSubscriptionMappersHandler.toCmNotificationSubscriptionDmiInEvent(
dmiCmNotificationSubscriptionPredicates);
}
private void acceptAndPublishCmNotificationSubscriptionNcmpOutEventPerDmi(final String subscriptionId,
- final String dmiPluginName) {
+ final String dmiPluginName) {
dmiCmNotificationSubscriptionCacheHandler.updateDmiCmNotificationSubscriptionStatusPerDmi(subscriptionId,
dmiPluginName, CmNotificationSubscriptionStatus.ACCEPTED);
dmiCmNotificationSubscriptionCacheHandler.persistIntoDatabasePerDmi(subscriptionId, dmiPluginName);
}
-}
+
+ private void sendSubscriptionDeleteRequestToDmi(final String subscriptionId) {
+ final Map<String, DmiCmNotificationSubscriptionDetails> dmiCmNotificationSubscriptionDetailsMap =
+ dmiCmNotificationSubscriptionCacheHandler.get(subscriptionId);
+ dmiCmNotificationSubscriptionDetailsMap.forEach((dmiPluginName, dmiCmNotificationSubscriptionDetails) -> {
+ final CmNotificationSubscriptionDmiInEvent cmNotificationSubscriptionDmiInEvent =
+ cmNotificationSubscriptionMappersHandler.toCmNotificationSubscriptionDmiInEvent(
+ dmiCmNotificationSubscriptionDetails.getDmiCmNotificationSubscriptionPredicates());
+ cmNotificationSubscriptionEventsHandler.publishCmNotificationSubscriptionDmiInEvent(subscriptionId,
+ dmiPluginName, "subscriptionDeleteRequest", cmNotificationSubscriptionDmiInEvent);
+ });
+ }
+}
\ No newline at end of file
def loggingEvent = getLoggingEvent()
assert loggingEvent.level == Level.INFO
and: 'the log indicates the task completed successfully'
- assert loggingEvent.formattedMessage == 'Subscription for source some-resource with subscription id test-id ...'
+ assert loggingEvent.formattedMessage == 'Subscription create request for source some-resource with subscription id test-id ...'
and: 'the subscription handler service is called once'
- 1 * mockCmNotificationSubscriptionHandlerService.processSubscriptionCreateRequest(_)
+ 1 * mockCmNotificationSubscriptionHandlerService.processSubscriptionCreateRequest('test-id',_)
}
+ def 'Consume valid CmNotificationSubscriptionNcmpInEvent delete message'() {
+ given: 'a cmNotificationSubscription event'
+ def jsonData = TestUtils.getResourceFileContent('cmSubscription/cmNotificationSubscriptionNcmpInEvent.json')
+ def testEventSent = jsonObjectMapper.convertJsonString(jsonData, CmNotificationSubscriptionNcmpInEvent.class)
+ def testCloudEventSent = CloudEventBuilder.v1()
+ .withData(objectMapper.writeValueAsBytes(testEventSent))
+ .withId('sub-id')
+ .withType('subscriptionDeleteRequest')
+ .withSource(URI.create('some-resource'))
+ .withExtension('correlationid', 'test-cmhandle1').build()
+ def consumerRecord = new ConsumerRecord<String, CloudEvent>('topic-name', 0, 0, 'event-key', testCloudEventSent)
+ when: 'the valid event is consumed'
+ objectUnderTest.consumeSubscriptionEvent(consumerRecord)
+ then: 'an event is logged with level INFO'
+ def loggingEvent = getLoggingEvent()
+ assert loggingEvent.level == Level.INFO
+ and: 'the log indicates the task completed successfully'
+ assert loggingEvent.formattedMessage == 'Subscription delete request for source some-resource with subscription id test-id ...'
+ and: 'the subscription handler service is called once'
+ 1 * mockCmNotificationSubscriptionHandlerService.processSubscriptionDeleteRequest('test-id',_)
+ }
+
+
def getLoggingEvent() {
return logger.list[1]
}
def testEventConsumed = jsonObjectMapper.convertJsonString(jsonData, CmNotificationSubscriptionNcmpInEvent.class)
def testListOfDeltaPredicates = [new DmiCmNotificationSubscriptionPredicate(['ch1'].toSet(), DatastoreType.PASSTHROUGH_OPERATIONAL, ['/a/b'].toSet())]
mockCmNotificationSubscriptionPersistenceService.isUniqueSubscriptionId("test-id") >> true
+ and: 'relevant details is extracted from the event'
+ def subscriptionId = testEventConsumed.getData().getSubscriptionId()
+ def predicates = testEventConsumed.getData().getPredicates()
and: 'the cache handler returns for relevant subscription id'
1 * mockDmiCmNotificationSubscriptionCacheHandler.get("test-id") >> testSubscriptionDetailsMap
and: 'the delta predicates is returned'
1 * mockCmNotificationSubscriptionMappersHandler
.toCmNotificationSubscriptionDmiInEvent(testListOfDeltaPredicates) >> testDmiInEvent
when: 'the valid and unique event is consumed'
- objectUnderTest.processSubscriptionCreateRequest(testEventConsumed)
+ objectUnderTest.processSubscriptionCreateRequest(subscriptionId, predicates)
then: 'the subscription cache handler is called once'
1 * mockDmiCmNotificationSubscriptionCacheHandler.add('test-id',_)
and: 'the events handler method to publish DMI event is called correct number of times with the correct parameters'
and: 'the delta predicates is returned'
1 * mockCmNotificationSubscriptionDelta.getDelta(_) >> noDeltaPredicates
when: 'the valid and unique event is consumed'
- objectUnderTest.processSubscriptionCreateRequest(testEventConsumed)
+ objectUnderTest.processSubscriptionCreateRequest('test-id', noDeltaPredicates)
then: 'the subscription cache handler is called once'
1 * mockDmiCmNotificationSubscriptionCacheHandler.add('test-id', _)
and: 'the subscription details are updated in the cache'
def jsonData = TestUtils.getResourceFileContent('cmSubscription/cmNotificationSubscriptionNcmpInEvent.json')
def testEventConsumed = jsonObjectMapper.convertJsonString(jsonData, CmNotificationSubscriptionNcmpInEvent.class)
mockCmNotificationSubscriptionPersistenceService.isUniqueSubscriptionId('test-id') >> false
+ and: 'relevant details is extracted from the event'
+ def subscriptionId = testEventConsumed.getData().getSubscriptionId()
+ def predicates = testEventConsumed.getData().getPredicates()
and: 'the NCMP out in event mapper returns an event for rejected request'
def testNcmpOutEvent = new CmNotificationSubscriptionNcmpOutEvent()
1 * mockCmNotificationSubscriptionMappersHandler.toCmNotificationSubscriptionNcmpOutEventForRejectedRequest(
"test-id",_) >> testNcmpOutEvent
when: 'the valid but non-unique event is consumed'
- objectUnderTest.processSubscriptionCreateRequest(testEventConsumed)
+ objectUnderTest.processSubscriptionCreateRequest(subscriptionId, predicates)
then: 'the events handler method to publish DMI event is never called'
0 * mockCmNotificationSubscriptionEventsHandler.publishCmNotificationSubscriptionDmiInEvent(_,_,_,_)
and: 'the events handler method to publish NCMP out event is called once'
1 * mockCmNotificationSubscriptionEventsHandler.publishCmNotificationSubscriptionNcmpOutEvent(
'test-id', 'subscriptionCreateResponse', testNcmpOutEvent, false)
}
+
+ def 'Consume valid CmNotificationSubscriptionNcmpInEvent delete message'() {
+ given: 'a cmNotificationSubscriptionNcmp in event for delete'
+ def jsonData = TestUtils.getResourceFileContent('cmSubscription/cmNotificationSubscriptionNcmpInEvent.json')
+ def testEventConsumed = jsonObjectMapper.convertJsonString(jsonData, CmNotificationSubscriptionNcmpInEvent.class)
+ and: 'relevant details is extracted from the event'
+ def subscriptionId = testEventConsumed.getData().getSubscriptionId()
+ def predicates = testEventConsumed.getData().getPredicates()
+ and: 'the cache handler returns for relevant subscription id'
+ 1 * mockDmiCmNotificationSubscriptionCacheHandler.get('test-id') >> testSubscriptionDetailsMap
+ when: 'the valid and unique event is consumed'
+ objectUnderTest.processSubscriptionDeleteRequest(subscriptionId, predicates)
+ then: 'the subscription cache handler is called once'
+ 1 * mockDmiCmNotificationSubscriptionCacheHandler.add('test-id', predicates)
+ and: 'the mapper handler to get DMI in event is called once'
+ 1 * mockCmNotificationSubscriptionMappersHandler.toCmNotificationSubscriptionDmiInEvent(_)
+ and: 'the events handler method to publish DMI event is called correct number of times with the correct parameters'
+ testSubscriptionDetailsMap.size() * mockCmNotificationSubscriptionEventsHandler.publishCmNotificationSubscriptionDmiInEvent(
+ 'test-id', 'dmi-1', 'subscriptionDeleteRequest', _)
+ and: 'we schedule to send the response after configured time from the cache'
+ 1 * mockCmNotificationSubscriptionEventsHandler.publishCmNotificationSubscriptionNcmpOutEvent(
+ 'test-id', 'subscriptionDeleteResponse', null, true)
+ }
}
Code Review / cps.git / commitdiff