<commit_before>/*
 * Copyright 2015 Open Networking Laboratory
 *
 * 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.
 */
package org.onosproject.incubator.store.config.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.BooleanNode;
import com.fasterxml.jackson.databind.node.DoubleNode;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.LongNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.fasterxml.jackson.databind.node.ShortNode;
import com.fasterxml.jackson.databind.node.TextNode;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
import org.onlab.util.KryoNamespace;
import org.onosproject.incubator.net.config.Config;
import org.onosproject.incubator.net.config.ConfigApplyDelegate;
import org.onosproject.incubator.net.config.ConfigFactory;
import org.onosproject.incubator.net.config.NetworkConfigEvent;
import org.onosproject.incubator.net.config.NetworkConfigStore;
import org.onosproject.incubator.net.config.NetworkConfigStoreDelegate;
import org.onosproject.store.AbstractStore;
import org.onosproject.store.serializers.KryoNamespaces;
import org.onosproject.store.service.ConsistentMap;
import org.onosproject.store.service.MapEvent;
import org.onosproject.store.service.MapEventListener;
import org.onosproject.store.service.Serializer;
import org.onosproject.store.service.StorageService;
import org.onosproject.store.service.Versioned;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

import static org.onosproject.incubator.net.config.NetworkConfigEvent.Type.*;

/**
 * Implementation of a distributed network configuration store.
 */
@Component(immediate = true)
@Service
public class DistributedNetworkConfigStore
        extends AbstractStore<NetworkConfigEvent, NetworkConfigStoreDelegate>
        implements NetworkConfigStore {

    private final Logger log = LoggerFactory.getLogger(getClass());

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected StorageService storageService;

    private ConsistentMap<ConfigKey, ObjectNode> configs;

    private final Map<String, ConfigFactory> factoriesByConfig = Maps.newConcurrentMap();
    private final ObjectMapper mapper = new ObjectMapper();
    private final ConfigApplyDelegate applyDelegate = new InternalApplyDelegate();
    private final MapEventListener<ConfigKey, ObjectNode> listener = new InternalMapListener();

    @Activate
    public void activate() {
        KryoNamespace.Builder kryoBuilder = new KryoNamespace.Builder()
                .register(KryoNamespaces.API)
                .register(ConfigKey.class, ObjectNode.class, ArrayNode.class,
                          JsonNodeFactory.class, LinkedHashMap.class,
                          TextNode.class, BooleanNode.class,
                          LongNode.class, DoubleNode.class, ShortNode.class);

        configs = storageService.<ConfigKey, ObjectNode>consistentMapBuilder()
                .withSerializer(Serializer.using(kryoBuilder.build()))
                .withName("onos-network-configs")
                .build();
        configs.addListener(listener);
        log.info("Started");
    }

    @Deactivate
    public void deactivate() {
        configs.removeListener(listener);
        log.info("Stopped");
    }

    @Override
    public void addConfigFactory(ConfigFactory configFactory) {
        factoriesByConfig.put(configFactory.configClass().getName(), configFactory);
        notifyDelegate(new NetworkConfigEvent(CONFIG_REGISTERED, configFactory.configKey(),
                                              configFactory.configClass()));
    }

    @Override
    public void removeConfigFactory(ConfigFactory configFactory) {
        factoriesByConfig.remove(configFactory.configClass().getName());
        notifyDelegate(new NetworkConfigEvent(CONFIG_UNREGISTERED, configFactory.configKey(),
                                              configFactory.configClass()));
    }

    @Override
    @SuppressWarnings("unchecked")
    public <S, C extends Config<S>> ConfigFactory<S, C> getConfigFactory(Class<C> configClass) {
        return (ConfigFactory<S, C>) factoriesByConfig.get(configClass.getName());
    }

    @Override
    @SuppressWarnings("unchecked")
    public <S> Set<S> getSubjects(Class<S> subjectClass) {
        ImmutableSet.Builder<S> builder = ImmutableSet.builder();
        configs.keySet().forEach(k -> {
            if (subjectClass.isInstance(k.subject)) {
                builder.add((S) k.subject);
            }
        });
        return builder.build();
    }

    @Override
    @SuppressWarnings("unchecked")
    public <S, C extends Config<S>> Set<S> getSubjects(Class<S> subjectClass, Class<C> configClass) {
        ImmutableSet.Builder<S> builder = ImmutableSet.builder();
        String cName = configClass.getName();
        configs.keySet().forEach(k -> {
            if (subjectClass.isInstance(k.subject) && cName.equals(k.configClass)) {
                builder.add((S) k.subject);
            }
        });
        return builder.build();
    }

    @Override
    @SuppressWarnings("unchecked")
    public <S> Set<Class<? extends Config<S>>> getConfigClasses(S subject) {
        ImmutableSet.Builder<Class<? extends Config<S>>> builder = ImmutableSet.builder();
        configs.keySet().forEach(k -> {
            if (Objects.equals(subject, k.subject) && delegate != null) {
                builder.add(factoriesByConfig.get(k.configClass).configClass());
            }
        });
        return builder.build();
    }

    @Override
    public <S, T extends Config<S>> T getConfig(S subject, Class<T> configClass) {
        Versioned<ObjectNode> json = configs.get(key(subject, configClass));
        return json != null ? createConfig(subject, configClass, json.value()) : null;
    }


    @Override
    public <S, C extends Config<S>> C createConfig(S subject, Class<C> configClass) {
        Versioned<ObjectNode> json = configs.computeIfAbsent(key(subject, configClass),
                                                             k -> mapper.createObjectNode());
        return createConfig(subject, configClass, json.value());
    }

    @Override
    public <S, C extends Config<S>> C applyConfig(S subject, Class<C> configClass, ObjectNode json) {
        return createConfig(subject, configClass,
                            configs.putAndGet(key(subject, configClass), json).value());
    }

    @Override
    public <S, C extends Config<S>> void clearConfig(S subject, Class<C> configClass) {
        configs.remove(key(subject, configClass));
    }

    /**
     * Produces a config from the specified subject, config class and raw JSON.
     *
     * @param subject     config subject
     * @param configClass config class
     * @param json        raw JSON data
     * @return config object or null of no factory found or if the specified
     * JSON is null
     */<commit_msg>Added retry to the config store, to recover from CM timeout exception.<commit_after>/*
 * Copyright 2015 Open Networking Laboratory
 *
 * 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.
 */
package org.onosproject.incubator.store.config.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.BooleanNode;
import com.fasterxml.jackson.databind.node.DoubleNode;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.LongNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.fasterxml.jackson.databind.node.ShortNode;
import com.fasterxml.jackson.databind.node.TextNode;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
import org.onlab.util.KryoNamespace;
import org.onlab.util.Tools;
import org.onosproject.incubator.net.config.Config;
import org.onosproject.incubator.net.config.ConfigApplyDelegate;
import org.onosproject.incubator.net.config.ConfigFactory;
import org.onosproject.incubator.net.config.NetworkConfigEvent;
import org.onosproject.incubator.net.config.NetworkConfigStore;
import org.onosproject.incubator.net.config.NetworkConfigStoreDelegate;
import org.onosproject.store.AbstractStore;
import org.onosproject.store.serializers.KryoNamespaces;
import org.onosproject.store.service.ConsistentMap;
import org.onosproject.store.service.ConsistentMapException;
import org.onosproject.store.service.MapEvent;
import org.onosproject.store.service.MapEventListener;
import org.onosproject.store.service.Serializer;
import org.onosproject.store.service.StorageService;
import org.onosproject.store.service.Versioned;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

import static org.onosproject.incubator.net.config.NetworkConfigEvent.Type.*;

/**
 * Implementation of a distributed network configuration store.
 */
@Component(immediate = true)
@Service
public class DistributedNetworkConfigStore
        extends AbstractStore<NetworkConfigEvent, NetworkConfigStoreDelegate>
        implements NetworkConfigStore {

    private static final int MAX_BACKOFF = 10;

    private final Logger log = LoggerFactory.getLogger(getClass());

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected StorageService storageService;

    private ConsistentMap<ConfigKey, ObjectNode> configs;

    private final Map<String, ConfigFactory> factoriesByConfig = Maps.newConcurrentMap();
    private final ObjectMapper mapper = new ObjectMapper();
    private final ConfigApplyDelegate applyDelegate = new InternalApplyDelegate();
    private final MapEventListener<ConfigKey, ObjectNode> listener = new InternalMapListener();

    @Activate
    public void activate() {
        KryoNamespace.Builder kryoBuilder = new KryoNamespace.Builder()
                .register(KryoNamespaces.API)
                .register(ConfigKey.class, ObjectNode.class, ArrayNode.class,
                          JsonNodeFactory.class, LinkedHashMap.class,
                          TextNode.class, BooleanNode.class,
                          LongNode.class, DoubleNode.class, ShortNode.class);

        configs = storageService.<ConfigKey, ObjectNode>consistentMapBuilder()
                .withSerializer(Serializer.using(kryoBuilder.build()))
                .withName("onos-network-configs")
                .build();
        configs.addListener(listener);
        log.info("Started");
    }

    @Deactivate
    public void deactivate() {
        configs.removeListener(listener);
        log.info("Stopped");
    }

    @Override
    public void addConfigFactory(ConfigFactory configFactory) {
        factoriesByConfig.put(configFactory.configClass().getName(), configFactory);
        notifyDelegate(new NetworkConfigEvent(CONFIG_REGISTERED, configFactory.configKey(),
                                              configFactory.configClass()));
    }

    @Override
    public void removeConfigFactory(ConfigFactory configFactory) {
        factoriesByConfig.remove(configFactory.configClass().getName());
        notifyDelegate(new NetworkConfigEvent(CONFIG_UNREGISTERED, configFactory.configKey(),
                                              configFactory.configClass()));
    }

    @Override
    @SuppressWarnings("unchecked")
    public <S, C extends Config<S>> ConfigFactory<S, C> getConfigFactory(Class<C> configClass) {
        return (ConfigFactory<S, C>) factoriesByConfig.get(configClass.getName());
    }

    @Override
    @SuppressWarnings("unchecked")
    public <S> Set<S> getSubjects(Class<S> subjectClass) {
        ImmutableSet.Builder<S> builder = ImmutableSet.builder();
        configs.keySet().forEach(k -> {
            if (subjectClass.isInstance(k.subject)) {
                builder.add((S) k.subject);
            }
        });
        return builder.build();
    }

    @Override
    @SuppressWarnings("unchecked")
    public <S, C extends Config<S>> Set<S> getSubjects(Class<S> subjectClass, Class<C> configClass) {
        ImmutableSet.Builder<S> builder = ImmutableSet.builder();
        String cName = configClass.getName();
        configs.keySet().forEach(k -> {
            if (subjectClass.isInstance(k.subject) && cName.equals(k.configClass)) {
                builder.add((S) k.subject);
            }
        });
        return builder.build();
    }

    @Override
    @SuppressWarnings("unchecked")
    public <S> Set<Class<? extends Config<S>>> getConfigClasses(S subject) {
        ImmutableSet.Builder<Class<? extends Config<S>>> builder = ImmutableSet.builder();
        configs.keySet().forEach(k -> {
            if (Objects.equals(subject, k.subject) && delegate != null) {
                builder.add(factoriesByConfig.get(k.configClass).configClass());
            }
        });
        return builder.build();
    }

    @Override
    public <S, T extends Config<S>> T getConfig(S subject, Class<T> configClass) {
        // FIXME: There has to be a better way to absorb the timeout exceptions.
        Versioned<ObjectNode> json = null;
        try {
            json = configs.get(key(subject, configClass));
        } catch (ConsistentMapException.Timeout e) {
            Tools.randomDelay(MAX_BACKOFF);
            json = configs.get(key(subject, configClass));
        }
        return json != null ? createConfig(subject, configClass, json.value()) : null;
    }


    @Override
    public <S, C extends Config<S>> C createConfig(S subject, Class<C> configClass) {
        Versioned<ObjectNode> json = configs.computeIfAbsent(key(subject, configClass),
                                                             k -> mapper.createObjectNode());
        return createConfig(subject, configClass, json.value());
    }

    @Override
    public <S, C extends Config<S>> C applyConfig(S subject, Class<C> configClass, ObjectNode json) {
        return createConfig(subject, configClass,
                            configs.putAndGet(key(subject, configClass), json).value());
    }

    @Override
    public <S, C extends Config<S>> void clearConfig(S subject, Class<C> configClass) {
        configs.remove(key(subject, configClass));
    }

    /**
     * Produces a config from the specified subject, config class and raw JSON.
     *
     * @param subject     config subject
     * @param configClass config class
     * @param json        raw JSON data
     * @return config object or null of no factory found or if the specified
     * JSON is null
     */
{'fraction_non_alphanumeric': 0.10252356339312861, 'fraction_numerical': 0.0013377926421404682, 'mean_word_length': 3.30298273155416, 'pattern_counts': {'":': 0, '<': 85, '<?xml version=': 0, '>': 93, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 18}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7595672', 'n_tokens_mistral': 4530, 'n_tokens_neox': 4383, 'n_words': 990}