package elephantdb.hadoop;

import elephantdb.DomainSpec;
import elephantdb.persistence.LocalPersistenceFactory;
import java.io.File;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.UUID;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.ChecksumFileSystem;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;


public class LocalElephantManager {
        FileSystem _fs;
        File _dirFlag;
        String _localRoot;
        DomainSpec _spec;
        Map<String, Object> _persistenceOptions;
        
        public LocalElephantManager(FileSystem fs, DomainSpec spec, Map<String, Object> persistenceOptions, List<String> tmpDirs) throws IOException {
            _localRoot = selectAndFlagRoot(tmpDirs);
            _fs = fs;
            _spec = spec;
            _persistenceOptions = persistenceOptions;
        }

        /**
         * Creates a temporary directory, downloads the remotePath (tied to the FS), and returns it. If remotePath is null or doesn't exist,
         * creates an empty local elephant and closes it.
         */
        public String downloadRemoteShard(String id, String remotePath) throws IOException {
            LocalPersistenceFactory fact = _spec.getLPFactory();
            String returnDir = localTmpDir(id);
            if(remotePath==null || !_fs.exists(new Path(remotePath))) {
                fact.createPersistence(
                            returnDir,
                            _persistenceOptions)
                         .close();
            } else {
               ((ChecksumFileSystem)_fs).copyToLocalFile(new Path(remotePath), new Path(returnDir), false);
            }
            return returnDir;
        }

        public String localTmpDir(String id) {
            return _localRoot + "/" + id;
        }

        public void progress() {
            _dirFlag.setLastModified(System.currentTimeMillis());
        }


        public void cleanup() throws IOException {
            FileSystem.getLocal(new Configuration()).delete(new Path(_localRoot), true);
            _dirFlag.delete();
        }

        private void clearStaleFlags(List<String> tmpDirs) {
            //delete any flags more than an hour old
            for(String tmp: tmpDirs) {
                File flagDir = new File(flagDir(tmp));
                flagDir.mkdirs();
                for(File flag: flagDir.listFiles()) {
                    if(flag.lastModified() < System.currentTimeMillis() - 1000*60*60) {
                        flag.delete();
                    }
                }
            }
        }

        private String flagDir(String tmpDir) {
            return tmpDir + "/flags";
        }

        private String selectAndFlagRoot(List<String> tmpDirs) throws IOException {
            clearStaleFlags(tmpDirs);
            Map<String, Integer> flagCounts = new HashMap<String, Integer>();
            for(String tmp: tmpDirs) {
                File flagDir = new File(flagDir(tmp));
                flagDir.mkdirs();
                flagCounts.put(tmp, flagDir.list().length);
            }
            String best = null;
            Integer bestCount = null;
            for(Entry<String, Integer> e: flagCounts.entrySet()) {
                if(best==null || e.getValue() < bestCount) {
                    best = e.getKey();
                    bestCount = e.getValue();
                }
            }
            String token = UUID.randomUUID().toString();
            _dirFlag = new File(flagDir(best) + "/" + token);
            _dirFlag.createNewFile();
            return best + "/" + token;
        }
}

@@ -1642,16 +1642,76 @@
  else %7B%0A
+                if(_fs instanceof ChecksumFileSystem) %7B%0A    
         
@@ -1806,24 +1806,151 @@
 r), false);%0A
+                %7D else %7B%0A                    _fs.copyToLocalFile(new Path(remotePath), new Path(returnDir));%0A                %7D%0A
             

fix localelephantmanager to work with s3