package kinesisFHM.relationConsumer;

import java.util.List;

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

	private static final String inputStreamName="RelationStream";
	private static final String outputStreamName="UpdateStream";	
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {

		if (args.length < 4) {
			System.err.println("Usage: RelationConsumer credentialFileName awsProfileName inputBatchSize threadsleep [minDep]");
			System.exit(-1);
		}
		
		String credentialFileName = args[0];
		String awsProfileName = args[1];		
		int inputBatchSize = Integer.parseInt(args[2]);
		long threadsleep = Long.parseLong(args[3]);
		
		float minDep = 0.5f;
		if (args.length == 5)
			minDep = Float.parseFloat(args[4]); 
		RelationConsumerThread.mindep = minDep;
		
		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, Math.max(1,  inputShards.size()/10))) {
				for (Shard shard : inputShards) {
					RelationConsumerThread t = new RelationConsumerThread(	inputStreamName, 
																			shard, 
																			amazonKinesisClient, 
																			inputReporter,
																			outputReporter,
																			outputStreamName,
																			threadsleep,
																			inputBatchSize);
					t.setName("RelationConsumerThread_" + 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);
		}		
	}

}
