package com.alibaba.otter.canal.client.adapter.config.common;

import java.io.IOException;
import java.io.Reader;
import java.util.*;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.util.Assert;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.constructor.Constructor;
import org.yaml.snakeyaml.nodes.MappingNode;
import org.yaml.snakeyaml.parser.ParserException;
import org.yaml.snakeyaml.reader.UnicodeReader;

/**
 * Base class for YAML factories.
 *
 * @author Dave Syer
 * @author Juergen Hoeller
 * @since 4.1
 */
public abstract class YamlProcessor {

    private final Log             logger           = LogFactory.getLog(getClass());

    private ResolutionMethod      resolutionMethod = ResolutionMethod.OVERRIDE;

    private Resource[]            resources        = new Resource[0];

    private List<DocumentMatcher> documentMatchers = Collections.emptyList();

    private boolean               matchDefault     = true;

    /**
     * A map of document matchers allowing callers to selectively use only some of
     * the documents in a YAML resource. In YAML documents are separated by
     * <code>---<code> lines, and each document is converted to properties before
     * the match is made. E.g.
     *
     * <pre class="code">
     * environment: dev
     * url: http://dev.bar.com
     * name: Developer Setup
     * ---
     * environment: prod
     * url:http://foo.bar.com
     * name: My Cool App
     * </pre>
     *
     * when mapped with
     *
     * <pre class="code">
     * setDocumentMatchers(properties -> ("prod"
     *     .equals(properties.getProperty("environment")) ? MatchStatus.FOUND : MatchStatus.NOT_FOUND));
     * </pre>
     *
     * would end up as
     *
     * <pre class="code">
     * environment=prod
     * url=http://foo.bar.com
     * name=My Cool App
     * </pre>
     */
    public void setDocumentMatchers(DocumentMatcher... matchers) {
        this.documentMatchers = Arrays.asList(matchers);
    }

    /**
     * Flag indicating that a document for which all the
     * {@link #setDocumentMatchers(DocumentMatcher...) document matchers} abstain
     * will nevertheless match. Default is {@code true}.
     */
    public void setMatchDefault(boolean matchDefault) {
        this.matchDefault = matchDefault;
    }

    /**
     * Method to use for resolving resources. Each resource will be converted to a
     * Map, so this property is used to decide which map entries to keep in the
     * final output from this factory. Default is {@link ResolutionMethod#OVERRIDE}.
     */
    public void setResolutionMethod(ResolutionMethod resolutionMethod) {
        Assert.notNull(resolutionMethod, "ResolutionMethod must not be null");
        this.resolutionMethod = resolutionMethod;
    }

    /**
     * Set locations of YAML {@link Resource resources} to be loaded.
     *
     * @see ResolutionMethod
     */
    public void setResources(Resource... resources) {
        this.resources = resources;
    }

    /**
     * Provide an opportunity for subclasses to process the Yaml parsed from the
     * supplied resources. Each resource is parsed in turn and the documents inside
     * checked against the {@link #setDocumentMatchers(DocumentMatcher...)
     * matchers}. If a document matches it is passed into the callback, along with
     * its representation as Properties. Depending on the
     * {@link #setResolutionMethod(ResolutionMethod)} not all of the documents will
     * be parsed.
     *
     * @param callback a callback to delegate to once matching documents are found
     * @see #createYaml()
     */
    protected void process(MatchCallback callback) {
        Yaml yaml = createYaml();
        for (Resource resource : this.resources) {
            boolean found = process(callback, yaml, resource);
            if (this.resolutionMethod == ResolutionMethod.FIRST_FOUND && found) {
                return;
            }
        }
    }

    /**
     * Create the {@link Yaml} instance to use.
     */
    protected Yaml createYaml() {
        return new Yaml(new StrictMapAppenderConstructor());
    }

    private boolean process(MatchCallback callback, Yaml yaml, Resource resource) {
        int count = 0;
        try {
            if (logger.isDebugEnabled()) {
                logger.debug("Loading from YAML: " + resource);
            }
            try (Reader reader = new UnicodeReader(resource.getInputStream())) {
                for (Object object : yaml.loadAll(reader)) {
                    if (object != null && process(asMap(object), callback)) {
                        count++;
                        if (this.resolutionMethod == ResolutionMethod.FIRST_FOUND) {
                            break;
                        }
                    }
                }
                if (logger.isDebugEnabled()) {
                    logger.debug(
                            "Loaded " + count + " document" + (count > 1 ? "s" : "") + " from YAML resource: " + resource);
                }
            }
        } catch (IOException ex) {
            handleProcessError(resource, ex);
        }
        return (count > 0);
    }

    private void handleProcessError(Resource resource, IOException ex) {
        if (this.resolutionMethod != ResolutionMethod.FIRST_FOUND
            && this.resolutionMethod != ResolutionMethod.OVERRIDE_AND_IGNORE) {
            throw new IllegalStateException(ex);
        }
        if (logger.isWarnEnabled()) {
            logger.warn("Could not load map from " + resource + ": " + ex.getMessage());
        }
    }

    @SuppressWarnings("unchecked")
    private Map<String, Object> asMap(Object object) {
        // YAML can have numbers as keys
        Map<String, Object> result = new LinkedHashMap<>();
        if (!(object instanceof Map)) {
            // A document can be a text literal
            result.put("document", object);
            return result;
        }

        Map<Object, Object> map = (Map<Object, Object>) object;
        for (Map.Entry<Object, Object> entry : map.entrySet()) {
            Object value = entry.getValue();
            if (value instanceof Map) {
                value = asMap(value);
            }
            Object key = entry.getKey();
            if (key instanceof CharSequence) {
                result.put(key.toString(), value);
            } else {
                // It has to be a map key in this case
                result.put("[" + key.toString() + "]", value);
            }
        }
        return result;
    }

    private boolean process(Map<String, Object> map, MatchCallback callback) {
        @SuppressWarnings("serial")
        Properties properties = new Properties() {

            @Override
            public String getProperty(String key) {
                Object value = get(key);
                return (value != null ? value.toString() : null);
            }
        };
        properties.putAll(getFlattenedMap(map));

        if (this.documentMatchers.isEmpty()) {
            if (logger.isDebugEnabled()) {
                logger.debug("Merging document (no matchers set): " + map);
            }
            callback.process(properties, map);
            return true;
        }

        MatchStatus result = MatchStatus.ABSTAIN;
        for (DocumentMatcher matcher : this.documentMatchers) {
            MatchStatus match = matcher.matches(properties);
            result = MatchStatus.getMostSpecific(match, result);
            if (match == MatchStatus.FOUND) {
                if (logger.isDebugEnabled()) {
                    logger.debug("Matched document with document matcher: " + properties);
                }
                callback.process(properties, map);
                return true;
            }
        }

        if (result == MatchStatus.ABSTAIN && this.matchDefault) {
            if (logger.isDebugEnabled()) {
                logger.debug("Matched document with default matcher: " + map);
            }
            callback.process(properties, map);
            return true;
        }

        if (logger.isDebugEnabled()) {
            logger.debug("Unmatched document: " + map);
        }
        return false;
    }

    /**
     * Return a flattened version of the given map, recursively following any nested
     * Map or Collection values. Entries from the resulting map retain the same
     * order as the source. When called with the Map from a {@link MatchCallback}
     * the result will contain the same values as the {@link MatchCallback}
     * Properties.
     *
     * @param source the source map
     * @return a flattened map
     * @since 4.1.3
     */
    protected final Map<String, Object> getFlattenedMap(Map<String, Object> source) {
        Map<String, Object> result = new LinkedHashMap<>();
        buildFlattenedMap(result, source, null);
        return result;
    }

    private static boolean containsText(CharSequence str) {
        int strLen = str.length();
        for (int i = 0; i < strLen; i++) {
            if (!Character.isWhitespace(str.charAt(i))) {
                return true;
            }
        }
        return false;
    }

    private void buildFlattenedMap(Map<String, Object> result, Map<String, Object> source, String path) {
        for (Map.Entry<String, Object> entry : source.entrySet()) {
            String key = entry.getKey();
            if (path != null && !path.isEmpty() && containsText(path)) {
                if (key.startsWith("[")) {
                    key = path + key;
                } else {
                    key = path + '.' + key;
                }
            }
            Object value = entry.getValue();
            if (value instanceof String) {
                result.put(key, value);
            } else if (value instanceof Map) {
                // Need a compound key
                @SuppressWarnings("unchecked")
                Map<String, Object> map = (Map<String, Object>) value;
                buildFlattenedMap(result, map, key);
            } else if (value instanceof Collection) {
                // Need a compound key
                @SuppressWarnings("unchecked")
                Collection<Object> collection = (Collection<Object>) value;
                int count = 0;
                for (Object object : collection) {
                    buildFlattenedMap(result, Collections.singletonMap("[" + (count++) + "]", object), key);
                }
            } else {
                result.put(key, (value != null ? value : ""));
            }
        }
    }

    /**
     * Callback interface used to process the YAML parsing results.
     */
    public interface MatchCallback {

        /**
         * Process the given representation of the parsing results.
         *
         * @param properties the properties to process (as a flattened representation
         *     with indexed keys in case of a collection or map)
         * @param map the result map (preserving the original value structure in the
         *     YAML document)
         */
        void process(Properties properties, Map<String, Object> map);
    }

    /**
     * Strategy interface used to test if properties match.
     */
    public interface DocumentMatcher {

        /**
         * Test if the given properties match.
         *
         * @param properties the properties to test
         * @return the status of the match
         */
        MatchStatus matches(Properties properties);
    }

    /**
     * Status returned from {@link DocumentMatcher#matches(Properties)}
     */
    public enum MatchStatus {

                             /**
                              * A match was found.
                              */
                             FOUND,

                             /**
                              * No match was found.
                              */
                             NOT_FOUND,

                             /**
                              * The matcher should not be considered.
                              */
                             ABSTAIN;

        /**
         * Compare two {@link MatchStatus} items, returning the most specific status.
         */
        public static MatchStatus getMostSpecific(MatchStatus a, MatchStatus b) {
            return (a.ordinal() < b.ordinal() ? a : b);
        }
    }

    /**
     * Method to use for resolving resources.
     */
    public enum ResolutionMethod {

                                  /**
                                   * Replace values from earlier in the list.
                                   */
                                  OVERRIDE,

                                  /**
                                   * Replace values from earlier in the list, ignoring any failures.
                                   */
                                  OVERRIDE_AND_IGNORE,

                                  /**
                                   * Take the first resource in the list that exists and use just that.
                                   */
                                  FIRST_FOUND
    }

    /**
     * A specialized {@link Constructor} that checks for duplicate keys.
     */
    protected static class StrictMapAppenderConstructor extends Constructor {

        // Declared as public for use in subclasses
        public StrictMapAppenderConstructor(){
            super();
        }

        @Override
        protected Map<Object, Object> constructMapping(MappingNode node) {
            try {
                return super.constructMapping(node);
            } catch (IllegalStateException ex) {
                throw new ParserException("while parsing MappingNode",
                    node.getStartMark(),
                    ex.getMessage(),
                    node.getEndMark());
            }
        }

        @Override
        protected Map<Object, Object> createDefaultMap() {
            final Map<Object, Object> delegate = super.createDefaultMap();
            return new AbstractMap<Object, Object>() {

                @Override
                public Object put(Object key, Object value) {
                    if (delegate.containsKey(key)) {
                        throw new IllegalStateException("Duplicate key: " + key);
                    }
                    return delegate.put(key, value);
                }

                @Override
                public Set<Entry<Object, Object>> entrySet() {
                    return delegate.entrySet();
                }
            };
        }
    }
}
