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 java.util.concurrent.ExecutorService;

import net.sf.ehcache.Cache;

import net.sf.ehcache.CacheManager;

import org.apache.log4j.Logger;

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.Executor;

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

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

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

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

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

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

import eu.dnetlib.validator.service.impls.persistance.MemoryRegistry;

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

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

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 ExecutorService to submit and execute jobs.

 * 

 * @author Manos Karvounis

 * @author Nikon Gasparis

 * @see ExecutorService

 */

@Deprecated

public class MemoryThreadValidator implements Validator {

	private CacheManager cacheManager;

	private ProvidersManager providersManager;

	private ListenersManager listenersManager;

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

	private final ExecutorService executor;

	private final Executor taskExecutor;

	private final long timeout, generalTimeout;

	private Boolean dnetWorkflow;

	private MemoryRegistry<Rule> rules;	

	/**

	 * Creates a new validator.

	 * 

	 * @param executor

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

	 * @param taskExecutor

	 *            The executor used to execute the jobs.

	 * @param timeout

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

	 *            task is composed of a single Validation Object applied on all

	 *            the rules of the job.

	 * @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 MemoryThreadValidator(ExecutorService executor, Executor taskExecutor, long timeout, long generalTimeout) {

		super();

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

		this.executor = executor;

		this.taskExecutor = taskExecutor;

		this.timeout = timeout;

		this.generalTimeout = generalTimeout;

		rules = new MemoryRegistry<Rule>(RegistryType.rules);

	}

	public void init() {

	}

	@Override

	public <T extends Serializable> void addToRegistry(int objid, T obj, String registryName) {

		if (registryName.equals(RegistryType.rules)) {

			Rule rule = (Rule) obj;

			log.debug("Adding to registry rule " + rule.getId());

			rules.addObject(rule.getId(), rule);

		} else

			throw new UnsupportedOperationException("You may not add an object to a non-existing registry");

	}

	@Override

	public Serializable getFromRegistry(int objid, String registryName) throws ValidatorException {

		if (registryName.equals(RegistryType.rules)) {

			return rules.getObject(objid);

		} else

			throw new ValidatorException("The registry "+registryName+" does not exist");

	}

	@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(DnetProvider.MDSTORE_ID )!= null) {

				props = new Properties();

				props.setProperty(DnetProvider.WORKER_ID, Integer.toString(workerId));

				props.setProperty(DnetProvider.WORKERS, Integer.toString(workers));

				props.setProperty(DnetProvider.MDSTORE_ID, job.providerProps.getProperty(DnetProvider.MDSTORE_ID));

				props.setProperty(DnetProvider.BATCH_SIZE, job.providerProps.getProperty(DnetProvider.BATCH_SIZE));

				props.setProperty(DnetProvider.RECORDS, job.providerProps.getProperty(DnetProvider.RECORDS));

			}

			workerId++;

			prov.setConfiguration(props);

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

			if (set == null)

				set = "none";

			ExecutorSubmitter submitter = new ThreadExecutorSubmitter(job.id, job.rules, set, prov, mainListener, taskExecutor, timeout, generalTimeout, listeners);

			executor.submit(submitter);

		}  

	}

	@Override

	public void shutdown() throws ValidatorException {

		log.info("Shutting down validator");

		executor.shutdownNow();

		this.rules = null;

	}

	@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 remove as soon as spring validator works ok

//		crisListener.setExecutor(executor);

		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);

			ThreadExecutorSubmitter submitter = new ThreadExecutorSubmitter(job.id, entry.getValue(), entry.getKey(), prv, mainListener, taskExecutor, timeout, generalTimeout, listeners);

			executor.submit(submitter);

		}

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

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

		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 ThreadExecutorSubmitter(job.id, entry.getValue(), entry.getKey(), prv, mainListener, taskExecutor, timeout, generalTimeout, listeners));

		}

		//TODO remove as soon as spring validator works ok

//		crisListener.setSubmittersForReferential(submittersForReferential);

		log.error("Sumbitters created for referential check: " + submittersForReferential.size());

	}

	@Override

	public void start() throws ValidatorException {

		log.debug("Starting validator");

	}

	/**

	 * Contains the names of the 3 containers (registries) used by this

	 * validator implementation.

	 * 

	 * @author Manos Karvounis

	 * @author Nikon Gasparis

	 * 

	 * @see Validator#addToRegistry(int, Serializable, String)

	 */

	public interface RegistryType {

		public static final String rules = "RULES";

	}

	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;

	}

}