import lombok.extern.slf4j.Slf4j;
import org.onap.cps.events.EventsPublisher;
import org.onap.cps.ncmp.api.impl.events.cmsubscription.mapper.CmNotificationSubscriptionNcmpOutEventMapper;
-import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.DmiCmNotificationSubscriptionDetails;
import org.onap.cps.ncmp.events.cmsubscription_merge1_0_0.ncmp_to_client.CmNotificationSubscriptionNcmpOutEvent;
import org.onap.cps.utils.JsonObjectMapper;
import org.springframework.beans.factory.annotation.Value;
private final EventsPublisher<CloudEvent> eventsPublisher;
private final JsonObjectMapper jsonObjectMapper;
- private final Map<String, Map<String, DmiCmNotificationSubscriptionDetails>> cmNotificationSubscriptionCache;
private final CmNotificationSubscriptionNcmpOutEventMapper cmNotificationSubscriptionNcmpOutEventMapper;
+ private final DmiCmNotificationSubscriptionCacheHandler dmiCmNotificationSubscriptionCacheHandler;
private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
private static final Map<String, ScheduledFuture<?>> scheduledTasksPerSubscriptionId = new ConcurrentHashMap<>();
* or published now
*/
public void publishCmNotificationSubscriptionNcmpOutEvent(final String subscriptionId, final String eventType,
- final CmNotificationSubscriptionNcmpOutEvent cmNotificationSubscriptionNcmpOutEvent,
- final boolean isScheduledEvent) {
+ final CmNotificationSubscriptionNcmpOutEvent
+ cmNotificationSubscriptionNcmpOutEvent,
+ final boolean isScheduledEvent) {
if (isScheduledEvent && !scheduledTasksPerSubscriptionId.containsKey(subscriptionId)) {
final ScheduledFuture<?> scheduledFuture =
cmNotificationSubscriptionNcmpOutEvent);
log.info("Published CmNotificationSubscriptionEvent on demand for subscriptionId : {}", subscriptionId);
}
-
}
private ScheduledFuture<?> scheduleAndPublishCmNotificationSubscriptionNcmpOutEvent(final String subscriptionId,
- final String eventType) {
+ final String eventType) {
final CmNotificationSubscriptionNcmpOutEventPublishingTask
cmNotificationSubscriptionNcmpOutEventPublishingTask =
new CmNotificationSubscriptionNcmpOutEventPublishingTask(cmNotificationSubscriptionNcmpOutEventTopic,
- subscriptionId, eventType, eventsPublisher, jsonObjectMapper, cmNotificationSubscriptionCache,
- cmNotificationSubscriptionNcmpOutEventMapper);
+ subscriptionId, eventType, eventsPublisher, jsonObjectMapper,
+ cmNotificationSubscriptionNcmpOutEventMapper, dmiCmNotificationSubscriptionCacheHandler);
return scheduledExecutorService.schedule(cmNotificationSubscriptionNcmpOutEventPublishingTask,
cmNotificationSubscriptionDmiOutEventTimeoutInMs, TimeUnit.MILLISECONDS);
}
private void publishCmNotificationSubscriptionNcmpOutEventNow(final String subscriptionId, final String eventType,
- final CmNotificationSubscriptionNcmpOutEvent cmNotificationSubscriptionNcmpOutEvent) {
+ final CmNotificationSubscriptionNcmpOutEvent
+ cmNotificationSubscriptionNcmpOutEvent) {
final CloudEvent cmNotificationSubscriptionNcmpOutEventAsCloudEvent =
buildAndGetCmNotificationNcmpOutEventAsCloudEvent(jsonObjectMapper, subscriptionId, eventType,
cmNotificationSubscriptionNcmpOutEvent);
eventsPublisher.publishCloudEvent(cmNotificationSubscriptionNcmpOutEventTopic, subscriptionId,
cmNotificationSubscriptionNcmpOutEventAsCloudEvent);
+ dmiCmNotificationSubscriptionCacheHandler
+ .removeAcceptedAndRejectedDmiCmNotificationSubscriptionEntries(subscriptionId);
}
protected static CloudEvent buildAndGetCmNotificationNcmpOutEventAsCloudEvent(
final CmNotificationSubscriptionNcmpOutEvent cmNotificationSubscriptionNcmpOutEvent) {
return CloudEventBuilder.v1().withId(UUID.randomUUID().toString()).withType(eventType)
- .withSource(URI.create("NCMP")).withDataSchema(URI.create("org.onap.ncmp.cm.subscription:1.0.0"))
- .withExtension("correlationid", subscriptionId)
- .withData(jsonObjectMapper.asJsonBytes(cmNotificationSubscriptionNcmpOutEvent)).build();
+ .withSource(URI.create("NCMP")).withDataSchema(URI.create("org.onap.ncmp.cm.subscription:1.0.0"))
+ .withExtension("correlationid", subscriptionId)
+ .withData(jsonObjectMapper.asJsonBytes(cmNotificationSubscriptionNcmpOutEvent)).build();
}
}
private final String eventType;
private final EventsPublisher<CloudEvent> eventsPublisher;
private final JsonObjectMapper jsonObjectMapper;
- private final Map<String, Map<String, DmiCmNotificationSubscriptionDetails>> cmNotificationSubscriptionCache;
private final CmNotificationSubscriptionNcmpOutEventMapper cmNotificationSubscriptionNcmpOutEventMapper;
+ private final DmiCmNotificationSubscriptionCacheHandler dmiCmNotificationSubscriptionCacheHandler;
/**
* Delegating the responsibility of publishing CmNotificationSubscriptionNcmpOutEvent as a separate task which will
@Override
public void run() {
final Map<String, DmiCmNotificationSubscriptionDetails> dmiCmNotificationSubscriptionDetailsMap =
- cmNotificationSubscriptionCache.get(subscriptionId);
+ dmiCmNotificationSubscriptionCacheHandler.get(subscriptionId);
final CmNotificationSubscriptionNcmpOutEvent cmNotificationSubscriptionNcmpOutEvent =
cmNotificationSubscriptionNcmpOutEventMapper.toCmNotificationSubscriptionNcmpOutEvent(subscriptionId,
dmiCmNotificationSubscriptionDetailsMap);
eventsPublisher.publishCloudEvent(topicName, subscriptionId,
buildAndGetCmNotificationNcmpOutEventAsCloudEvent(jsonObjectMapper, subscriptionId, eventType,
cmNotificationSubscriptionNcmpOutEvent));
+ dmiCmNotificationSubscriptionCacheHandler
+ .removeAcceptedAndRejectedDmiCmNotificationSubscriptionEntries(subscriptionId);
}
}
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
import lombok.RequiredArgsConstructor;
import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.CmNotificationSubscriptionStatus;
import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.DmiCmNotificationSubscriptionDetails;
/**
* Adds new subscription to the subscription cache.
*
- * @param subscriptionId subscription Id
+ * @param subscriptionId subscription id
* @param predicates subscription request predicates
*/
public void add(final String subscriptionId, final List<Predicate> predicates) {
cmNotificationSubscriptionCache.put(subscriptionId, createDmiCmNotificationSubscriptionsPerDmi(predicates));
}
+ /**
+ * Get cm notification subscription cache entry via subscription id.
+ *
+ * @param subscriptionId subscription id
+ * @return map of dmi cm notification subscriptions per dmi
+ */
+ public Map<String, DmiCmNotificationSubscriptionDetails> get(final String subscriptionId) {
+ return cmNotificationSubscriptionCache.get(subscriptionId);
+ }
+
+
+ /**
+ * Remove cache entries with CmNotificationSubscriptionStatus ACCEPTED/REJECTED via subscription id.
+ *
+ * @param subscriptionId subscription id as key in CM notification Subscription cache.
+ */
+ public void removeAcceptedAndRejectedDmiCmNotificationSubscriptionEntries(final String subscriptionId) {
+ final Map<String, DmiCmNotificationSubscriptionDetails> dmiCmNotificationSubscriptionsPerDmi =
+ cmNotificationSubscriptionCache.get(subscriptionId);
+ final Map<String, DmiCmNotificationSubscriptionDetails> updatedDmiCmNotificationSubscriptionsPerDmi =
+ dmiCmNotificationSubscriptionsPerDmi.entrySet().stream().filter(
+ dmiCmNotificationSubscription ->
+ !isAcceptedOrRejected(dmiCmNotificationSubscription.getValue()))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ cmNotificationSubscriptionCache.put(subscriptionId, updatedDmiCmNotificationSubscriptionsPerDmi);
+ }
+
/**
* Creates map of subscription details per DMI.
*
*
*/
public void updateDmiCmNotificationSubscriptionStatusPerDmi(
- final String subscriptionId, final String dmiServiceName, final CmNotificationSubscriptionStatus status) {
+ final String subscriptionId, final String dmiServiceName, final CmNotificationSubscriptionStatus status) {
cmNotificationSubscriptionCache.get(subscriptionId).get(dmiServiceName)
- .setCmNotificationSubscriptionStatus(status);
+ .setCmNotificationSubscriptionStatus(status);
}
/**
*/
public void persistIntoDatabasePerDmi(final String subscriptionId, final String dmiServiceName) {
final List<DmiCmNotificationSubscriptionPredicate> dmiCmNotificationSubscriptionPredicateList =
- cmNotificationSubscriptionCache.get(subscriptionId).get(dmiServiceName)
- .getDmiCmNotificationSubscriptionPredicates();
+ cmNotificationSubscriptionCache.get(subscriptionId).get(dmiServiceName)
+ .getDmiCmNotificationSubscriptionPredicates();
for (final DmiCmNotificationSubscriptionPredicate dmiCmNotificationSubscriptionPredicate:
- dmiCmNotificationSubscriptionPredicateList) {
+ dmiCmNotificationSubscriptionPredicateList) {
final DatastoreType datastoreType = dmiCmNotificationSubscriptionPredicate.getDatastoreType();
final Set<String> cmHandles = dmiCmNotificationSubscriptionPredicate.getTargetCmHandleIds();
final Set<String> xpaths = dmiCmNotificationSubscriptionPredicate.getXpaths();
}
return targetCmHandlesByDmiServiceNames;
}
+
+ private boolean isAcceptedOrRejected(
+ final DmiCmNotificationSubscriptionDetails dmiCmNotificationSubscription) {
+ return dmiCmNotificationSubscription.getCmNotificationSubscriptionStatus().toString().equals("ACCEPTED")
+ || dmiCmNotificationSubscription.getCmNotificationSubscriptionStatus().toString().equals("REJECTED");
+ }
}
\ No newline at end of file
import io.cloudevents.CloudEvent
import org.onap.cps.events.EventsPublisher
import org.onap.cps.ncmp.api.impl.events.cmsubscription.mapper.CmNotificationSubscriptionNcmpOutEventMapper
-import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.DmiCmNotificationSubscriptionDetails
import org.onap.cps.ncmp.api.impl.events.mapper.CloudEventMapper
import org.onap.cps.ncmp.events.cmsubscription_merge1_0_0.ncmp_to_client.CmNotificationSubscriptionNcmpOutEvent
import org.onap.cps.ncmp.events.cmsubscription_merge1_0_0.ncmp_to_client.Data
def mockEventsPublisher = Mock(EventsPublisher)
def jsonObjectMapper = new JsonObjectMapper(new ObjectMapper())
- def mockCmNotificationSubscriptionCache = Mock(Map<String, Map<String, DmiCmNotificationSubscriptionDetails>>)
def mockCmNotificationSubscriptionNcmpOutEventMapper = Mock(CmNotificationSubscriptionNcmpOutEventMapper)
+ def mockDmiCmNotificationSubscriptionCacheHandler = Mock(DmiCmNotificationSubscriptionCacheHandler)
- def objectUnderTest = new CmNotificationSubscriptionNcmpOutEventProducer(mockEventsPublisher, jsonObjectMapper, mockCmNotificationSubscriptionCache, mockCmNotificationSubscriptionNcmpOutEventMapper)
+ def objectUnderTest = new CmNotificationSubscriptionNcmpOutEventProducer(mockEventsPublisher, jsonObjectMapper,
+ mockCmNotificationSubscriptionNcmpOutEventMapper, mockDmiCmNotificationSubscriptionCacheHandler)
def 'Create and #scenario Cm Notification Subscription NCMP out event'() {
given: 'a cm subscription response for the client'
assert CloudEventMapper.toTargetEvent(cmNotificationSubscriptionNcmpOutEventAsCloudEvent, CmNotificationSubscriptionNcmpOutEvent) == cmNotificationSubscriptionNcmpOutEvent
}
}
+ then: 'the cache handler is called once to remove accepted and rejected entries in cache'
+ 1 * mockDmiCmNotificationSubscriptionCacheHandler.removeAcceptedAndRejectedDmiCmNotificationSubscriptionEntries(subscriptionId)
}
import io.cloudevents.core.builder.CloudEventBuilder
import org.apache.kafka.clients.consumer.ConsumerRecord
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.service.CmNotificationSubscriptionPersistenceService
import org.onap.cps.ncmp.api.impl.inventory.InventoryPersistence
import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle
assert testCache.containsKey(subscriptionId)
}
+ def 'Get cache entry via subscription id'() {
+ given: 'the cache contains value for some-id'
+ testCache.put('some-id',[:])
+ when: 'the get method is called'
+ def result = objectUnderTest.get('some-id')
+ then: 'correct value is returned as expected'
+ assert result == [:]
+ }
+
+ def 'Remove accepted and rejected entries from cache via subscription id'() {
+ given: 'a map as the value for cache entry for some-id'
+ def testMap = [:]
+ testMap.put("dmi-1",
+ new DmiCmNotificationSubscriptionDetails([],CmNotificationSubscriptionStatus.ACCEPTED))
+ testMap.put("dmi-2",
+ new DmiCmNotificationSubscriptionDetails([],CmNotificationSubscriptionStatus.REJECTED))
+ testMap.put("dmi-3",
+ new DmiCmNotificationSubscriptionDetails([],CmNotificationSubscriptionStatus.PENDING))
+ testCache.put("test-id", testMap)
+ assert testCache.get("test-id").size() == 3
+ when: 'the method to remove accepted and rejected entries for test-id is called'
+ objectUnderTest.removeAcceptedAndRejectedDmiCmNotificationSubscriptionEntries("test-id")
+ then: 'all entries with status accepted/rejected are no longer present for test-id'
+ testCache.get("test-id").each { key, testResultMap ->
+ assert testResultMap.cmNotificationSubscriptionStatus != CmNotificationSubscriptionStatus.ACCEPTED
+ || testResultMap.cmNotificationSubscriptionStatus != CmNotificationSubscriptionStatus.REJECTED
+ }
+ and: 'the size of the map for cache entry test-id is as expected'
+ assert testCache.get("test-id").size() == 1
+ }
+
def 'Create map for DMI cm notification subscription per DMI service name'() {
given: 'list of predicates from the create subscription event'
def predicates = cmNotificationSubscriptionNcmpInEvent.getData().getPredicates()