package kinesisFHM.graphConsumer;

import java.awt.Dimension;
import java.awt.GraphicsEnvironment;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.PrintStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map.Entry;

import javax.swing.JFrame;

import kinesisFHM.commonTypes.CNetAugmentation;
import kinesisFHM.commonTypes.CNetType;
import kinesisFHM.commonTypes.DependencyGraph;
import kinesisFHM.utilities.CloudWatchReporter;

import com.amazonaws.auth.profile.ProfilesConfigFile;
import com.amazonaws.services.cloudwatch.AmazonCloudWatchClient;
import com.amazonaws.services.kinesis.AmazonKinesisClient;
import com.amazonaws.services.kinesis.model.Shard;

import edu.uci.ics.jung.algorithms.layout.FRLayout;
import edu.uci.ics.jung.algorithms.layout.Layout;
import edu.uci.ics.jung.graph.DirectedSparseGraph;
import edu.uci.ics.jung.graph.Graph;
import edu.uci.ics.jung.visualization.VisualizationViewer;
import edu.uci.ics.jung.visualization.control.DefaultModalGraphMouse;
import edu.uci.ics.jung.visualization.decorators.ToStringLabeller;
import edu.uci.ics.jung.visualization.renderers.Renderer.VertexLabel.Position;

public class GraphConsumer {

	public final String graphInputStreamName="GraphUpdateStream";
	public final String traceInputStreamName="TraceStream";
	public final String graphOutputFileName="kinesisFHM.graph.dot";

	public AmazonKinesisClient amazonKinesisClient;

	public CloudWatchReporter graphInputReporter;
	public CloudWatchReporter traceInputReporter;
	
	// this will hold the dependency graph
	public DependencyGraph DG = new DependencyGraph();
	
	// This is the stuff for visualization!
	
	// Graph<V, E> where V is the type of the vertices 
	// and E is the type of the edges
	public Graph<String, String> g = new DirectedSparseGraph<String, String>();
	public VisualizationViewer<String,String> vv; 

	// this will hold the list of tuples representing the augmented CNet
	public HashMap<CNetAugmentation, Long> cnetaugs = new HashMap<CNetAugmentation, Long>();

	public boolean graphTurn;
	public boolean traceTurn;
	public Short graphRunning;
	public Short traceRunning;

	public long graphThreadsleep;
	public long traceThreadsleep;
	
	public float graphTraceRatio;
	
	private boolean tracePreviousTurn = true;

	public void switchTurn() {
		if (tracePreviousTurn) {
			traceTurn = false;
			System.err.println("== Previous turn was traceConsumers, switching off the traceConsumers (graphTraceRation = " + graphTraceRatio + ")");
			while (traceRunning > 0) {
//				System.err.println("== Waiting for " + traceRunning + " traceConsumers to finish");
				synchronized(this) {
					try { wait(traceThreadsleep); } catch (InterruptedException e) { }
				}
			}
			persist();
			graphTurn = true;
		}
		if (!tracePreviousTurn) {
			graphTurn = false;
			System.err.println("== Previous turn was graphConsumers, switching off the graphConsumers (graphTraceRation = " + graphTraceRatio + ")");
			while (graphRunning > 0) {
//				System.err.println("== Waiting for " + graphRunning + " graphConsumers to finish");
				synchronized(this) {
					try { wait(graphThreadsleep); } catch (InterruptedException e) { }
				}
			}
			persist();
			traceTurn = true;
		}
		tracePreviousTurn = !tracePreviousTurn;
	}
	
	private void persist() {
		try {
			System.err.print("== Persisting");
			// persist the dependency graph and augmented Cnet
			File f = new File("graphconsumer.persistedCounts");
			ObjectOutputStream oos;
			try {
				oos = new ObjectOutputStream(new FileOutputStream(f, false));
				oos.writeObject(DG);
				oos.writeObject(cnetaugs);
				oos.writeObject(g);
				oos.close();
			} catch (FileNotFoundException e) {
				e.printStackTrace();
			} catch (IOException e) {
				e.printStackTrace();
			} catch (Exception e) { }
			
			// write the augmented cnet as a dot/graphviz file
			PrintStream out = new PrintStream(new File("kinesisFHM.graph.forward.dot"));
			out.println("digraph \"Augmented CNet\" {");
			out.println("  rankdir=LR;");
			out.println("  size=\"8,8\";");

			for (Entry<CNetAugmentation, Long> cnetaug : cnetaugs.entrySet()) {
				if (cnetaug.getKey().getDir() == CNetType.SUCC) {
					for (String v : cnetaug.getKey().getOthers()) {
						out.println("  " + cnetaug.getKey().getEvent() + " -> " + v + " [label=\"SUCC: " + cnetaug.getValue() + "\"]; ");
					}
				}
			}
			out.println("}");
			out.close();

			out = new PrintStream(new File("kinesisFHM.graph.backward.dot"));
			out.println("digraph \"Augmented CNet\" {");
			out.println("  rankdir=LR;");
			out.println("  size=\"8,8\";");

			for (Entry<CNetAugmentation, Long> cnetaug : cnetaugs.entrySet()) {
				if (cnetaug.getKey().getDir() == CNetType.PRED) {
					for (String v : cnetaug.getKey().getOthers()) {
						out.println("  " +  cnetaug.getKey().getEvent() + " -> "+ v + " [label=\"PRED: " + cnetaug.getValue() + "\"]; ");
					}					
				}
			}
			out.println("}");
			out.close();
			
			System.err.println(" ... done");
		} catch (IOException e) {
//			System.err.println(" ... exception " + e.getMessage());
		}
	}


	public void runme(String[] args) {

		if (args.length < 3) {
			System.err.println("Usage: GraphConsumer credentialFileName awsProfileName threadsleep [graphTraceRatio]");
			System.exit(-1);
		}
		
		String credentialFileName = args[0];
		String awsProfileName = args[1];		
		graphThreadsleep = Long.parseLong(args[2]);
		traceThreadsleep = graphThreadsleep;
		graphTraceRatio = 10f;
		if (args.length == 4) {
			graphTraceRatio = Float.parseFloat(args[3]);
		}

		amazonKinesisClient = new AmazonKinesisClient(new ProfilesConfigFile(credentialFileName).getCredentials(awsProfileName));
		AmazonCloudWatchClient amazonCloudWatchClient = new AmazonCloudWatchClient(new ProfilesConfigFile(credentialFileName).getCredentials(awsProfileName));
		graphInputReporter = new CloudWatchReporter(amazonCloudWatchClient, "ReadRecords", graphInputStreamName);
		traceInputReporter = new CloudWatchReporter(amazonCloudWatchClient, "ReadRecords", traceInputStreamName);
		graphInputReporter.start();
		traceInputReporter.start();
		
		List<Shard> traceInputShards = kinesisFHM.utilities.Functions.getInputShards(amazonKinesisClient, traceInputStreamName);
		List<Shard> graphInputShards = kinesisFHM.utilities.Functions.getInputShards(amazonKinesisClient, graphInputStreamName);
		
		if (traceInputShards != null && graphInputShards != null) {

			// read any persisted information from the last run
			File f = new File("graphconsumer.persistedCounts");
			ObjectInputStream ois;
			try {
				ois = new ObjectInputStream(new FileInputStream(f));
				DG = (DependencyGraph) ois.readObject();
				cnetaugs = (HashMap<CNetAugmentation, Long>) ois.readObject();
				g = (DirectedSparseGraph<String, String>) ois.readObject();
				ois.close();
			} catch (Exception e) { } 
			
			if (!GraphicsEnvironment.isHeadless()) {
				// The Layout<V, E> is parameterized by the vertex and edge types
				Layout<String, String> layout = new FRLayout<String, String>(g);
				layout.setSize(new Dimension(800,800)); // sets the initial size of the space
				vv = new VisualizationViewer<String,String>(layout);
				vv.setPreferredSize(new Dimension(800,800)); //Sets the viewing area size
		        vv.getRenderContext().setVertexLabelTransformer(new ToStringLabeller<String>());
		        vv.getRenderContext().setEdgeLabelTransformer(new ToStringLabeller<String>());
		        vv.getRenderer().getVertexLabelRenderer().setPosition(Position.CNTR);
		        DefaultModalGraphMouse<?, ?> gm = new DefaultModalGraphMouse<Object, Object>();
		        gm.setMode(DefaultModalGraphMouse.Mode.TRANSFORMING);
		        vv.setGraphMouse(gm);
		      
		        JFrame frame = new JFrame("Interactive Graph View");
				frame.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
				frame.getContentPane().add(vv); 
				frame.pack();
				frame.setVisible(true);				
			}
			graphTurn = false;
			traceTurn = false;

			graphRunning = 0;
			traceRunning = 0;

			for (int i=0; i < graphInputShards.size(); i++) {
				GraphConsumerThread t = new GraphConsumerThread( graphInputShards.get(i), this ); 
				t.setName("GraphConsumerThread_" + graphInputShards.get(i).getShardId());
				t.start();
			}
			for (int i = 0; i < traceInputShards.size(); i++) {
				TraceConsumerThread t = new TraceConsumerThread( traceInputShards.get(i), this ); 
				t.setName("TraceConsumerThread_" + traceInputShards.get(i).getShardId());
				t.start();
			}

			synchronized(this) {
				while (true) {
//					System.err.println("== Scheduler is going to swtich");
					switchTurn();
//					System.err.println("== Scheduler is done switching");
//					System.err.println("== graphTurn is now " + graphTurn + " and traceTurn is now " + traceTurn);
					try {
//						System.err.println("== Scheduler is going to wait");
						// we wait longer if the traceConsumers are running, they need much more time than the graphConsumers
						if (graphTurn) { wait(1000); } else { wait(Math.round(1000*graphTraceRatio) ); }
//						System.err.println("== Scheduler is done waiting");
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
//						System.err.println("== Scheduler has been interrupted");
					}
				}
			}
		}
		
	}
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		GraphConsumer gc = new GraphConsumer();
		gc.runme(args);
	}

}
