D-Net

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

import java.util.ArrayList;

import java.util.List;

import java.util.Map;

import org.apache.log4j.Logger;

import eu.dnetlib.api.enabling.ResultSetService;

import eu.dnetlib.api.enabling.ResultSetServiceException;

import eu.dnetlib.domain.EPR;

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

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

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

import eu.dnetlib.utils.EPRUtils;

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

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

import gr.uoa.di.driver.util.ServiceLocator;

public class DnetListener implements JobListener {

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

	private RecordXMLBuilder xmlBuilder;

	private BlackboardJob job;

	private BlackboardNotificationHandler<BlackboardServerHandler> blackboardHandler;

	private ServiceLocator<ResultSetService> resultSetServiceLocator = null;

	private ResultSetService resultSetService;

	private String resultSetPassID;

	private String resultSetFailID;

	private List<String> passBuffer;

	private List<String> failBuffer;

	private EPR eprPass; 

	private EPR eprFail; 

	public void init() { 

		try {

		// Get reference to service

		resultSetService = resultSetServiceLocator.getService();

		// Create a new result set

		eprPass = resultSetService.createPushRS(86400, 0);

		eprFail = resultSetService.createPushRS(86400, 0);

		// get result set ids

		resultSetPassID = eprPass.getParameter("ResourceIdentifier");

		resultSetFailID = eprFail.getParameter("ResourceIdentifier");

		// initializing buffers

		passBuffer = new ArrayList<String>();

		failBuffer = new ArrayList<String>();

		} catch (Exception e) {

			logger.error("Error initializing ResultSetService.", e);

		}

	}

    public void sendToPassQueue(String xmlString) {

    	logger.debug("received passed XMLresult");

    	// add records

    	passBuffer.add(xmlString);

    	if (passBuffer.size() > 30) {

    		try {

				resultSetService.populateRS(resultSetPassID, passBuffer);

				passBuffer.clear();

			} catch (ResultSetServiceException e) {

				logger.error("Error populating ResultSetService.", e);

			}    		

    	}

    	logger.debug("XML: " + xmlString);

    }

    public void sendToFailQueue(String xmlString) {

    	logger.debug("received failed XMLresult");

    	// add records

    	failBuffer.add(xmlString);

    	if (failBuffer.size() > 30) {

    		try {

				resultSetService.populateRS(resultSetFailID, failBuffer);

				failBuffer.clear();

			} catch (ResultSetServiceException e) {

				logger.error("Error populating ResultSetService.", e);

			}    		

    	}

    	logger.debug("XML: " + xmlString);

    }

	@Override

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

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

		@SuppressWarnings("unchecked")

		String xmlString = xmlBuilder.buildXml((List<Map<String, String>>) recordContext.get("veloList"),record, (Map<String,String>) recordContext.get("recordValidationResult"));

		if ( (Integer) recordContext.get("score") > 0)

			this.sendToPassQueue(xmlString);

		else

			this.sendToFailQueue(xmlString);

		logger.debug("XML: " + xmlString);

		blackboardHandler.getBlackboardHandler().ongoing(job);

	}

	@Override

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

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

		@SuppressWarnings("unchecked")

		String xmlString = xmlBuilder.buildXml((List<Map<String, String>>) recordContext.get("veloList"),record, (Map<String,String>) recordContext.get("recordValidationResult"));

		if ( (Integer) recordContext.get("score") > 0)

			this.sendToPassQueue(xmlString);

		else

			this.sendToFailQueue(xmlString);

		logger.debug("XML: " + xmlString);

		blackboardHandler.getBlackboardHandler().ongoing(job);

	}

	@Override

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

		//send remaining xmls from buffers

    	if (passBuffer.size() > 0) {

    		try {

				resultSetService.populateRS(resultSetPassID, passBuffer);

				passBuffer.clear();

			} catch (ResultSetServiceException e) {

				logger.error("Error populating ResultSetService.", e);

			}    		

    	}

		if (failBuffer.size() > 0) {

    		try {

				resultSetService.populateRS(resultSetFailID, failBuffer);

				failBuffer.clear();

			} catch (ResultSetServiceException e) {

				logger.error("Error populating ResultSetService.", e);

			}    		

    	}

		// finally, close rs.

		resultSetService.closeRS(resultSetPassID);

		resultSetService.closeRS(resultSetFailID);

		// update job status

		job.getParameters().put("passEpr", EPRUtils.eprToXml(eprPass));

		job.getParameters().put("failEpr", EPRUtils.eprToXml(eprFail));

		blackboardHandler.getBlackboardHandler().done(job);

	}

	@Override

	public synchronized void failed(int jobId, Map<String, Object> jobContext, Throwable t) {

		resultSetService.closeRS(resultSetPassID);

		resultSetService.closeRS(resultSetFailID);

		// update job status

		blackboardHandler.getBlackboardHandler().failed(job, t);

	}

	public RecordXMLBuilder getXmlBuilder() {

		return xmlBuilder;

	}

	public void setXmlBuilder(RecordXMLBuilder xmlBuilder) {

		this.xmlBuilder = xmlBuilder;

	}

	public BlackboardJob getJob() {

		return job;

	}

	public void setJob(BlackboardJob job) {

		this.job = job;

	}

	public ServiceLocator<ResultSetService> getResultSetServiceLocator() {

		return resultSetServiceLocator;

	}

	public void setResultSetServiceLocator(

			ServiceLocator<ResultSetService> resultSetServiceLocator) {

		this.resultSetServiceLocator = resultSetServiceLocator;

	}

	public BlackboardNotificationHandler<BlackboardServerHandler> getBlackboardHandler() {

		return blackboardHandler;

	}

	public void setBlackboardHandler(BlackboardNotificationHandler<BlackboardServerHandler> blackboardHandler) {

		this.blackboardHandler = blackboardHandler;

	}

}