diff options
author | 2023-04-13 23:44:41 +0100 | |
---|---|---|
committer | 2023-05-23 17:53:05 +0100 | |
commit | e9ac24f8ad866f81d1ceace6d0b8f010930e9f74 (patch) | |
tree | 9bf8b9135b190eea558c51542a776172a41df24a /cps-ncmp-service | |
parent | 9b059e3ad4edc1ea8d2f62ec4bf82543a1e157c2 (diff) |
NCMP : forward bulk response messages to client topic
Issue-ID: CPS-1557
Signed-off-by: raviteja.karumuri <raviteja.karumuri@est.tech>
Change-Id: I0ea040e98987f992f46105afea0a171c4031d64f
Diffstat (limited to 'cps-ncmp-service')
5 files changed, 279 insertions, 11 deletions
diff --git a/cps-ncmp-service/pom.xml b/cps-ncmp-service/pom.xml index 0a732ef87..b87fe6436 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 000000000..088e96564 --- /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 000000000..2a332d003 --- /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 000000000..65c43a011 --- /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 000000000..49eb273f5 --- /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 |