/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You 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 geode.kafka;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class GeodeConnectorConfig {

    //GeodeKafka Specific Configuration
    /**
     * Identifier for each task
     */
    public static final String TASK_ID = "GEODE_TASK_ID"; //One config per task
    /**
     * Specifies which Locators to connect to Apache Geode
     */
    public static final String LOCATORS = "locators";
    public static final String DEFAULT_LOCATOR = "localhost[10334]";
    public static final String SECURITY_CLIENT_AUTH_INIT = "security-client-auth-init";

    protected final int taskId;
    protected List<LocatorHostPort> locatorHostPorts;
    private String securityClientAuthInit;

    protected GeodeConnectorConfig() {
        taskId = 0;
    }

    public GeodeConnectorConfig(Map<String, String> connectorProperties) {
        taskId = Integer.parseInt(connectorProperties.get(TASK_ID));
        locatorHostPorts = parseLocators(connectorProperties.get(GeodeConnectorConfig.LOCATORS));
        securityClientAuthInit = connectorProperties.get(SECURITY_CLIENT_AUTH_INIT);
    }


    public static Map<String, List<String>> parseTopicToRegions(String combinedBindings) {
        //It's the same formatting, so parsing is the same going topic to region or region to topic
        return parseRegionToTopics(combinedBindings);
    }

    /**
     * Given a string of the form [region:topic,...] will produce a map where the key is the
     * regionName and the value is a list of topicNames to push values to
     *
     * @param combinedBindings a string with similar form to "[region:topic,...], [region2:topic2,...]
     * @return mapping of regionName to list of topics to update
     */
    public static Map<String, List<String>> parseRegionToTopics(String combinedBindings) {
        if (combinedBindings == null || combinedBindings.equals("")){
            return new HashMap();
        }
        List<String> bindings = parseBindings(combinedBindings);
        return bindings.stream().map(binding -> {
            String[] regionToTopicsArray = parseBinding(binding);
            return regionToTopicsArray;
        }).collect(Collectors.toMap(regionToTopicsArray -> regionToTopicsArray[0], regionToTopicsArray -> parseStringByComma(regionToTopicsArray[1])));
    }

    public static List<String> parseBindings(String bindings) {
        return Arrays.stream(bindings.split("](\\s)*,")).map((s) -> {
            s = s.replaceAll("\\[", "");
            s = s.replaceAll("\\]", "");
            s = s.trim();
            return s;
        }).collect(Collectors.toList());
    }

    private static String[] parseBinding(String binding) {
        return binding.split(":");
    }

    //Used to parse a string of topics or regions
    public static List<String> parseStringByComma(String string) {
        return parseStringBy(string, ",");
    }

    public static List<String> parseStringBy(String string, String regex) {
        return Arrays.stream(string.split(regex)).map((s) -> s.trim()).collect(Collectors.toList());
    }

    public static String reconstructString(Collection<String> strings) {
        return strings.stream().collect(Collectors.joining("],["));
    }

    List<LocatorHostPort> parseLocators(String locators) {
        return Arrays.stream(locators.split(",")).map((s) -> {
            String locatorString = s.trim();
            return parseLocator(locatorString);
        }).collect(Collectors.toList());
    }

    private LocatorHostPort parseLocator(String locatorString) {
        String[] splits = locatorString.split("\\[");
        String locator = splits[0];
        int port = Integer.parseInt(splits[1].replace("]", ""));
        return new LocatorHostPort(locator, port);
    }

    public int getTaskId() {
        return taskId;
    }

    public List<LocatorHostPort> getLocatorHostPorts() {
        return locatorHostPorts;
    }

    public String getSecurityClientAuthInit() {
        return securityClientAuthInit;
    }
}