--- /dev/null
+{
+ "$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
--- /dev/null
+{
+ "$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
<minimum-coverage>0.96</minimum-coverage>
</properties>
<dependencies>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>cps-service</artifactId>
<artifactId>hazelcast-spring</artifactId>
</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>
<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>
--- /dev/null
+/*
+ * ============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"));
+ };
+ }
+}
--- /dev/null
+/*
+ * ============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());
+ }
+}
--- /dev/null
+/*
+ * ============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
+ }
+}
--- /dev/null
+{
+ "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
Code Review / cps.git / commitdiff