D-Net

package eu.dnetlib.validator.service.impls;

import java.io.Serializable;

import java.util.ArrayList;

import java.util.List;

import java.util.Map;

import java.util.Properties;

import java.util.Set;

import java.util.UUID;

import net.sf.ehcache.Cache;

import net.sf.ehcache.CacheManager;

import org.apache.log4j.Logger;

import org.quartz.spi.ThreadPool;

import org.springframework.core.task.SyncTaskExecutor;

import org.springframework.core.task.TaskExecutor;

import eu.dnetlib.validator.engine.Validator;

import eu.dnetlib.validator.engine.ValidatorException;

import eu.dnetlib.validator.engine.data.Provider;

import eu.dnetlib.validator.engine.data.Rule;

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

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

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

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

import eu.dnetlib.validator.service.impls.listeners.CrisListener;

import eu.dnetlib.validator.service.impls.listeners.ListenersManager;

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

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

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

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

/**

 * A simple validator that stores rules, providers, and executing/pending jobs

 * in memory. It uses an SyncTaskExecutor to submit and execute jobs.

 * 

 * @author Nikon Gasparis

 * @see ThreadPool

 */

public class SpringValidator implements Validator {

	private CacheManager cacheManager;

	private ProvidersManager providersManager;

	private ListenersManager listenersManager;

	private transient Logger log = Logger.getLogger(SpringValidator.class);

	private final TaskExecutor jobExecutor;

	private final SyncTaskExecutor taskExecutor;

	private final long generalTimeout;

	private Boolean dnetWorkflow;

	/**

	 * Creates a new validator.

	 * 

	 * @param jobExecutor

	 *            The module used to run the providers and submit jobs.

	 * @param taskExecutor

	 *            The executor used to execute the jobs.

	 * @param generalTimeout

	 *            How long to wait for a job to be executed (in seconds). A job

	 *            is composed of the application of all the rules on all of the

	 *            Validation Objects.

	 */

	public SpringValidator(TaskExecutor jobExecutor, SyncTaskExecutor taskExecutor, long generalTimeout) {

		super();

		log.info("Creating a new Validator");

		this.generalTimeout = generalTimeout;

		this.taskExecutor = taskExecutor;

		this.jobExecutor = jobExecutor;

	}

	@Override

	public <T extends Serializable> void addToRegistry(int objid, T obj,

			String registryName) {

		throw new UnsupportedOperationException(

				"You may not add new registries to this Validator implementation");

	}

	@Override

	public Serializable getFromRegistry(int objid, String registryName)

			throws ValidatorException {

		throw new UnsupportedOperationException(

				"You may not add new registries to this Validator implementation");

	}

	@Override

	public <T extends Serializable> void addRegistry(String name) {

		throw new UnsupportedOperationException(

				"You may not add new registries to this Validator implementation");

	}

	@Override

	public void submitJob(Job job, int workers, JobListener... listeners)

			throws ValidatorException {

		log.debug("Submitting job " + job.id);

		List<Provider> providers = new ArrayList<Provider>();

		try {

			log.debug("Creating a new provider instance");

			for (int i = 0; i < workers; i++) {

				Provider prv = null;

				if (job.providerId == 1)

					prv = providersManager.createOaipmhRecordProvider();

				else if (job.providerId == 2)

					prv = providersManager.createOaipmhSinglePageVerbProvider();

				else if (job.providerId == 3)

					prv = providersManager.createDnetProvider();

				providers.add(prv);

			}

		} catch (Exception e) {

			log.error("Error creating provider instance", e);

		}

		ValidatorJobMainListener mainListener = new ValidatorJobMainListener();

		// mainListener.setWorkers(workers);

		Integer workerId = 0;

		for (Provider prov : providers) {

			Properties props = job.providerProps;

			if (job.providerProps.getProperty(DnetProviderNew.MDSTORE_ID) != null) {

				props = new Properties();

				props.setProperty(DnetProviderNew.WORKER_ID,

						Integer.toString(workerId));

				props.setProperty(DnetProviderNew.WORKERS,

						Integer.toString(workers));

				props.setProperty(DnetProviderNew.MDSTORE_ID,

						job.providerProps.getProperty(DnetProviderNew.MDSTORE_ID));

				props.setProperty(DnetProviderNew.BATCH_SIZE,

						job.providerProps.getProperty(DnetProviderNew.BATCH_SIZE));

				props.setProperty(DnetProviderNew.RECORDS,

						job.providerProps.getProperty(DnetProviderNew.RECORDS));

			}

			workerId++;

			prov.setConfiguration(props);

			String set = props.getProperty(OAIPMHRecordProvider.SET);

			if (set == null)

				set = "none";

			JobWorker jobWorker = new JobWorker(job.id, job.rules, set, prov,

					mainListener, taskExecutor, generalTimeout, listeners);

			jobExecutor.execute(jobWorker);

		}

	}

	@Override

	public void submitJobForCris(Job job,

			Map<String, Set<Rule>> rulesPerEntity,

			Map<String, Set<Rule>> rulesPerEntityRef, JobListener... listeners)

			throws ValidatorException {

		CrisProvider prv = providersManager.createCrisProvider();

		log.debug("creating cris  - timeout: " + prv.getTimeout());

		String cacheName = UUID.randomUUID().toString();

		Properties props = job.providerProps;

		if (!rulesPerEntityRef.isEmpty()) {

			cacheManager.addCache(new Cache(cacheName, 20000, true, true,

					10000, 10000));

			log.debug("caches: ");

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

				log.debug("name: " + cache);

			}

			prv.setCache(cacheManager.getCache(cacheName));

		}

		prv.setEntities(rulesPerEntity.keySet());

		prv.setBaseUrl(props.getProperty(OAIPMHRecordProvider.BASEURL));

		prv.setMetadataPrefix(props

				.getProperty(OAIPMHRecordProvider.METADATA_PREFIX));

		prv.setRecords(props.getProperty(OAIPMHRecordProvider.RECORDS));

		ValidatorJobMainListener mainListener = new ValidatorJobMainListener();

		// mainListener.setWorkers(rulesPerEntity.keySet().size() +

		// rulesPerEntityRef.keySet().size());

		CrisListener crisListener = listenersManager.createCrisListener();

		crisListener.setWorkersFirstPhase(rulesPerEntity.keySet().size());

		//TODO UPDATE LISTENER

		crisListener.setExecutor(jobExecutor);

		//		crisListener.setExecutor(taskExecutor);

		crisListener.setProvider(prv);

		crisListener.setWorkers(rulesPerEntity.keySet().size()

				+ rulesPerEntityRef.keySet().size());

		if (!rulesPerEntityRef.isEmpty()) {

			crisListener.setCacheManager(cacheManager);

			crisListener.setCacheName(cacheName);

		}

		mainListener.setCrisListener(crisListener);

		for (Map.Entry<String, Set<Rule>> entry : rulesPerEntity.entrySet()) {

			log.debug("Cris# set: " + entry.getKey() + " - Rules size: "

					+ entry.getValue().size());

			prv.setSet(entry.getKey());

			props.setProperty(OAIPMHRecordProvider.SET, entry.getKey());

			prv.setConfiguration(props);

			JobWorker jobWorker = new JobWorker(job.id, entry.getValue(), entry.getKey(), prv,

					mainListener, taskExecutor, generalTimeout,

					listeners);

			jobExecutor.execute(jobWorker);

		}

		log.error("Creating Sumbitters for referential check..");

		List<JobWorker> submittersForReferential = new ArrayList<JobWorker>();

		for (Map.Entry<String, Set<Rule>> entry : rulesPerEntityRef.entrySet()) {

			log.error("Ref Submitter set: " + entry.getKey() + " and rules: "

					+ entry.getValue() + " and new rule ids: "

					+ entry.getValue().size());

			// logger.debug("Cris# set: "+ entry.getKey() + " - Rules: " +

			// entry.getValue());

			prv.setSet(entry.getKey());

			props.setProperty(OAIPMHRecordProvider.SET, entry.getKey());

			prv.setConfiguration(props);

			submittersForReferential.add(new JobWorker(job.id,

					entry.getValue(), entry.getKey(), prv, mainListener,

					taskExecutor, generalTimeout, listeners));

		}

		//TODO UPDATE LISTENER

//		crisListener.setSubmittersForReferential(submittersForReferential);

		crisListener.setSubmittersForReferential(submittersForReferential);

		log.error("Sumbitters created for referential check: "

				+ submittersForReferential.size());

	}

	@Override

	public void start() throws ValidatorException {

		log.debug("Starting validator");

	}

	private class ValidatorJobMainListener implements JobListener {

		private CrisListener crisListener = null;

		public void setCrisListener(CrisListener crisListener) {

			this.crisListener = crisListener;

		}

		@Override

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

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

			// TODO Auto-generated method stub

		}

		@Override

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

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

			// TODO Auto-generated method stub

		}

		@Override

		public synchronized void finished(int jobId,

				Map<String, Object> jobContext) {

			if (crisListener != null)

				crisListener.finished(jobId, jobContext);

			log.debug("Job " + jobId

					+ " has finished, removing it from the registry");

		}

		@Override

		public synchronized void failed(int jobId,

				Map<String, Object> jobContext, Throwable t) {

			if (crisListener != null)

				crisListener.failed(jobId, jobContext, t);

			log.debug("A job has finished-failed, removing it from the registry");

		}

	}

	public Boolean getDnetWorkflow() {

		return dnetWorkflow;

	}

	public void setDnetWorkflow(Boolean dnetWorkflow) {

		this.dnetWorkflow = dnetWorkflow;

	}

	public ProvidersManager getProvidersManager() {

		return providersManager;

	}

	public void setProvidersManager(ProvidersManager providersManager) {

		this.providersManager = providersManager;

	}

	public CacheManager getCacheManager() {

		return cacheManager;

	}

	public void setCacheManager(CacheManager cacheManager) {

		this.cacheManager = cacheManager;

	}

	public ListenersManager getListenersManager() {

		return listenersManager;

	}

	public void setListenersManager(ListenersManager listenersManager) {

		this.listenersManager = listenersManager;

	}

	@Override

	public void shutdown() throws ValidatorException {

		// TODO Auto-generated method stub

	}

}