/*
 * 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 org.apache.geode.distributed.internal;

import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_MANAGER;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_POST_PROCESSOR;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileFilter;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.UnaryOperator;
import java.util.stream.Collectors;

import javax.xml.parsers.ParserConfigurationException;
import javax.xml.transform.TransformerException;

import joptsimple.internal.Strings;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.filefilter.DirectoryFileFilter;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.Logger;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.NodeList;
import org.xml.sax.SAXException;

import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.DiskStore;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.configuration.CacheConfig;
import org.apache.geode.distributed.ConfigurationPersistenceService;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.internal.locks.DLockService;
import org.apache.geode.internal.cache.ClusterConfigurationLoader;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalRegionArguments;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.persistence.PersistentMemberManager;
import org.apache.geode.internal.cache.persistence.PersistentMemberPattern;
import org.apache.geode.internal.cache.xmlcache.CacheXmlGenerator;
import org.apache.geode.internal.config.JAXBService;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.management.internal.configuration.callbacks.ConfigurationChangeListener;
import org.apache.geode.management.internal.configuration.domain.Configuration;
import org.apache.geode.management.internal.configuration.domain.SharedConfigurationStatus;
import org.apache.geode.management.internal.configuration.domain.XmlEntity;
import org.apache.geode.management.internal.configuration.messages.ConfigurationResponse;
import org.apache.geode.management.internal.configuration.messages.SharedConfigurationStatusResponse;
import org.apache.geode.management.internal.configuration.utils.XmlUtils;

public class InternalConfigurationPersistenceService implements ConfigurationPersistenceService {
  private static final Logger logger = LogService.getLogger();

  /**
   * Name of the directory where the shared configuration artifacts are stored
   */
  public static final String CLUSTER_CONFIG_ARTIFACTS_DIR_NAME = "cluster_config";

  private static final String CLUSTER_CONFIG_DISK_STORE_NAME = "cluster_config";

  public static final String CLUSTER_CONFIG_DISK_DIR_PREFIX = "ConfigDiskDir_";

  /**
   * Name of the lock service used for shared configuration
   */
  private static final String SHARED_CONFIG_LOCK_SERVICE_NAME = "__CLUSTER_CONFIG_LS";

  /**
   * Name of the lock for locking the shared configuration
   */
  private static final String SHARED_CONFIG_LOCK_NAME = "__CLUSTER_CONFIG_LOCK";

  /**
   * Name of the region which is used to store the configuration information
   */
  public static final String CONFIG_REGION_NAME = "_ConfigurationRegion";
  private static final String CACHE_CONFIG_VERSION = "1.0";

  private final Path configDirPath;
  private final Path configDiskDirPath;

  private final Set<PersistentMemberPattern> newerSharedConfigurationLocatorInfo = new HashSet<>();
  private final AtomicReference<SharedConfigurationStatus> status = new AtomicReference<>();

  private final InternalCache cache;
  private final DistributedLockService sharedConfigLockingService;
  private final JAXBService jaxbService;

  public InternalConfigurationPersistenceService(InternalCache cache, Path workingDirectory,
      JAXBService jaxbService) {
    this(cache,
        sharedConfigLockService(cache.getDistributedSystem()),
        jaxbService,
        workingDirectory.resolve(CLUSTER_CONFIG_ARTIFACTS_DIR_NAME),
        workingDirectory
            .resolve(CLUSTER_CONFIG_DISK_DIR_PREFIX + cache.getDistributedSystem().getName()));
  }

  @VisibleForTesting
  public InternalConfigurationPersistenceService(JAXBService jaxbService) {
    this(null, null, jaxbService, null, null);
  }

  private InternalConfigurationPersistenceService(InternalCache cache,
      DistributedLockService sharedConfigLockingService, JAXBService jaxbService,
      Path configDirPath, Path configDiskDirPath) {
    this.cache = cache;
    this.configDirPath = configDirPath;
    this.configDiskDirPath = configDiskDirPath;
    this.sharedConfigLockingService = sharedConfigLockingService;
    status.set(SharedConfigurationStatus.NOT_STARTED);
    this.jaxbService = jaxbService;
  }

  /**
   * Gets or creates (if not created) shared configuration lock service
   */
  private static DistributedLockService sharedConfigLockService(DistributedSystem ds) {
    DistributedLockService sharedConfigDls =
        DLockService.getServiceNamed(SHARED_CONFIG_LOCK_SERVICE_NAME);
    try {
      if (sharedConfigDls == null) {
        sharedConfigDls = DLockService.create(SHARED_CONFIG_LOCK_SERVICE_NAME,
            (InternalDistributedSystem) ds, true, true);
      }
    } catch (IllegalArgumentException ignore) {
      return DLockService.getServiceNamed(SHARED_CONFIG_LOCK_SERVICE_NAME);
    }
    return sharedConfigDls;
  }

  public JAXBService getJaxbService() {
    return jaxbService;
  }

  /**
   * Adds/replaces the xml entity in the shared configuration we don't need to trigger the change
   * listener for this modification, so it's ok to operate on the original configuration object
   */
  public void addXmlEntity(XmlEntity xmlEntity, String[] groups) {
    lockSharedConfiguration();
    try {
      Region<String, Configuration> configRegion = getConfigurationRegion();
      if (groups == null || groups.length == 0) {
        groups = new String[] {ConfigurationPersistenceService.CLUSTER_CONFIG};
      }
      for (String group : groups) {
        Configuration configuration = configRegion.get(group);
        if (configuration == null) {
          configuration = new Configuration(group);
        }
        String xmlContent = configuration.getCacheXmlContent();
        if (xmlContent == null || xmlContent.isEmpty()) {
          xmlContent = generateInitialXmlContent();
        }
        try {
          final Document doc = XmlUtils.createAndUpgradeDocumentFromXml(xmlContent);
          XmlUtils.addNewNode(doc, xmlEntity);
          configuration.setCacheXmlContent(XmlUtils.prettyXml(doc));
          configRegion.put(group, configuration);
        } catch (Exception e) {
          logger.error("error updating cluster configuration for group {}", group, e);
        }
      }
    } finally {
      unlockSharedConfiguration();
    }
  }

  /**
   * Deletes the xml entity from the shared configuration.
   */
  public void deleteXmlEntity(final XmlEntity xmlEntity, String[] groups) {
    lockSharedConfiguration();
    try {
      Region<String, Configuration> configRegion = getConfigurationRegion();
      // No group is specified, so delete in every single group if it exists.
      if (groups == null) {
        Set<String> groupSet = configRegion.keySet();
        groups = groupSet.toArray(new String[groupSet.size()]);
      }
      for (String group : groups) {
        Configuration configuration = configRegion.get(group);
        if (configuration != null) {
          String xmlContent = configuration.getCacheXmlContent();
          try {
            if (xmlContent != null && !xmlContent.isEmpty()) {
              Document doc = XmlUtils.createAndUpgradeDocumentFromXml(xmlContent);
              XmlUtils.deleteNode(doc, xmlEntity);
              configuration.setCacheXmlContent(XmlUtils.prettyXml(doc));
              configRegion.put(group, configuration);
            }
          } catch (Exception e) {
            logger.error("error updating cluster configuration for group {}", group, e);
          }
        }
      }
    } finally {
      unlockSharedConfiguration();
    }
  }

  /**
   * we don't need to trigger the change listener for this modification, so it's ok to operate on
   * the original configuration object
   */
  public void modifyXmlAndProperties(Properties properties, XmlEntity xmlEntity, String[] groups) {
    lockSharedConfiguration();
    try {
      if (groups == null) {
        groups = new String[] {ConfigurationPersistenceService.CLUSTER_CONFIG};
      }
      Region<String, Configuration> configRegion = getConfigurationRegion();
      for (String group : groups) {
        Configuration configuration = configRegion.get(group);
        if (configuration == null) {
          configuration = new Configuration(group);
        }

        if (xmlEntity != null) {
          String xmlContent = configuration.getCacheXmlContent();
          if (xmlContent == null || xmlContent.isEmpty()) {
            StringWriter sw = new StringWriter();
            PrintWriter pw = new PrintWriter(sw);
            CacheXmlGenerator.generateDefault(pw);
            xmlContent = sw.toString();
          }
          try {
            Document doc = XmlUtils.createAndUpgradeDocumentFromXml(xmlContent);
            // Modify the cache attributes
            XmlUtils.modifyRootAttributes(doc, xmlEntity);
            // Change the xml content of the configuration and put it the config region
            configuration.setCacheXmlContent(XmlUtils.prettyXml(doc));
          } catch (Exception e) {
            logger.error("error updating cluster configuration for group {}", group, e);
          }
        }

        if (properties != null) {
          configuration.getGemfireProperties().putAll(properties);
        }
        configRegion.put(group, configuration);
      }
    } finally {
      unlockSharedConfiguration();
    }
  }

  /**
   * Add jar information into the shared configuration and save the jars in the file system used
   * when deploying jars
   */
  public void addJarsToThisLocator(List<String> jarFullPaths, String[] groups) throws IOException {
    lockSharedConfiguration();
    try {
      if (groups == null) {
        groups = new String[] {ConfigurationPersistenceService.CLUSTER_CONFIG};
      }
      Region<String, Configuration> configRegion = getConfigurationRegion();
      for (String group : groups) {
        Configuration configuration = configRegion.get(group);

        if (configuration == null) {
          configuration = new Configuration(group);
          createConfigDirIfNecessary(group);
        }

        Path groupDir = configDirPath.resolve(group);
        Set<String> jarNames = new HashSet<>();
        for (String jarFullPath : jarFullPaths) {
          File stagedJar = new File(jarFullPath);
          jarNames.add(stagedJar.getName());
          Path filePath = groupDir.resolve(stagedJar.getName());
          FileUtils.copyFile(stagedJar, filePath.toFile());
        }

        // update the record after writing the jars to the file system, since the listener
        // will need the jars on file to upload to other locators. Need to update the jars
        // using a new copy of the Configuration so that the change listener will pick up the jar
        // name changes.
        String memberId = cache.getMyId().getId();

        Configuration configurationCopy = new Configuration(configuration);
        configurationCopy.addJarNames(jarNames);
        configRegion.put(group, configurationCopy, memberId);
      }
    } finally {
      unlockSharedConfiguration();
    }
  }

  /**
   * Removes the jar files from the shared configuration. used when undeploy jars
   *
   * @param jarNames Names of the jar files.
   * @param groups Names of the groups which had the jar file deployed.
   * @return true on success.
   */
  public boolean removeJars(final String[] jarNames, String[] groups) {
    lockSharedConfiguration();
    boolean success = true;
    try {
      Region<String, Configuration> configRegion = getConfigurationRegion();
      if (groups == null) {
        groups = configRegion.keySet().stream().toArray(String[]::new);
      }
      for (String group : groups) {
        Configuration configuration = configRegion.get(group);
        if (configuration == null) {
          break;
        }

        for (String jarRemoved : jarNames) {
          File jar = getPathToJarOnThisLocator(group, jarRemoved).toFile();
          if (jar.exists()) {
            try {
              FileUtils.forceDelete(jar);
            } catch (IOException e) {
              logger.error(
                  "Exception occurred while attempting to delete a jar from the filesystem: {}",
                  jarRemoved, e);
            }
          }
        }

        Configuration configurationCopy = new Configuration(configuration);
        configurationCopy.removeJarNames(jarNames);
        configRegion.put(group, configurationCopy);
      }
    } catch (Exception e) {
      logger.info("Exception occurred while deleting the jar files", e);
      success = false;
    } finally {
      unlockSharedConfiguration();
    }
    return success;
  }

  // Only used when a locator is initially starting up
  public void downloadJarFromOtherLocators(String groupName, String jarName)
      throws IllegalStateException, IOException {
    logger.info("Getting Jar files from other locators");
    DistributionManager dm = cache.getDistributionManager();
    DistributedMember me = cache.getMyId();
    List<DistributedMember> locators =
        new ArrayList<>(dm.getAllHostedLocatorsWithSharedConfiguration().keySet());
    locators.remove(me);

    createConfigDirIfNecessary(groupName);

    if (locators.isEmpty()) {
      throw new IllegalStateException(
          "Request to download jar " + jarName + " but no other locators are present");
    }

    downloadJarFromLocator(groupName, jarName, locators.get(0));
  }

  // used in the cluster config change listener when jarnames are changed in the internal region
  public void downloadJarFromLocator(String groupName, String jarName,
      DistributedMember sourceLocator) throws IllegalStateException, IOException {
    logger.info("Downloading jar {} from locator {}", jarName, sourceLocator.getName());

    createConfigDirIfNecessary(groupName);

    File jarFile = downloadJar(sourceLocator, groupName, jarName);

    File jarToWrite = getPathToJarOnThisLocator(groupName, jarName).toFile();
    Files.copy(jarFile.toPath(), jarToWrite.toPath(), StandardCopyOption.REPLACE_EXISTING);
  }

  /**
   * Retrieve a deployed jar from a locator. The retrieved file is staged in a temporary location.
   *
   * @param locator the DistributedMember
   * @param groupName the group to use when retrieving the jar
   * @param jarName the name of the deployed jar
   * @return a File referencing the downloaded jar. The File is downloaded to a temporary location.
   */
  public File downloadJar(DistributedMember locator, String groupName, String jarName)
      throws IOException {
    ClusterConfigurationLoader loader = new ClusterConfigurationLoader();
    return loader.downloadJar(locator, groupName, jarName);
  }

  /**
   * Creates the shared configuration service
   *
   * @param loadSharedConfigFromDir when set to true, loads the configuration from the share_config
   *        directory
   */
  void initSharedConfiguration(boolean loadSharedConfigFromDir) throws IOException {
    status.set(SharedConfigurationStatus.STARTED);
    Region<String, Configuration> configRegion = getConfigurationRegion();
    lockSharedConfiguration();
    try {
      removeInvalidXmlConfigurations(configRegion);
      if (loadSharedConfigFromDir) {
        logger.info("Reading cluster configuration from '{}' directory",
            InternalConfigurationPersistenceService.CLUSTER_CONFIG_ARTIFACTS_DIR_NAME);
        loadSharedConfigurationFromDir(configDirPath.toFile());
      } else {
        persistSecuritySettings(configRegion);
        // for those groups that have jar files, need to download the jars from other locators
        // if it doesn't exist yet
        for (Entry<String, Configuration> stringConfigurationEntry : configRegion.entrySet()) {
          Configuration config = stringConfigurationEntry.getValue();
          for (String jar : config.getJarNames()) {
            if (!getPathToJarOnThisLocator(stringConfigurationEntry.getKey(), jar).toFile()
                .exists()) {
              downloadJarFromOtherLocators(stringConfigurationEntry.getKey(), jar);
            }
          }
        }
      }
    } finally {
      unlockSharedConfiguration();
    }

    status.set(SharedConfigurationStatus.RUNNING);
  }

  void removeInvalidXmlConfigurations(Region<String, Configuration> configRegion)
      throws IOException {
    for (Map.Entry<String, Configuration> entry : configRegion.entrySet()) {
      String group = entry.getKey();
      Configuration configuration = entry.getValue();
      String configurationXml = configuration.getCacheXmlContent();
      if (configurationXml != null && !configurationXml.isEmpty()) {
        try {
          Document document = XmlUtils.createDocumentFromXml(configurationXml);
          boolean removedInvalidReceivers = removeInvalidGatewayReceivers(document);
          boolean removedDuplicateReceivers = removeDuplicateGatewayReceivers(document);
          if (removedInvalidReceivers || removedDuplicateReceivers) {
            configuration.setCacheXmlContent(XmlUtils.prettyXml(document));
            configRegion.put(group, configuration);
          }
        } catch (SAXException | TransformerException | ParserConfigurationException e) {
          throw new IOException("Unable to parse existing cluster configuration from disk. ", e);
        }
      }
    }
  }

  boolean removeInvalidGatewayReceivers(Document document) throws TransformerException {
    boolean modified = false;
    NodeList receiverNodes = document.getElementsByTagName("gateway-receiver");
    for (int i = receiverNodes.getLength() - 1; i >= 0; i--) {
      Element receiverElement = (Element) receiverNodes.item(i);

      // Check hostname-for-senders
      String hostNameForSenders = receiverElement.getAttribute("hostname-for-senders");
      if (StringUtils.isNotBlank(hostNameForSenders)) {
        receiverElement.getParentNode().removeChild(receiverElement);
        logger.info("Removed invalid cluster configuration gateway-receiver element="
            + XmlUtils.prettyXml(receiverElement));
        modified = true;
      }

      // Check bind-address
      String bindAddress = receiverElement.getAttribute("bind-address");
      if (StringUtils.isNotBlank(bindAddress) && !bindAddress.equals("0.0.0.0")) {
        receiverElement.getParentNode().removeChild(receiverElement);
        logger.info("Removed invalid cluster configuration gateway-receiver element="
            + XmlUtils.prettyXml(receiverElement));
        modified = true;
      }
    }
    return modified;
  }

  boolean removeDuplicateGatewayReceivers(Document document) throws TransformerException {
    boolean modified = false;
    NodeList receiverNodes = document.getElementsByTagName("gateway-receiver");
    while (receiverNodes.getLength() > 1) {
      Element receiverElement = (Element) receiverNodes.item(0);
      receiverElement.getParentNode().removeChild(receiverElement);
      logger.info("Removed duplicate cluster configuration gateway-receiver element="
          + XmlUtils.prettyXml(receiverElement));
      modified = true;
      receiverNodes = document.getElementsByTagName("gateway-receiver");
    }
    return modified;
  }

  private void persistSecuritySettings(final Region<String, Configuration> configRegion) {
    Properties securityProps = cache.getDistributedSystem().getSecurityProperties();

    Configuration clusterPropertiesConfig =
        configRegion.get(ConfigurationPersistenceService.CLUSTER_CONFIG);
    if (clusterPropertiesConfig == null) {
      clusterPropertiesConfig = new Configuration(ConfigurationPersistenceService.CLUSTER_CONFIG);
      configRegion.put(ConfigurationPersistenceService.CLUSTER_CONFIG, clusterPropertiesConfig);
    }
    // put security-manager and security-post-processor in the cluster config
    Properties clusterProperties = clusterPropertiesConfig.getGemfireProperties();

    if (securityProps.containsKey(SECURITY_MANAGER)) {
      clusterProperties.setProperty(SECURITY_MANAGER, securityProps.getProperty(SECURITY_MANAGER));
    }
    if (securityProps.containsKey(SECURITY_POST_PROCESSOR)) {
      clusterProperties.setProperty(SECURITY_POST_PROCESSOR,
          securityProps.getProperty(SECURITY_POST_PROCESSOR));
    }
  }

  /**
   * Creates a ConfigurationResponse based on the configRequest, configuration response contains the
   * requested shared configuration This method locks the ConfigurationPersistenceService
   */
  public ConfigurationResponse createConfigurationResponse(Set<String> groups) {
    ConfigurationResponse configResponse = null;

    boolean isLocked = lockSharedConfiguration();
    try {
      if (isLocked) {
        configResponse = new ConfigurationResponse();
        groups.add(ConfigurationPersistenceService.CLUSTER_CONFIG);
        logger.info("Building up configuration response with following configurations: {}", groups);

        for (String group : groups) {
          Configuration configuration = getConfiguration(group);
          configResponse.addConfiguration(configuration);
          if (configuration != null) {
            configResponse.addJar(group, configuration.getJarNames());
          }
        }

        return configResponse;
      }
    } finally {
      unlockSharedConfiguration();
    }
    return configResponse;
  }

  /**
   * Create a response containing the status of the Shared configuration and information about other
   * locators containing newer shared configuration data (if at all)
   *
   * @return {@link SharedConfigurationStatusResponse} containing the
   *         {@link SharedConfigurationStatus}
   */
  SharedConfigurationStatusResponse createStatusResponse() {
    SharedConfigurationStatusResponse response = new SharedConfigurationStatusResponse();
    response.setStatus(getStatus());
    response.addWaitingLocatorInfo(newerSharedConfigurationLocatorInfo);
    return response;
  }

  /**
   * For tests only. TODO: clean this up and remove from production code
   * <p>
   * Throws {@code AssertionError} wrapping any exception thrown by operation.
   */
  public void destroySharedConfiguration() {
    try {
      Region<String, Configuration> configRegion = getConfigurationRegion();
      if (configRegion != null) {
        configRegion.destroyRegion();
      }
      DiskStore configDiskStore = cache.findDiskStore(CLUSTER_CONFIG_ARTIFACTS_DIR_NAME);
      if (configDiskStore != null) {
        configDiskStore.destroy();
      }
      FileUtils.deleteDirectory(configDirPath.toFile());
    } catch (Exception exception) {
      throw new AssertionError(exception);
    }
  }

  public Path getPathToJarOnThisLocator(String groupName, String jarName) {
    return configDirPath.resolve(groupName).resolve(jarName);
  }

  public Configuration getConfiguration(String groupName) {
    return getConfigurationRegion().get(groupName);
  }

  public void setConfiguration(String groupName, Configuration configuration) {
    getConfigurationRegion().put(groupName, configuration);
  }

  public boolean hasXmlConfiguration() {
    Region<String, Configuration> configRegion = getConfigurationRegion();
    return configRegion.values().stream().anyMatch(c -> c.getCacheXmlContent() != null);
  }

  public Map<String, Configuration> getEntireConfiguration() {
    Set<String> keys = getConfigurationRegion().keySet();
    return getConfigurationRegion().getAll(keys);
  }

  public Set<String> getGroups() {
    return getConfigurationRegion().keySet();
  }

  public Path getClusterConfigDirPath() {
    return configDirPath;
  }

  /**
   * Gets the current status of the ConfigurationPersistenceService If the status is started , it
   * determines if the shared configuration is waiting for new configuration on other locators
   *
   * @return {@link SharedConfigurationStatus}
   */
  public SharedConfigurationStatus getStatus() {
    SharedConfigurationStatus scStatus = status.get();
    if (scStatus == SharedConfigurationStatus.STARTED) {
      PersistentMemberManager pmm = cache.getPersistentMemberManager();
      Map<String, Set<PersistentMemberID>> waitingRegions = pmm.getWaitingRegions();
      if (!waitingRegions.isEmpty()) {
        status.compareAndSet(SharedConfigurationStatus.STARTED,
            SharedConfigurationStatus.WAITING);
        Set<PersistentMemberID> persMemIds =
            waitingRegions.get(Region.SEPARATOR_CHAR + CONFIG_REGION_NAME);
        for (PersistentMemberID persMemId : persMemIds) {
          newerSharedConfigurationLocatorInfo.add(new PersistentMemberPattern(persMemId));
        }
      }
    }
    return status.get();
  }

  // configDir is the dir that has all the groups structure underneath it.
  public void loadSharedConfigurationFromDir(File configDir) throws IOException {
    lockSharedConfiguration();
    try {
      File[] groupNames = configDir.listFiles((FileFilter) DirectoryFileFilter.INSTANCE);
      boolean needToCopyJars = true;
      if (configDir.getAbsolutePath().equals(configDirPath)) {
        needToCopyJars = false;
      }

      logger.info("loading the cluster configuration: ");
      Map<String, Configuration> sharedConfiguration = new HashMap<>();
      for (File groupName : groupNames) {
        Configuration configuration = readConfiguration(groupName);
        logger.info(configuration.getConfigName() + " xml content: \n"
            + configuration.getCacheXmlContent());
        logger.info(configuration.getConfigName() + " properties: "
            + configuration.getGemfireProperties().size());
        logger.info(configuration.getConfigName() + " jars: "
            + Strings.join(configuration.getJarNames(), ", "));
        sharedConfiguration.put(groupName.getName(), configuration);
        if (needToCopyJars && configuration.getJarNames().size() > 0) {
          Path groupDirPath = createConfigDirIfNecessary(configuration.getConfigName()).toPath();
          for (String jarName : configuration.getJarNames()) {
            Files.copy(groupName.toPath().resolve(jarName), groupDirPath.resolve(jarName));
          }
        }
      }
      Region<String, Configuration> clusterRegion = getConfigurationRegion();
      clusterRegion.clear();

      String memberId = cache.getMyId().getId();
      clusterRegion.putAll(sharedConfiguration, memberId);

      // Overwrite the security settings using the locator's properties, ignoring whatever
      // in the import
      persistSecuritySettings(clusterRegion);
    } finally {
      unlockSharedConfiguration();
    }
  }

  // Write the content of xml and properties into the file system for exporting purpose
  public void writeConfigToFile(final Configuration configuration, File rootDir)
      throws IOException {
    File configDir = createConfigDirIfNecessary(rootDir, configuration.getConfigName());

    File propsFile = new File(configDir, configuration.getPropertiesFileName());
    BufferedWriter bw = new BufferedWriter(new FileWriter(propsFile));
    configuration.getGemfireProperties().store(bw, null);
    bw.close();

    File xmlFile = new File(configDir, configuration.getCacheXmlFileName());
    FileUtils.writeStringToFile(xmlFile, configuration.getCacheXmlContent(), "UTF-8");

    // copy the jars if the rootDir is different than the configDirPath
    if (rootDir.getAbsolutePath().equals(configDirPath)) {
      return;
    }

    File locatorConfigDir = configDirPath.resolve(configuration.getConfigName()).toFile();
    if (locatorConfigDir.exists()) {
      File[] jarFiles = locatorConfigDir.listFiles(x -> x.getName().endsWith(".jar"));
      for (File file : jarFiles) {
        Files.copy(file.toPath(), configDir.toPath().resolve(file.getName()));
      }
    }
  }

  public boolean lockSharedConfiguration() {
    return sharedConfigLockingService.lock(SHARED_CONFIG_LOCK_NAME, -1, -1);
  }

  public void unlockSharedConfiguration() {
    sharedConfigLockingService.unlock(SHARED_CONFIG_LOCK_NAME);
  }

  /**
   * Gets the region containing the shared configuration data. The region is created , if it does
   * not exist already. Note : this could block if this locator contains stale persistent
   * configuration data.
   *
   * @return {@link Region} ConfigurationRegion, this should never be null
   */
  public Region<String, Configuration> getConfigurationRegion() {
    Region<String, Configuration> configRegion = cache.getRegion(CONFIG_REGION_NAME);

    try {
      if (configRegion == null) {
        File diskDir = configDiskDirPath.toFile();

        if (!diskDir.exists() && !diskDir.mkdirs()) {
          throw new IOException("Cannot create directory at " + configDiskDirPath);
        }

        File[] diskDirs = {diskDir};
        cache.createDiskStoreFactory().setDiskDirs(diskDirs).setAutoCompact(true)
            .setMaxOplogSize(10).create(CLUSTER_CONFIG_DISK_STORE_NAME);

        AttributesFactory<String, Configuration> regionAttrsFactory = new AttributesFactory<>();
        regionAttrsFactory.setDataPolicy(DataPolicy.PERSISTENT_REPLICATE);
        regionAttrsFactory.setCacheListener(new ConfigurationChangeListener(this, cache));
        regionAttrsFactory.setDiskStoreName(CLUSTER_CONFIG_DISK_STORE_NAME);
        regionAttrsFactory.setScope(Scope.DISTRIBUTED_ACK);
        InternalRegionArguments internalArgs = new InternalRegionArguments();
        internalArgs.setIsUsedForMetaRegion(true);
        internalArgs.setMetaRegionWithTransactions(false);

        configRegion = cache.createVMRegion(CONFIG_REGION_NAME, regionAttrsFactory.create(),
            internalArgs);
      }
    } catch (RuntimeException e) {
      if (configRegion == null) {
        status.set(SharedConfigurationStatus.STOPPED);
      }
      // throw RuntimeException as is
      throw e;
    } catch (Exception e) {
      if (configRegion == null) {
        status.set(SharedConfigurationStatus.STOPPED);
      }
      // turn all other exceptions into runtime exceptions
      throw new RuntimeException("Error occurred while initializing cluster configuration", e);
    }

    return configRegion;
  }

  /**
   * Reads the configuration information from the shared configuration directory and returns a
   * {@link Configuration} object
   *
   * @return {@link Configuration}
   */
  private Configuration readConfiguration(File groupConfigDir) throws IOException {
    Configuration configuration = new Configuration(groupConfigDir.getName());
    File cacheXmlFull = new File(groupConfigDir, configuration.getCacheXmlFileName());
    File propertiesFull = new File(groupConfigDir, configuration.getPropertiesFileName());

    configuration.setCacheXmlFile(cacheXmlFull);
    configuration.setPropertiesFile(propertiesFull);

    Set<String> jarFileNames = Arrays.stream(groupConfigDir.list())
        .filter((String filename) -> filename.endsWith(".jar")).collect(Collectors.toSet());
    configuration.addJarNames(jarFileNames);
    return configuration;
  }

  /**
   * Creates a directory for this configuration if it doesn't already exist.
   */
  private File createConfigDirIfNecessary(final String configName) throws IOException {
    return createConfigDirIfNecessary(configDirPath.toFile(), configName);
  }

  private File createConfigDirIfNecessary(File clusterConfigDir, final String configName)
      throws IOException {
    if (!clusterConfigDir.exists() && !clusterConfigDir.mkdirs()) {
      throw new IOException("Cannot create directory : " + configDirPath);
    }
    Path configDirPath = clusterConfigDir.toPath().resolve(configName);

    File configDir = configDirPath.toFile();
    if (!configDir.exists() && !configDir.mkdir()) {
      throw new IOException("Cannot create directory : " + configDirPath);
    }

    return configDir;
  }

  private String generateInitialXmlContent() {
    StringWriter sw = new StringWriter();
    PrintWriter pw = new PrintWriter(sw);
    CacheXmlGenerator.generateDefault(pw);
    return sw.toString();
  }

  @Override
  public CacheConfig getCacheConfig(String group) {
    return getCacheConfig(group, false);
  }


  @Override
  public CacheConfig getCacheConfig(String group, boolean createNew) {
    if (group == null) {
      group = CLUSTER_CONFIG;
    }
    Configuration configuration = getConfiguration(group);
    if (configuration == null) {
      if (createNew) {
        return new CacheConfig(CACHE_CONFIG_VERSION);
      }
      return null;
    }
    String xmlContent = configuration.getCacheXmlContent();
    // group existed, so we should create a blank one to start with
    if (xmlContent == null || xmlContent.isEmpty()) {
      if (createNew) {
        return new CacheConfig(CACHE_CONFIG_VERSION);
      }
      return null;
    }

    return jaxbService.unMarshall(xmlContent);
  }

  @Override
  public void updateCacheConfig(String group, UnaryOperator<CacheConfig> mutator) {
    if (group == null) {
      group = CLUSTER_CONFIG;
    }
    lockSharedConfiguration();
    try {
      CacheConfig cacheConfig = getCacheConfig(group, true);
      cacheConfig = mutator.apply(cacheConfig);
      if (cacheConfig == null) {
        // mutator returns a null config, indicating no change needs to be persisted
        return;
      }
      Configuration configuration = getConfiguration(group);
      if (configuration == null) {
        configuration = new Configuration(group);
      }
      configuration.setCacheXmlContent(jaxbService.marshall(cacheConfig));
      getConfigurationRegion().put(group, configuration);
    } finally {
      unlockSharedConfiguration();
    }
  }
}