package kinesisFHM.utilities;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.PrintWriter;
import java.sql.Timestamp;
import java.util.Date;

import com.amazonaws.services.cloudwatch.AmazonCloudWatchClient;
import com.amazonaws.services.cloudwatch.model.Dimension;
import com.amazonaws.services.cloudwatch.model.MetricDatum;
import com.amazonaws.services.cloudwatch.model.PutMetricDataRequest;
import com.amazonaws.services.cloudwatch.model.StandardUnit;

public class CloudWatchReporter extends Thread implements Runnable {

	private String metricName;
	private String streamName;
	private AmazonCloudWatchClient amazonCloudWatchClient;
	
	private int count;
	private File logFile;
	
	public CloudWatchReporter(AmazonCloudWatchClient amazonCloudWatchClient, String metricName, String streamName) {
		this.metricName = metricName;
		this.streamName = streamName;
		this.amazonCloudWatchClient = amazonCloudWatchClient;
		this.count = 0;
		this.setDaemon(true);
		
		this.logFile = new File("kinesisFHM." + metricName + "." + streamName + ".csv");
		
		try {
			if (!logFile.exists()) {
				PrintWriter out = new PrintWriter(new BufferedWriter(new FileWriter(logFile, false)));
				out.println("Time, Value");
				out.close();
			}
		} catch (IOException e) { }
	}
	
	public synchronized void add(int amount) {
		count += amount;
	}
	
	private synchronized void report() {
		try {
			amazonCloudWatchClient.putMetricData(new PutMetricDataRequest()
				.withNamespace("kinesisFHM")
				.withMetricData(new MetricDatum()
					.withMetricName(metricName)
					.withTimestamp(new Timestamp(System.currentTimeMillis()))
					.withUnit(StandardUnit.Count)
					.withValue(new Double(count))
					.withDimensions(new Dimension()
						.withName("StreamName")
						.withValue(streamName))));

			PrintWriter out = new PrintWriter(new BufferedWriter(new FileWriter(logFile, true)));
			Date d = new Date();
			out.println(d + ", " + count);
			out.close();

			count=0;

		} catch (Exception e) {
			System.err.println(this.getName() + " " + e.getMessage());
		}
	}
	
	public void run() {
		while (true) {
			try {
				sleep(10000);
				report();
			} catch (Exception e) {}
		}
	}
}
