package kinesisFHM.eventConsumer;

import java.util.List;

import kinesisFHM.utilities.CloudWatchReporter;
import kinesisFHM.utilities.Functions;

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 EventConsumer {

	private static final String inputStreamName="EventStream";
	private static final String relationOutputStreamName="RelationStream";	
	private static final String traceOutputStreamName="TraceStream";
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {

		if (args.length <= 6) {
			System.err.println("Usage: EventConsumer credentialFileName awsProfileName numTraceOutputShards numRelationOutputShards inputBatchSize threadsleep realTime");
			System.exit(-1);
		}
		
		String credentialFileName = args[0];
		String awsProfileName = args[1];		
		int numTraceOutputShards = Integer.parseInt(args[2]);
		int numRelationOutputShards = Integer.parseInt(args[3]);
		int inputBatchSize = Integer.parseInt(args[4]);
		long threadsleep = Long.parseLong(args[5]);
		
		boolean realTime = false;
		if (args.length == 7) 
			realTime = Boolean.parseBoolean(args[6]);
		
		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 traceOutputReporter = new CloudWatchReporter(amazonCloudWatchClient, "PutRecords", traceOutputStreamName);
		CloudWatchReporter relationOutputReporter = new CloudWatchReporter(amazonCloudWatchClient, "PutRecords", relationOutputStreamName);
		inputReporter.start();
		traceOutputReporter.start();
		relationOutputReporter.start();
		
		List<Shard> inputShards = Functions.getInputShards(amazonKinesisClient, inputStreamName);
		if (inputShards != null) {
			if (Functions.isOutputStreamActive(amazonKinesisClient, relationOutputStreamName, numRelationOutputShards) && 
					Functions.isOutputStreamActive(amazonKinesisClient, traceOutputStreamName, numTraceOutputShards)) {
				for (Shard shard : inputShards) {
					EventConsumerThread t = new EventConsumerThread(inputStreamName, shard, amazonKinesisClient, 
																	traceOutputReporter, relationOutputReporter, inputReporter,
																	relationOutputStreamName, traceOutputStreamName, 
																	inputShards.size(), realTime, threadsleep, inputBatchSize);
					t.setName("EventConsumerThread_"+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);
		}
	}
	
}
