Reject create request with duplicated subscriptionId
[cps.git] / cps-ncmp-service / src / main / java / org / onap / cps / ncmp / api / impl / events / cmsubscription / CmNotificationSubscriptionDmiOutEventConsumer.java
1 /*
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
8  *
9  *        http://www.apache.org/licenses/LICENSE-2.0
10  *
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.
16  *
17  *  SPDX-License-Identifier: Apache-2.0
18  *  ============LICENSE_END=========================================================
19  */
20
21 package org.onap.cps.ncmp.api.impl.events.cmsubscription;
22
23 import static org.onap.cps.ncmp.api.impl.events.mapper.CloudEventMapper.toTargetEvent;
24
25 import io.cloudevents.CloudEvent;
26 import lombok.RequiredArgsConstructor;
27 import lombok.extern.slf4j.Slf4j;
28 import org.apache.kafka.clients.consumer.ConsumerRecord;
29 import org.onap.cps.ncmp.api.impl.events.cmsubscription.model.CmNotificationSubscriptionStatus;
30 import org.onap.cps.ncmp.events.cmnotificationsubscription_merge1_0_0.dmi_to_ncmp.CmNotificationSubscriptionDmiOutEvent;
31 import org.springframework.kafka.annotation.KafkaListener;
32 import org.springframework.stereotype.Component;
33
34 @Component
35 @Slf4j
36 @RequiredArgsConstructor
37 public class CmNotificationSubscriptionDmiOutEventConsumer {
38
39     private final DmiCmNotificationSubscriptionCacheHandler dmiCmNotificationSubscriptionCacheHandler;
40
41     /**
42      * Consume the Cm Notification Subscription event from the dmi-plugin.
43      *
44      * @param cmNotificationSubscriptionDmiOutEventConsumerRecord the event to be consumed
45      */
46     @KafkaListener(topics = "${app.ncmp.avc.subscription-response-topic}",
47             containerFactory = "cloudEventConcurrentKafkaListenerContainerFactory")
48     public void consumeCmNotificationSubscriptionDmiOutEvent(
49             final ConsumerRecord<String, CloudEvent> cmNotificationSubscriptionDmiOutEventConsumerRecord) {
50         final CloudEvent cloudEvent = cmNotificationSubscriptionDmiOutEventConsumerRecord.value();
51         final CmNotificationSubscriptionDmiOutEvent cmNotificationSubscriptionDmiOutEvent =
52                 toTargetEvent(cloudEvent, CmNotificationSubscriptionDmiOutEvent.class);
53         final String correlationId = String.valueOf(cloudEvent.getExtension("correlationid"));
54         if ("subscriptionCreateResponse".equals(cloudEvent.getType()) && cmNotificationSubscriptionDmiOutEvent != null
55                     && correlationId != null) {
56             handleCmSubscriptionCreate(correlationId, cmNotificationSubscriptionDmiOutEvent);
57         }
58     }
59
60     private void handleCmSubscriptionCreate(final String correlationId,
61             final CmNotificationSubscriptionDmiOutEvent cmNotificationSubscriptionDmiOutEvent) {
62         final String subscriptionId = correlationId.split("#")[0];
63         final String dmiPluginName = correlationId.split("#")[1];
64
65         if ("ACCEPTED".equals(cmNotificationSubscriptionDmiOutEvent.getData().getStatusMessage())) {
66             handleCacheStatusPerDmi(subscriptionId, dmiPluginName, CmNotificationSubscriptionStatus.ACCEPTED);
67             dmiCmNotificationSubscriptionCacheHandler.persistIntoDatabasePerDmi(subscriptionId, dmiPluginName);
68         }
69
70         if ("REJECTED".equals(cmNotificationSubscriptionDmiOutEvent.getData().getStatusMessage())) {
71             handleCacheStatusPerDmi(subscriptionId, dmiPluginName, CmNotificationSubscriptionStatus.REJECTED);
72         }
73
74         log.info("Cm Subscription with id : {} handled by the dmi-plugin : {} has the status : {}", subscriptionId,
75                 dmiPluginName, cmNotificationSubscriptionDmiOutEvent.getData().getStatusMessage());
76     }
77
78     private void handleCacheStatusPerDmi(final String subscriptionId, final String dmiPluginName,
79                                          final CmNotificationSubscriptionStatus cmNotificationSubscriptionStatus) {
80         dmiCmNotificationSubscriptionCacheHandler.updateDmiCmNotificationSubscriptionStatusPerDmi(subscriptionId,
81             dmiPluginName, cmNotificationSubscriptionStatus);
82     }
83 }