package kinesisFHM.eventGenerator;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import kinesisFHM.commonTypes.Event;

import org.apache.commons.math3.distribution.PoissonDistribution;

import com.amazonaws.services.kinesis.model.PutRecordsRequest;
import com.amazonaws.services.kinesis.model.PutRecordsRequestEntry;
import com.amazonaws.services.kinesis.model.PutRecordsResult;

public class EventGeneratorThread extends Thread implements Runnable {

	private String[] myCaseIDs;
	private long[] myCaseNums;
	private Iterator<String>[] myIterators;
	private EventGenerator eg;
	
	// maximum supported by AWS is 500
	private static final int outputBatchSize = 500;
	
	public EventGeneratorThread(String[] caseIDs, EventGenerator eg) {
		super();
		
		this.eg = eg;
		setNewCaseIDs(caseIDs);
		
		System.err.println("Created EventGeneratorThread " + this.getName() + " for sending " + caseIDs.length + " cases." );
	}
	
	protected void setNewCaseIDs(String[] newCaseIDs) {
		myCaseIDs = newCaseIDs;
		myIterators = new Iterator[myCaseIDs.length];
		myCaseNums = new long[myCaseIDs.length];
		for (int i=0; i<myCaseIDs.length; i++) {
			myIterators[i] = eg.iterators.get(myCaseIDs[i]);
			myCaseNums[i] = eg.casenums.get(myCaseIDs[i]) + 1;
		}		
	}
	
	private ArrayList<Event> getNextEvents() {
		Long now = System.currentTimeMillis();
		ArrayList<Event> events = new ArrayList<Event>();
		for (int i=0; i<myCaseIDs.length; i++) {
			if (myIterators[i].hasNext()) {
				events.add(new Event(myCaseIDs[i] + "_" + Long.toString(myCaseNums[i]), myIterators[i].next(), now) );
			} else {
				if (myCaseNums[i] < eg.maxInstances - 1) {
					myCaseNums[i]++;
					eg.casenums.put(myCaseIDs[i], myCaseNums[i]);
//					System.err.println(this.getName() + " Starting next instance of caseID " + caseIDs[i]);
					myIterators[i] = eg.cases.get(myCaseIDs[i]).values().iterator();
					eg.iterators.put(myCaseIDs[i], myIterators[i]);
					events.add(new Event(myCaseIDs[i] + "_" + Long.toString(myCaseNums[i]), myIterators[i].next(), now) );					
				}
			}
		}
		return events;
	}

	public void run() {

		boolean output_throttled = false;
		
		while (true) {
			int interarrivalTime = new PoissonDistribution(eg.lambda).sample();
			try {
				sleep(interarrivalTime);
			} catch (InterruptedException e) {
				System.err.println(this.getName() + " Premature wakeup");
			};
			
			ArrayList<Event> eventList = getNextEvents();
			int writeCount = eventList.size();
			while (!eventList.isEmpty()) {
//				System.err.println(this.getName() + " Event Stream Output Queue size = " + eventList.size());
				List <PutRecordsRequestEntry> putRecordsRequestEntryList  = new ArrayList<>(); 

				for (int i=0; i<Math.min(eventList.size(), outputBatchSize); i++) {
					Event event = eventList.get(i);
					PutRecordsRequestEntry putRecordsRequestEntry  = new PutRecordsRequestEntry();
					putRecordsRequestEntry.setData(ByteBuffer.wrap(event.getBase64()));
					putRecordsRequestEntry.setPartitionKey(event.getCaseID());
					putRecordsRequestEntryList.add(putRecordsRequestEntry); 
//					System.err.println(this.getName() + " Added event" + event + " to " + streamName);
				}
				PutRecordsRequest putRecordsRequest  = new PutRecordsRequest();
				putRecordsRequest.setStreamName( eg.streamName );
				putRecordsRequest.setRecords(putRecordsRequestEntryList);
				PutRecordsResult putRecordsResult  = eg.amazonKinesisClient.putRecords(putRecordsRequest);
				int internalErrorCount = 0;
				for(int j=putRecordsResult.getRecords().size()-1; j >= 0 ; j--) {
					if (putRecordsResult.getRecords().get(j).getErrorCode() == null)
						eventList.remove(j);
					else {
						if (putRecordsResult.getRecords().get(j).getErrorCode().equals("ProvisionedThroughputExceededException")) {
							// slowing down by lengthening our sleep period
							output_throttled = true;
						} else {
							internalErrorCount++;
						}
					}
				}
				if (internalErrorCount > 0)
					System.err.println(this.getName() + " INTERNAL ERROR : " + internalErrorCount);
				if (output_throttled) 
					try { Thread.sleep(interarrivalTime/10); } catch (InterruptedException e) {}
			}
			if (!output_throttled) {
				System.err.print("\r" + this.getName() + " Emitted " + myCaseIDs.length + " events with Lambda " + eg.lambda + "  and interarrivalTime " + interarrivalTime + "                     ");
			} else {
				System.err.print("\r" + this.getName() + " Emitted  " + myCaseIDs.length + " events with Lambda " + eg.lambda + "  and interarrivalTime " + interarrivalTime + " === Throttled ===  ");
			}
			output_throttled=false;
//			System.err.println(this.getName() + " Done emitting events");
			// log the puts to cloudwatch
			eg.outputReporter.add(writeCount);
		}
	}
}
