diff options
author | 2023-01-13 17:13:25 +0000 | |
---|---|---|
committer | 2023-01-23 10:09:41 +0000 | |
commit | 7d25644c516869c299ea8d93ea915750e5b07203 (patch) | |
tree | c1ca85b53298326c041ce9914bac1ff7c12a7d8c /cps-ncmp-service/src | |
parent | 556e54f5905440e4c9c4fc7fbf9c027877e51517 (diff) |
[NCMP] Consume & Forward to client topic
-Consumes event from dmi-cm-events
-Immediately forwards to static topic (topic selection for events comes
later from subscription information)
-Added Kafka test
-SHOULD BE MERGED BEFORE DMI PART
Issue-ID: CPS-138
Signed-off-by: JosephKeenan <joseph.keenan@est.tech>
Change-Id: I0a426381e2c3f9173b8d3916960c05722ad4f77d
Signed-off-by: seanbeirne <sean.beirne@est.tech>
Diffstat (limited to 'cps-ncmp-service/src')
7 files changed, 259 insertions, 7 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 a9e7164fd..bc6624dee 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 @@ -1,6 +1,6 @@ /* * ============LICENSE_START======================================================= - * Copyright (c) 2022 Nordix Foundation. + * 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. @@ -45,7 +45,9 @@ public class NcmpAsyncRequestResponseEventConsumer { * * @param dmiAsyncRequestResponseEvent the event to be consumed and produced. */ - @KafkaListener(topics = "${app.ncmp.async-m2m.topic}") + @KafkaListener( + topics = "${app.ncmp.async-m2m.topic}", + properties = {"spring.json.value.default.type=org.onap.cps.ncmp.event.model.DmiAsyncRequestResponseEvent"}) public void consumeAndForward(final DmiAsyncRequestResponseEvent dmiAsyncRequestResponseEvent) { log.debug("Consuming event {} ...", dmiAsyncRequestResponseEvent); diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/notifications/avc/AvcEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/notifications/avc/AvcEventConsumer.java new file mode 100644 index 000000000..79a36bf50 --- /dev/null +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/notifications/avc/AvcEventConsumer.java @@ -0,0 +1,53 @@ +/* + * ============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.notifications.avc; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.onap.cps.ncmp.event.model.AvcEvent; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.stereotype.Component; + +/** + * Listener for AVC events. + */ +@Component +@Slf4j +@RequiredArgsConstructor +@ConditionalOnProperty(name = "notification.enabled", havingValue = "true", matchIfMissing = true) +public class AvcEventConsumer { + + private final AvcEventProducer avcEventProducer; + + /** + * Consume the specified event. + * + * @param avcEvent the event to be consumed and produced. + */ + @KafkaListener( + topics = "dmi-cm-events", + 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); + } +} diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/notifications/avc/AvcEventMapper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/notifications/avc/AvcEventMapper.java new file mode 100644 index 000000000..531de4641 --- /dev/null +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/notifications/avc/AvcEventMapper.java @@ -0,0 +1,50 @@ +/* + * ============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.notifications.avc; + +import java.util.UUID; +import org.mapstruct.Mapper; +import org.mapstruct.Mapping; +import org.mapstruct.Named; +import org.onap.cps.ncmp.event.model.AvcEvent; + + +/** + * Mapper for converting incoming {@link AvcEvent} to outgoing {@link AvcEvent}. + */ +@Mapper(componentModel = "spring") +public interface AvcEventMapper { + + @Mapping(source = "eventTime", target = "eventTime") + @Mapping(source = "eventId", target = "eventId", qualifiedByName = "avcEventId") + @Mapping(source = "eventCorrelationId", target = "eventCorrelationId") + @Mapping(source = "eventSchema", target = "eventSchema") + @Mapping(source = "eventSchemaVersion", target = "eventSchemaVersion") + @Mapping(source = "eventTarget", target = "eventTarget") + @Mapping(source = "eventType", target = "eventType") + AvcEvent toOutgoingAvcEvent(AvcEvent incomingAvcEvent); + + @Named("avcEventId") + static String getAvcEventId(String eventId) { + return UUID.randomUUID().toString(); + } + +} diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/notifications/avc/AvcEventProducer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/notifications/avc/AvcEventProducer.java new file mode 100644 index 000000000..049f66100 --- /dev/null +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/notifications/avc/AvcEventProducer.java @@ -0,0 +1,52 @@ +/* + * ============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.notifications.avc; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.onap.cps.ncmp.event.model.AvcEvent; +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; + + /** + * 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(), "cm-events"); + kafkaTemplate.send("cm-events", outgoingAvcEvent.getEventId(), outgoingAvcEvent); + } +} diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/notifications/avc/AvcEventProducerIntegrationSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/notifications/avc/AvcEventProducerIntegrationSpec.groovy new file mode 100644 index 000000000..0089f777d --- /dev/null +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/notifications/avc/AvcEventProducerIntegrationSpec.groovy @@ -0,0 +1,83 @@ +/* + * ============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.notifications.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.async.NcmpAsyncRequestResponseEventMapper +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' + kafkaConsumer.subscribe(['cm-events'] as List<String>) + 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.eventTarget == convertedAvcEvent.getEventTarget() + } + +}
\ No newline at end of file diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/kafka/MessagingBaseSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/kafka/MessagingBaseSpec.groovy index f7c41ecdf..bb0ce8745 100644 --- a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/kafka/MessagingBaseSpec.groovy +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/kafka/MessagingBaseSpec.groovy @@ -1,6 +1,6 @@ /* * ============LICENSE_START======================================================= - * Copyright (c) 2022 Nordix Foundation. + * 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. @@ -33,14 +33,14 @@ import spock.lang.Specification class MessagingBaseSpec extends Specification { - static { - Runtime.getRuntime().addShutdownHook(new Thread(kafkaTestContainer::stop)) - } - def setupSpec() { kafkaTestContainer.start() } + def cleanupSpec() { + kafkaTestContainer.stop() + } + static kafkaTestContainer = new KafkaContainer(DockerImageName.parse('registry.nordix.org/onaptest/confluentinc/cp-kafka:6.2.1').asCompatibleSubstituteFor('confluentinc/cp-kafka')) def producerConfigProperties() { diff --git a/cps-ncmp-service/src/test/resources/sampleAvcInputEvent.json b/cps-ncmp-service/src/test/resources/sampleAvcInputEvent.json new file mode 100644 index 000000000..d7d252b9a --- /dev/null +++ b/cps-ncmp-service/src/test/resources/sampleAvcInputEvent.json @@ -0,0 +1,12 @@ +{ + "eventId": "4cb32729-85e3-44d1-aa6e-c923b9b059a5", + "eventCorrelationId": "68f15800-8ed4-4bae-9e53-27a9e03e1911", + "eventTime": "2022-12-12T14:29:23.876+0000", + "eventTarget": "NCMP", + "eventType": "org.onap.cps.ncmp.event.model.AvcEvent", + "eventSchema": "urn:cps:org.onap.cps.ncmp.event.model.AvcEvent", + "eventSchemaVersion": "v1", + "event": { + "payload": "Hello world!" + } +}
\ No newline at end of file |