package kinesisFHM.graphConsumer;

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.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.SortedMap;

import kinesisFHM.commonTypes.CNetAugmentation;
import kinesisFHM.commonTypes.CNetType;
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;

public class TraceConsumerThread extends Thread implements Runnable {

	private String tracefname;
	private Properties traceInputProps;
	private Shard traceInputShard;

	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 TraceConsumerThread(Shard traceInputShard, GraphConsumer gcmain) {
		this.gcmain = gcmain;
		this.traceInputShard = traceInputShard;
		this.tracefname = gcmain.traceInputStreamName + "." + traceInputShard.getShardId();
	}

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

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


	private SortedMap<Timestamp, String> decodeTraceRecord(Record record) {
		byte[] data = Base64.decodeBase64(record.getData().array());
		ByteArrayInputStream bais = new ByteArrayInputStream(data);
		ObjectInputStream ois;
		SortedMap<Timestamp, String> trace = null;
		try {
			ois = new ObjectInputStream( bais );
			trace = (SortedMap<Timestamp, String>) ois.readObject();
		} catch (Exception e) {
			System.err.println(this.getName() + " " + e.getMessage());
		}
		return trace;
	}

	private void processTraceRecord(Record record) {
		SortedMap<Timestamp, String> tr = decodeTraceRecord(record);

		if (tr != null) {
			// this is now the time-sorted trace
			ArrayList<String> trace = new ArrayList<String>( tr.values() );

			for (int current = 0; current < trace.size()-1; current++) {
				// get the current event
				String ev1 = trace.get(current);				
				// Hold the candidate successors
				Set<String> candidates = new HashSet<String>();
				// Rest of the trace after the current event
				List<String> rest = trace.subList(current+1, trace.size());
				// for each successor of the current event
				for (String s: gcmain.DG.getTargets(ev1)) {
					// if the rest of the trace contains the successor
					if (rest.contains(s)) {
						int succind = rest.indexOf(s); 
						// check whether any of its predecessors are more recent than the current event
						// i.e. whether the rest of the trace up to the successor contains any of the 
						// successor's predecessors
						Set<String> preds = new HashSet<String>(gcmain.DG.getCauses(s));
						if (!preds.removeAll(rest.subList(0, succind))) {
							candidates.add(s);
						}
					}
				}
				if (candidates.size() > 0)
					addToOutput(ev1, candidates, CNetType.SUCC);	
			}

			for (int current = 1; current < trace.size(); current++) {
				// get the current event
				String ev1 = trace.get(current);				
				// Hold the candidate predecessors
				Set<String> candidates = new HashSet<String>();
				// Beginning of the trace before the current event
				List<String> rest = trace.subList(0, current);
				// for each predecessor of the current event
				for (String p: gcmain.DG.getCauses(ev1)) {
					// if the beginning of the trace contains the predecessor
					if (rest.contains(p)) {
						int predind = rest.lastIndexOf(p); 
						// check whether any of its successors are less recent than the current event
						// i.e. whether the rest of the trace up to the predecessor contains any of the 
						// predecessor's successors
						Set<String> succs = new HashSet<String>(gcmain.DG.getTargets(p));
						if (!succs.removeAll(rest.subList(predind+1, rest.size()))) {
							candidates.add(p);
						}
					}
				}
				if (candidates.size() > 0)
					addToOutput(ev1, candidates, CNetType.PRED);	
			}
		} else {
			//			System.err.println(this.getName() + " Received invalid trace record");
		}
	}

	private void addToOutput(String ev1, Collection<String> ev2, CNetType dir) {

		CNetAugmentation cnetaug = new CNetAugmentation(ev1, ev2, dir);
		Long count = gcmain.cnetaugs.get(cnetaug);

		if (count != null) {
			count++;
			gcmain.cnetaugs.put(cnetaug, count);
		} else {
			synchronized(gcmain.cnetaugs) {
				gcmain.cnetaugs.put(cnetaug, new Long(1));
			}
		}
	}

	public void run() {

		try {
			loadTraceProps();
			String traceCheckPoint = traceInputProps.getProperty(traceInputShard.getShardId());

			GetShardIteratorRequest getTraceShardIteratorRequest = new GetShardIteratorRequest();
			getTraceShardIteratorRequest.setStreamName(gcmain.traceInputStreamName);
			getTraceShardIteratorRequest.setShardId(traceInputShard.getShardId());

			// FIXME: DEBUG
			//			traceCheckPoint = null;

			if (traceCheckPoint != null) {
				getTraceShardIteratorRequest.setShardIteratorType("AFTER_SEQUENCE_NUMBER");
				getTraceShardIteratorRequest.setStartingSequenceNumber(traceCheckPoint);
			} else {
				getTraceShardIteratorRequest.setShardIteratorType("TRIM_HORIZON");
			}
			GetShardIteratorResult getTraceShardIteratorResult = gcmain.amazonKinesisClient.getShardIterator(getTraceShardIteratorRequest);
			String traceShardIterator = getTraceShardIteratorResult.getShardIterator();

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

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

			while (true) {
				try {
					if (gcmain.traceTurn) {
						synchronized(gcmain.traceRunning) {	
							gcmain.traceRunning++; 
						}
						// Create a new getRecordsRequest with an existing shardIterator 
						GetRecordsRequest getTraceRecordsRequest = new GetRecordsRequest();
						getTraceRecordsRequest.setShardIterator(traceShardIterator);
						getTraceRecordsRequest.setLimit(inputBatchSize); 

						GetRecordsResult traceResult = null;

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

						long starttime = System.currentTimeMillis();
						for (Record record : traceResult.getRecords()) {
							processTraceRecord(record);
							traceCheckPoint = record.getSequenceNumber();
						}
						long endtime = System.currentTimeMillis();

						// FIXME: DEBUG
						//				traceCheckPoint = null;

						if (traceCheckPoint != null) {
							//						System.err.println(this.getName() + " Checkpoint");
							traceInputProps.setProperty(traceInputShard.getShardId(), traceCheckPoint);
							saveTraceProps();
						}

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

						if ( (traceResult.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.traceThreadsleep > 50)
								gcmain.traceThreadsleep*=0.95;
						if ( traceResult.getRecords().size() < inputBatchSize/2 ) {
							// slowing down by reading smaller batches and extending our sleep period
							inputBatchSize*=0.95;
							if (gcmain.traceThreadsleep < 400) {
								gcmain.traceThreadsleep*=1.05;
								if (gcmain.graphTraceRatio > 0.1)
									gcmain.graphTraceRatio*=0.95;
							}
						}
						throttled=false;

						traceShardIterator = traceResult.getNextShardIterator();
						synchronized(gcmain.traceRunning) {
							gcmain.traceRunning--;
						}
					}	
					try { Thread.sleep(gcmain.traceThreadsleep); } catch (InterruptedException exception) { }
				} catch (Exception e) { System.err.println(this.getName() + " " + e.getMessage()); }	
			}
		} finally {
			saveTraceProps();
		}
	}

}
