package com.fourcasters.forec.reconciler.server.persist;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map.Entry;
import java.util.concurrent.Future;

import com.fourcasters.forec.reconciler.server.ApplicationInterface;
import com.fourcasters.forec.reconciler.server.SelectorTask;
import com.fourcasters.forec.reconciler.server.persist.TradeTaskFactory.FullTask;
import com.fourcasters.forec.reconciler.server.persist.TradeTaskFactory.SingleTradeTask;

public class TransactionManager implements TransactionPhaseListener {

	private int tasksToRun;
	private final TradeTaskFactory taskFactory;
	private final LinkedHashMap<Integer, Transaction> transactions = new LinkedHashMap<>();
	private final ApplicationInterface application;

	public TransactionManager(TradeTaskFactory taskFactory, ApplicationInterface application) {
		super();
		this.tasksToRun = 0;
		this.taskFactory = taskFactory;
		this.application = application;
	}

	public void onFullTransaction(Integer transId, String tradesInMessage) {
		final FullTask task = taskFactory.newFullReconciliationTask(tradesInMessage, this, transId);

		Transaction t = transactions.get(transId);
		if (t == null) {
			final Deque<Runnable> tasks = new ArrayDeque<>(16);
			t = new Transaction(tasks);
			transactions.put(transId, t);
		}
		t.add(task);
		onTaskStart();
	}

	public void onSingleTransaction(int transId, String tradesInMessage) {
		final SingleTradeTask task = taskFactory.newSingleTradeTask(tradesInMessage, this, transId);
		final Deque<Runnable> tasks = new ArrayDeque<>(1);
		tasks.add(task);
		final Transaction t = new Transaction(tasks);
		transactions.put(transId, t);
		onTaskStart();
	}

	private final SelectorTask POLLING_TASK = new SelectorTask() {
		@Override
		public void run() {
			//TODO avoid iterator allocation using toArray(E[])
			final Iterator<Entry<Integer, Transaction>> it = transactions.entrySet().iterator();
			Entry<Integer, Transaction> e = it.next();
			while (e.getValue().completed && it.hasNext()) {
				transactions.remove(e.getKey());
				e = it.next();
			}

			final Transaction t = e.getValue();
			final Runnable task = t.nextTask();
			if (task != null) {
				final Future<?> future = application.executor().submit(task);
				application.futureTasks().add(future);
			}
			else {
				application.selectorTasks().add(this);
			}
		}
	};

	private final SelectorTask DECREASE_TASK_COUNT = new SelectorTask() {
		@Override
		public void run() {
			tasksToRun--;
		}
	};


	int numberOfTransactions() {
		return transactions.size();
	}

	public int tasksToRun() {
		return tasksToRun;
	}

	static class Transaction {

		private final Deque<Runnable> tasks;
		private boolean completed;

		private Transaction(Deque<Runnable> tasks) {
			completed = false;
			this.tasks = tasks;
		}

		private void add(Runnable task) {
			tasks.offer(task);
		}

		private Runnable nextTask() {
			assert tasks.size() > 0;
			return tasks.poll();
		}

		void complete() {
			completed = true;
		}
	}

	@Override
	public void onTransactionStart(int transId) {

	}

	@Override
	public void onTransactionEnd(int transId) {
		application.selectorTasks().add(new SelectorTask() {
			@Override
			public void run() {
				final Transaction t = transactions.get(transId);
				if (t != null) {
					t.completed = true;
				}
			}
		});
	}

	@Override
	public void onTaskEnd() {
		application.selectorTasks().add(DECREASE_TASK_COUNT);
	}

	@Override
	public void onTaskStart() {
		tasksToRun++;
		//add polling task to selector task queue
		application.selectorTasks().add(POLLING_TASK);
	}

}

@@ -1983,16 +1983,43 @@
 ator();%0A
+%09%09%09while (it.hasNext()) %7B%0A%09
 %09%09%09Entry
@@ -2059,23 +2059,22 @@
 t();%0A%09%09%09
-while
+%09if
  (
+!
 e.getVal
@@ -2091,90 +2091,14 @@
 eted
- && it.hasNext()) %7B%0A%09%09%09%09transactions.remove(e.getKey());%0A%09%09%09%09e = it.next();%0A%09%09%09%7D%0A%0A
+) %7B%0A%09%09
 %09%09%09f
@@ -2131,24 +2131,26 @@
 Value();%0A%09%09%09
+%09%09
 final Runnab
@@ -2176,16 +2176,18 @@
 k();%0A%09%09%09
+%09%09
 if (task
@@ -2198,24 +2198,26 @@
 null) %7B%0A%09%09%09%09
+%09
+%09
 final Future
@@ -2266,24 +2266,26 @@
 (task);%0A%09%09%09%09
+%09%09
 application.
@@ -2311,24 +2311,26 @@
 uture);%0A
+%09%09
 %09%09%09%7D%0A%09%09%09
 else %7B%0A%09
@@ -2325,15 +2325,19 @@
 %0A%09%09%09
+%09%09
 else %7B%0A
+%09%09
 %09%09%09%09
@@ -2375,16 +2375,58 @@
 (this);%0A
+%09%09%09%09%09%7D%0A%09%09%09%09%09break;%0A%09%09%09%09%7D%0A%09%09%09%09it.remove();%0A
 %09%09%09%7D%0A%09%09%7D

Fix concurrent modification exception