package kinesisFHM.utilities;

import java.util.List;

import com.amazonaws.services.kinesis.AmazonKinesisClient;
import com.amazonaws.services.kinesis.model.CreateStreamRequest;
import com.amazonaws.services.kinesis.model.DescribeStreamRequest;
import com.amazonaws.services.kinesis.model.DescribeStreamResult;
import com.amazonaws.services.kinesis.model.ListStreamsRequest;
import com.amazonaws.services.kinesis.model.ListStreamsResult;
import com.amazonaws.services.kinesis.model.ResourceNotFoundException;
import com.amazonaws.services.kinesis.model.Shard;

public class Functions {
	
	public static List<Shard> getInputShards(AmazonKinesisClient amazonKinesisClient, String inputStreamName) {
		System.err.println("Getting input shards");

		ListStreamsRequest listStreamsRequest = new ListStreamsRequest();
		listStreamsRequest.setLimit(10000);
		ListStreamsResult listStreamsResult = amazonKinesisClient.listStreams(listStreamsRequest);

		if (!listStreamsResult.getStreamNames().contains( inputStreamName )) {
			throw new RuntimeException( "Stream " + inputStreamName + " does not exist");
		}

		DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest();
		describeStreamRequest.setStreamName( inputStreamName );
		describeStreamRequest.setLimit(10000);
		
		List<Shard> inputShards = null;

		long endTime = System.currentTimeMillis() + ( 10 * 60 * 1000 );
		while ( System.currentTimeMillis() < endTime ) {
			try {
				System.err.println("  trying ...");
				DescribeStreamResult describeStreamResponse = amazonKinesisClient.describeStream( describeStreamRequest );
				String streamStatus = describeStreamResponse.getStreamDescription().getStreamStatus();
				if ( streamStatus.equals( "ACTIVE" ) ) {
					inputShards = describeStreamResponse.getStreamDescription().getShards();
					break;
				}
				try { Thread.sleep( 5 * 1000 ); }
				catch ( Exception e ) {}
			}
			catch ( ResourceNotFoundException e ) {}
		}
		if ( System.currentTimeMillis() >= endTime ) {
			System.err.println( "Stream " + inputStreamName + " is not active" );
		}
		return inputShards;
	}
	
	public static boolean isOutputStreamActive(AmazonKinesisClient amazonKinesisClient, String streamName, int numOutputShards) {
		System.err.println("Checking output stream " + streamName );
		
		ListStreamsRequest listStreamsRequest = new ListStreamsRequest();
		listStreamsRequest.setLimit(10000);
		ListStreamsResult listStreamsResult = amazonKinesisClient.listStreams(listStreamsRequest);

		if (!listStreamsResult.getStreamNames().contains( streamName )) {
			CreateStreamRequest createStreamRequest = new CreateStreamRequest();
			createStreamRequest.setStreamName( streamName );
			createStreamRequest.setShardCount( numOutputShards );
			amazonKinesisClient.createStream( createStreamRequest );
			System.err.println("  creating ...");
			try { Thread.sleep(20 * 1000); } 
			catch ( Exception e ) {}
		}

		DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest();
		describeStreamRequest.setStreamName( streamName );

		long endTime = System.currentTimeMillis() + ( 10 * 60 * 1000 );
		while ( System.currentTimeMillis() < endTime ) {

			try {
				System.err.println("  waiting for active ...");
				DescribeStreamResult describeStreamResponse = amazonKinesisClient.describeStream( describeStreamRequest );
				String streamStatus = describeStreamResponse.getStreamDescription().getStreamStatus();
				if ( streamStatus.equals( "ACTIVE" ) ) {
					break;
				}
				try { Thread.sleep( 5 * 1000 ); }
				catch ( Exception e ) {}
			}
			catch ( ResourceNotFoundException e ) {}
		}
		if ( System.currentTimeMillis() >= endTime ) {
			System.err.println( "Stream " + streamName + " never went active" );
			return false;
		}
		return true;
	}
	

}
