/* * ============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.events.avcsubscription; import com.hazelcast.map.IMap; import java.util.Collection; import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.common.header.Headers; import org.onap.cps.ncmp.api.impl.event.avc.ForwardedSubscriptionEventCacheConfig; import org.onap.cps.ncmp.api.impl.event.avc.ResponseTimeoutTask; import org.onap.cps.ncmp.api.impl.events.EventsPublisher; import org.onap.cps.ncmp.api.impl.utils.DmiServiceNameOrganizer; import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle; import org.onap.cps.ncmp.api.inventory.InventoryPersistence; import org.onap.cps.ncmp.event.model.SubscriptionEvent; import org.onap.cps.spi.exceptions.OperationNotYetSupportedException; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; @Component @Slf4j @RequiredArgsConstructor public class SubscriptionEventForwarder { private final InventoryPersistence inventoryPersistence; private final EventsPublisher eventsPublisher; private final IMap> forwardedSubscriptionEventCache; private final SubscriptionEventResponseOutcome subscriptionEventResponseOutcome; private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor(); @Value("${app.ncmp.avc.subscription-forward-topic-prefix}") private String dmiAvcSubscriptionTopicPrefix; @Value("${ncmp.timers.subscription-forwarding.dmi-response-timeout-ms:30000}") private int dmiResponseTimeoutInMs; /** * Forward subscription event. * * @param subscriptionEvent the event to be forwarded */ public void forwardCreateSubscriptionEvent(final SubscriptionEvent subscriptionEvent, final Headers eventHeaders) { final List cmHandleTargets = subscriptionEvent.getEvent().getPredicates().getTargets(); if (cmHandleTargets == null || cmHandleTargets.isEmpty() || cmHandleTargets.stream().anyMatch(id -> ((String) id).contains("*"))) { throw new OperationNotYetSupportedException( "CMHandle targets are required. \"Wildcard\" operations are not yet supported"); } final List cmHandleTargetsAsStrings = cmHandleTargets.stream().map( Objects::toString).collect(Collectors.toList()); final Collection yangModelCmHandles = inventoryPersistence.getYangModelCmHandles(cmHandleTargetsAsStrings); final Map>> dmiPropertiesPerCmHandleIdPerServiceName = DmiServiceNameOrganizer.getDmiPropertiesPerCmHandleIdPerServiceName(yangModelCmHandles); final Set dmisToRespond = new HashSet<>(dmiPropertiesPerCmHandleIdPerServiceName.keySet()); if (dmisToRespond.isEmpty()) { final String clientID = subscriptionEvent.getEvent().getSubscription().getClientID(); final String subscriptionName = subscriptionEvent.getEvent().getSubscription().getName(); subscriptionEventResponseOutcome.sendResponse(clientID, subscriptionName, true); } else { startResponseTimeout(subscriptionEvent, dmisToRespond); forwardEventToDmis(dmiPropertiesPerCmHandleIdPerServiceName, subscriptionEvent, eventHeaders); } } private void startResponseTimeout(final SubscriptionEvent subscriptionEvent, final Set dmisToRespond) { final String subscriptionClientId = subscriptionEvent.getEvent().getSubscription().getClientID(); final String subscriptionName = subscriptionEvent.getEvent().getSubscription().getName(); final String subscriptionEventId = subscriptionClientId + subscriptionName; forwardedSubscriptionEventCache.put(subscriptionEventId, dmisToRespond, ForwardedSubscriptionEventCacheConfig.SUBSCRIPTION_FORWARD_STARTED_TTL_SECS, TimeUnit.SECONDS); final ResponseTimeoutTask responseTimeoutTask = new ResponseTimeoutTask(forwardedSubscriptionEventCache, subscriptionEventResponseOutcome, subscriptionClientId, subscriptionName); try { executorService.schedule(responseTimeoutTask, dmiResponseTimeoutInMs, TimeUnit.MILLISECONDS); } catch (final RuntimeException ex) { log.info("Caught exception in ScheduledExecutorService for ResponseTimeoutTask. StackTrace: {}", ex.toString()); } } private void forwardEventToDmis(final Map>> dmiNameCmHandleMap, final SubscriptionEvent subscriptionEvent, final Headers eventHeaders) { dmiNameCmHandleMap.forEach((dmiName, cmHandlePropertiesMap) -> { subscriptionEvent.getEvent().getPredicates().setTargets(Collections.singletonList(cmHandlePropertiesMap)); final String eventKey = createEventKey(subscriptionEvent, dmiName); final String dmiAvcSubscriptionTopic = dmiAvcSubscriptionTopicPrefix + dmiName; eventsPublisher.publishEvent(dmiAvcSubscriptionTopic, eventKey, eventHeaders, subscriptionEvent); }); } private String createEventKey(final SubscriptionEvent subscriptionEvent, final String dmiName) { return subscriptionEvent.getEvent().getSubscription().getClientID() + "-" + subscriptionEvent.getEvent().getSubscription().getName() + "-" + dmiName; } }