package com.altamiracorp.lumify.core.bootstrap.lib;

import com.altamiracorp.lumify.core.config.Configuration;
import com.altamiracorp.lumify.core.exception.LumifyException;
import com.altamiracorp.lumify.core.util.LumifyLogger;
import com.altamiracorp.lumify.core.util.LumifyLoggerFactory;
import com.google.common.io.Files;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.util.HashSet;

public class HdfsLibCacheLoader extends LibLoader {
    private static final LumifyLogger LOGGER = LumifyLoggerFactory.getLogger(HdfsLibCacheLoader.class);

    @Override
    public void loadLibs(Configuration configuration) {
        LOGGER.debug("Loading libs using %s", HdfsLibCacheLoader.class.getName());

        String hdfsLibCacheDirectory = configuration.get(Configuration.HDFS_LIB_CACHE_DIRECTORY, null);
        if (hdfsLibCacheDirectory == null) {
            LOGGER.warn("skipping HDFS libcache. Configuration parameter %s not found", Configuration.HDFS_LIB_CACHE_DIRECTORY);
            return;
        }

        FileSystem hdfsFileSystem = getFileSystem(configuration);
        File libCacheDirectory = ensureLocalLibCacheDirectory();

        try {
            syncLibCache(hdfsFileSystem, new Path(hdfsLibCacheDirectory), libCacheDirectory);
        } catch (Exception ex) {
            throw new LumifyException(String.format("Could not sync HDFS libcache. %s -> %s", hdfsLibCacheDirectory, libCacheDirectory.getAbsolutePath()), ex);
        }

        addLibDirectory(libCacheDirectory);
    }

    private File ensureLocalLibCacheDirectory() {
        File libCacheDirectory = Files.createTempDir();
        libCacheDirectory.deleteOnExit();
        if (!libCacheDirectory.exists()) {
            if (!libCacheDirectory.mkdirs()) {
                throw new LumifyException("Could not mkdir " + libCacheDirectory.getAbsolutePath());
            }
        }
        return libCacheDirectory;
    }

    private FileSystem getFileSystem(Configuration configuration) {
        FileSystem hdfsFileSystem;
        try {
            String hdfsRootDir = configuration.get(Configuration.HADOOP_URL);
            org.apache.hadoop.conf.Configuration hadoopConfiguration = new org.apache.hadoop.conf.Configuration();
            hdfsFileSystem = FileSystem.get(new URI(hdfsRootDir), hadoopConfiguration, "hadoop");
        } catch (Exception ex) {
            throw new LumifyException("Could not open HDFS file system.", ex);
        }
        return hdfsFileSystem;
    }

    private static void syncLibCache(FileSystem fs, Path source, File dest) throws IOException {
        if (!fs.exists(source)) {
            throw new LumifyException(String.format("Could not sync directory %s. Directory does not exist.", source));
        }

        HashSet<String> foundFiles = addFilesFromHdfs(fs, source, dest);
        removeOldFiles(dest, foundFiles);
    }

    private static HashSet<String> addFilesFromHdfs(FileSystem fs, Path source, File dest) throws IOException {
        HashSet<String> foundFiles = new HashSet<String>();
        RemoteIterator<LocatedFileStatus> sourceFiles = fs.listFiles(source, true);
        while (sourceFiles.hasNext()) {
            LocatedFileStatus sourceFile = sourceFiles.next();
            String relativePath = sourceFile.getPath().toString().substring(source.toString().length());
            foundFiles.add(relativePath);
            File destFile = new File(dest, relativePath);
            if (sourceFile.isDirectory()) {
                destFile.mkdirs();
            } else {
                fs.copyToLocalFile(sourceFile.getPath(), new Path(destFile.getAbsolutePath()));
                new File(destFile.getParent(), "." + destFile.getName() + ".crc").deleteOnExit();
            }
            destFile.deleteOnExit();
        }
        return foundFiles;
    }

    private static void removeOldFiles(File file, HashSet<String> foundFiles) {
        if (file.isHidden() || file.getAbsolutePath().startsWith(".")) {
            return;
        }

        if (file.isDirectory()) {
            File[] files = file.listFiles();
            if (files == null) {
                throw new LumifyException(String.format("Could not list files of directory %s", file.getAbsolutePath()));
            }
            for (File f : files) {
                removeOldFiles(f, foundFiles);
            }
            return;
        }

        boolean foundFile = false;
        for (String f : foundFiles) {
            if (file.getAbsolutePath().endsWith(f)) {
                foundFile = true;
                break;
            }
        }

        if (!foundFile) {
            LOGGER.info("Removing old libcache file: %s", file.getAbsolutePath());
            file.delete();
        }
    }
}

@@ -558,34 +558,8 @@
 URI;
-%0Aimport java.util.HashSet;
 %0A%0Apu
@@ -2916,37 +2916,8 @@
     
- HashSet%3CString%3E foundFiles =
  add
@@ -2953,50 +2953,8 @@
 t);%0A
-        removeOldFiles(dest, foundFiles);%0A
     
@@ -2975,31 +2975,20 @@
  static 
-HashSet%3CString%3E
+void
  addFile
@@ -3061,68 +3061,8 @@
 n %7B%0A
-        HashSet%3CString%3E foundFiles = new HashSet%3CString%3E();%0A
     
@@ -3353,50 +3353,8 @@
 ));%0A
-            foundFiles.add(relativePath);%0A
     
@@ -3766,956 +3766,8 @@
   %7D%0A
-        return foundFiles;%0A    %7D%0A%0A    private static void removeOldFiles(File file, HashSet%3CString%3E foundFiles) %7B%0A        if (file.isHidden() %7C%7C file.getAbsolutePath().startsWith(%22.%22)) %7B%0A            return;%0A        %7D%0A%0A        if (file.isDirectory()) %7B%0A            File%5B%5D files = file.listFiles();%0A            if (files == null) %7B%0A                throw new LumifyException(String.format(%22Could not list files of directory %25s%22, file.getAbsolutePath()));%0A            %7D%0A            for (File f : files) %7B%0A                removeOldFiles(f, foundFiles);%0A            %7D%0A            return;%0A        %7D%0A%0A        boolean foundFile = false;%0A        for (String f : foundFiles) %7B%0A            if (file.getAbsolutePath().endsWith(f)) %7B%0A                foundFile = true;%0A                break;%0A            %7D%0A        %7D%0A%0A        if (!foundFile) %7B%0A            LOGGER.info(%22Removing old libcache file: %25s%22, file.getAbsolutePath());%0A            file.delete();%0A        %7D%0A
     

remove remove