diff options
Diffstat (limited to 'cps-ncmp-service/src/main/java/org')
7 files changed, 321 insertions, 41 deletions
diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/config/kafka/KafkaTemplateConfig.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/config/kafka/KafkaTemplateConfig.java new file mode 100644 index 0000000000..b76f86ebeb --- /dev/null +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/config/kafka/KafkaTemplateConfig.java @@ -0,0 +1,127 @@ +/* + * ============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.config.kafka; + +import io.cloudevents.CloudEvent; +import java.util.Map; +import lombok.RequiredArgsConstructor; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.springframework.boot.autoconfigure.kafka.KafkaProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Primary; +import org.springframework.kafka.annotation.EnableKafka; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.support.serializer.JsonDeserializer; +import org.springframework.kafka.support.serializer.JsonSerializer; + +/** + * kafka Configuration for legacy and cloud events. + * + * @param <T> valid legacy event to be published over the wire. + */ +@Configuration +@EnableKafka +@RequiredArgsConstructor +public class KafkaTemplateConfig<T> { + + private final KafkaProperties kafkaProperties; + + /** + * This sets the strategy for creating legacy Kafka producer instance from kafka properties defined into + * application.yml and replaces value-serializer by JsonSerializer. + * + * @return legacy event producer instance. + */ + @Bean + public ProducerFactory<String, T> legacyEventProducerFactory() { + final Map<String, Object> producerConfigProperties = kafkaProperties.buildProducerProperties(); + producerConfigProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); + return new DefaultKafkaProducerFactory<>(producerConfigProperties); + } + + /** + * The ConsumerFactory implementation is to produce new legacy instance for provided kafka properties defined + * into application.yml and replaces deserializer-value by JsonDeserializer. + * + * @return an instance of legacy consumer factory. + */ + @Bean + public ConsumerFactory<String, T> legacyEventConsumerFactory() { + final Map<String, Object> consumerConfigProperties = kafkaProperties.buildConsumerProperties(); + consumerConfigProperties.put("spring.deserializer.value.delegate.class", JsonDeserializer.class); + return new DefaultKafkaConsumerFactory<>(consumerConfigProperties); + } + + /** + * This sets the strategy for creating cloud Kafka producer instance from kafka properties defined into + * application.yml with CloudEventSerializer. + * + * @return cloud event producer instance. + */ + @Bean + public ProducerFactory<String, CloudEvent> cloudEventProducerFactory() { + final Map<String, Object> producerConfigProperties = kafkaProperties.buildProducerProperties(); + return new DefaultKafkaProducerFactory<>(producerConfigProperties); + } + + /** + * The ConsumerFactory implementation to produce new legacy instance for provided kafka properties defined + * into application.yml having CloudEventDeserializer as deserializer-value. + * + * @return an instance of cloud consumer factory. + */ + @Bean + public ConsumerFactory<String, CloudEvent> cloudEventConsumerFactory() { + final Map<String, Object> consumerConfigProperties = kafkaProperties.buildConsumerProperties(); + return new DefaultKafkaConsumerFactory<>(consumerConfigProperties); + } + + /** + * A legacy Kafka event template for executing high-level operations. The legacy producer factory ensure this. + * + * @return an instance of legacy Kafka template. + */ + @Bean + @Primary + public KafkaTemplate<String, T> legacyEventKafkaTemplate() { + final KafkaTemplate<String, T> kafkaTemplate = new KafkaTemplate<>(legacyEventProducerFactory()); + kafkaTemplate.setConsumerFactory(legacyEventConsumerFactory()); + return kafkaTemplate; + } + + /** + * A cloud Kafka event template for executing high-level operations. The cloud producer factory ensure this. + * + * @return an instance of cloud Kafka template. + */ + @Bean + public KafkaTemplate<String, CloudEvent> cloudEventKafkaTemplate() { + final KafkaTemplate<String, CloudEvent> kafkaTemplate = new KafkaTemplate<>(cloudEventProducerFactory()); + kafkaTemplate.setConsumerFactory(cloudEventConsumerFactory()); + return kafkaTemplate; + } + +} diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/EventsPublisher.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/EventsPublisher.java index d92316dc58..7b28b4cd5f 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/EventsPublisher.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/EventsPublisher.java @@ -20,6 +20,7 @@ package org.onap.cps.ncmp.api.impl.events; +import io.cloudevents.CloudEvent; import java.util.Map; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -42,7 +43,12 @@ import org.springframework.util.concurrent.ListenableFutureCallback; @RequiredArgsConstructor public class EventsPublisher<T> { - private final KafkaTemplate<String, T> eventKafkaTemplate; + /** Once all cps events will be modified to cloud compliant, will remove legacyKafkaEventTemplate with + it's java configuration file KafkaTemplateConfig. **/ + @Deprecated(forRemoval = true) + private final KafkaTemplate<String, T> legacyKafkaEventTemplate; + + private final KafkaTemplate<String, CloudEvent> cloudEventKafkaTemplate; /** * Generic Event publisher. @@ -54,7 +60,8 @@ public class EventsPublisher<T> { */ @Deprecated public void publishEvent(final String topicName, final String eventKey, final T event) { - final ListenableFuture<SendResult<String, T>> eventFuture = eventKafkaTemplate.send(topicName, eventKey, event); + final ListenableFuture<SendResult<String, T>> eventFuture + = legacyKafkaEventTemplate.send(topicName, eventKey, event); eventFuture.addCallback(handleCallback(topicName)); } @@ -70,7 +77,7 @@ public class EventsPublisher<T> { final ProducerRecord<String, T> producerRecord = new ProducerRecord<>(topicName, null, eventKey, event, eventHeaders); - final ListenableFuture<SendResult<String, T>> eventFuture = eventKafkaTemplate.send(producerRecord); + final ListenableFuture<SendResult<String, T>> eventFuture = legacyKafkaEventTemplate.send(producerRecord); eventFuture.addCallback(handleCallback(topicName)); } diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseOutcome.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseOutcome.java index a74682571b..1bfc4ab28b 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseOutcome.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/events/avcsubscription/SubscriptionEventResponseOutcome.java @@ -22,8 +22,6 @@ package org.onap.cps.ncmp.api.impl.events.avcsubscription; import java.io.Serializable; import java.util.Collection; -import java.util.HashMap; -import java.util.Iterator; import java.util.List; import java.util.Map; import lombok.RequiredArgsConstructor; @@ -104,29 +102,12 @@ public class SubscriptionEventResponseOutcome { private SubscriptionEventResponse toSubscriptionEventResponse( final List<Collection<Serializable>> cmHandleIdToStatus, final String subscriptionClientId, final String subscriptionName) { - final Map<String, SubscriptionStatus> cmHandleIdToStatusMap = new HashMap<>(); + final Map<String, SubscriptionStatus> cmHandleIdToStatusMap = + DataNodeHelper.getCmHandleIdToStatusMap(cmHandleIdToStatus); + final SubscriptionEventResponse subscriptionEventResponse = new SubscriptionEventResponse(); subscriptionEventResponse.setClientId(subscriptionClientId); subscriptionEventResponse.setSubscriptionName(subscriptionName); - - for (final Collection<Serializable> cmHandleToStatusBucket: cmHandleIdToStatus) { - final Iterator<Serializable> bucketIterator = cmHandleToStatusBucket.iterator(); - while (bucketIterator.hasNext()) { - final String item = (String) bucketIterator.next(); - if ("PENDING".equals(item)) { - cmHandleIdToStatusMap.put((String) bucketIterator.next(), - SubscriptionStatus.PENDING); - } - if ("REJECTED".equals(item)) { - cmHandleIdToStatusMap.put((String) bucketIterator.next(), - SubscriptionStatus.REJECTED); - } - if ("ACCEPTED".equals(item)) { - cmHandleIdToStatusMap.put((String) bucketIterator.next(), - SubscriptionStatus.ACCEPTED); - } - } - } subscriptionEventResponse.setCmHandleIdToStatus(cmHandleIdToStatusMap); return subscriptionEventResponse; diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionPersistence.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionPersistence.java index f240c4510d..27d4266566 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionPersistence.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionPersistence.java @@ -39,4 +39,11 @@ public interface SubscriptionPersistence { * @return the DataNode as collection. */ Collection<DataNode> getDataNodesForSubscriptionEvent(); + + /** + * Get data nodes by xpath. + * + * @return the DataNode as collection. + */ + Collection<DataNode> getCmHandlesForSubscriptionEvent(String clientId, String subscriptionName); } diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionPersistenceImpl.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionPersistenceImpl.java index 9a063d6dfd..d2b1237a4d 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionPersistenceImpl.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionPersistenceImpl.java @@ -22,11 +22,18 @@ package org.onap.cps.ncmp.api.impl.subscriptions; import static org.onap.cps.ncmp.api.impl.constants.DmiRegistryConstants.NO_TIMESTAMP; +import java.io.Serializable; +import java.util.Arrays; import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import java.util.Optional; +import java.util.stream.Collectors; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.onap.cps.api.CpsDataService; +import org.onap.cps.ncmp.api.impl.utils.DataNodeHelper; import org.onap.cps.ncmp.api.impl.yangmodels.YangModelSubscriptionEvent; import org.onap.cps.spi.FetchDescendantsOption; import org.onap.cps.spi.model.DataNode; @@ -41,35 +48,86 @@ public class SubscriptionPersistenceImpl implements SubscriptionPersistence { private static final String SUBSCRIPTION_DATASPACE_NAME = "NCMP-Admin"; private static final String SUBSCRIPTION_ANCHOR_NAME = "AVC-Subscriptions"; private static final String SUBSCRIPTION_REGISTRY_PARENT = "/subscription-registry"; - private final JsonObjectMapper jsonObjectMapper; private final CpsDataService cpsDataService; @Override public void saveSubscriptionEvent(final YangModelSubscriptionEvent yangModelSubscriptionEvent) { - final String subscriptionEventJsonData = - createSubscriptionEventJsonData(jsonObjectMapper.asJsonString(yangModelSubscriptionEvent)); + final String clientId = yangModelSubscriptionEvent.getClientId(); + final String subscriptionName = yangModelSubscriptionEvent.getSubscriptionName(); + final Collection<DataNode> dataNodes = cpsDataService.getDataNodes(SUBSCRIPTION_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME, SUBSCRIPTION_REGISTRY_PARENT, FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS); + + if (isSubscriptionRegistryEmptyOrNonExist(dataNodes, clientId, subscriptionName)) { + saveSubscriptionEventYangModel(createSubscriptionEventJsonData( + jsonObjectMapper.asJsonString(yangModelSubscriptionEvent))); + } else { + findDeltaCmHandlesAddOrUpdateInDatabase(yangModelSubscriptionEvent, clientId, subscriptionName, dataNodes); + } + } + + private void findDeltaCmHandlesAddOrUpdateInDatabase(final YangModelSubscriptionEvent yangModelSubscriptionEvent, + final String clientId, final String subscriptionName, + final Collection<DataNode> dataNodes) { + final Map<String, SubscriptionStatus> cmHandleIdsFromYangModel = + extractCmHandleFromYangModelAsMap(yangModelSubscriptionEvent); + final Map<String, SubscriptionStatus> cmHandleIdsFromDatabase = + extractCmHandleFromDbAsMap(dataNodes); + + final Map<String, SubscriptionStatus> newCmHandles = + mapDifference(cmHandleIdsFromYangModel, cmHandleIdsFromDatabase); + traverseCmHandleList(newCmHandles, clientId, subscriptionName, true); + + final Map<String, SubscriptionStatus> existingCmHandles = + mapDifference(cmHandleIdsFromYangModel, newCmHandles); + traverseCmHandleList(existingCmHandles, clientId, subscriptionName, false); + } + + private boolean isSubscriptionRegistryEmptyOrNonExist(final Collection<DataNode> dataNodes, + final String clientId, final String subscriptionName) { final Optional<DataNode> dataNodeFirst = dataNodes.stream().findFirst(); - final boolean isCreateOperation = - dataNodeFirst.isPresent() && dataNodeFirst.get().getChildDataNodes().isEmpty(); - saveOrUpdateSubscriptionEventYangModel(subscriptionEventJsonData, isCreateOperation); + return ((dataNodeFirst.isPresent() && dataNodeFirst.get().getChildDataNodes().isEmpty()) + || getCmHandlesForSubscriptionEvent(clientId, subscriptionName).isEmpty()); + } + + private void traverseCmHandleList(final Map<String, SubscriptionStatus> cmHandleMap, + final String clientId, + final String subscriptionName, + final boolean isAddListElementOperation) { + final List<YangModelSubscriptionEvent.TargetCmHandle> cmHandleList = + targetCmHandlesAsList(cmHandleMap); + for (final YangModelSubscriptionEvent.TargetCmHandle targetCmHandle : cmHandleList) { + final String targetCmHandleAsJson = + createTargetCmHandleJsonData(jsonObjectMapper.asJsonString(targetCmHandle)); + addOrReplaceCmHandlePredicateListElement(targetCmHandleAsJson, clientId, subscriptionName, + isAddListElementOperation); + } } - private void saveOrUpdateSubscriptionEventYangModel(final String subscriptionEventJsonData, - final boolean isCreateOperation) { - if (isCreateOperation) { - log.info("SubscriptionEventJsonData to be saved into DB {}", subscriptionEventJsonData); - cpsDataService.saveListElements(SUBSCRIPTION_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME, - SUBSCRIPTION_REGISTRY_PARENT, subscriptionEventJsonData, NO_TIMESTAMP); + private void addOrReplaceCmHandlePredicateListElement(final String targetCmHandleAsJson, + final String clientId, + final String subscriptionName, + final boolean isAddListElementOperation) { + if (isAddListElementOperation) { + log.info("targetCmHandleAsJson to be added into DB {}", targetCmHandleAsJson); + cpsDataService.saveListElements(SUBSCRIPTION_DATASPACE_NAME, + SUBSCRIPTION_ANCHOR_NAME, createCmHandleXpathPredicates(clientId, subscriptionName), + targetCmHandleAsJson, NO_TIMESTAMP); } else { - log.info("SubscriptionEventJsonData to be updated into DB {}", subscriptionEventJsonData); - cpsDataService.updateDataNodeAndDescendants(SUBSCRIPTION_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME, - SUBSCRIPTION_REGISTRY_PARENT, subscriptionEventJsonData, NO_TIMESTAMP); + log.info("targetCmHandleAsJson to be updated into DB {}", targetCmHandleAsJson); + cpsDataService.updateNodeLeaves(SUBSCRIPTION_DATASPACE_NAME, + SUBSCRIPTION_ANCHOR_NAME, createCmHandleXpathPredicates(clientId, subscriptionName), + targetCmHandleAsJson, NO_TIMESTAMP); } } + private void saveSubscriptionEventYangModel(final String subscriptionEventJsonData) { + log.info("SubscriptionEventJsonData to be saved into DB {}", subscriptionEventJsonData); + cpsDataService.saveListElements(SUBSCRIPTION_DATASPACE_NAME, SUBSCRIPTION_ANCHOR_NAME, + SUBSCRIPTION_REGISTRY_PARENT, subscriptionEventJsonData, NO_TIMESTAMP); + } + @Override public Collection<DataNode> getDataNodesForSubscriptionEvent() { return cpsDataService.getDataNodes(SUBSCRIPTION_DATASPACE_NAME, @@ -77,7 +135,58 @@ public class SubscriptionPersistenceImpl implements SubscriptionPersistence { FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS); } + @Override + public Collection<DataNode> getCmHandlesForSubscriptionEvent(final String clientId, final String subscriptionName) { + return cpsDataService.getDataNodesForMultipleXpaths(SUBSCRIPTION_DATASPACE_NAME, + SUBSCRIPTION_ANCHOR_NAME, Arrays.asList(createCmHandleXpath(clientId, subscriptionName)), + FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS); + } + + private static Map<String, SubscriptionStatus> extractCmHandleFromDbAsMap(final Collection<DataNode> dataNodes) { + final List<Map<String, Serializable>> dataNodeLeaves = DataNodeHelper.getDataNodeLeaves(dataNodes); + final List<Collection<Serializable>> cmHandleIdToStatus = DataNodeHelper.getCmHandleIdToStatus(dataNodeLeaves); + return DataNodeHelper.getCmHandleIdToStatusMap(cmHandleIdToStatus); + } + + private static Map<String, SubscriptionStatus> extractCmHandleFromYangModelAsMap( + final YangModelSubscriptionEvent yangModelSubscriptionEvent) { + return yangModelSubscriptionEvent.getPredicates().getTargetCmHandles() + .stream().collect(Collectors.toMap( + YangModelSubscriptionEvent.TargetCmHandle::getCmHandleId, + YangModelSubscriptionEvent.TargetCmHandle::getStatus)); + } + + private static List<YangModelSubscriptionEvent.TargetCmHandle> targetCmHandlesAsList( + final Map<String, SubscriptionStatus> newCmHandles) { + return newCmHandles.entrySet().stream().map(entry -> + new YangModelSubscriptionEvent.TargetCmHandle(entry.getKey(), + entry.getValue())).collect(Collectors.toList()); + } + private static String createSubscriptionEventJsonData(final String yangModelSubscriptionAsJson) { return "{\"subscription\":[" + yangModelSubscriptionAsJson + "]}"; } + + private static String createTargetCmHandleJsonData(final String targetCmHandleAsJson) { + return "{\"targetCmHandles\":[" + targetCmHandleAsJson + "]}"; + } + + private static String createCmHandleXpathPredicates(final String clientId, final String subscriptionName) { + return "/subscription-registry/subscription[@clientID='" + clientId + + "' and @subscriptionName='" + subscriptionName + "']/predicates"; + } + + private static String createCmHandleXpath(final String clientId, final String subscriptionName) { + return "/subscription-registry/subscription[@clientID='" + clientId + + "' and @subscriptionName='" + subscriptionName + "']"; + } + + private static <K, V> Map<K, V> mapDifference(final Map<? extends K, ? extends V> left, + final Map<? extends K, ? extends V> right) { + final Map<K, V> difference = new HashMap<>(); + difference.putAll(left); + difference.putAll(right); + difference.entrySet().removeAll(right.entrySet()); + return difference; + } } diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionStatus.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionStatus.java index 0b4f91fac3..ce3b88ba03 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionStatus.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/subscriptions/SubscriptionStatus.java @@ -20,8 +20,36 @@ package org.onap.cps.ncmp.api.impl.subscriptions; +import java.io.Serializable; +import java.util.Iterator; +import java.util.Map; + public enum SubscriptionStatus { ACCEPTED, REJECTED, - PENDING + PENDING; + + + /** + * Populates a map with a key of cm handle id and a value of subscription status. + * + * @param resultMap the map is being populated + * @param bucketIterator to iterate over the collection + */ + public static void populateCmHandleToSubscriptionStatusMap(final Map<String, SubscriptionStatus> resultMap, + final Iterator<Serializable> bucketIterator) { + final String item = (String) bucketIterator.next(); + if ("PENDING".equals(item)) { + resultMap.put((String) bucketIterator.next(), + SubscriptionStatus.PENDING); + } + if ("REJECTED".equals(item)) { + resultMap.put((String) bucketIterator.next(), + SubscriptionStatus.REJECTED); + } + if ("ACCEPTED".equals(item)) { + resultMap.put((String) bucketIterator.next(), + SubscriptionStatus.ACCEPTED); + } + } } diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/DataNodeHelper.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/DataNodeHelper.java index 1648ac4fbb..8d44592ae2 100644 --- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/DataNodeHelper.java +++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/impl/utils/DataNodeHelper.java @@ -22,12 +22,15 @@ package org.onap.cps.ncmp.api.impl.utils; import java.io.Serializable; import java.util.Collection; +import java.util.HashMap; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.stream.Collectors; import java.util.stream.Stream; import lombok.AccessLevel; import lombok.NoArgsConstructor; +import org.onap.cps.ncmp.api.impl.subscriptions.SubscriptionStatus; import org.onap.cps.spi.model.DataNode; @NoArgsConstructor(access = AccessLevel.PRIVATE) @@ -72,4 +75,22 @@ public class DataNodeHelper { || col.contains("REJECTED")) .collect(Collectors.toList()); } + + /** + * The cm handle and status is returned as a map. + * + * @param cmHandleIdToStatus as a list of collection + * @return a map of cm handle id to status + */ + public static Map<String, SubscriptionStatus> getCmHandleIdToStatusMap( + final List<Collection<Serializable>> cmHandleIdToStatus) { + final Map<String, SubscriptionStatus> resultMap = new HashMap<>(); + for (final Collection<Serializable> cmHandleToStatusBucket: cmHandleIdToStatus) { + final Iterator<Serializable> bucketIterator = cmHandleToStatusBucket.iterator(); + while (bucketIterator.hasNext()) { + SubscriptionStatus.populateCmHandleToSubscriptionStatusMap(resultMap, bucketIterator); + } + } + return resultMap; + } } |