diff options
Diffstat (limited to 'cps-ncmp-service/src/main/java')
4 files changed, 159 insertions, 2 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 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); + } +} |