From d6c207d0ade19f19216ebd642d2511a768dfb4ab Mon Sep 17 00:00:00 2001 From: "raviteja.karumuri" Date: Thu, 13 Apr 2023 21:36:55 +0100 Subject: [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 Change-Id: I6dd8e9c9d9dbec06a2c9464fc2d382fa89f6fc32 --- .../NcmpAsyncRequestResponseEventConsumer.java | 5 +- .../NcmpAsyncRequestResponseEventProducer.java | 46 ------------ .../ncmp/api/impl/events/avc/AvcEventConsumer.java | 13 +++- .../ncmp/api/impl/events/avc/AvcEventProducer.java | 56 -------------- ...AsyncRequestResponseEventIntegrationSpec.groovy | 12 +-- .../impl/events/avc/AvcEventConsumerSpec.groovy | 85 ++++++++++++++++++++++ .../avc/AvcEventProducerIntegrationSpec.groovy | 84 --------------------- 7 files changed, 106 insertions(+), 195 deletions(-) delete mode 100644 cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventProducer.java delete mode 100644 cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avc/AvcEventProducer.java create mode 100644 cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumerSpec.groovy delete mode 100644 cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventProducerIntegrationSpec.groovy (limited to 'cps-ncmp-service/src') 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 bc6624dee6..0ac0fb92de 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 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 8ab6db9045..0000000000 --- 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 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 3b5b5aaa0c..83ad5e5704 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 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 15cbeb8891..0000000000 --- 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 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 51162c74e0..bcf75a29b2 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(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/AvcEventConsumerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumerSpec.groovy new file mode 100644 index 0000000000..d57527a454 --- /dev/null +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventConsumerSpec.groovy @@ -0,0 +1,85 @@ +/* + * ============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 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 +import org.onap.cps.utils.JsonObjectMapper +import org.spockframework.spring.SpringBean +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.boot.test.context.SpringBootTest +import org.springframework.test.annotation.DirtiesContext +import org.testcontainers.spock.Testcontainers + +import java.time.Duration + +@SpringBootTest(classes = [EventsPublisher, AvcEventConsumer, ObjectMapper, JsonObjectMapper]) +@Testcontainers +@DirtiesContext +class AvcEventConsumerSpec extends MessagingBaseSpec { + + @SpringBean + AvcEventMapper avcEventMapper = Mappers.getMapper(AvcEventMapper.class) + + @SpringBean + EventsPublisher eventsPublisher = new EventsPublisher(kafkaTemplate) + + @SpringBean + AvcEventConsumer acvEventConsumer = new AvcEventConsumer(eventsPublisher, avcEventMapper) + + @Autowired + JsonObjectMapper jsonObjectMapper + + def kafkaConsumer = new KafkaConsumer<>(consumerConfigProperties('ncmp-group')) + + def 'Consume and forward valid message'() { + given: 'consumer has a subscription on a topic' + def cmEventsTopicName = 'cm-events' + acvEventConsumer.cmEventsTopicName = cmEventsTopicName + kafkaConsumer.subscribe([cmEventsTopicName] as List) + and: 'an event is sent' + def jsonData = TestUtils.getResourceFileContent('sampleAvcInputEvent.json') + def testEventSent = jsonObjectMapper.convertJsonString(jsonData, AvcEvent.class) + when: 'the event is consumed' + acvEventConsumer.consumeAndForward(testEventSent) + and: 'the topic is polled' + def records = kafkaConsumer.poll(Duration.ofMillis(1500)) + then: 'poll returns one record' + assert records.size() == 1 + and: 'record can be converted to AVC event' + def record = records.iterator().next() + def convertedAvcEvent = jsonObjectMapper.convertJsonString(record.value(), AvcEvent) + and: 'consumed forwarded NCMP event id differs from DMI event id' + assert testEventSent.eventId != convertedAvcEvent.getEventId() + and: 'correlation id matches' + assert testEventSent.eventCorrelationId == convertedAvcEvent.getEventCorrelationId() + and: 'timestamps match' + assert testEventSent.eventTime == convertedAvcEvent.getEventTime() + and: 'target matches' + assert testEventSent.eventSource == convertedAvcEvent.getEventSource() + } + +} \ No newline at end of file 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/AvcEventProducerIntegrationSpec.groovy deleted file mode 100644 index 64d1f64e54..0000000000 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/events/avc/AvcEventProducerIntegrationSpec.groovy +++ /dev/null @@ -1,84 +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 com.fasterxml.jackson.databind.ObjectMapper -import org.apache.kafka.clients.consumer.KafkaConsumer -import org.mapstruct.factory.Mappers -import org.onap.cps.ncmp.api.kafka.MessagingBaseSpec -import org.onap.cps.ncmp.event.model.AvcEvent -import org.onap.cps.ncmp.utils.TestUtils -import org.onap.cps.utils.JsonObjectMapper -import org.spockframework.spring.SpringBean -import org.springframework.beans.factory.annotation.Autowired -import org.springframework.boot.test.context.SpringBootTest -import org.springframework.test.annotation.DirtiesContext -import org.testcontainers.spock.Testcontainers - -import java.time.Duration - -@SpringBootTest(classes = [AvcEventProducer, AvcEventConsumer, ObjectMapper, JsonObjectMapper]) -@Testcontainers -@DirtiesContext -class AvcEventProducerIntegrationSpec extends MessagingBaseSpec { - - @SpringBean - AvcEventMapper avcEventMapper = Mappers.getMapper(AvcEventMapper.class) - - @SpringBean - AvcEventProducer avcEventProducer = new AvcEventProducer(kafkaTemplate, avcEventMapper) - - @SpringBean - AvcEventConsumer acvEventConsumer = new AvcEventConsumer(avcEventProducer) - - @Autowired - JsonObjectMapper jsonObjectMapper - - def kafkaConsumer = new KafkaConsumer<>(consumerConfigProperties('ncmp-group')) - - 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) - and: 'an event is sent' - def jsonData = TestUtils.getResourceFileContent('sampleAvcInputEvent.json') - def testEventSent = jsonObjectMapper.convertJsonString(jsonData, AvcEvent.class) - when: 'the event is consumed' - acvEventConsumer.consumeAndForward(testEventSent) - and: 'the topic is polled' - def records = kafkaConsumer.poll(Duration.ofMillis(1500)) - then: 'poll returns one record' - assert records.size() == 1 - and: 'record can be converted to AVC event' - def record = records.iterator().next() - def convertedAvcEvent = jsonObjectMapper.convertJsonString(record.value(), AvcEvent) - and: 'consumed forwarded NCMP event id differs from DMI event id' - assert testEventSent.eventId != convertedAvcEvent.getEventId() - and: 'correlation id matches' - assert testEventSent.eventCorrelationId == convertedAvcEvent.getEventCorrelationId() - and: 'timestamps match' - assert testEventSent.eventTime == convertedAvcEvent.getEventTime() - and: 'target matches' - assert testEventSent.eventSource == convertedAvcEvent.getEventSource() - } - -} \ No newline at end of file -- cgit 1.2.3-korg