diff options
9 files changed, 323 insertions, 14 deletions
diff --git a/cps-application/src/main/resources/application.yml b/cps-application/src/main/resources/application.yml index b5b10b0f74..f7fe46e4e0 100644 --- a/cps-application/src/main/resources/application.yml +++ b/cps-application/src/main/resources/application.yml @@ -1,7 +1,7 @@ # ============LICENSE_START=======================================================
# Copyright (C) 2021 Pantheon.tech
# Modifications Copyright (C) 2021-2022 Bell Canada
-# Modifications Copyright (C) 2021-2022 Nordix Foundation
+# Modifications Copyright (C) 2021-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.
@@ -84,16 +84,15 @@ spring: properties:
spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
- spring.json.value.default.type: org.onap.cps.ncmp.event.model.DmiAsyncRequestResponseEvent
spring.json.use.type.headers: false
jackson:
- default-property-inclusion: NON_NULL
- serialization:
- FAIL_ON_EMPTY_BEANS: false
+ default-property-inclusion: NON_NULL
+ serialization:
+ FAIL_ON_EMPTY_BEANS: false
sql:
- init:
- mode: ALWAYS
+ init:
+ mode: ALWAYS
app:
ncmp:
async-m2m:
@@ -104,6 +103,7 @@ app: events:
topic: ${LCM_EVENTS_TOPIC:ncmp-events}
+
notification:
enabled: true
data-updated:
diff --git a/cps-ncmp-events/src/main/resources/schemas/avc-event-schema-v1.json b/cps-ncmp-events/src/main/resources/schemas/avc-event-schema-v1.json new file mode 100644 index 0000000000..6db03f6ebc --- /dev/null +++ b/cps-ncmp-events/src/main/resources/schemas/avc-event-schema-v1.json @@ -0,0 +1,57 @@ +{ + "$schema": "https://json-schema.org/draft/2019-09/schema", + "$id": "urn:cps:org.onap.cps.ncmp.events:avc-event-schema:v1", + "$ref": "#/definitions/AvcEvent", + "definitions": { + "AvcEvent": { + "description": "The payload for AVC event.", + "type": "object", + "properties": { + "eventId": { + "description": "The unique id identifying the event generated by DMI for this AVC event.", + "type": "string" + }, + "eventCorrelationId": { + "description": "The request id passed by NCMP for this AVC event.", + "type": "string" + }, + "eventTime": { + "description": "The time of the AVC event. The expected format is 'yyyy-MM-dd'T'HH:mm:ss.SSSZ'.", + "type": "string" + }, + "eventTarget": { + "description": "The target of the AVC event.", + "type": "string" + }, + "eventType": { + "description": "The type of the AVC event.", + "type": "string" + }, + "eventSchema": { + "description": "The event schema for AVC events.", + "type": "string" + }, + "eventSchemaVersion": { + "description": "The event schema version for AVC events.", + "type": "string" + }, + "event": { + "$ref": "#/definitions/Event" + } + }, + "required": [ + "eventId", + "eventCorrelationId", + "eventTime", + "eventTarget", + "eventType", + "eventSchema", + "eventSchemaVersion" + ] + }, + "Event": { + "description": "The AVC event content.", + "type": "object" + } + } +}
\ No newline at end of file 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 a9e7164fd7..bc6624dee6 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 0000000000..79a36bf506 --- /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 0000000000..531de46414 --- /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 0000000000..049f66100b --- /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 0000000000..0089f777d3 --- /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 f7c41ecdf2..bb0ce8745d 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 0000000000..d7d252b9aa --- /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 |