package org.hive2hive.core.file.watcher;

import java.io.File;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.hive2hive.core.api.interfaces.IFileManager;
import org.hive2hive.core.exceptions.NoPeerConnectionException;
import org.hive2hive.core.exceptions.NoSessionException;
import org.hive2hive.core.file.FileUtil;
import org.hive2hive.core.log.H2HLogger;
import org.hive2hive.core.log.H2HLoggerFactory;
import org.hive2hive.core.processes.framework.RollbackReason;
import org.hive2hive.core.processes.framework.exceptions.InvalidProcessStateException;
import org.hive2hive.core.processes.framework.interfaces.IProcessComponent;
import org.hive2hive.core.processes.framework.interfaces.IProcessComponentListener;

public class DeleteFileBuffer extends BaseFileBuffer {

	private static final H2HLogger logger = H2HLoggerFactory.getLogger(DeleteFileBuffer.class);
	private static final long MAX_DELETION_PROCESS_DURATION_MS = 30000; // timeout to omit blocks

	public DeleteFileBuffer(IFileManager fileManager, File root) {
		super(fileManager, root);
	}

	/**
	 * Process the files in the buffer after the buffering time exceeded.
	 * 
	 * @param bufferedFiles
	 */
	protected void processBuffer(IFileBufferHolder buffer) {
		List<File> bufferedFiles = buffer.getFileBuffer();
		Set<File> syncFiles = buffer.getSyncFiles();

		Set<File> toRemove = new HashSet<File>();
		for (File file : bufferedFiles) {
			if (!syncFiles.contains(file)) {
				// has already been deleted
				toRemove.add(file);
			}
		}
		bufferedFiles.removeAll(toRemove);

		// sort first
		FileUtil.sortPreorder(bufferedFiles);
		// reverse the sorting
		Collections.reverse(bufferedFiles);

		// delete individual files
		for (File toDelete : bufferedFiles) {
			try {
				logger.debug("Starting to delete buffered file " + toDelete);
				IProcessComponent delete = fileManager.delete(toDelete);
				BlockingListener listener = new BlockingListener();
				delete.attachListener(listener);
				if (!fileManager.isAutostart())
					delete.start();

				// wait for execution
				listener.waitForFinish();
			} catch (NoSessionException | NoPeerConnectionException | InvalidProcessStateException e) {
				logger.error(e.getMessage());
			}
		}

		logger.debug("Buffer with " + bufferedFiles.size() + " files processed.");
	}

	/**
	 * Listener blocking until the process is finished
	 */
	private class BlockingListener implements IProcessComponentListener {

		private final CountDownLatch latch;

		public BlockingListener() {
			this.latch = new CountDownLatch(1);
		}

		@Override
		public void onSucceeded() {
			latch.countDown();
		}

		@Override
		public void onFailed(RollbackReason reason) {
			latch.countDown();
		}

		@Override
		public void onFinished() {
			latch.countDown();
		}

		public void waitForFinish() {
			try {
				latch.await(MAX_DELETION_PROCESS_DURATION_MS, TimeUnit.MILLISECONDS);
			} catch (InterruptedException e) {
				logger.error("Could not wait for process to finish");
			}
		}

	}
}

@@ -160,90 +160,8 @@
 Set;
-%0Aimport java.util.concurrent.CountDownLatch;%0Aimport java.util.concurrent.TimeUnit;
 %0A%0Aim
@@ -468,70 +468,8 @@
 ry;%0A
-import org.hive2hive.core.processes.framework.RollbackReason;%0A
 impo
@@ -630,92 +630,8 @@
 ent;
-%0Aimport org.hive2hive.core.processes.framework.interfaces.IProcessComponentListener;
 %0A%0Apu
@@ -1787,101 +1787,8 @@
 e);%0A
-%09%09%09%09BlockingListener listener = new BlockingListener();%0A%09%09%09%09delete.attachListener(listener);%0A
 %09%09%09%09
@@ -1844,61 +1844,24 @@
 ();%0A
-%0A
 %09%09%09%09
-// wait for execution%0A%09%09%09%09listener.waitForFinish
+delete.await
 ();%0A
@@ -1941,32 +1941,60 @@
 ssStateException
+%0A%09%09%09%09%09%7C InterruptedException
  e) %7B%0A%09%09%09%09logger
@@ -2110,703 +2110,7 @@
 ;%0A%09%7D
-%0A%0A%09/**%0A%09 * Listener blocking until the process is finished%0A%09 */%0A%09private class BlockingListener implements IProcessComponentListener %7B%0A%0A%09%09private final CountDownLatch latch;%0A%0A%09%09public BlockingListener() %7B%0A%09%09%09this.latch = new CountDownLatch(1);%0A%09%09%7D%0A%0A%09%09@Override%0A%09%09public void onSucceeded() %7B%0A%09%09%09latch.countDown();%0A%09%09%7D%0A%0A%09%09@Override%0A%09%09public void onFailed(RollbackReason reason) %7B%0A%09%09%09latch.countDown();%0A%09%09%7D%0A%0A%09%09@Override%0A%09%09public void onFinished() %7B%0A%09%09%09latch.countDown();%0A%09%09%7D%0A%0A%09%09public void waitForFinish() %7B%0A%09%09%09try %7B%0A%09%09%09%09latch.await(MAX_DELETION_PROCESS_DURATION_MS, TimeUnit.MILLISECONDS);%0A%09%09%09%7D catch (InterruptedException e) %7B%0A%09%09%09%09logger.error(%22Could not wait for process to finish%22);%0A%09%09%09%7D%0A%09%09%7D%0A%0A%09%7D
 %0A%7D%0A

Use the await #69 in the fileobserver