2 * ============LICENSE_START=======================================================
3 * Copyright (C) 2024 Nordix Foundation
4 * ================================================================================
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
17 * SPDX-License-Identifier: Apache-2.0
18 * ============LICENSE_END=========================================================
21 package org.onap.cps.ncmp.api.impl.events.cmsubscription;
23 import io.cloudevents.CloudEvent;
24 import io.cloudevents.core.builder.CloudEventBuilder;
27 import java.util.UUID;
28 import java.util.concurrent.ConcurrentHashMap;
29 import java.util.concurrent.Executors;
30 import java.util.concurrent.ScheduledExecutorService;
31 import java.util.concurrent.ScheduledFuture;
32 import java.util.concurrent.TimeUnit;
33 import lombok.RequiredArgsConstructor;
34 import lombok.extern.slf4j.Slf4j;
35 import org.onap.cps.events.EventsPublisher;
36 import org.onap.cps.ncmp.api.impl.events.cmsubscription.mapper.CmNotificationSubscriptionNcmpOutEventMapper;
37 import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.DmiCmNotificationSubscriptionDetails;
38 import org.onap.cps.ncmp.events.cmsubscription_merge1_0_0.ncmp_to_client.CmNotificationSubscriptionNcmpOutEvent;
39 import org.onap.cps.utils.JsonObjectMapper;
40 import org.springframework.beans.factory.annotation.Value;
41 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
42 import org.springframework.stereotype.Component;
46 @RequiredArgsConstructor
47 @ConditionalOnProperty(name = "notification.enabled", havingValue = "true", matchIfMissing = true)
48 public class CmNotificationSubscriptionNcmpOutEventProducer {
50 @Value("${app.ncmp.avc.subscription-outcome-topic}")
51 private String cmNotificationSubscriptionNcmpOutEventTopic;
53 @Value("${ncmp.timers.subscription-forwarding.dmi-response-timeout-ms}")
54 private Integer cmNotificationSubscriptionDmiOutEventTimeoutInMs;
56 private final EventsPublisher<CloudEvent> eventsPublisher;
57 private final JsonObjectMapper jsonObjectMapper;
58 private final Map<String, Map<String, DmiCmNotificationSubscriptionDetails>> cmNotificationSubscriptionCache;
59 private final CmNotificationSubscriptionNcmpOutEventMapper cmNotificationSubscriptionNcmpOutEventMapper;
60 private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
61 private static final Map<String, ScheduledFuture<?>> scheduledTasksPerSubscriptionId = new ConcurrentHashMap<>();
64 * Publish the event to the client who requested the subscription with key as subscription id and event is Cloud
67 * @param subscriptionId Cm Subscription Id
68 * @param eventType Type of event
69 * @param cmNotificationSubscriptionNcmpOutEvent Cm Notification Subscription Event for the
71 * @param isScheduledEvent Determines if the event is to be scheduled
74 public void publishCmNotificationSubscriptionNcmpOutEvent(final String subscriptionId, final String eventType,
75 final CmNotificationSubscriptionNcmpOutEvent cmNotificationSubscriptionNcmpOutEvent,
76 final boolean isScheduledEvent) {
78 if (isScheduledEvent && !scheduledTasksPerSubscriptionId.containsKey(subscriptionId)) {
79 final ScheduledFuture<?> scheduledFuture =
80 scheduleAndPublishCmNotificationSubscriptionNcmpOutEvent(subscriptionId, eventType);
81 scheduledTasksPerSubscriptionId.putIfAbsent(subscriptionId, scheduledFuture);
82 log.debug("Scheduled the CmNotificationSubscriptionEvent for subscriptionId : {}", subscriptionId);
84 cancelScheduledTaskForSubscriptionId(subscriptionId);
85 publishCmNotificationSubscriptionNcmpOutEventNow(subscriptionId, eventType,
86 cmNotificationSubscriptionNcmpOutEvent);
87 log.info("Published CmNotificationSubscriptionEvent on demand for subscriptionId : {}", subscriptionId);
92 private ScheduledFuture<?> scheduleAndPublishCmNotificationSubscriptionNcmpOutEvent(final String subscriptionId,
93 final String eventType) {
94 final CmNotificationSubscriptionNcmpOutEventPublishingTask
95 cmNotificationSubscriptionNcmpOutEventPublishingTask =
96 new CmNotificationSubscriptionNcmpOutEventPublishingTask(cmNotificationSubscriptionNcmpOutEventTopic,
97 subscriptionId, eventType, eventsPublisher, jsonObjectMapper, cmNotificationSubscriptionCache,
98 cmNotificationSubscriptionNcmpOutEventMapper);
99 return scheduledExecutorService.schedule(cmNotificationSubscriptionNcmpOutEventPublishingTask,
100 cmNotificationSubscriptionDmiOutEventTimeoutInMs, TimeUnit.MILLISECONDS);
103 private void cancelScheduledTaskForSubscriptionId(final String subscriptionId) {
105 final ScheduledFuture<?> scheduledFuture = scheduledTasksPerSubscriptionId.get(subscriptionId);
106 if (scheduledFuture != null) {
107 scheduledFuture.cancel(true);
108 scheduledTasksPerSubscriptionId.remove(subscriptionId);
114 private void publishCmNotificationSubscriptionNcmpOutEventNow(final String subscriptionId, final String eventType,
115 final CmNotificationSubscriptionNcmpOutEvent cmNotificationSubscriptionNcmpOutEvent) {
116 final CloudEvent cmNotificationSubscriptionNcmpOutEventAsCloudEvent =
117 buildAndGetCmNotificationNcmpOutEventAsCloudEvent(jsonObjectMapper, subscriptionId, eventType,
118 cmNotificationSubscriptionNcmpOutEvent);
119 eventsPublisher.publishCloudEvent(cmNotificationSubscriptionNcmpOutEventTopic, subscriptionId,
120 cmNotificationSubscriptionNcmpOutEventAsCloudEvent);
123 protected static CloudEvent buildAndGetCmNotificationNcmpOutEventAsCloudEvent(
124 final JsonObjectMapper jsonObjectMapper, final String subscriptionId, final String eventType,
125 final CmNotificationSubscriptionNcmpOutEvent cmNotificationSubscriptionNcmpOutEvent) {
127 return CloudEventBuilder.v1().withId(UUID.randomUUID().toString()).withType(eventType)
128 .withSource(URI.create("NCMP")).withDataSchema(URI.create("org.onap.ncmp.cm.subscription:1.0.0"))
129 .withExtension("correlationid", subscriptionId)
130 .withData(jsonObjectMapper.asJsonBytes(cmNotificationSubscriptionNcmpOutEvent)).build();