package com.orangemile.chronicle;

import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.LogManager;

import com.higherfrequencytrading.chronicle.Excerpt;
import com.higherfrequencytrading.chronicle.impl.IndexedChronicle;

public class ReplayChronicle {

	private static final Logger log = LogManager.getLogger( ReplayChronicle.class.getName() );

	/**
	 * 
	 * @param chronicles
	 * @param timeZeroInMs
	 * @param speedFactor
	 * @param replayCallback
	 */
	public void replay(String [] chronicles, long timeZeroInMs, int speedFactor, ReplayHandler callback ) {
		if ( callback == null )
			throw new RuntimeException("Callback is required");
		Future<?>[] futures = new Future[chronicles.length];
		ExecutorService executorService = Executors.newFixedThreadPool(chronicles.length);
		for (int i = 0; i < chronicles.length; i++) {
			ReplayThread r = new ReplayThread(chronicles[i], timeZeroInMs, speedFactor, callback);
			futures[i] = executorService.submit(r);
		}
		for (Future<?> f : futures) {
			try {
				f.get();
			} catch (Exception e) {
				log.error(e, e);
			}
		}		
	}
	
	private static class ReplayThread implements Runnable {

		private String path;
		private long lastTime;
		private int speedFactor;
		private ReplayHandler callback;
		
		public ReplayThread( String path, long timeZero, int speedFactor, ReplayHandler callback ) {
			this.path = path;
			this.lastTime = timeZero;
			this.speedFactor = speedFactor;
			this.callback = callback;
		}
		
		public void run() {
			Thread.currentThread().setName(path);
			try {
				IndexedChronicle c = new IndexedChronicle(path);
				log.debug("number of entries in chronicle: " + c.size());
				Excerpt ptr = c.createExcerpt();
				long relativeTime = 0;
				while (ptr.hasNextIndex()) {
					if (!ptr.nextIndex()) {
						log.debug(path + ": there are more entries, but unable to progress");
						break;
					}
					long time = ptr.readLong();
					// if start time is not defined, assume start is the first entry
					if (lastTime == 0) {  
						lastTime = time;
						Date firstEntryDate = new Date(lastTime);
						log.debug("Time zero: " + firstEntryDate);
						callback.replay(ptr, time, relativeTime);
					} else {
						long diff = (time - lastTime) / speedFactor;
						log.debug("sleeping " + diff + ", current-relative-time: " + relativeTime );
						Thread.sleep(diff);
						relativeTime += diff;
						lastTime = time;
						callback.replay(ptr, time, relativeTime);
					}
				}
				log.debug("Chronicle (" + path + ") ended!");
				c.close();
			} catch ( Exception e ) {
				throw new RuntimeException(e);
			}
		}
	}

}
