package kinesisFHM.updateConsumer;

import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.List;
import java.util.Properties;

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;

public class UpdateConsumer {

	private String inputStreamName="UpdateStream";
	private String outputStreamName="GraphUpdateStream";	
	
	protected Long totalRecordCount;

	private Properties props;
	
	private boolean loadProps() {
		props = new Properties();
		try {
			props.load(new FileReader("updateconsumer.properties"));
			return true;
		} catch (Exception e) {
			return false;
		}	
	}

	protected void saveProps() {
		try {
			props.setProperty("totalRecordCount", totalRecordCount.toString());
			props.store(new FileWriter("updateconsumer.properties"), null);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void runme(String[] args) {

		if (args.length < 4) {
			System.err.println("Usage: UpdateConsumer credentialFileName awsProfileName inputBatchSize threadsleep [defaultThreshold]");
			System.exit(-1);
		}
		
		String credentialFileName = args[0];
		String awsProfileName = args[1];
		int inputBatchSize = Integer.parseInt(args[2]);
		long threadsleep = Long.parseLong(args[3]);
		
		Float defaultThreshold = 0.5f;
		if (args.length == 5) {
			defaultThreshold = Float.parseFloat(args[4]);
		}
		UpdateConsumerThread.defaultThreshold = defaultThreshold;

		if (loadProps())
			try {
				totalRecordCount = Long.parseLong(props.getProperty("totalRecordCount"));
			} catch (NumberFormatException e) {
				totalRecordCount = 0l;
			}
		else
			totalRecordCount = 0l;
		
		AmazonKinesisClient amazonKinesisClient = new AmazonKinesisClient(new ProfilesConfigFile(credentialFileName).getCredentials(awsProfileName));
		AmazonCloudWatchClient amazonCloudWatchClient = new AmazonCloudWatchClient(new ProfilesConfigFile(credentialFileName).getCredentials(awsProfileName));
		CloudWatchReporter inputReporter = new CloudWatchReporter(amazonCloudWatchClient, "ReadRecords", inputStreamName);
		CloudWatchReporter outputReporter = new CloudWatchReporter(amazonCloudWatchClient, "PutRecords", outputStreamName);
		inputReporter.start();
		outputReporter.start();

		List<Shard> inputShards = kinesisFHM.utilities.Functions.getInputShards(amazonKinesisClient, inputStreamName);
		if (inputShards != null) {
			if (kinesisFHM.utilities.Functions.isOutputStreamActive(amazonKinesisClient, outputStreamName, inputShards.size())) {
				for (Shard shard : inputShards) {
					UpdateConsumerThread t = new UpdateConsumerThread(	inputStreamName, 
																			shard, 
																			amazonKinesisClient, 
																			inputReporter,
																			outputReporter,
																			outputStreamName,
																			this,
																			threadsleep,
																			inputBatchSize);
					t.setName("UpdateConsumerThread_" + shard.getShardId());
					t.start();
				}
			} else {
				System.err.println("Output stream not active and cannot be created");
				System.exit(-1);
			}
		} else {
			System.err.println("Input stream not active");
			System.exit(-1);
		}
	}
	
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		UpdateConsumer updateConsumer = new UpdateConsumer();
		updateConsumer.runme(args);
		Object o = new Object();
		
		while (true) {
			synchronized(o) {
				try {
					o.wait(1000);
				} catch (InterruptedException e) { }
			} 
		}
	}
	
}
