content: package backtype.storm.utils;
  
  import backtype.storm.Config;
  import backtype.storm.generated.ComponentCommon;
  import backtype.storm.generated.ComponentObject;
  import backtype.storm.generated.StormTopology;
  import clojure.lang.IFn;
  import clojure.lang.RT;
  import com.netflix.curator.framework.CuratorFramework;
  import com.netflix.curator.framework.CuratorFrameworkFactory;
  import com.netflix.curator.retry.RetryNTimes;
  import java.io.ByteArrayInputStream;
  import java.io.ByteArrayOutputStream;
  import java.io.File;
  import java.io.FileInputStream;
  import java.io.FileOutputStream;
  import java.io.InputStreamReader;
  import java.io.IOException;
  import java.io.ObjectInputStream;
  import java.io.ObjectOutputStream;
  import java.net.URL;
  import java.nio.ByteBuffer;
  import java.nio.channels.Channels;
  import java.nio.channels.WritableByteChannel;
  import java.util.ArrayList;
  import java.util.Enumeration;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.TreeMap;
  import java.util.UUID;
  import org.apache.commons.lang.StringUtils;
  import org.apache.thrift7.TException;
  import org.yaml.snakeyaml.Yaml;
  
  public class Utils {
  	public static final String DEFAULT_STREAM_ID = "default";
  
  	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 findAndReadConfigFile(String name, boolean mustExist) {
  		try {
  			String higopath=System.getenv("higo_conf_path");
  			if(higopath!=null)
  			{
  				File f=new File(higopath,name);
  				if(f.exists())
  				{
  					   InputStreamReader isr = new InputStreamReader (new FileInputStream (f));
  					   Yaml yaml = new Yaml();
  						Map ret = (Map) yaml.load(isr);
  						if (ret != null)
  						{
  							return new HashMap(ret);
  						}
  				}
  			}
  			
  			
  			Enumeration resources = Thread.currentThread()
  					.getContextClassLoader().getResources(name);
  			if (!resources.hasMoreElements()) {
  				if (mustExist)
  					throw new RuntimeException(
  							"Could not find config file on classpath " + name);
  				else
  					return new HashMap();
  			}
  			URL resource = (URL) resources.nextElement();
  			Yaml yaml = new Yaml();
  			Map ret = (Map) yaml.load(new InputStreamReader(resource
  					.openStream()));
  			if (ret == null)
  			{
  				ret = new HashMap();
  			}
  
  //			if (resources.hasMoreElements()) {
  //				throw new RuntimeException("Found multiple " + name
  //						+ " resources " + resource + ":"
  //						+ resources.nextElement());
  //			}
  			return new HashMap(ret);
  
  		} catch (IOException e) {
  			throw new RuntimeException(e);
  		}
  	}
  
  	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 Map readStormConfig(String filename) {
  		Map ret = readStormConfig();
  		Map storm = findAndReadConfigFile(filename, false);
  		ret.putAll(storm);
  		return ret;
  	}
  
  
  
  	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();
  	}
  
  	public static boolean isSystemId(String id) {
  		return id.startsWith("__");
  	}
  
  	public static <K, V> Map<V, K> reverseMap(Map<K, V> map) {
  		Map<V, K> ret = new HashMap<V, K>();
  		for (K key : map.keySet()) {
  			ret.put(map.get(key), key);
  		}
  		return ret;
  	}
  
  	public static ComponentCommon getComponentCommon(StormTopology topology,
  			String id) {
  		if (topology.get_spouts().containsKey(id)) {
  			return topology.get_spouts().get(id).get_common();
  		}
  		if (topology.get_bolts().containsKey(id)) {
  			return topology.get_bolts().get(id).get_common();
  		}
  		if (topology.get_state_spouts().containsKey(id)) {
  			return topology.get_state_spouts().get(id).get_common();
  		}
  		throw new IllegalArgumentException("Could not find component with id "
  				+ id);
  	}
  
  	public static Integer getInt(Object o) {
  		if (o instanceof Long) {
  			return ((Long) o).intValue();
  		} else if (o instanceof Integer) {
  			return (Integer) o;
  		} else if (o instanceof Short) {
  			return ((Short) o).intValue();
  		} else {
  			throw new IllegalArgumentException("Don't know how to convert " + o
  					+ " + to int");
  		}
  	}
  
  	public static long randomLong() {
  		return UUID.randomUUID().getLeastSignificantBits();
  	}
  
  	public static CuratorFramework newCurator(Map conf, List<String> servers,
  			Object port, String root) {
  		List<String> serverPorts = new ArrayList<String>();
  		for (String zkServer : (List<String>) servers) {
  			serverPorts.add(zkServer + ":" + Utils.getInt(port));
  		}
  		String zkStr = StringUtils.join(serverPorts, ",") + root;
  		try {
  			CuratorFramework ret = CuratorFrameworkFactory
  					.newClient(
  							zkStr,
  							Utils.getInt(conf
  									.get(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT)),
  							15000,
  							new RetryNTimes(
  									Utils.getInt(conf
  											.get(Config.STORM_ZOOKEEPER_RETRY_TIMES)),
  									Utils.getInt(conf
  											.get(Config.STORM_ZOOKEEPER_RETRY_INTERVAL))));
  			return ret;
  		} catch (IOException e) {
  			throw new RuntimeException(e);
  		}
  	}
  
  	public static CuratorFramework newCurator(Map conf, List<String> servers,
  			Object port) {
  		return newCurator(conf, servers, port, "");
  	}
  
  	public static CuratorFramework newCuratorStarted(Map conf,
  			List<String> servers, Object port, String root) {
  		CuratorFramework ret = newCurator(conf, servers, port, root);
  		ret.start();
  		return ret;
  	}
  
  	public static CuratorFramework newCuratorStarted(Map conf,
  			List<String> servers, Object port) {
  		CuratorFramework ret = newCurator(conf, servers, port);
  		ret.start();
  		return ret;
  	}
  
  	/**
  	 * 
  	 (defn integer-divided [sum num-pieces] (let [base (int (/ sum
  	 * num-pieces)) num-inc (mod sum num-pieces) num-bases (- num-pieces
  	 * num-inc)] (if (= num-inc 0) {base num-bases} {base num-bases (inc base)
  	 * num-inc} )))
  	 * 
  	 * @param sum
  	 * @param numPieces
  	 * @return
  	 */
  
  	public static TreeMap<Integer, Integer> integerDivided(int sum,
  			int numPieces) {
  		int base = sum / numPieces;
  		int numInc = sum % numPieces;
  		int numBases = numPieces - numInc;
  		TreeMap<Integer, Integer> ret = new TreeMap<Integer, Integer>();
  		ret.put(base, numBases);
  		if (numInc != 0) {
  			ret.put(base + 1, numInc);
  		}
  		return ret;
  	}
  
  	public static byte[] toByteArray(ByteBuffer buffer) {
  		byte[] ret = new byte[buffer.remaining()];
  		buffer.get(ret, 0, ret.length);
  		return ret;
  	}
  }
  