D-Net

package eu.dnetlib.validator.service.impls.listeners;

import java.util.List;

import java.util.Map;

import java.util.concurrent.ExecutorService;

import eu.dnetlib.validator.service.impls.executors.JobWorker;

import net.sf.ehcache.CacheManager;

import org.apache.log4j.Logger;

import org.springframework.core.task.TaskExecutor;

import org.springframework.stereotype.Controller;

import eu.dnetlib.validator.engine.execution.CompletedTask;

import eu.dnetlib.validator.engine.execution.JobListener;

import eu.dnetlib.validator.service.impls.executors.ThreadExecutorSubmitter;

import eu.dnetlib.validator.service.impls.providers.CrisProvider;

@Controller

public class CrisListener implements JobListener {

	private static Logger logger = Logger.getLogger(CrisListener.class);

	private int workersFirstPhase = 0;

	private int workersFinished = 0;

	private int workers;

//	private ExecutorService executor;

	private TaskExecutor executor;

	private List<JobWorker> submittersForReferential;

//	private List<ThreadExecutorSubmitter> submittersForReferential;

	private CrisProvider provider;

	private CacheManager cacheManager = null;

	private String cacheName = null;

	public void beginCrisReferentialCheck() {

		provider.restartResultSets();

		for (JobWorker submiter : this.submittersForReferential) {

			executor.execute(submiter);

		}

//		for (ThreadExecutorSubmitter submiter : this.submittersForReferential) {

//			executor.submit(submiter);

//		}

	}

	@Override

	public void currentResults(List<CompletedTask> tasks, int jobId,

			Object record, Map<String, Object> recordContext, Throwable t) {

		// TODO Auto-generated method stub

	}

	@Override

	public void currentResults(List<CompletedTask> tasks, int jobId,

			Object record, Map<String, Object> recordContext) {

		// TODO Auto-generated method stub

	}

	@Override

	public synchronized void finished(int jobId, Map<String, Object> jobContext) {

		try { 

			workersFinished ++;

			if (workersFinished == workersFirstPhase) {

				logger.debug("1st phase finished.. Beginning second one..");

				this.beginCrisReferentialCheck();

			}

			if (workersFinished == workers) {

				if (cacheName != null) {

					cacheManager.removeCache(cacheName);

					logger.debug("caches after removal of cache: " + cacheName);

				}

	//			for (String cache : cacheManager.getCacheNames()) {

	//				logger.debug("name: " + cache);

	//			}

			}

		} catch (Exception e) {

			logger.error("Error while finalizing successful cris job");

		}

	}

	@Override

	public synchronized void failed(int jobId, Map<String, Object> jobContext,

			Throwable t) {

		try { 

			workersFinished ++;

			if (workersFinished == (workers-workersFirstPhase)) {

				if (cacheName != null) {

					cacheManager.removeCache(cacheName);

					logger.debug("caches after removal of cache: " + cacheName);

				}

	//			for (String cache : cacheManager.getCacheNames()) {

	//				logger.debug("name: " + cache);

	//			}

			}

		} catch (Exception e) {

			logger.error("Error while finalizing failed cris job");

		}

	}

	public void setWorkersFirstPhase(int workersFirstPhase) {

		this.workersFirstPhase = workersFirstPhase;

	}

	public void setWorkers(int workers) {

		this.workers = workers;

	}

//	public void setExecutor(ExecutorService executor) {

//		this.executor = executor;

//	}

//	public void setSubmittersForReferential(

//			List<ThreadExecutorSubmitter> submittersForReferential) {

//		this.submittersForReferential = submittersForReferential;

//	}

	public void setExecutor(TaskExecutor executor) {

		this.executor = executor;

	}

	public void setSubmittersForReferential(List<JobWorker> submittersForReferential) {

		this.submittersForReferential = submittersForReferential;

	}

	public void setProvider(CrisProvider provider) {

		this.provider = provider;

	}

	public void setCacheManager(CacheManager cacheManager) {

		this.cacheManager = cacheManager;

	}

	public void setCacheName(String cacheName) {

		this.cacheName = cacheName;

	}

}