diff options
author | 2023-04-13 21:36:55 +0100 | |
---|---|---|
committer | 2023-04-17 09:36:41 +0000 | |
commit | d6c207d0ade19f19216ebd642d2511a768dfb4ab (patch) | |
tree | fbe44145dee8d594d29ed7750f6bac6dc4da8e3c /cps-ncmp-service/src | |
parent | 3019cdb495e3b2024c9519c9a8ab89c18a150095 (diff) |
[CPS] Removing the duplicate producer code
# Using the existing 'EventPublisher' class and deleting the duplicated producer classes.
Issue-ID: CPS-1557
Signed-off-by: raviteja.karumuri <raviteja.karumuri@est.tech>
Change-Id: I6dd8e9c9d9dbec06a2c9464fc2d382fa89f6fc32
Diffstat (limited to 'cps-ncmp-service/src')
-rw-r--r-- | cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventConsumer.java | 5 | ||||
-rw-r--r-- | cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventProducer.java | 46 | ||||
-rw-r--r-- | cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumer.java | 13 | ||||
-rw-r--r-- | cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avc/AvcEventProducer.java | 56 | ||||
-rw-r--r-- | cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/CpsAsyncRequestResponseEventIntegrationSpec.groovy | 12 | ||||
-rw-r--r-- | cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumerSpec.groovy (renamed from cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventProducerIntegrationSpec.groovy) | 15 |
6 files changed, 29 insertions, 118 deletions
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventConsumer.java index bc6624dee..0ac0fb92d 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventConsumer.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventConsumer.java @@ -22,6 +22,7 @@ package org.onap.cps.ncmp.api.impl.async; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.onap.cps.ncmp.api.impl.events.EventsPublisher; import org.onap.cps.ncmp.event.model.DmiAsyncRequestResponseEvent; import org.onap.cps.ncmp.event.model.NcmpAsyncRequestResponseEvent; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; @@ -37,7 +38,7 @@ import org.springframework.stereotype.Component; @ConditionalOnProperty(name = "notification.enabled", havingValue = "true", matchIfMissing = true) public class NcmpAsyncRequestResponseEventConsumer { - private final NcmpAsyncRequestResponseEventProducer ncmpAsyncRequestResponseEventProducer; + private final EventsPublisher<NcmpAsyncRequestResponseEvent> eventsPublisher; private final NcmpAsyncRequestResponseEventMapper ncmpAsyncRequestResponseEventMapper; /** @@ -53,7 +54,7 @@ public class NcmpAsyncRequestResponseEventConsumer { final NcmpAsyncRequestResponseEvent ncmpAsyncRequestResponseEvent = ncmpAsyncRequestResponseEventMapper.toNcmpAsyncEvent(dmiAsyncRequestResponseEvent); - ncmpAsyncRequestResponseEventProducer.sendMessage( + eventsPublisher.publishEvent(ncmpAsyncRequestResponseEvent.getEventTarget(), ncmpAsyncRequestResponseEvent.getEventId(), ncmpAsyncRequestResponseEvent); } } diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventProducer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventProducer.java deleted file mode 100644 index 8ab6db904..000000000 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventProducer.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * ============LICENSE_START======================================================= - * Copyright (C) 2022 Nordix Foundation - * ================================================================================ - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - * SPDX-License-Identifier: Apache-2.0 - * ============LICENSE_END========================================================= - */ - -package org.onap.cps.ncmp.api.impl.async; - -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.onap.cps.ncmp.event.model.NcmpAsyncRequestResponseEvent; -import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.stereotype.Service; - -@Slf4j -@Service -@RequiredArgsConstructor -public class NcmpAsyncRequestResponseEventProducer { - - private final KafkaTemplate<String, NcmpAsyncRequestResponseEvent> kafkaTemplate; - - - /** - * Sends message to the configured topic with a message key. - * - * @param eventId message key - * @param ncmpAsyncRequestResponseEvent message payload - */ - public void sendMessage(final String eventId, final NcmpAsyncRequestResponseEvent ncmpAsyncRequestResponseEvent) { - kafkaTemplate.send(ncmpAsyncRequestResponseEvent.getEventTarget(), eventId, ncmpAsyncRequestResponseEvent); - } -} diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumer.java index 3b5b5aaa0..83ad5e570 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumer.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumer.java @@ -22,7 +22,9 @@ package org.onap.cps.ncmp.api.impl.events.avc; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.onap.cps.ncmp.api.impl.events.EventsPublisher; import org.onap.cps.ncmp.event.model.AvcEvent; +import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component; @@ -36,7 +38,13 @@ import org.springframework.stereotype.Component; @ConditionalOnProperty(name = "notification.enabled", havingValue = "true", matchIfMissing = true) public class AvcEventConsumer { - private final AvcEventProducer avcEventProducer; + + @Value("${app.ncmp.avc.cm-events-topic}") + private String cmEventsTopicName; + + private final EventsPublisher<AvcEvent> eventsPublisher; + private final AvcEventMapper avcEventMapper; + /** * Consume the specified event. @@ -48,6 +56,7 @@ public class AvcEventConsumer { properties = {"spring.json.value.default.type=org.onap.cps.ncmp.event.model.AvcEvent"}) public void consumeAndForward(final AvcEvent avcEvent) { log.debug("Consuming AVC event {} ...", avcEvent); - avcEventProducer.sendMessage(avcEvent); + final AvcEvent outgoingAvcEvent = avcEventMapper.toOutgoingAvcEvent(avcEvent); + eventsPublisher.publishEvent(cmEventsTopicName, outgoingAvcEvent.getEventId(), outgoingAvcEvent); } } diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avc/AvcEventProducer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avc/AvcEventProducer.java deleted file mode 100644 index 15cbeb889..000000000 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avc/AvcEventProducer.java +++ /dev/null @@ -1,56 +0,0 @@ -/* - * ============LICENSE_START======================================================= - * Copyright (C) 2023 Nordix Foundation - * ================================================================================ - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - * SPDX-License-Identifier: Apache-2.0 - * ============LICENSE_END========================================================= - */ - -package org.onap.cps.ncmp.api.impl.events.avc; - -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.onap.cps.ncmp.event.model.AvcEvent; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.stereotype.Service; - -/** - * Producer for AVC events. - */ -@Slf4j -@Service -@RequiredArgsConstructor -public class AvcEventProducer { - - private final KafkaTemplate<String, AvcEvent> kafkaTemplate; - - private final AvcEventMapper avcEventMapper; - - @Value("${app.ncmp.avc.cm-events-topic}") - private String cmEventsTopic; - - /** - * Sends message to the configured topic with a message key. - * - * @param incomingAvcEvent message payload - */ - public void sendMessage(final AvcEvent incomingAvcEvent) { - // generate new event id while keeping other data - final AvcEvent outgoingAvcEvent = avcEventMapper.toOutgoingAvcEvent(incomingAvcEvent); - log.debug("Forwarding AVC event {} to topic {} ", outgoingAvcEvent.getEventId(), cmEventsTopic); - kafkaTemplate.send(cmEventsTopic, outgoingAvcEvent.getEventId(), outgoingAvcEvent); - } -} diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/CpsAsyncRequestResponseEventIntegrationSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/CpsAsyncRequestResponseEventIntegrationSpec.groovy index 51162c74e..bcf75a29b 100644 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/CpsAsyncRequestResponseEventIntegrationSpec.groovy +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/CpsAsyncRequestResponseEventIntegrationSpec.groovy @@ -1,6 +1,6 @@ /* * ============LICENSE_START======================================================= - * Copyright (c) 2022 Nordix Foundation. + * Copyright (c) 2022-2023 Nordix Foundation. * ================================================================================ * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,6 +23,7 @@ package org.onap.cps.ncmp.api.impl.async import com.fasterxml.jackson.databind.ObjectMapper import org.apache.kafka.clients.consumer.KafkaConsumer import org.mapstruct.factory.Mappers +import org.onap.cps.ncmp.api.impl.events.EventsPublisher import org.onap.cps.ncmp.api.kafka.MessagingBaseSpec import org.onap.cps.ncmp.event.model.DmiAsyncRequestResponseEvent import org.onap.cps.ncmp.event.model.NcmpAsyncRequestResponseEvent @@ -36,14 +37,15 @@ import org.testcontainers.spock.Testcontainers import java.time.Duration -@SpringBootTest(classes = [NcmpAsyncRequestResponseEventProducer, NcmpAsyncRequestResponseEventConsumer, ObjectMapper, JsonObjectMapper]) +@SpringBootTest(classes = [EventsPublisher, NcmpAsyncRequestResponseEventConsumer, ObjectMapper, JsonObjectMapper]) @Testcontainers @DirtiesContext class NcmpAsyncRequestResponseEventProducerIntegrationSpec extends MessagingBaseSpec { @SpringBean - NcmpAsyncRequestResponseEventProducer cpsAsyncRequestResponseEventProducerService = - new NcmpAsyncRequestResponseEventProducer(kafkaTemplate); + EventsPublisher cpsAsyncRequestResponseEventPublisher = + new EventsPublisher<NcmpAsyncRequestResponseEvent>(kafkaTemplate); + @SpringBean NcmpAsyncRequestResponseEventMapper ncmpAsyncRequestResponseEventMapper = @@ -51,7 +53,7 @@ class NcmpAsyncRequestResponseEventProducerIntegrationSpec extends MessagingBase @SpringBean NcmpAsyncRequestResponseEventConsumer ncmpAsyncRequestResponseEventConsumer = - new NcmpAsyncRequestResponseEventConsumer(cpsAsyncRequestResponseEventProducerService, + new NcmpAsyncRequestResponseEventConsumer(cpsAsyncRequestResponseEventPublisher, ncmpAsyncRequestResponseEventMapper) @Autowired diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventProducerIntegrationSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumerSpec.groovy index 64d1f64e5..d57527a45 100644 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventProducerIntegrationSpec.groovy +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumerSpec.groovy @@ -23,6 +23,7 @@ package org.onap.cps.ncmp.api.impl.events.avc import com.fasterxml.jackson.databind.ObjectMapper import org.apache.kafka.clients.consumer.KafkaConsumer import org.mapstruct.factory.Mappers +import org.onap.cps.ncmp.api.impl.events.EventsPublisher import org.onap.cps.ncmp.api.kafka.MessagingBaseSpec import org.onap.cps.ncmp.event.model.AvcEvent import org.onap.cps.ncmp.utils.TestUtils @@ -35,19 +36,19 @@ import org.testcontainers.spock.Testcontainers import java.time.Duration -@SpringBootTest(classes = [AvcEventProducer, AvcEventConsumer, ObjectMapper, JsonObjectMapper]) +@SpringBootTest(classes = [EventsPublisher, AvcEventConsumer, ObjectMapper, JsonObjectMapper]) @Testcontainers @DirtiesContext -class AvcEventProducerIntegrationSpec extends MessagingBaseSpec { +class AvcEventConsumerSpec extends MessagingBaseSpec { @SpringBean AvcEventMapper avcEventMapper = Mappers.getMapper(AvcEventMapper.class) @SpringBean - AvcEventProducer avcEventProducer = new AvcEventProducer(kafkaTemplate, avcEventMapper) + EventsPublisher eventsPublisher = new EventsPublisher<AvcEvent>(kafkaTemplate) @SpringBean - AvcEventConsumer acvEventConsumer = new AvcEventConsumer(avcEventProducer) + AvcEventConsumer acvEventConsumer = new AvcEventConsumer(eventsPublisher, avcEventMapper) @Autowired JsonObjectMapper jsonObjectMapper @@ -56,9 +57,9 @@ class AvcEventProducerIntegrationSpec extends MessagingBaseSpec { def 'Consume and forward valid message'() { given: 'consumer has a subscription on a topic' - def cmEventsTopic = 'cm-events' - avcEventProducer.cmEventsTopic = cmEventsTopic - kafkaConsumer.subscribe([cmEventsTopic] as List<String>) + def cmEventsTopicName = 'cm-events' + acvEventConsumer.cmEventsTopicName = cmEventsTopicName + kafkaConsumer.subscribe([cmEventsTopicName] as List<String>) and: 'an event is sent' def jsonData = TestUtils.getResourceFileContent('sampleAvcInputEvent.json') def testEventSent = jsonObjectMapper.convertJsonString(jsonData, AvcEvent.class) |