diff options
Diffstat (limited to 'cps-ncmp-service/src/main/java/org')
3 files changed, 172 insertions, 0 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 new file mode 100644 index 0000000000..4e5c57ba57 --- /dev/null +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventConsumer.java @@ -0,0 +1,55 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (c) 2022 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.onap.cps.ncmp.event.model.DmiAsyncRequestResponseEvent; +import org.onap.cps.ncmp.event.model.NcmpAsyncRequestResponseEvent; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.stereotype.Component; + +/** + * Listener for cps-ncmp async request response events. + */ +@Component +@Slf4j +@RequiredArgsConstructor +public class NcmpAsyncRequestResponseEventConsumer { + + private final NcmpAsyncRequestResponseEventProducer ncmpAsyncRequestResponseEventProducer; + private final NcmpAsyncRequestResponseEventMapper ncmpAsyncRequestResponseEventMapper; + + /** + * Consume the specified event. + * + * @param dmiAsyncRequestResponseEvent the event to be consumed and produced. + */ + @KafkaListener(topics = "${app.ncmp.async-m2m.topic}") + public void consumeAndForward(final DmiAsyncRequestResponseEvent dmiAsyncRequestResponseEvent) { + log.debug("Consuming event {} ...", dmiAsyncRequestResponseEvent); + + final NcmpAsyncRequestResponseEvent ncmpAsyncRequestResponseEvent = + ncmpAsyncRequestResponseEventMapper.toNcmpAsyncEvent(dmiAsyncRequestResponseEvent); + ncmpAsyncRequestResponseEventProducer.sendMessage( + ncmpAsyncRequestResponseEvent.getEventId(), ncmpAsyncRequestResponseEvent); + } +} diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventMapper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventMapper.java new file mode 100644 index 0000000000..5d8ac7f841 --- /dev/null +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventMapper.java @@ -0,0 +1,71 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (C) 2022 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 java.time.ZonedDateTime; +import java.time.format.DateTimeFormatter; +import java.util.UUID; +import org.mapstruct.AfterMapping; +import org.mapstruct.Mapper; +import org.mapstruct.Mapping; +import org.mapstruct.MappingTarget; +import org.mapstruct.Named; +import org.onap.cps.ncmp.event.model.DmiAsyncRequestResponseEvent; +import org.onap.cps.ncmp.event.model.NcmpAsyncRequestResponseEvent; + +/** + * Mapper for converting DmiAsyncRequestResponseEvent to NcmpAsyncRequestResponseEvent. + */ +@Mapper(componentModel = "spring") +public interface NcmpAsyncRequestResponseEventMapper { + + @Mapping(source = "eventId", target = "eventId", qualifiedByName = "ncmpAsyncEventId") + @Mapping(source = "eventTime", target = "eventTime", qualifiedByName = "currentTime") + @Mapping(source = "eventId", target = "forwardedEvent.eventId") + @Mapping(source = "eventCorrelationId", target = "forwardedEvent.eventCorrelationId") + @Mapping(source = "eventSchema", target = "forwardedEvent.eventSchema") + @Mapping(source = "eventSource", target = "forwardedEvent.eventSource") + @Mapping(source = "eventTarget", target = "forwardedEvent.eventTarget") + @Mapping(source = "eventTime", target = "forwardedEvent.eventTime") + @Mapping(source = "eventType", target = "forwardedEvent.eventType") + @Mapping(source = "eventContent.responseStatus", target = "forwardedEvent.responseStatus") + @Mapping(source = "eventContent.responseCode", target = "forwardedEvent.responseCode") + @Mapping(source = "eventContent.responseDataSchema", target = "forwardedEvent.responseDataSchema") + NcmpAsyncRequestResponseEvent toNcmpAsyncEvent(DmiAsyncRequestResponseEvent dmiAsyncRequestResponseEvent); + + @Named("ncmpAsyncEventId") + static String getNcmpAsyncEventId(String eventId) { + return UUID.randomUUID().toString(); + } + + @Named("currentTime") + static String getFormattedCurrentTime(String eventTime) { + return ZonedDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ")); + } + + @AfterMapping + default void mapAdditionalProperties(DmiAsyncRequestResponseEvent dmiAsyncRequestResponseEvent, + @MappingTarget NcmpAsyncRequestResponseEvent ncmpAsyncRequestResponseEvent) { + ncmpAsyncRequestResponseEvent.getForwardedEvent().setAdditionalProperty("response-data", + dmiAsyncRequestResponseEvent.getEventContent().getResponseData().getAdditionalProperties()); + } + +} diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventProducer.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventProducer.java new file mode 100644 index 0000000000..8ab6db9045 --- /dev/null +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/async/NcmpAsyncRequestResponseEventProducer.java @@ -0,0 +1,46 @@ +/* + * ============LICENSE_START======================================================= + * Copyright (C) 2022 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.onap.cps.ncmp.event.model.NcmpAsyncRequestResponseEvent; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.stereotype.Service; + +@Slf4j +@Service +@RequiredArgsConstructor +public class NcmpAsyncRequestResponseEventProducer { + + private final KafkaTemplate<String, NcmpAsyncRequestResponseEvent> kafkaTemplate; + + + /** + * Sends message to the configured topic with a message key. + * + * @param eventId message key + * @param ncmpAsyncRequestResponseEvent message payload + */ + public void sendMessage(final String eventId, final NcmpAsyncRequestResponseEvent ncmpAsyncRequestResponseEvent) { + kafkaTemplate.send(ncmpAsyncRequestResponseEvent.getEventTarget(), eventId, ncmpAsyncRequestResponseEvent); + } +} |