package kinesisFHM.eventGenerator;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.ObjectInputStream;
import java.sql.Timestamp;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.SortedMap;
import java.util.StringTokenizer;
import java.util.TreeMap;

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.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.S3Object;

public class EventGenerator {

	private static int caseLimit = 10000;
	
	protected final String streamName="EventStream";
	
	protected Hashtable<String, SortedMap<Timestamp, String>> cases = new Hashtable<String, SortedMap<Timestamp, String>>();
	protected Hashtable<String, Iterator<String>> iterators = new Hashtable<String, Iterator<String>>();
	protected Hashtable<String, Long> casenums = new Hashtable<String, Long>();
	
	protected AmazonKinesisClient amazonKinesisClient;
	private AmazonCloudWatchClient amazonCloudWatchClient;
	protected CloudWatchReporter outputReporter;
	
	protected int lambda;
	protected int maxInstances;

	
	private void runMe(String[] args) {
		String credentialFileName = args[0];
		String awsProfileName = args[1];
		String bucketname = args[2];
		String objectname = args[3];
		Integer numShards = Integer.parseInt(args[4]);
		Integer maxCases = Integer.parseInt(args[5]);
		lambda = Integer.parseInt(args[6]);
		maxInstances = Integer.parseInt(args[7]);
		if (maxInstances == 0) maxInstances = 1;
		if (maxInstances == -1) maxInstances = Integer.MAX_VALUE;
		
		// We do **not** assume that events within cases are sorted by time, we do that upon reading them

		AmazonS3 s3Client = new AmazonS3Client(new ProfilesConfigFile(credentialFileName).getCredentials(awsProfileName));     
		S3Object S3object = s3Client.getObject(new GetObjectRequest(bucketname, objectname));
		InputStream objectData = S3object.getObjectContent();
		BufferedReader reader = new BufferedReader(new InputStreamReader(objectData));

		String line;
		try {
			line = reader.readLine();
			while (line != null) {

				StringTokenizer tokenizer = new StringTokenizer(line, "\t");

				if (tokenizer.countTokens() == 3) {
					String caseID = tokenizer.nextToken();
					String eventID = tokenizer.nextToken();
					Timestamp timeStamp = Timestamp.valueOf(tokenizer.nextToken());

					SortedMap<Timestamp, String> currentCase = cases.get(caseID);
					if (currentCase == null) {
						if (cases.size() < caseLimit) {
							currentCase = new TreeMap<Timestamp, String>();
							cases.put(caseID,  currentCase);
//							System.err.println("New Case " + caseID + " : ");
						} else {
							break;
						}
					}
					currentCase.put(timeStamp, eventID);
				}
				line = reader.readLine();
			}
			objectData.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		System.err.println("Read " + cases.size() + " cases");

		File f = new File("EventGenerator.persistedCounts");
		ObjectInputStream ois;
		Hashtable<String, Long> restoredCaseNums = new Hashtable<String, Long>();
		try {
			ois = new ObjectInputStream(new FileInputStream(f));
			restoredCaseNums = (Hashtable<String, Long>) ois.readObject();
		} catch (FileNotFoundException e) {
			System.err.println(e.getMessage());			
		} catch (IOException e) {
			System.err.println(e.getMessage());
		} catch (ClassNotFoundException e) {
			System.err.println(e.getMessage());
		}
		System.err.println("Read " + restoredCaseNums.size() + " restored case numbers");
		
		// set up the iterators and case count numbers for cycling over the traces
		for (String caseID : cases.keySet()) {
			iterators.put(caseID, cases.get(caseID).values().iterator());
			if (restoredCaseNums.containsKey(caseID))
				casenums.put(caseID, restoredCaseNums.get(caseID) + 1);
			else
				casenums.put(caseID,  0l);
		}

		caseLimit = Math.min(caseLimit,  cases.size());
		int numThreads = Math.min(maxCases, Runtime.getRuntime().availableProcessors());
		System.err.println("Detected " + numThreads + " threads");
		int casesPerThread = maxCases / numThreads;
		System.err.println("Handling " + casesPerThread + " cases per thread");
		
		String[] keyArray = cases.keySet().toArray(new String[]{});
		String[][] caseIDs = new String[numThreads][casesPerThread];
		for (int i=0; i<(numThreads * casesPerThread); i++) {
			caseIDs[ i%numThreads ][ i/numThreads ] = keyArray[i];
		}
		
		amazonKinesisClient = new AmazonKinesisClient(new ProfilesConfigFile(credentialFileName).getCredentials(awsProfileName));
		amazonCloudWatchClient = new AmazonCloudWatchClient(new ProfilesConfigFile(credentialFileName).getCredentials(awsProfileName));
		outputReporter = new CloudWatchReporter(amazonCloudWatchClient, "PutRecords", streamName);
		outputReporter.start();		

		EventGeneratorPersistOnShutdown persistOnShutdown = new EventGeneratorPersistOnShutdown(this);
		Runtime.getRuntime().addShutdownHook(persistOnShutdown);
		
		if (kinesisFHM.utilities.Functions.isOutputStreamActive(amazonKinesisClient,  streamName,  numShards)) {	
			EventGeneratorThread[] threads = new EventGeneratorThread[numThreads];
			for (int i=0; i<numThreads; i++) {				
				EventGeneratorThread t = new EventGeneratorThread(caseIDs[i], this);
				t.setName("EventGeneratorThread" + i);
				t.start();
				threads[i] = t;
			}
			BufferedReader inputReader = new BufferedReader(new InputStreamReader(System.in));

			try {
				line = inputReader.readLine();
				while (line != null) {
					String[] parts = line.split("=");
					if (parts.length == 2)
						if (parts[0].equals("l") || parts[0].equals("n")) 
							try {
								int num = Integer.parseInt(parts[1]);
								if (parts[0].equals("l"))
									if (num >= 0)
										lambda = num;
									else
										System.err.println("Invalid lambda specified");
								if (parts[0].equals("n"))
									if (num > 0 && num <= caseLimit) {
										maxCases = num;
										casesPerThread = maxCases / numThreads;
										System.err.println("Handling " + casesPerThread + " cases per thread");
										String[][] newCaseIDs = new String[numThreads][casesPerThread];
										for (int i=0; i<(numThreads * casesPerThread); i++) {
											newCaseIDs[ i%numThreads ][ i/numThreads ] = keyArray[i];
										}
										for (int i=0; i<numThreads; i++) {
											threads[i].setNewCaseIDs(newCaseIDs[i]);
										}
									}
									else 
										System.err.println("Invalid numCases specified");
							} catch (Exception e) { System.err.println("Invalid number format"); }
						else
							System.err.println("Invalid spec");
					else
						System.err.println("Invalid entry");
					line = inputReader.readLine();
				}
			} catch (Exception e) {}

		} else {
			System.err.println("Output stream not active and cannot be created");
			System.exit(-1);
		}
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {

		if (args.length < 8) {
			System.err.println("Usage: EventGenerator credentialFileName awsProfileName bucket object numShards maxCases lambda maxInstances");
			System.exit(-1);
		}
		
		EventGenerator eg = new EventGenerator();
		eg.runMe(args);
	}
}
