package kinesisFHM.graphConsumer;

import java.awt.GraphicsEnvironment;
import java.io.ByteArrayInputStream;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.sql.Timestamp;
import java.util.Properties;
import javax.swing.event.ChangeEvent;

import kinesisFHM.commonTypes.GraphUpdate;
import kinesisFHM.commonTypes.WfRType;

import org.apache.commons.codec.binary.Base64;

import com.amazonaws.services.cloudwatch.model.Dimension;
import com.amazonaws.services.cloudwatch.model.MetricDatum;
import com.amazonaws.services.cloudwatch.model.PutMetricDataRequest;
import com.amazonaws.services.cloudwatch.model.StandardUnit;
import com.amazonaws.services.kinesis.model.ExpiredIteratorException;
import com.amazonaws.services.kinesis.model.GetRecordsRequest;
import com.amazonaws.services.kinesis.model.GetRecordsResult;
import com.amazonaws.services.kinesis.model.GetShardIteratorRequest;
import com.amazonaws.services.kinesis.model.GetShardIteratorResult;
import com.amazonaws.services.kinesis.model.ProvisionedThroughputExceededException;
import com.amazonaws.services.kinesis.model.Record;
import com.amazonaws.services.kinesis.model.Shard;

import edu.uci.ics.jung.graph.util.EdgeType;

public class GraphConsumerThread extends Thread implements Runnable {

	private String graphfname;
	private Properties graphInputProps;
	private Shard graphInputShard;
	
	private int inputBatchSize = 100;
	// maximum supported by AWS is 100000
	private static final int maxInputBatchSize = 10000;

	private boolean throttled;

	// it's the object that runs in the main thread
	private GraphConsumer gcmain;
	
	public GraphConsumerThread(Shard graphInputShard, GraphConsumer gcmain ) {
		this.gcmain = gcmain;
		this.graphInputShard = graphInputShard;
		this.graphfname = gcmain.graphInputStreamName + "." + graphInputShard.getShardId();
	}

	private void loadGraphProps() {
		graphInputProps = new Properties();
		try {
			graphInputProps.load(new FileReader(graphfname));
		} catch (FileNotFoundException e) {
			System.err.println(this.getName() + " " + e.getMessage());
		} catch (IOException e) {
			System.err.println(this.getName() + " " + e.getMessage());
		}	
	}

	private void saveGraphProps() {
		try {
			graphInputProps.store(new FileWriter(graphfname), null);
		} catch (IOException e) {
			System.err.println(this.getName() + " " + e.getMessage());
		}
	}

	private GraphUpdate decodeGraphRecord(Record record) {
		byte[] data = Base64.decodeBase64(record.getData().array());
		ByteArrayInputStream bais = new ByteArrayInputStream(data);
		ObjectInputStream ois;
		GraphUpdate upd = null;
		try {
			ois = new ObjectInputStream( bais );
			upd = (GraphUpdate) ois.readObject();
		} catch (Exception e) {
			System.err.println(this.getName() + " " + e.getMessage());
		}
		return upd;
	}

	private void processGraphRecord(Record record) {
		GraphUpdate upd = decodeGraphRecord(record);
		if (upd != null) {
			
			synchronized(gcmain.DG) {

				GraphUpdate.GraphUpdateType type = upd.getOp();
				String ev1 = upd.getEvent1();
				String ev2 = upd.getEvent2();
				String edg = ev1 + "-->" + ev2;

				switch (type) {
				case ADD : {
					gcmain.DG.add(ev1, ev2, WfRType.COUNT);
					if (!GraphicsEnvironment.isHeadless()) {
						gcmain.g.addEdge( edg, ev1, ev2, EdgeType.DIRECTED );
						gcmain.vv.stateChanged(new ChangeEvent(this));
					}
					break;
				}
				case REMOVE : {
					gcmain.DG.remove(upd.getEvent1(),  upd.getEvent2(), WfRType.COUNT);
					if (!GraphicsEnvironment.isHeadless()) {
						gcmain.g.removeEdge( edg );
						gcmain.vv.stateChanged(new ChangeEvent(this));
					}
					break;
				}
				default:
//					System.err.println(this.getName() + " Received illegal graph operation");
					break;
				}
			}
			
		} else {
//			System.err.println(this.getName() + " Received invalid graph record");
		}

	}

	public void run() {
		
		try {
			loadGraphProps();
			String graphCheckPoint = graphInputProps.getProperty(graphInputShard.getShardId());

			GetShardIteratorRequest getGraphShardIteratorRequest = new GetShardIteratorRequest();
			getGraphShardIteratorRequest.setStreamName(gcmain.graphInputStreamName);
			getGraphShardIteratorRequest.setShardId(graphInputShard.getShardId());

			// FIXME: DEBUG
//			graphCheckPoint = null;
			
			if (graphCheckPoint != null) {
				getGraphShardIteratorRequest.setShardIteratorType("AFTER_SEQUENCE_NUMBER");
				getGraphShardIteratorRequest.setStartingSequenceNumber(graphCheckPoint);
			} else {
				getGraphShardIteratorRequest.setShardIteratorType("TRIM_HORIZON");
			}
			GetShardIteratorResult getGraphShardIteratorResult = gcmain.amazonKinesisClient.getShardIterator(getGraphShardIteratorRequest);
			String graphShardIterator = getGraphShardIteratorResult.getShardIterator();

			System.err.println(this.getName() + " GraphConsumerThread started");

			long previous_behind = Long.MAX_VALUE;
			long current_behind = 0l;
			int count = 0;

			while (true) {
				try {
				if (gcmain.graphTurn){
					synchronized(gcmain.graphRunning) { 
						gcmain.graphRunning++;
					}
					
					// Create a new getRecordsRequest with an existing shardIterator 
					GetRecordsRequest getGraphRecordsRequest = new GetRecordsRequest();
					getGraphRecordsRequest.setShardIterator(graphShardIterator);
					getGraphRecordsRequest.setLimit(inputBatchSize); 

					GetRecordsResult graphResult = null;
					while (graphResult == null) {
						try {
							graphResult = gcmain.amazonKinesisClient.getRecords(getGraphRecordsRequest);
//							System.err.println(this.getName() + " Received " + graphResult.getRecords().size() + " graph records");
						} catch (ExpiredIteratorException e) {
//							System.err.println(this.getName() + " Iterator expired, getting new one");
							getGraphShardIteratorResult = gcmain.amazonKinesisClient.getShardIterator(getGraphShardIteratorRequest);
							graphShardIterator = getGraphShardIteratorResult.getShardIterator();
							getGraphRecordsRequest = new GetRecordsRequest();
							getGraphRecordsRequest.setShardIterator(graphShardIterator);
							getGraphRecordsRequest.setLimit(inputBatchSize); 
							graphResult = gcmain.amazonKinesisClient.getRecords(getGraphRecordsRequest);
//							System.err.println(this.getName() + " Received " + graphResult.getRecords().size() + " graph records");
						} catch (ProvisionedThroughputExceededException e) {
							System.err.println(this.getName() + " Read throughput exceeded, waiting a turn");
							throttled = true;
							if (gcmain.graphThreadsleep > 50)
								gcmain.graphThreadsleep*=1.05;
							else 
								if (inputBatchSize < 20) 
									inputBatchSize++;
								else 
									inputBatchSize = Math.min( maxInputBatchSize, Math.round(inputBatchSize*1.05f) );
							try { Thread.sleep(gcmain.graphThreadsleep/10); } catch (InterruptedException ee) { }
						}
					}
					// log the reads to cloudwatch
					gcmain.graphInputReporter.add(graphResult.getRecords().size());

					long starttime = System.currentTimeMillis();
					for (Record record : graphResult.getRecords()) {
						processGraphRecord(record);
						graphCheckPoint = record.getSequenceNumber();
					}
					long endtime = System.currentTimeMillis();

					// FIXME: DEBUG
					//				graphCheckPoint = null;

					if (graphCheckPoint != null) {
//						System.err.println(this.getName() + " Checkpoint");
						graphInputProps.setProperty(graphInputShard.getShardId(), graphCheckPoint);
						saveGraphProps();
					}

					previous_behind = current_behind;
					current_behind = graphResult.getMillisBehindLatest();
					if (!throttled) {
						if (count>0) count--;
						System.err.println(this.getName() + " Processed " + graphResult.getRecords().size() + "/" + inputBatchSize + " records in " + (endtime-starttime) + " millis at " + current_behind + " millis behind tip, threadsleep " + gcmain.graphThreadsleep + " count " + count);
					}
					else {
						System.err.println(this.getName() + " Processed " + graphResult.getRecords().size() + "/" + inputBatchSize + " records in " + (endtime-starttime) + " millis at " + current_behind + " millis behind tip, threadsleep " + gcmain.graphThreadsleep + " count " + count + " === Throttled ===");
						count=5;
					}

					if ( (graphResult.getRecords().size() == inputBatchSize) && (current_behind > 10000) && (current_behind > previous_behind) && !throttled & count==0)
						// speeding up by shortening our sleep period and reading bigger batches
						if (inputBatchSize < 20) 
							inputBatchSize++;
						else 
							inputBatchSize = Math.min( maxInputBatchSize, Math.round(inputBatchSize*1.05f) );
						if (gcmain.graphThreadsleep > 50)
							gcmain.graphThreadsleep*=0.95;
						else
					if ( graphResult.getRecords().size() < inputBatchSize/2 ) {
						// slowing down by reading smaller batches and extending our sleep period
						inputBatchSize*=0.95;
						if (gcmain.graphThreadsleep < 400) {
							gcmain.graphThreadsleep*=1.05;
							if (gcmain.graphTraceRatio < 10)
								gcmain.graphTraceRatio*=1.05;
						}
					}
					throttled=false;

					graphShardIterator = graphResult.getNextShardIterator();
					synchronized(gcmain.graphRunning) { 
						gcmain.graphRunning--;
					}
				}
				try {
					Thread.sleep(gcmain.graphThreadsleep);
				} 
				catch (InterruptedException exception) { }
			} catch (Exception e) { System.err.println(this.getName() + " " + e.getMessage()); }
			}
		} finally {
			saveGraphProps();
		}
	}

}
