package backtype.storm.utils;

import backtype.storm.generated.ComponentObject;
import clojure.lang.IFn;
import clojure.lang.RT;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.InputStreamReader;
import java.io.InputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import org.apache.thrift7.TException;
import org.jvyaml.YAML;

public class Utils {
    public static final int DEFAULT_STREAM_ID = 1;

    public static byte[] serialize(Object obj) {
        try {
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(bos);
            oos.writeObject(obj);
            oos.close();
            return bos.toByteArray();
        } catch(IOException ioe) {
            throw new RuntimeException(ioe);
        }
    }

    public static Object deserialize(byte[] serialized) {
        try {
            ByteArrayInputStream bis = new ByteArrayInputStream(serialized);
            ObjectInputStream ois = new ObjectInputStream(bis);
            Object ret = ois.readObject();
            ois.close();
            return ret;
        } catch(IOException ioe) {
            throw new RuntimeException(ioe);
        } catch(ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }

    public static <T> String join(Iterable<T> coll, String sep) {
        Iterator<T> it = coll.iterator();
        String ret = "";
        while(it.hasNext()) {
            ret = ret + it.next();
            if(it.hasNext()) {
                ret = ret + sep;
            }
        }
        return ret;
    }

    public static void sleep(long millis) {
        try {
            Time.sleep(millis);
        } catch(InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public static Map readYamlConfig(String path) {
        try {
            Map ret = (Map) YAML.load(new FileReader(path));
            if(ret==null) ret = new HashMap();
            return new HashMap(ret);
        } catch (FileNotFoundException ex) {
            throw new RuntimeException(ex);
        }
    }

    public static Map findAndReadConfigFile(String name, boolean mustExist) {
        // It's important to use Utils.class here instead of Object.class here.
        // If Object.class is used, sbt can't find defaults.yaml
        InputStream is = Utils.class.getResourceAsStream("/" + name);
        if(is==null) {
            if(mustExist) throw new RuntimeException("Could not find config file on classpath " + name);
            else return new HashMap();
        }
        Map ret = (Map) YAML.load(new InputStreamReader(is));
        if(ret==null) ret = new HashMap();
        return new HashMap(ret);
    }

    public static Map findAndReadConfigFile(String name) {
       return findAndReadConfigFile(name, true);
    }

    public static Map readDefaultConfig() {
        return findAndReadConfigFile("defaults.yaml", true);
    }

    public static Map readStormConfig() {
        Map ret = readDefaultConfig();
        Map storm = findAndReadConfigFile("storm.yaml", false);
        ret.putAll(storm);
        return ret;
    }

    public static Object getSetComponentObject(ComponentObject obj) {
        if(obj.getSetField()==ComponentObject._Fields.SERIALIZED_JAVA) {
            return Utils.deserialize(obj.get_serialized_java());
        } else {
            return obj.get_shell();
        }
    }

    public static <S, T> T get(Map<S, T> m, S key, T def) {
        T ret = m.get(key);
        if(ret==null) {
            ret = def;
        }
        return ret;
    }
    
    public static List<Object> tuple(Object... values) {
        List<Object> ret = new ArrayList<Object>();
        for(Object v: values) {
            ret.add(v);
        }
        return ret;
    }

    public static void downloadFromMaster(Map conf, String file, String localFile) throws IOException, TException {
        NimbusClient client = NimbusClient.getConfiguredClient(conf);
        String id = client.getClient().beginFileDownload(file);
        WritableByteChannel out = Channels.newChannel(new FileOutputStream(localFile));
        while(true) {
            ByteBuffer chunk = client.getClient().downloadChunk(id);
            int written = out.write(chunk);
            if(written==0) break;
        }
        out.close();
    }
    
    public static IFn loadClojureFn(String namespace, String name) {
        try {
          clojure.lang.Compiler.eval(RT.readString("(require '" + namespace + ")"));
        } catch (Exception e) {
          //if playing from the repl and defining functions, file won't exist
        }
        return (IFn) RT.var(namespace, name).deref();        
    }
}

@@ -446,32 +446,53 @@
 ctOutputStream;%0A
+import java.net.URL;%0A
 import java.nio.
@@ -608,24 +608,54 @@
 .ArrayList;%0A
+import java.util.Enumeration;%0A
 import java.
@@ -2669,238 +2669,168 @@
     
-// It's important to use Utils.class here instead of Object.class here.%0A        // If Object.class is used, sbt can't find defaults.yaml%0A        InputStream is = Utils.class.getResourceAsStream(%22/%22 + name);%0A        if(is==null) %7B%0A
+try %7B%0A            Enumeration resources = Thread.currentThread().getContextClassLoader().getResources(name);%0A            if(!resources.hasMoreElements()) %7B%0A    
     
@@ -2942,16 +2942,20 @@
         
+    
 else ret
@@ -2973,34 +2973,100 @@
 hMap();%0A        
+    
 %7D%0A
+            URL resource = (URL) resources.nextElement();%0A    
         Map ret 
@@ -3109,14 +3109,37 @@
 der(
-is));%0A
+resource.openStream()));%0A    
     
@@ -3189,32 +3189,294 @@
     
-return new HashMap(ret);
+    %0A            if(resources.hasMoreElements()) %7B%0A                throw new RuntimeException(%22Found multiple %22 + name + %22 resources%22);%0A            %7D%0A            return new HashMap(ret);%0A            %0A        %7D catch (IOException e) %7B%0A            throw new RuntimeException(e);%0A        %7D
 %0A   

throw error if detect multiple defaults.yaml or storm.yaml files of the same name