[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

2 files changed, 15 insertions, 12 deletions

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)