package storm.petrel;

import java.io.InputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.FileInputStream;
import java.util.Map;

import org.apache.thrift7.TBase;

import org.yaml.snakeyaml.Yaml;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;

public class GenericTopology
{
    private static StormTopology readTopology() throws IOException
    {
        InputStream stream = ResourceLoader.getResourceAsStream("resources/topology.ser", GenericTopology.class);
        assert stream != null;
        
        try
        {
            // Create the reader
            ThriftReader thriftIn = new ThriftReader(stream, new ThriftReader.TBaseCreator() {
              @Override public TBase create() {
                return new StormTopology();
              }
            });
            
            // Read objects
            assert thriftIn.hasNext();
            TBase base = thriftIn.read();
            assert base != null;
            StormTopology topology = (StormTopology) base;
            assert topology != null;
            
            return topology;
        }
        finally
        {
            // Close stream
            stream.close();
        }
    }

    public static void main(String[] args) throws Exception
    {
        // Read topology file created by petrelbuilder.
        StormTopology topology = readTopology();

        // Get topology-specific configuration from the JAR.
        Config conf = new Config();
        InputStream stream = ResourceLoader.getResourceAsStream("resources/__topology__.yaml", GenericTopology.class);
        try
        {
            Yaml yaml = new Yaml();
            Map localConf = (Map) yaml.load(new InputStreamReader(stream));
            conf.putAll(localConf);
        }
        finally
        {
            // Close stream
            stream.close();
        }

        // Read __submitter__.yaml to get submitter user and hostname
        // information. Include it in the config.
        stream = ResourceLoader.getResourceAsStream("resources/__submitter__.yaml", GenericTopology.class);
        try
        {
            Yaml yaml = new Yaml();
            Map submitterConf = (Map) yaml.load(new InputStreamReader(stream));
            conf.putAll(submitterConf);
        }
        finally
        {
            // Close stream
            stream.close();
        }
        
        if (args!=null && args.length > 0)
        {
            StormSubmitter.submitTopology(args[0], conf, topology);
        }
        else
        {
            // Force some conservative settings to try and avoid overloading a
            // local machine's CPU or memory.
            conf.setDebug(true);
            conf.setNumWorkers(1);
            conf.setMaxTaskParallelism(1);

            final LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("test topology", conf, topology);

            Runtime.getRuntime().addShutdownHook(new Thread() {
                public void run() {
                    System.out.println("Shutting down local topology");
                    cluster.shutdown();
                }
            });    

            while (true)
            {
                System.out.println("Topology is running. Press ^C to stop it.");
                Thread.sleep(60000);
            }
        }
    }
}

@@ -110,40 +110,8 @@
 er;%0A
-import java.io.FileInputStream;%0A
 impo

Remove unused imports (again)