D-Net

{

  "type_source": "SVN", 

  "goal": "package -U source:jar", 

  "url": "http://svn-public.driver.research-infrastructures.eu/driver/dnet40/modules/uoa-validator-service/trunk", 

  "deploy_repository": "dnet4-snapshots", 

  "version": "4", 

  "mail": "antleb@di.uoa.gr, kiatrop@di.uoa.gr, nikonas@di.uoa.gr", 

  "deploy_repository_url": "http://maven.research-infrastructures.eu/nexus/content/repositories/dnet4-snapshots", 

  "name": "uoa-validator-service"

}

package eu.dnetlib.validator.service.impl;

import eu.dnetlib.domain.functionality.validator.JobForValidation;

import eu.dnetlib.domain.functionality.validator.RuleSet;

import eu.dnetlib.domain.functionality.validator.StoredJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardNotificationHandler;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;

import eu.dnetlib.validator.engine.ValidatorException;

import java.util.List;

public interface ValidatorManager {

    StoredJob beginDataJobForWorkflow(String datasource, String guidelines, String groupBy, int records, int workers, BlackboardJob bJob, BlackboardNotificationHandler<BlackboardServerHandler> blackboardHandler, int jobStatusUpdateInterval, boolean outputEpr, boolean blacklistedRecords, String blacklistGuidelines) throws ValidatorException;

    StoredJob getStoredJob(int jobId, String groupBy) throws ValidatorException;

    List<StoredJob> getStoredJobs(String userMail, String jobType,

                                         Integer offset, Integer limit, String dateFrom, String dateTo) throws ValidatorException;

    List<StoredJob> getStoredJobs(String userMail, String jobType,

                                  Integer offset, Integer limit, String dateFrom, String dateTo, String jobStatus) throws ValidatorException;

    List<RuleSet> getRuleSets() throws ValidatorException;

    void submitJob(JobForValidation job) throws ValidatorException;

    int getStoredJobsTotalNumber(String userMail, String jobType) throws ValidatorException;

    int getStoredJobsTotalNumber(String userMail, String jobType, String jobStatus) throws ValidatorException;

}

package eu.dnetlib.validator.service.impl;

import eu.dnetlib.domain.functionality.validator.CustomProperties;

import eu.dnetlib.domain.functionality.validator.JobForValidation;

import eu.dnetlib.domain.functionality.validator.RuleSet;

import eu.dnetlib.domain.functionality.validator.StoredJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardNotificationHandler;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;

import eu.dnetlib.validator.commons.dao.DaoException;

import eu.dnetlib.validator.commons.dao.jobs.JobsDAO;

import eu.dnetlib.validator.commons.dao.rules.RulesDAO;

import eu.dnetlib.validator.commons.dao.rules.RulesetsDAO;

import eu.dnetlib.validator.engine.Validator;

import eu.dnetlib.validator.engine.ValidatorException;

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

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

import eu.dnetlib.validator.service.impls.ValidatorRestore;

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

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

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

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

import eu.dnetlib.validator.service.impls.rules.ChainRule;

import eu.dnetlib.validator.service.impls.rules.RuleTypes;

import org.apache.log4j.Logger;

import org.springframework.transaction.annotation.Propagation;

import org.springframework.transaction.annotation.Transactional;

import java.util.*;

@Transactional

public class ValidatorManagerImpl implements ValidatorManager {

	private Validator validator;

	private JobsDAO jobsDao;

	private RulesetsDAO rulesetsDao;

	private RulesDAO rulesDao;

	private ValidatorRestore valRestore;

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

	private ListenersManager listenersManager;

//	this.restartJobs();

	public void start() {

		logger.info("Initializing Validator Manager module");

		if (valRestore.isAutoRestoreOnStartup()) {

			logger.info("auto-restoring OpenAire Validator is enabled..");

			List<StoredJob> jobs = valRestore.deleteUncompleted();

			valRestore.restartJobs(jobs);

			logger.info("finished restoring OpenAire Validator..");

		} else {

			logger.info("auto-restoring OpenAire Validator is disabled..");

		}

	}

	public StoredJob getStoredJob(int jobId, String groupBy)

			throws ValidatorException {

		try {

			logger.info("Getting job summary for job " + jobId +" with groupBy: "+groupBy);

			return jobsDao.getJobSummary(jobId,groupBy);

		} catch (Exception e) {

			logger.error("error getting job summary for job " + jobId, e);

			throw new ValidatorException(e);

		}

	}

	public List<StoredJob> getStoredJobs(String userMail, String jobType,

			Integer offset, Integer limit, String dateFrom, String dateTo)

			throws ValidatorException {

		return this.getStoredJobs(userMail, jobType, offset, limit, dateFrom, dateTo, null);

	}

	public List<StoredJob> getStoredJobs(String userMail, String jobType,

										 Integer offset, Integer limit, String dateFrom, String dateTo, String jobStatus)

			throws ValidatorException {

		try {

			logger.debug("Getting jobs of user " + userMail);

			return jobsDao.getJobs(userMail, jobType, offset, limit, dateFrom, dateTo, jobStatus);

		} catch (Exception e) {

			logger.error("error Getting jobs of user " + userMail, e);

			throw new ValidatorException(e);

		}

	}

	public int getStoredJobsTotalNumber(String userMail, String jobType)

			throws ValidatorException {

		return this.getStoredJobsTotalNumber(userMail, jobType, null);

	}

	public int getStoredJobsTotalNumber(String userMail, String jobType, String jobStatus)

			throws ValidatorException {

		try {

			logger.debug("Getting jobs total sum of user " + userMail);

			return jobsDao.getJobsTotalNumber(userMail, jobType, jobStatus);

		} catch (Exception e) {

			logger.error("error Getting jobs sum of user " + userMail, e);

			throw new ValidatorException(e);

		}

	}

	public List<RuleSet> getRuleSets() throws ValidatorException {

		try {

			logger.info("Getting rulesets");

			return rulesetsDao.getRuleSets();

		} catch (Exception e) {

			logger.error("error Getting rulesets ", e);

			throw new ValidatorException(e);

		}

	}

	@Transactional(readOnly = false, propagation = Propagation.REQUIRES_NEW)

	public StoredJob beginDataJobForWorkflow(String mdstoreId, String guidelines, String groupBy, int records, int workers, BlackboardJob bJob, BlackboardNotificationHandler<BlackboardServerHandler> blackboardHandler, int jobStatusUpdateInterval, boolean outputEpr, boolean blacklistedRecords, String blacklistGuidelines) throws ValidatorException{

		try {

			logger.info("Submitting data job for workflow");

//			String desiredCompatibility = "openaire3.0";

			Set<Integer> ruleIdsForBlacklist = new HashSet<Integer>();

			Set<Integer> ruleIdsContent = new HashSet<Integer>();

			for (RuleSet ruleset : rulesetsDao.getRuleSets()) {

				if (ruleset.getGuidelinesAcronym().equals(guidelines)) {

					ruleIdsContent = ruleset.getContentRulesIds();

				}	

				if (blacklistedRecords && ruleset.getGuidelinesAcronym().equals(blacklistGuidelines)) {

					ruleIdsForBlacklist = ruleset.getContentRulesIds();

					ruleIdsContent.addAll(ruleIdsForBlacklist);

				}

			}

			Properties props = new Properties();

			props.setProperty(DnetProvider.MDSTORE_ID, mdstoreId);

			props.setProperty(DnetProvider.BATCH_SIZE, "50");

			if (bJob.getParameters().get("batchSize") != null )

				props.setProperty(DnetProvider.BATCH_SIZE, bJob.getParameters().get("batchSize"));

			props.setProperty(DnetProvider.RECORDS, Integer.toString(records));

			StoredJob newJob = new StoredJob();

			newJob.setUserEmail("Workflow Service");

			newJob.setValidationType("C");

			newJob.setDesiredCompatibilityLevel(guidelines);

			newJob.setContentJobStatus("ongoing");

			newJob.setUsageJobStatus("none");

			newJob.setJobType("Workflow Request");

			newJob.setDuration("--");

			newJob.setBaseUrl(mdstoreId);

			newJob.setRules(ruleIdsContent);

			newJob.setRecords(records);

			newJob.setValidationSet("dnet-workflow");

			newJob.setGroupByXpath(groupBy);

			newJob.setMetadataPrefix("oai_dc");

			newJob.setId(-1);

			int jobIdStored = jobsDao.save(newJob);

			Set<Rule> rulesContent = new HashSet<Rule>();

			logger.debug("Selected content rules number: " + ruleIdsContent.size());

			for (Integer ruleId : ruleIdsContent){

				eu.dnetlib.domain.functionality.validator.Rule tempRule=rulesDao.get(ruleId);

				//special behaviour type of rule is chain

				if(tempRule.getType().equals("ChainRule")) {

					ChainRule<Rule> chainRule = this.handleChain(tempRule);

					rulesContent.add(chainRule);

				}

				else {

					rulesContent.add((Rule) this.getRuleClassInstanceByType(tempRule.getType(), tempRule.getConfiguration(), tempRule.getId()));

				}

			}

			Job jobContent = new Job(jobIdStored, 3, rulesContent, props);

			ValidatorJobListener listenerContent = listenersManager.createListener();

			listenerContent.setJobSubmittedId(jobIdStored);

			listenerContent.setJobSubmittedUser("Workflow Service");

			listenerContent.setGroupBy_xpath(groupBy);

			listenerContent.setValidationType("content");

			listenerContent.setInternalJobsSum(workers);

			listenerContent.setBlacklistRuleIds(ruleIdsForBlacklist);

			DnetListener dnetListener = listenersManager.createDnetListener();

			dnetListener.setJob(bJob);

			dnetListener.setBlackboardHandler(blackboardHandler);

			dnetListener.setInternalJobsSum(workers);

			dnetListener.setValidationJobId(jobIdStored);

			dnetListener.setJobStatusUpdateInterval(jobStatusUpdateInterval);

			if (blacklistedRecords) {

				dnetListener.setEnableOutputToDisk(true);

			}

			if (outputEpr)

				dnetListener.setEnableOutputToRS(true);

			validator.submitJob(jobContent, workers, listenerContent, dnetListener);

			return newJob;

		} catch (Exception e) {

			logger.error("Error Submitting content job", e);

			throw new ValidatorException(e);

		}

	}

	@Transactional(readOnly = false, propagation = Propagation.REQUIRED)

	public void submitJob(JobForValidation job) throws ValidatorException {

		try {

			logger.info("Submiting validation job requested by user: " + job.getUserEmail());

			if (job.isRegistration()) {

				logger.debug("initiating preregistration validations on repo " + job.getBaseUrl() + " for user " + job.getUserEmail() + "and desired compatibility: " + job.getDesiredCompatibilityLevel());

				for (RuleSet ruleset : rulesetsDao.getRuleSets()) {

					if (ruleset.getGuidelinesAcronym().equals(job.getDesiredCompatibilityLevel())) {

						job.setSelectedContentRules(ruleset.getContentRulesIds());

						job.setSelectedUsageRules(ruleset.getUsageRulesIds());

					}	

				}	

			}

			if ( job.getDesiredCompatibilityLevel().equalsIgnoreCase("openaire2.0") || 

					job.getDesiredCompatibilityLevel().equalsIgnoreCase("openaire3.0") || 

					job.getDesiredCompatibilityLevel().equalsIgnoreCase("driver") ) {

				logger.debug("Chosen set: OpenAIRE For Literature Repositories");

				logger.debug("Setting METADATA_PREFIX to: oai_dc");

				job.setMetadataPrefix("oai_dc");

			} else if ( job.getDesiredCompatibilityLevel().equalsIgnoreCase("openaire2.0_data")) {

				logger.debug("Chosen set: OpenAIRE For Data Archives");

				logger.debug("Setting METADATA_PREFIX to: oai_datacite");

				job.setMetadataPrefix("oai_datacite");			

			} else if ( job.getDesiredCompatibilityLevel().equalsIgnoreCase("openaire3.0_cris")) {

				logger.debug("Chosen set: OpenAIRE For Cris");

				logger.debug("Setting METADATA_PREFIX to: oai_CERIF_openaire");

				job.setMetadataPrefix("oai_CERIF_openaire");			

			}

			StoredJob newJob = new StoredJob(job);

			int usageJobs = 0;

			int contentJobs = 0;

			Set<Integer> totalRules = new HashSet<Integer>();

			Map<String,Set<Integer>> verbRuleMap = null;

			String validationType = "";

			if (job.getSelectedContentRules() != null) {

				contentJobs = 1;

				validationType +="C";

				totalRules.addAll(job.getSelectedContentRules());

				newJob.setContentJobStatus("ongoing");

			} else {

				newJob.setContentJobStatus("none");

			}

			if (job.getSelectedUsageRules() != null) {

				logger.debug("Creating map for provider information");

				verbRuleMap= new HashMap<String,Set<Integer>>();

				Set<Integer> old,temp = null;

				for (Integer id : job.getSelectedUsageRules()){

					eu.dnetlib.domain.functionality.validator.Rule ruleStored = rulesDao.get(id);

					logger.debug("Checking for verb : "+ruleStored.getProvider_information());

			        if((old=verbRuleMap.get(ruleStored.getProvider_information())) == null){

			        	logger.debug("Verb doesn't exist");

			            temp = new HashSet<Integer>();

			            temp.add(ruleStored.getId());

			            verbRuleMap.put(ruleStored.getProvider_information(),temp);

			        }else{

			        	logger.debug("Verb exists");

			            old.add(ruleStored.getId());

			        }

				}

				usageJobs = verbRuleMap.size();

				validationType +="U";

				totalRules.addAll(job.getSelectedUsageRules());

				newJob.setUsageJobStatus("ongoing");

			} else {

				newJob.setUsageJobStatus("none");

			}

			Map<String, Set<Rule>> entityChosenRulesMap = null;

			Map<String, Set<Rule>> entityChosenRulesMapReferential = null;

			if (job.isCris()) {

				entityChosenRulesMap = new HashMap<String, Set<Rule>>();

				entityChosenRulesMapReferential = new HashMap<String, Set<Rule>>();

				this.prepareCrisJobs(job, entityChosenRulesMap, entityChosenRulesMapReferential);

				newJob.setGroupByXpath("//header/setSpec");

				contentJobs = entityChosenRulesMap.keySet().size() + entityChosenRulesMapReferential.keySet().size();

				//TODO move in uoa-domain

				newJob.setCris(true);

				newJob.setSelectedCrisEntities(job.getSelectedCrisEntities());

				newJob.setCrisReferentialChecks(job.isCrisReferentialChecks());

			}

			newJob.setValidationType(validationType);

			if (job.isRegistration()) {

				newJob.setJobType("Registration Request");

				//TODO move in uoa-domain

				newJob.setActivationId(job.getActivationId());

				newJob.setRegistration(true);

				newJob.setUpdateExisting(job.isUpdateExisting());

				newJob.setOfficialName(job.getOfficialName());

				newJob.setAdminEmails(job.getAdminEmails());

				newJob.setDatasourceId(job.getDatasourceId());

				newJob.setInterfaceId(job.getInterfaceId());

				newJob.setInterfaceIdOld(job.getInterfaceIdOld());

				newJob.setRepoType(job.getRepoType());

			} else

				newJob.setJobType("Compatibility Test");

			newJob.setDuration("--");

			newJob.setRules(totalRules);

			newJob.setId(-1);

			int jobIdStored = jobsDao.save(newJob);

			RegistrationListener regListener = null;

			if (job.isRegistration()) {

				regListener = listenersManager.createRegListener();

				regListener.setActivationId(job.getActivationId());

				regListener.setBaseUrl(job.getBaseUrl());

				regListener.setActivationId(job.getActivationId());

				regListener.setAdminEmails(job.getAdminEmails());

				regListener.setOfficialName(job.getOfficialName());

				regListener.setUserMail(job.getUserEmail());

				regListener.setDatasourceId(job.getDatasourceId());

				regListener.setInterfaceId(job.getInterfaceId());

				regListener.setValidationSet(job.getValidationSet());

				regListener.setDesiredCompLevel(job.getDesiredCompatibilityLevel());

				regListener.setRepoType(job.getRepoType());

				regListener.setInterfaceIdOld(job.getInterfaceIdOld());

				regListener.setUpdateExisting(job.isUpdateExisting());	

				regListener.setTotalJobs(usageJobs + contentJobs);

			}

			CompatibilityTestListener compTestListener = listenersManager.createCompTestListener();

			compTestListener.setValidationSet(job.getValidationSet());

			compTestListener.setGuidelines(job.getDesiredCompatibilityLevel());

			compTestListener.setTotalJobs(usageJobs + contentJobs);

//CONTENT

			if (job.getSelectedContentRules() != null) {

				Set<Rule> rulesContent = new HashSet<Rule>();

				Properties props = new Properties();

				props.setProperty(OAIPMHRecordProvider.BASEURL, job.getBaseUrl());

				props.setProperty(OAIPMHRecordProvider.METADATA_PREFIX, newJob.getMetadataPrefix());

				props.setProperty(OAIPMHRecordProvider.RECORDS,Integer.toString(job.getRecords()));

				props.setProperty(OAIPMHRecordProvider.SET,job.getValidationSet());

				Job jobContent = null;

				ValidatorJobListener listenerContent = listenersManager.createListener();

				listenerContent.setJobSubmittedId(jobIdStored);

				listenerContent.setJobSubmittedUser(job.getUserEmail());

				listenerContent.setGroupBy_xpath(newJob.getGroupByXpath());

				listenerContent.setValidationType("content");

				listenerContent.setInternalJobsSum(contentJobs);

				if (job.isCris()) {

					logger.debug("Submiting job for cris.");

					logger.debug("Total content jobs: " + contentJobs);

//					logger.debug("Selected content rules number: " + job.getSelectedContentRules().size());

					jobContent = new Job(jobIdStored, 4, rulesContent, props);

					validator.submitJobForCris(jobContent, entityChosenRulesMap, entityChosenRulesMapReferential, listenerContent, compTestListener);

				} else {

					logger.debug("Selected content rules number: " + job.getSelectedContentRules().size());

					for (Integer ruleId : job.getSelectedContentRules()){

						eu.dnetlib.domain.functionality.validator.Rule tempRule=rulesDao.get(ruleId);

						//special behaviour type of rule is chain

						if(tempRule.getType().equals("ChainRule")) {

							ChainRule<Rule> chainRule = this.handleChain(tempRule);

							rulesContent.add(chainRule);

						}

						else {

							rulesContent.add((Rule) this.getRuleClassInstanceByType(tempRule.getType(), tempRule.getConfiguration(), tempRule.getId()));

						}

					}

					jobContent = new Job(jobIdStored, 1, rulesContent, props);

					if (job.isRegistration()) {

						validator.submitJob(jobContent, 1, listenerContent, regListener);

					} else {

						validator.submitJob(jobContent, 1, listenerContent, compTestListener);

					}	

				}

			}

			//USAGE	

			if (job.getSelectedUsageRules() != null) {

				ValidatorJobListener listenerUsage = listenersManager.createListener();

				listenerUsage.setJobSubmittedId(jobIdStored);

				listenerUsage.setValidationType("usage");

				listenerUsage.setJobSubmittedUser(job.getUserEmail());

				listenerUsage.setInternalJobsSum(usageJobs);

				List <Job> jobsUsage = new ArrayList<Job>();

				for (Map.Entry<String, Set<Integer>> entry : verbRuleMap.entrySet()) {

					Properties pros = new Properties();

					pros.setProperty(OAIPMHSinglePageVerbProvider.VERB,entry.getKey());

					pros.setProperty(OAIPMHSinglePageVerbProvider.BASEURL, job.getBaseUrl());

					Set<Rule> rulesUsage = new HashSet<Rule>();

					for (Integer ruleId : entry.getValue()){

						eu.dnetlib.domain.functionality.validator.Rule tempRule = rulesDao.get(ruleId);

						logger.debug("prepare to add rule to registry with regexp: "+tempRule.getConfiguration().getProperty("regexp"));

						rulesUsage.add((Rule)this.getRuleClassInstanceByType(tempRule.getType(), tempRule.getConfiguration(), tempRule.getId()));

					}

					jobsUsage.add(new Job(jobIdStored, 2, rulesUsage, pros));

				}

				for (Job jobUsage : jobsUsage ) {

					if (job.isRegistration()) {

						validator.submitJob(jobUsage, 1, listenerUsage, regListener);

					} else {

						validator.submitJob(jobUsage, 1, listenerUsage, compTestListener);

					}

				}

			}			

		} catch (Exception e) {

			logger.error("error submitting job ", e);

			throw new ValidatorException(e);

		}

	}

	private void prepareCrisJobs(JobForValidation job,

			Map<String, Set<Rule>> entityChosenRulesMap,

			Map<String, Set<Rule>> entityChosenRulesMapReferential) throws ValidatorException, DaoException {

		//getting rules per entity and leave only chosen ones

		logger.debug("Selected Entities: " + job.getSelectedCrisEntities());

		for (String entity : RuleTypes.getEntities().keySet()) {

			logger.debug("checking entity: " + entity);

			Set<Rule> rulesBasic = new HashSet<Rule>();

			Set<Rule> rulesReferential = new HashSet<Rule>();

			if (job.getSelectedCrisEntities().contains(entity)) {

				logger.debug("entity: " + entity + " is selected");

				for (eu.dnetlib.domain.functionality.validator.Rule rule : rulesDao.getAllRulesByJobTypeEntityType("content", entity)) {

					if (job.getSelectedContentRules().contains(rule.getId())) {

						eu.dnetlib.domain.functionality.validator.Rule tempRule=rulesDao.get(rule.getId());

						if (rule.getName().contains("Relationship")) {

							if (job.isCrisReferentialChecks()) {

								rulesReferential.add((Rule) this.getRuleClassInstanceByType(tempRule.getType(), tempRule.getConfiguration(), tempRule.getId()));

							} 

						} else {

							rulesBasic.add((Rule) this.getRuleClassInstanceByType(tempRule.getType(), tempRule.getConfiguration(), tempRule.getId()));

						}

					}

				}

				logger.debug("Basic rules: " + rulesBasic.size());

				logger.debug("Referential rules: " + rulesReferential.size());

				entityChosenRulesMap.put(RuleTypes.getSetOfEntity(entity), rulesBasic);

				if (job.isCrisReferentialChecks() && !rulesReferential.isEmpty())

					entityChosenRulesMapReferential.put(RuleTypes.getSetOfEntity(entity), rulesReferential);

			} else {

				logger.debug("entity: " + entity + " is not selected");

			}

		}

		logger.debug("Return entities: " + entityChosenRulesMap.keySet());

	}

	public Rule getRuleClassInstanceByType(String type, CustomProperties pros, int id) throws ValidatorException {

		logger.debug("getting rule object of type "+type);

		String classname = RuleTypes.getClassOfType(type);

		if (classname == null){

			logger.debug("error getting rule object of type "+type+" classname=null");

			return null;

		}

		else {

			try {

				Class<?> clazz = Class.forName(classname);

				logger.debug("classname: "+clazz.getName());

				Properties properties = new Properties();

				properties.putAll(pros.getProperties());

				return (Rule) clazz.getConstructor(new Class[]{Properties.class,Integer.TYPE}).newInstance(properties,id);

			} catch (Exception e) {

				logger.debug("error getting rule object: "+e);

				return null;

			} 

		}

	}

	private ChainRule<Rule> handleChain(eu.dnetlib.domain.functionality.validator.Rule tempRule) throws ValidatorException, DaoException {

		logger.debug("chain rule found");

		List<Rule> rules_chain = new ArrayList<Rule>();

		eu.dnetlib.domain.functionality.validator.Rule tempRuleChain1=rulesDao.get(Integer.parseInt(tempRule.getConfiguration().getProperty("rule_1")));

		if(tempRuleChain1.getType().equals("ChainRule")) {

			ChainRule<Rule> chainRule1 = this.handleChain(tempRuleChain1);

			rules_chain.add(chainRule1);

		} else {

			rules_chain.add((Rule) this.getRuleClassInstanceByType(tempRuleChain1.getType(), tempRuleChain1.getConfiguration(),tempRuleChain1.getId()));

		}

		eu.dnetlib.domain.functionality.validator.Rule tempRuleChain2=rulesDao.get(Integer.parseInt(tempRule.getConfiguration().getProperty("rule_2")));

		if(tempRuleChain2.getType().equals("ChainRule")) {

			ChainRule<Rule> chainRule2 = this.handleChain(tempRuleChain2);

			rules_chain.add(chainRule2);

		} else {

			rules_chain.add((Rule) this.getRuleClassInstanceByType(tempRuleChain2.getType(), tempRuleChain2.getConfiguration(), tempRuleChain2.getId()));

		}

		Properties chainPros = new Properties();

		chainPros.setProperty(ChainRule.TYPE,tempRule.getConfiguration().getProperty(ChainRule.TYPE));

		ChainRule<Rule> chainRule = new ChainRule<Rule>(chainPros, tempRule.getId(), rules_chain);

		return chainRule;

	}

	public Validator getValidator() {

		return validator;

	}

	public ListenersManager getListenersManager() {

		return listenersManager;

	}

	public void setListenersManager(ListenersManager listenersManager) {

		this.listenersManager = listenersManager;

	}

	public void setValidator(Validator validator) {

		this.validator = validator;

	}

	public JobsDAO getJobsDao() {

		return jobsDao;

	}

	public void setJobsDao(JobsDAO jobsDao) {

		this.jobsDao = jobsDao;

	}

	public RulesetsDAO getRulesetsDao() {

		return rulesetsDao;

	}

	public void setRulesetsDao(RulesetsDAO rulesetsDao) {

		this.rulesetsDao = rulesetsDao;

	}

	public RulesDAO getRulesDao() {

		return rulesDao;

	}

	public void setRulesDao(RulesDAO rulesDao) {

		this.rulesDao = rulesDao;

	}

	public ValidatorRestore getValRestore() {

		return valRestore;

	}

	public void setValRestore(ValidatorRestore valRestore) {

		this.valRestore = valRestore;

	}

}

package eu.dnetlib.validator.service.impl;

import eu.dnetlib.api.functionality.ValidatorService;

import eu.dnetlib.api.functionality.ValidatorServiceException;

import eu.dnetlib.domain.ActionType;

import eu.dnetlib.domain.ResourceType;

import eu.dnetlib.domain.enabling.Notification;

import eu.dnetlib.domain.functionality.validator.JobForValidation;

import eu.dnetlib.domain.functionality.validator.RuleSet;

import eu.dnetlib.domain.functionality.validator.StoredJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardNotificationHandler;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;

import eu.dnetlib.validator.engine.ValidatorException;

import gr.uoa.di.driver.app.DriverServiceImpl;

import gr.uoa.di.driver.enabling.issn.NotificationListener;

import java.util.List;

import org.apache.log4j.Logger;

public class ValidatorServiceImpl extends DriverServiceImpl implements

		ValidatorService {

	private BlackboardNotificationHandler<BlackboardServerHandler> blackboardNotificationHandler = null;

	private Logger logger = Logger.getLogger(ValidatorServiceImpl.class);

	private ValidatorManager valManager;

	// do NOT call directly. It will be called by the

	// InitializingServiceRegistrationManager

	// after it sets the service EPR and id.

	@Override

	public void init() {

		super.init();

		if (this.blackboardNotificationHandler != null) { 

			this.subscribe(ActionType.UPDATE,

				ResourceType.VALIDATORSERVICERESOURCETYPE, this.getServiceEPR()

						.getParameter("serviceId"),

				"RESOURCE_PROFILE/BODY/BLACKBOARD/LAST_REQUEST",

				new NotificationListener() {

					@Override

					public void processNotification(Notification notification) {

						blackboardNotificationHandler.notified(

								notification.getSubscriptionId(),

								notification.getTopic(),

								notification.getIsId(),

								notification.getMessage());

						logger.debug(notification.getSubscriptionId());

						logger.debug(notification.getTopic());

						logger.debug(notification.getIsId());

						logger.debug(notification.getMessage());

					}

				});

		}

	}

	@Override

	public StoredJob getStoredJob(int jobId, String groupBy)

			throws ValidatorServiceException {

		try {

			logger.info("received request for job " + jobId);

			return valManager.getStoredJob(jobId, groupBy);

		} catch (ValidatorException e) {

			throw new ValidatorServiceException(e);

		}

	}

	@Override

	public List<StoredJob> getStoredJobs(String userMail, String jobType,

										 Integer offset, Integer limit, String dateFrom, String dateTo)

			throws ValidatorServiceException {

		try {

			logger.debug("received request for jobs of user " + userMail);

			return valManager.getStoredJobs(userMail, jobType, offset, limit, dateFrom, dateTo);

		} catch (ValidatorException e) {

			throw new ValidatorServiceException(e);

		}

	}

	@Override

	public List<StoredJob> getStoredJobsNew(String userMail, String jobType,

										 Integer offset, Integer limit, String dateFrom, String dateTo, String jobStatus)

			throws ValidatorServiceException {

		try {

			logger.debug("received request for jobs of user " + userMail);

			return valManager.getStoredJobs(userMail, jobType, offset, limit, dateFrom, dateTo, jobStatus);

		} catch (ValidatorException e) {

			throw new ValidatorServiceException(e);

		}

	}

	@Override

	public int getStoredJobsTotalNumber(String userMail, String jobType)

			throws ValidatorServiceException {

		try {

			logger.debug("received request for total number of jobs of user " + userMail);

			return valManager.getStoredJobsTotalNumber(userMail, jobType);

		} catch (ValidatorException e) {

			throw new ValidatorServiceException(e);

		}

	}

	@Override

	public int getStoredJobsTotalNumberNew(String userMail, String jobType, String jobStatus)

			throws ValidatorServiceException {

		try {

			logger.debug("received request for total number of jobs of user " + userMail);

			return valManager.getStoredJobsTotalNumber(userMail, jobType, jobStatus);

		} catch (ValidatorException e) {

			throw new ValidatorServiceException(e);

		}

	}

	@Override

	public List<RuleSet> getRuleSets() throws ValidatorServiceException {

		try {

			logger.info("received request for rulesets ");

			return valManager.getRuleSets();

		} catch (ValidatorException e) {

			throw new ValidatorServiceException(e);

		}

	}

	@Override

	public void submitValidationJob(JobForValidation job)

			throws ValidatorServiceException {

		try {

			logger.info("received request to submit job");

			this.valManager.submitJob(job);

		} catch (ValidatorException e) {

			throw new ValidatorServiceException(e);

		}

	}

	public BlackboardNotificationHandler<BlackboardServerHandler> getBlackboardNotificationHandler() {

		return blackboardNotificationHandler;

	}

	public ValidatorManager getValManager() {

		return valManager;

	}

	public void setValManager(ValidatorManager valManager) {

		this.valManager = valManager;

	}

	public void setBlackboardNotificationHandler(

			BlackboardNotificationHandler<BlackboardServerHandler> blackboardNotificationHandler) {

		this.blackboardNotificationHandler = blackboardNotificationHandler;

	}

}

package eu.dnetlib.validator.service.impl;

import eu.dnetlib.enabling.tools.blackboard.BlackboardJob;

import eu.dnetlib.enabling.tools.blackboard.BlackboardNotificationHandler;

import eu.dnetlib.enabling.tools.blackboard.BlackboardServerHandler;

import org.apache.log4j.Logger;

public class NotificationListener extends

        BlackboardNotificationHandler<BlackboardServerHandler> {

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

    private ValidatorManager valManager;

    @Override

    protected void processJob(BlackboardJob job) {

        super.processJob(job);

        logger.debug("New Job!");

//		String validationSet = job.getParameters().get("validationSet");

//		String validationType = job.getParameters().get("validationType");

        String type = job.getParameters().get("type"); //OAI

        String mdstoreId = job.getParameters().get("mdstoreId");

        String guidelines = job.getParameters().get("guidelines");

        String groupBy = job.getParameters().get("groupBy");

        String records = job.getParameters().get("records");

        String outputEpr = job.getParameters().get("outputEpr");

        if (outputEpr == null) {

            outputEpr = "false";

        }

        String blacklistGuidelines = job.getParameters().get("blacklistGuidelines");

        if (blacklistGuidelines == null)

            blacklistGuidelines = guidelines;

        String blacklistedRecords = job.getParameters().get("blacklistedRecords");

        if (blacklistedRecords == null) {

            blacklistedRecords = "false";

        }

        String submissionDate = job.getParameters().get("submissionDate");

        String jobStatusUpdateInterval = job.getParameters().get("jobStatusUpdateInterval");

        //the datasource id

        String datasourceId = job.getParameters().get("datasourceId");

        //the datasource baseUrl, optional for type=OAI, mandatory for type=DNET

        String baseUrl = job.getParameters().get("baseUrl");

        //the namespace prefix of the datasource, optional

        String datasourceNamespacePrefix = job.getParameters().get("datasourceNamespacePrefix");

        //the datasource name, optional

        String datasourceName = job.getParameters().get("datasourceName");

        //json map of other parameters that may be of interest, optional

        String extra_param = job.getParameters().get("extra_param");

        try {

            int workers = 1;

            int recordsInt = 0;

            int jobStatusUpdateIntervalInt = 100;

            if (records == null || records.equalsIgnoreCase("all"))

                recordsInt = -1;

            else

                recordsInt = Integer.parseInt(records);

            if (recordsInt > -1 && recordsInt < 100)

                workers = 1;

            else

                workers = 4;

            if (job.getParameters().get("workers") != null)

                workers = Integer.parseInt(job.getParameters().get("workers"));

            if (jobStatusUpdateInterval != null)

                jobStatusUpdateIntervalInt = Integer.parseInt(jobStatusUpdateInterval);

            if (recordsInt != 0) {

                logger.debug("New Job of type: " + type + " - workers: " + workers + " - recordsInt: " + recordsInt);

                logger.debug("workers: " + workers);

                logger.debug("recordsInt: " + recordsInt);

                logger.debug("jobStatusUpdateInterval: " + jobStatusUpdateIntervalInt);

                if (job.getAction().equalsIgnoreCase("VALIDATE")) {

                    if (type.equalsIgnoreCase("OAI")) {

                        logger.debug("type is oai");

                    } else if (type.equalsIgnoreCase("DNET")) {

                        logger.debug("type is dnet");

                        logger.debug("groupBy: " + groupBy);

//						this.getBlackboardHandler().ongoing(job);

                        valManager.beginDataJobForWorkflow(mdstoreId, guidelines, groupBy, recordsInt, workers, job, this, jobStatusUpdateIntervalInt, Boolean.parseBoolean(outputEpr), Boolean.parseBoolean(blacklistedRecords), blacklistGuidelines);

                    }

                } else {

                    throw new Exception("Unknown action");

                }

            } else {

                throw new Exception("There are no records in the MDStore.");

            }

        } catch (Exception e) {

            logger.error("Failed to begin the workflow for the job received", e);

            getBlackboardHandler().failed(job, e);

        }

    }

    public ValidatorManager getValManager() {

        return valManager;

    }

    public void setValManager(ValidatorManager valManager) {

        this.valManager = valManager;

    }

}

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

public interface ThreadSafeRegistry {

	public void lock();

	public void unlock();

	public void removeObject(int id);

}

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

import java.io.Serializable;

import java.util.HashMap;

import java.util.Map;

import eu.dnetlib.validator.engine.persistance.Registry;

/**

 * A Registry that stores the objects in an in-memory {@link Map}

 * @author Manos Karvounis

 * @param <T>

 */

public class MemoryRegistry<T extends Serializable> extends Registry<T>  {

	protected Map<Integer, T> registry = null;

	public MemoryRegistry(String name) {

		super(name);

		this.registry = new HashMap<Integer, T>();

	}

	@Override

	public T getObject(int id) {

		return registry.get(id);

	}

	@Override

	public void addObject(int id, T obj) {

		registry.put(id, obj);

	}

	public Map<Integer, T> getAllObjects() {

		return this.registry;

	}

}

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

import java.io.Serializable;

import java.util.concurrent.locks.Lock;

import java.util.concurrent.locks.ReentrantLock;

/**

 * A {@link MemoryRegistry} that may be accesses by multiple threads silmutaneously.

 * @author Manos Karvounis

 * @param <T>

 */

public class ThreadSafeMemoryRegistry<T extends Serializable> extends MemoryRegistry<T> implements ThreadSafeRegistry {

	private final Lock lock = new ReentrantLock();

	public ThreadSafeMemoryRegistry(String name) {

		super(name);

	}

	@Override

	public void lock() {

		lock.lock();

	}

	@Override

	public void unlock() {

		lock.unlock();

	}

	@Override

	public void removeObject(int id) {

		registry.remove(id);

	}

}

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

import java.io.BufferedInputStream;

import java.io.File;

import java.io.FileInputStream;

import java.io.IOException;

import java.io.InputStream;

import java.io.PrintWriter;

import javax.servlet.ServletException;

import javax.servlet.http.HttpServlet;

import javax.servlet.http.HttpServletRequest;

import javax.servlet.http.HttpServletResponse;

import org.apache.commons.io.IOUtils;

import org.apache.log4j.Logger;

import org.springframework.context.ApplicationContext;

import org.springframework.web.context.support.WebApplicationContextUtils;

public class BlacklistServlet extends HttpServlet {

	/**

	 * 

	 */

	private static final long serialVersionUID = 1788389574262137242L;

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

	private static final String REQUEST_GETBLACKLIST = "GetBlacklistedRecords";

//	private TasksDAO tasksDao;

	private String path;

	@Override

	public void init() throws ServletException {

		ApplicationContext context = WebApplicationContextUtils

                .getWebApplicationContext(getServletContext());

//		this.tasksDao = (TasksDAO) context.getBean("tasksDao");

//		this.path = "/tmp/validator-wf/";

		this.path = "/var/lib/dnet/validator/workflow_blacklists/";

//		this.path = context.getEnvironment().getProperty("blacklistedRecordsPath");

	}

	@Override

	public void doGet(HttpServletRequest req, HttpServletResponse resp)

			throws ServletException, IOException {

		this.handleRequest(req, resp);

	}

	@Override

	public void doPost(HttpServletRequest req, HttpServletResponse resp)

			throws ServletException, IOException {

		this.handleRequest(req, resp);

	}

	public void handleRequest(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {

		InputStream is = null;

		String dataSourceId = req.getParameter("datasourceId");

		String requestType = req.getParameter("request");

		try {

			if (requestType.equals(REQUEST_GETBLACKLIST)) {

				logger.debug("parameter: " + dataSourceId);

				File file = new File(path + dataSourceId);

				if (file.isFile()) {

					logger.debug("file exists");

					is = new BufferedInputStream(new FileInputStream(file));

					resp.setStatus(200);

					resp.setContentType("text/plain");

					resp.setContentLength((int) file.length());

//					InputStream fileStream = new FileInputStream(file);

//					InputStream gzipStream = new GZIPInputStream(fileStream);

//					Reader decoder = new InputStreamReader(gzipStream, "UTF-8");

//					BufferedReader buffered = new BufferedReader(decoder);

//					IOUtils.copyLarge(buffered, resp.getWriter());

					IOUtils.copyLarge(is, resp.getOutputStream());

//					tasksDao.get(4335);

				} else {

					PrintWriter out = resp.getWriter();

					resp.setStatus(400);

					resp.setContentType("text/plain");

					out.write("No file for this id: " + dataSourceId);

					out.close();

				}

			} else {

				PrintWriter out = resp.getWriter();

				resp.setStatus(500);

				resp.setContentType("text/plain");

				out.write("Request: " + requestType + " is not supported");

				out.close();

			}

		} catch (Exception e) {

			resp.setStatus(500);

			e.printStackTrace(resp.getWriter());

		} finally {

			if (is != null)

				is.close();

		}

	}

}

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

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

/**

 * Indicates that a Validatiob Object is actually plain text.

 * @author Manos Karvounis

 */

public interface TextValidationObject extends ValidationObject {

	public String getContentAsText();

}

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

import java.io.StringWriter;

import javax.xml.transform.Result;

import javax.xml.transform.Source;

import javax.xml.transform.Transformer;

import javax.xml.transform.TransformerConfigurationException;

import javax.xml.transform.TransformerException;

import javax.xml.transform.TransformerFactory;

import javax.xml.transform.TransformerFactoryConfigurationError;

import javax.xml.transform.dom.DOMSource;

import javax.xml.transform.stream.StreamResult;

import javax.xml.xpath.XPath;

import javax.xml.xpath.XPathConstants;

import javax.xml.xpath.XPathExpression;

import javax.xml.xpath.XPathExpressionException;

import javax.xml.xpath.XPathFactory;

import org.apache.log4j.Logger;

import org.w3c.dom.Document;

import org.w3c.dom.NodeList;

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

/**

 * Indicates that a Validation Object is actually an xml text.

 * @author Manos Karvounis

 */

public class XMLTextValidationObject implements TextValidationObject {

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

	private String id = "";

	private String status;

	private final Document doc;

	private XPathFactory factory = XPathFactory.newInstance();

	public XMLTextValidationObject(Document doc) {

		super();

		this.doc = doc;

	}

	@Override

	public String getId() {

		return this.id;

	}

	@Override

	public void setId(String id) {

		this.id = id;

	}

	public String getStatus() {

		return status;

	}

	public void setStatus(String status) {

		this.status = status;

	}

	@Override

	public Object getContentAsObject() {

		return this.doc;

	}

	@Override

	public String getContentAsText() {

		StringWriter sw = new StringWriter();

	    Transformer transformer = null;

		try {

			transformer = TransformerFactory.newInstance().newTransformer();

		    Source source = new DOMSource(doc);

		    Result output = new StreamResult(sw);

			transformer.transform(source, output);

		} catch (TransformerConfigurationException e) {

			log.error("", e);

		} catch (TransformerFactoryConfigurationError e) {

			log.error("", e);

		} catch (TransformerException e) {

			log.error("", e);

		}

	    return sw.toString();

	}

	public Document getDocument() {

		return this.doc;

	}

	/**

	 * Get the nodes retrieved by issuing the given XPATH query.

	 * @param xpath

	 * @return

	 * @throws DataException

	 */

	public NodeList getNodes(String xpath) throws DataException {

		log.debug("xpath: "+xpath);

		try {

			XPath xxpath = factory.newXPath();

			XPathExpression expr = xxpath.compile(xpath);

			return (NodeList) expr.evaluate(doc, XPathConstants.NODESET);

		} catch (XPathExpressionException e) {

			throw new DataException();

		}

	}

}

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

import java.util.ArrayList;

import java.util.Calendar;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Set;

import org.apache.log4j.Logger;

import org.springframework.core.task.SyncTaskExecutor;

import eu.dnetlib.validator.engine.ValidatorException;

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

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

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

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

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.engine.execution.Task;

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

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

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

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

/**

 * Retrieves the Validation Objects of a {@link Job} using the contained

 * Provider. A new Task in created for each Validation Object and all the rules

 * contained in the Job. Each task is submitted for execution at an

 * {@link Executor}

 * 

 * @author Nikon Gasparis

 *

 */

public class JobWorker implements Runnable {

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

	private final SyncTaskExecutor taskExecutor;

	public final long generalTimeout;

	public final String set;

	public final int jobId;

	public final Set<Rule> rules;

	public final Provider provider;

	public final JobListener validatorListener;

	public final JobListener[] listeners;

	protected Map<String, Object> jobContext;

	private Map<String, Object> recordContext;

	private int deletedRecords = 0;

	public JobWorker(int jobId, Set<Rule> rules, String set, Provider provider,

			JobListener validatorListener, SyncTaskExecutor taskExecutor,

			long generalTimeout, JobListener... listeners) {

		super();

		log.debug("Creating a new job Worker with generalTimeout " + generalTimeout);

		this.taskExecutor = taskExecutor;

		this.generalTimeout = generalTimeout;

		this.set = set;

		this.jobId = jobId;

		this.rules = rules;

		this.provider = provider;

		this.jobContext = new HashMap<String, Object>();

		this.listeners = listeners;

		this.validatorListener = validatorListener;

	}

	public void submit() throws JobWorkerException {

		long elapsed = 0;

		int count = 0;

		int record_limit = -1;

		log.debug("Submitting job");

		if (provider.getConfiguration().getProperty(

				OAIPMHRecordProvider.RECORDS) != null) {

			if (!provider.getConfiguration()

					.getProperty(OAIPMHRecordProvider.RECORDS).trim()

					.equals("-1")) {

				record_limit = Integer.parseInt(provider.getConfiguration()

						.getProperty(OAIPMHRecordProvider.RECORDS).trim());

			}

		}

		if (provider.getConfiguration().getProperty(DnetProvider.RECORDS) != null) {

			if (!provider.getConfiguration().getProperty(DnetProvider.RECORDS)

					.trim().equals("-1")) {

				record_limit = Integer.parseInt(provider.getConfiguration()

						.getProperty(DnetProvider.RECORDS).trim());

			}

		}

//		int batchSize = -1;

//		if (provider.getConfiguration().getProperty(DnetProvider.BATCH_SIZE) != null) {

//			 batchSize = Integer.parseInt(provider.getConfiguration()

//						.getProperty(DnetProvider.BATCH_SIZE));

//		}

		log.debug("Number of records to validate: " + record_limit

				+ " (-1 for all)");

		Boolean crisFlag = false;

		if (set != null && set.contains("_cris_"))

			crisFlag = true;

		boolean success = false;

		List<Task> tasks = new ArrayList<Task>();

		ResultSet<ValidationObject> ivobjs = null;

		try {

			if (!crisFlag) {

				log.debug("request to get validation objects without set..");

				ivobjs = provider.getValidationObjects();

			} else {

				log.debug("request to get validation objects with set: " + set);

				ivobjs = provider.getValidationObjects(set);

			}

		} catch (ProviderException e1) {

			log.error("Error while getting validation objects from provider",

					e1);

			for (JobListener listener : listeners) {

				listener.failed(jobId, this.jobContext, e1);

			}

		}

		try {

			long recordsTime = 0;

			while ((count < record_limit || record_limit == -1)

					&& ivobjs.next()) {

				this.recordContext = new HashMap<String, Object>();

				tasks.clear();

				ValidationObject vobj = ivobjs.get();

				log.debug("Checking if rules will be applied on object "

						+ vobj.getId());

				if (vobj.getStatus() != null

						&& vobj.getStatus().equalsIgnoreCase("deleted")) {

					log.debug("Object is deleted and will be ignored");

					deletedRecords++;

					continue;

				}

				if (crisFlag

						&& (vobj.getId().contains("cfEAddr") || vobj.getId()

								.contains("cfEquip"))) {

					log.debug("ignoring cfEaddr and cfEquip records from persons and datasets");

					continue;

				}

				log.debug("Applying rules on object " + vobj.getId());

				count++;

				for (Rule rule : rules) {

					if (rule != null) {

						rule.setValObjId(vobj.getId());

						rule.setProvider(provider);

						Task task = new Task(vobj, rule);

						tasks.add(task);

					}

				}

				TaskList ltasks = new TaskList(tasks);

				try {

					long time1 = Calendar.getInstance().getTimeInMillis();

//					Object lock = new Object();

					taskExecutor.execute(ltasks);

					long time2 = Calendar.getInstance().getTimeInMillis();

					log.debug("Task execution took " + ((time2 - time1))

							+ " milli seconds");

					elapsed += time2 - time1;

					log.debug("Elapsed time till now is " + elapsed/1000

							+ " seconds");

					recordsTime += time2 - time1;

//				    if (count % batchSize == 0) {	

//						log.debug("Elapsed time for " + batchSize + " records is  " + recordsTime

//								+ " milli seconds");

//						recordsTime = 0;

//					}

//					if (elapsed > this.generalTimeout)

//						throw new JobWorkerException("Job timed out");

					long time3 = Calendar.getInstance().getTimeInMillis();

					for (JobListener listener : listeners) {

						listener.currentResults(ltasks.getCtasks(), jobId,

								vobj.getContentAsObject(), this.recordContext);

					}

					long time4 = Calendar.getInstance().getTimeInMillis();

					log.debug("Informing listeners took " + ((time4 - time3))

							+ " milli seconds");

					success = true;

				} catch (Exception e) {

					log.error("an error occured while executing tasks: ", e);

					for (JobListener listener : listeners) {

						listener.currentResults(ltasks.getCtasks(), jobId,

								vobj.getContentAsObject(), this.recordContext,

								e);

					}

				}

			}

		} catch (Exception e) {

			log.error("data error", e);

			for (JobListener listener : listeners) {

				listener.failed(jobId, this.jobContext, e);

			}

		}

		if (success) {

			jobContext.put("deletedRecords", deletedRecords);

			for (JobListener listener : listeners) {

				listener.finished(jobId, this.jobContext);

			}

		} else {

			log.error("an error occured");

			for (JobListener listener : listeners) {

				if (ivobjs.getError() == null)

					listener.failed(jobId, this.jobContext,

							new JobWorkerException("All tasks failed"));

				else {

					listener.failed(jobId, this.jobContext,