diff options
7 files changed, 401 insertions, 11 deletions
diff --git a/cps-ncmp-events/src/main/resources/schemas/async/batch-event-headers-1.0.0.json b/cps-ncmp-events/src/main/resources/schemas/async/batch-event-headers-1.0.0.json new file mode 100644 index 0000000000..bbcadcd0f4 --- /dev/null +++ b/cps-ncmp-events/src/main/resources/schemas/async/batch-event-headers-1.0.0.json @@ -0,0 +1,55 @@ +{ + "$schema": "https://json-schema.org/draft/2019-09/schema", + "$id": "urn:cps:org.onap.cps.ncmp.events.async:batch-event-headers:1.0.0", + "$ref": "#/definitions/BatchEventHeaders", + "definitions": { + "BatchEventHeaders": { + "description": "The header information of the Batch event.", + "type": "object", + "javaType" : "org.onap.cps.ncmp.events.async.BatchEventHeadersV1", + "properties": { + "eventId": { + "description": "The unique id for identifying the event.", + "type": "string" + }, + "eventCorrelationId": { + "description": "The request id received by NCMP as an acknowledgement.", + "type": "string" + }, + "eventTime": { + "description": "The time of the event. It should be in RFC format ('yyyy-MM-dd'T'HH:mm:ss.SSSZ').", + "type": "string" + }, + "eventTarget": { + "description": "The destination topic to forward the consumed event.", + "type": "string" + }, + "eventSource": { + "description": "The source of the event.", + "type": "string" + }, + "eventType": { + "description": "The type of the Batch event.", + "type": "string" + }, + "eventSchema": { + "description": "The schema of the Batch event payload.", + "type": "string" + }, + "eventSchemaVersion": { + "description": "The schema version of the Batch event payload.", + "type": "string" + } + }, + "required": [ + "eventId", + "eventCorrelationId", + "eventTarget", + "eventType", + "eventSchema", + "eventSchemaVersion" + ], + "additionalProperties": false + } + } +}
\ No newline at end of file diff --git a/cps-ncmp-events/src/main/resources/schemas/async/batch-event-schema-1.0.0.json b/cps-ncmp-events/src/main/resources/schemas/async/batch-event-schema-1.0.0.json new file mode 100644 index 0000000000..da836ff167 --- /dev/null +++ b/cps-ncmp-events/src/main/resources/schemas/async/batch-event-schema-1.0.0.json @@ -0,0 +1,67 @@ +{ + "$schema": "https://json-schema.org/draft/2019-09/schema", + "$id": "urn:cps:org.onap.cps.ncmp.events.async:batch-event-schema:1.0.0", + "$ref": "#/definitions/BatchDataResponseEvent", + "definitions": { + "BatchDataResponseEvent": { + "description": "The payload of batch event.", + "type": "object", + "javaType" : "org.onap.cps.ncmp.events.async.BatchDataResponseEventV1", + "properties": { + "event": { + "description": "The payload content of the requested data.", + "type": "object", + "javaType" : "org.onap.cps.ncmp.events.async.BatchDataEvent", + "properties": { + "batch-responses": { + "description": "An array of batch responses which contains both success and failure", + "type": "array", + "items": { + "type": "object", + "properties": { + "operationId": { + "description": "Used to distinguish multiple operations using same cmhandleId", + "type": "string" + }, + "ids": { + "description": "Id's of the cmhandles", + "type": "array" + }, + "status-code": { + "description": "which says success or failure (0-99) are for success and (100-199) are for failure", + "type": "string" + }, + "status-message": { + "description": "Human readable message, Which says what the response has", + "type": "string" + }, + "data": { + "description": "Contains the requested data response.", + "type": "object", + "existingJavaType": "java.lang.Object", + "additionalProperties": false + } + }, + "required": [ + "operationId", + "ids", + "status-code", + "status-message" + ], + "additionalProperties": false + } + } + }, + "required": [ + "batch-responses" + ], + "additionalProperties": false + } + }, + "required": [ + "event" + ], + "additionalProperties": false + } + } +}
\ No newline at end of file diff --git a/cps-ncmp-service/pom.xml b/cps-ncmp-service/pom.xml index 0a732ef873..b87fe64366 100644 --- a/cps-ncmp-service/pom.xml +++ b/cps-ncmp-service/pom.xml @@ -38,6 +38,10 @@ </properties> <dependencies> <dependency> + <groupId>org.apache.commons</groupId> + <artifactId>commons-lang3</artifactId> + </dependency> + <dependency> <groupId>${project.groupId}</groupId> <artifactId>cps-service</artifactId> </dependency> @@ -67,6 +71,17 @@ </dependency> <!-- T E S T - D E P E N D E N C I E S --> <dependency> + <groupId>org.springframework.boot</groupId> + <artifactId>spring-boot-starter-test</artifactId> + <scope>test</scope> + <exclusions> + <exclusion> + <groupId>org.junit.vintage</groupId> + <artifactId>junit-vintage-engine</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> <groupId>org.spockframework</groupId> <artifactId>spock-core</artifactId> <scope>test</scope> @@ -86,16 +101,5 @@ <artifactId>spock</artifactId> <scope>test</scope> </dependency> - <dependency> - <groupId>org.springframework.boot</groupId> - <artifactId>spring-boot-starter-test</artifactId> - <scope>test</scope> - <exclusions> - <exclusion> - <groupId>org.junit.vintage</groupId> - <artifactId>junit-vintage-engine</artifactId> - </exclusion> - </exclusions> - </dependency> </dependencies> </project> diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/BatchRecordFilterStrategy.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/BatchRecordFilterStrategy.java new file mode 100644 index 0000000000..088e96564c --- /dev/null +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/BatchRecordFilterStrategy.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.async; + +import org.apache.commons.lang3.SerializationUtils; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.listener.adapter.RecordFilterStrategy; + +/** + * Batch Record filter strategy, which helps to filter the consumer records. + * + */ +@Configuration +public class BatchRecordFilterStrategy { + + /** + * Filtering the consumer records based on the eventType header, It + * returns boolean, true means filter the consumer record and false + * means not filter the consumer record. + * @return boolean value. + */ + @Bean + public RecordFilterStrategy<Object, Object> filterBatchDataResponseEvent() { + return consumedRecord -> { + final String headerValue = SerializationUtils + .deserialize(consumedRecord.headers().lastHeader("eventType").value()); + return !(headerValue != null + && headerValue.startsWith("org.onap.cps.ncmp.events.async.BatchDataResponseEvent")); + }; + } +} diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncBatchEventConsumer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncBatchEventConsumer.java new file mode 100644 index 0000000000..2a332d0037 --- /dev/null +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncBatchEventConsumer.java @@ -0,0 +1,64 @@ +/* + * ============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.async; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.SerializationUtils; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.onap.cps.ncmp.api.impl.events.EventsPublisher; +import org.onap.cps.ncmp.events.async.BatchDataResponseEventV1; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.stereotype.Component; + +/** + * Listener for cps-ncmp async batch events. + */ +@Component +@Slf4j +@RequiredArgsConstructor +@ConditionalOnProperty(name = "notification.enabled", havingValue = "true", matchIfMissing = true) +public class NcmpAsyncBatchEventConsumer { + + private final EventsPublisher<BatchDataResponseEventV1> eventsPublisher; + + /** + * Consume the BatchDataResponseEvent published by producer to topic 'async-m2m.topic' + * and publish the same to the client specified topic. + * + * @param batchEventConsumerRecord consuming event as a ConsumerRecord. + */ + @KafkaListener( + topics = "${app.ncmp.async-m2m.topic}", + filter = "filterBatchDataResponseEvent", + groupId = "ncmp-batch-event-group", + properties = {"spring.json.value.default.type=org.onap.cps.ncmp.events.async.BatchDataResponseEventV1"}) + public void consumeAndPublish(final ConsumerRecord<String, BatchDataResponseEventV1> batchEventConsumerRecord) { + log.info("Consuming event payload {} ...", batchEventConsumerRecord.value()); + final String eventTarget = SerializationUtils + .deserialize(batchEventConsumerRecord.headers().lastHeader("eventTarget").value()); + final String eventId = SerializationUtils + .deserialize(batchEventConsumerRecord.headers().lastHeader("eventId").value()); + eventsPublisher.publishEvent(eventTarget, eventId, batchEventConsumerRecord.headers(), + batchEventConsumerRecord.value()); + } +} diff --git a/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncBatchEventConsumerSpec.groovy b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncBatchEventConsumerSpec.groovy new file mode 100644 index 0000000000..65c43a011d --- /dev/null +++ b/cps-ncmp-service/src/test/groovy/org/onap/cps/ncmp/api/impl/async/NcmpAsyncBatchEventConsumerSpec.groovy @@ -0,0 +1,104 @@ +/* + * ============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.async + +import com.fasterxml.jackson.databind.ObjectMapper +import org.apache.commons.lang3.SerializationUtils +import org.apache.kafka.clients.consumer.ConsumerRecord +import org.apache.kafka.clients.consumer.KafkaConsumer +import org.apache.kafka.common.header.internals.RecordHeader +import org.onap.cps.ncmp.api.impl.events.EventsPublisher +import org.onap.cps.ncmp.api.kafka.MessagingBaseSpec +import org.onap.cps.ncmp.events.async.BatchDataResponseEventV1 +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.kafka.listener.adapter.RecordFilterStrategy +import org.springframework.test.annotation.DirtiesContext +import org.testcontainers.spock.Testcontainers + +import java.time.Duration + +@SpringBootTest(classes = [EventsPublisher, NcmpAsyncBatchEventConsumer, BatchRecordFilterStrategy,JsonObjectMapper, + ObjectMapper]) +@Testcontainers +@DirtiesContext +class NcmpAsyncBatchEventConsumerSpec extends MessagingBaseSpec { + + @SpringBean + EventsPublisher asyncBatchEventPublisher = new EventsPublisher<BatchDataResponseEventV1>(kafkaTemplate) + + @SpringBean + NcmpAsyncBatchEventConsumer asyncBatchEventConsumer = new NcmpAsyncBatchEventConsumer(asyncBatchEventPublisher) + + @Autowired + JsonObjectMapper jsonObjectMapper + + @Autowired + RecordFilterStrategy<Object, Object> recordFilterStrategy + + def kafkaConsumer = new KafkaConsumer<>(consumerConfigProperties('test')) + def static clientTopic = 'client-topic' + def static batchEventType = 'org.onap.cps.ncmp.events.async.BatchDataResponseEventV1' + + def 'Consume and publish event to client specified topic'() { + given: 'consumer subscribing to client topic' + kafkaConsumer.subscribe([clientTopic]) + and: 'consumer record for batch event' + def consumerRecordIn = createConsumerRecord(batchEventType) + when: 'the batch event is consumed and published to client specified topic' + asyncBatchEventConsumer.consumeAndPublish(consumerRecordIn) + and: 'the client specified topic is polled' + def consumerRecordOut = kafkaConsumer.poll(Duration.ofMillis(1500))[0] + then: 'verifying consumed event operationID is same as published event operationID' + def operationIdIn = consumerRecordIn.value.event.batchResponses[0].operationId + def operationIdOut = jsonObjectMapper.convertJsonString((String)consumerRecordOut.value(), BatchDataResponseEventV1.class).event.batchResponses[0].operationId + assert operationIdIn == operationIdOut + } + + def 'Filter an event with type #eventType'() { + given: 'consumer record for event with type #eventType' + def consumerRecord = createConsumerRecord(eventType) + when: 'while consuming the topic ncmp-async-m2m it executes the filter strategy' + def result = recordFilterStrategy.filter(consumerRecord) + then: 'the event is #description' + assert result == expectedResult + where: 'filter the event based on the eventType #eventType' + description | eventType || expectedResult + 'not filtered(the consumer will see the event)' | batchEventType || false + 'filtered(the consumer will not see the event)' | 'wrongType' || true + } + + def createConsumerRecord(eventTypeAsString) { + def jsonData = TestUtils.getResourceFileContent('batchDataEvent.json') + def testEventSent = jsonObjectMapper.convertJsonString(jsonData, BatchDataResponseEventV1.class) + def eventTarget = SerializationUtils.serialize(clientTopic) + def eventType = SerializationUtils.serialize(eventTypeAsString) + def eventId = SerializationUtils.serialize('12345') + def consumerRecord = new ConsumerRecord<String, Object>(clientTopic, 1, 1L, '123', testEventSent) + consumerRecord.headers().add(new RecordHeader('eventId', eventId)) + consumerRecord.headers().add(new RecordHeader('eventTarget', eventTarget)) + consumerRecord.headers().add(new RecordHeader('eventType', eventType)) + return consumerRecord + } +} diff --git a/cps-ncmp-service/src/test/resources/batchDataEvent.json b/cps-ncmp-service/src/test/resources/batchDataEvent.json new file mode 100644 index 0000000000..49eb273f58 --- /dev/null +++ b/cps-ncmp-service/src/test/resources/batchDataEvent.json @@ -0,0 +1,46 @@ +{ + "event":{ + "batch-responses":[ + { + "operationId":"1", + "ids":[ + "123", + "124" + ], + "status-code":1, + "status-message":"Batch operation success on the above cmhandle ids ", + "data":{ + "ietf-netconf-monitoring:netconf-state":{ + "schemas":{ + "schema":[ + { + "identifier":"ietf-tls-server", + "version":"2016-11-02", + "format":"ietf-netconf-monitoring:yang", + "namespace":"urn:ietf:params:xml:ns:yang:ietf-tls-server", + "location":[ + "NETCONF" + ] + } + ] + } + } + } + }, + { + "operationId":"101", + "ids":[ + "456", + "457" + ], + "status-code":101, + "status-message":"cmHandle(s) do not exist", + "data":{ + "error":{ + "message":"cmHandle(s) do not exist" + } + } + } + ] + } +}
\ No newline at end of file |