D-Net

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

import java.io.BufferedWriter;

import java.io.File;

import java.io.FileWriter;

import java.io.IOException;

import java.util.ArrayList;

import java.util.Calendar;

import java.util.List;

import java.util.Map;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.LinkedBlockingQueue;

import java.util.concurrent.atomic.AtomicInteger;

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

import eu.dnetlib.validator.service.impl.ValidatorManager;

import org.apache.log4j.Logger;

import org.springframework.core.task.TaskExecutor;

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

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

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

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

import javax.annotation.PostConstruct;

public class DnetListener implements JobListener {

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

    private final AtomicInteger activeThreads = new AtomicInteger(0);

    private Object allThreadsFinished = new Object();

    private RecordXMLBuilder xmlBuilder;

    private BlackboardJob job;

    private BlackboardNotificationHandler<BlackboardServerHandler> blackboardHandler;

    private TaskExecutor rsExecutor;

    private ServiceLocator<ResultSetService> resultSetServiceLocator = null;

    private ResultSetService resultSetService;

    private String outputResultSetID;

    private List<String> outputDiskBuffer;

    private EPR outputEpr;

    private int internalJobsFinished = 0;

    private int internalJobsSum = 1;

    private int objsValidated = 0;

    private int jobStatusUpdateInterval;

    private boolean enableOutputToRS = false;

    private boolean enableOutputToDisk = false;

    private BufferedWriter bw = null;

    private BlockingQueue<String> queue;

    private int validationJobId;

    private ValidatorManager validatorManager;

    private String groupBy;

    public void initOutputs() {

        logger.debug("initializing outputs");

        if (enableOutputToRS) {

            this.initEprOutput();

            job.getParameters().put("outputResultSetEpr", EPRUtils.eprToXml(outputEpr));

            logger.debug("output epr id successfully set: " + EPRUtils.eprToXml(outputEpr));

        }

        if (enableOutputToDisk) {

            this.initDiskOutput();

            logger.debug("disk output ok");

        } else {

            logger.debug("initializing disk disabled");

        }

    }

    @Override

    public synchronized void currentResults(List<CompletedTask> tasks, int jobId, Object record, Map<String, Object> recordContext, Throwable t) throws ValidatorException {

        try {

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

        } catch (Exception e) {

            logger.error("Error while proccessing record results");

            throw new ValidatorException("Error while proccessing record results", e);

        }

    }

    @Override

    public synchronized void currentResults(List<CompletedTask> tasks, int jobId, Object record, Map<String, Object> recordContext) throws ValidatorException {

        try {

            if (enableOutputToRS) {

                @SuppressWarnings("unchecked")

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

                this.sendToQueue(xmlString);

            }

            if (enableOutputToDisk) {

                if ((Integer) recordContext.get("recordBlacklistScore") < 100) {

                    bw.write(tasks.get(0).valobjId);

                    bw.newLine();

                }

//				outputDiskBuffer.add(tasks.get(0).valobjId);

//				if (outputDiskBuffer.size() % 1000 == 0) {

//					for (String id : outputDiskBuffer) {

//						bw.write(id);

//						bw.newLine();

//					}

//					outputDiskBuffer.clear();

//				}

            }

            objsValidated++;

            if (objsValidated % jobStatusUpdateInterval == 0) {

                job.getParameters().put("recordsTested", Integer.toString(objsValidated));

                blackboardHandler.getBlackboardHandler().ongoing(job);

            }

        } catch (Exception e) {

            logger.error("Error while proccessing record results");

            throw new ValidatorException("Error while proccessing record results", e);

        }

    }

    @Override

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

        try {

            logger.debug("Job " + jobId + " finished");

            internalJobsFinished++;

            if (internalJobsFinished == internalJobsSum) {

                logger.debug("internalJobsFinished == internalJobsSum");

                if (enableOutputToRS) {

                    // send remaining xmls from buffers

//					if (outputRSBuffer.size() > 0) {

//						try {

//							resultSetService.populateRS(outputResultSetID,

//									outputRSBuffer);

//							outputRSBuffer.clear();

//						} catch (ResultSetServiceException e) {

//							logger.error("Error populating ResultSetService.",

//									e);

//						}

//					}

                    try {

                        queue.put("finished");

                    } catch (InterruptedException e) {

                        logger.error("Error finalizing queue", e);

                    }

                    if (queue.size() > 0) {

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

                        activeThreads.getAndIncrement();

                        RSTask task = new RSTask(resultSetService, outputResultSetID, queue, activeThreads, allThreadsFinished);

                        rsExecutor.execute(task);

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

                        logger.debug("Populating RS took " + ((time2 - time1))

                                + " milli seconds");

                    }

                    // finally, close rs.

                    logger.debug("active threads to finish: " + activeThreads.get());

                    logger.debug("trying to close result set");

                    while (activeThreads.get() > 0) {

                        logger.debug("waiting active threads to finish. Remaining: " + activeThreads.get());

                        synchronized (allThreadsFinished) {

                            allThreadsFinished.wait();

                        }

                        logger.debug("retrying to finish. Remaining: " + activeThreads.get());

                    }

                    logger.debug("closing result set");

                    resultSetService.closeRS(outputResultSetID);

                }

                if (enableOutputToDisk) {

                    bw.write("FINISHED");

//					bw.newLine();

//					logger.debug("output disk buffer size: " + outputDiskBuffer.size());

//					if (outputDiskBuffer.size() > 0) {

//						for (String id : outputDiskBuffer) {

//							bw.write(id);

//							bw.newLine();

//						}

//						outputDiskBuffer.clear();

//					}

                    bw.close();

                }

                // job.getParameters().put("outputResultSetEpr",

                // EPRUtils.eprToXml(outputEpr));

                job.getParameters().put("jobId", Integer.toString(validationJobId));

                job.getParameters().put("recordsTested", Integer.toString(objsValidated));

//				 job.getParameters().put(

//				 "recordsInResultSet",

//				 Integer.toString(resultSetService

//				 .getNumberOfElements(outputResultSetID)));

                // update job status

                logger.info("Getting stored job");

                StoredJob storedJob = validatorManager.getStoredJob(jobId, this.groupBy);

                logger.info("Score: " + storedJob.getContentJobScore());

                job.getParameters().put("score", storedJob.getContentJobScore() + "");

                blackboardHandler.getBlackboardHandler().done(job);

            } else {

                logger.debug("Waiting "

                        + (internalJobsSum - internalJobsFinished)

                        + " job(s) to finish");

            }

        } catch (Exception e) {

            logger.error("Error while finalizing successfull workflow job", e);

        }

    }

    @Override

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

                                    Throwable t) {

        try {

            internalJobsFinished++;

            if (internalJobsFinished == internalJobsSum) {

                if (enableOutputToRS) {

                    resultSetService.closeRS(outputResultSetID);

                }

                if (enableOutputToDisk) {

                    outputDiskBuffer.clear();

                    bw.close();

                    File file = new File("/tmp/validator-wf/" + validationJobId);

//					File file = new File("/tmp/validator-wf/"

//							+ job.getParameters().get("datasourceId"));

                    if (file.exists()) {

                        file.delete();

                    }

                }

                job.getParameters().put("jobId",

                        Integer.toString(validationJobId));

                // update job status

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

            } else {

                logger.debug("Waiting "

                        + (internalJobsSum - internalJobsFinished)

                        + " job(s) to finish");

            }

        } catch (Exception e) {

            logger.error("Error while finalizing failed workflow job", e);

        }

    }

    private void sendToQueue(String xmlString) {

        logger.debug("received passed XMLresult");

        try {

            queue.put(xmlString);

        } catch (InterruptedException e1) {

            logger.error("Error putting in queue.", e1);

        }

        if (queue.size() > 50) {

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

            RSTask task = new RSTask(resultSetService, outputResultSetID, queue, activeThreads, allThreadsFinished);

//			synchronized (counterLock) {

//				activeThreads ++;

//			}

            activeThreads.getAndIncrement();

            rsExecutor.execute(task);

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

            logger.debug("Populating RS took " + ((time2 - time1))

                    + " milli seconds");

        }

    }

    private void initEprOutput() {

        try {

            logger.debug("Initializing ResultSetService.");

            // Get reference to service

            resultSetService = resultSetServiceLocator.getService();

            // Create a new result set

            outputEpr = resultSetService.createPushRS(86400, 0);

            // get result set ids

            outputResultSetID = outputEpr.getParameter("ResourceIdentifier");

            // initializing buffers

//            outputRSBuffer = new ArrayList<String>();

            queue = new LinkedBlockingQueue<String>();

        } catch (Exception e) {

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

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

        }

    }

    private void initDiskOutput() {

        try {

            logger.debug("Initializing FileOutputStream.");

//			File file = new File("/var/lib/dnet/validator/workflow_blacklists/"

//					+ validationJobId);

            String datasourceId;

            if (job.getParameters().get("datasourceId") != null) {

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

            } else if (job.getParameters().get("datasourceID") != null) {

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

            } else {

                datasourceId = "unknownId";

            }

            File file = new File("/var/lib/dnet/validator/workflow_blacklists/" + datasourceId);

            logger.debug("File: + /var/lib/dnet/validator/workflow_blacklists/" + datasourceId);

            if (file.exists()) {

                logger.debug("File will be replaced");

                file.delete();

            }

            file.createNewFile();

            bw = new BufferedWriter(new FileWriter(file.getAbsoluteFile()));

//			GZIPOutputStream zip = new GZIPOutputStream(new FileOutputStream(file.getAbsoluteFile()));

//			bw = new BufferedWriter(new OutputStreamWriter(zip, "UTF-8"));

            outputDiskBuffer = new ArrayList<String>();

        } catch (IOException e) {

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

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

        }

    }

    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;

    }

    public TaskExecutor getRsExecutor() {

        return rsExecutor;

    }

    public void setRsExecutor(TaskExecutor rsExecutor) {

        this.rsExecutor = rsExecutor;

    }

    public int getInternalJobsSum() {

        return internalJobsSum;

    }

    public void setInternalJobsSum(int internalJobsSum) {

        this.internalJobsSum = internalJobsSum;

    }

    public int getValidationJobId() {

        return validationJobId;

    }

    public void setValidationJobId(int validationJobId) {

        this.validationJobId = validationJobId;

    }

    public int getJobStatusUpdateInterval() {

        return jobStatusUpdateInterval;

    }

    public void setJobStatusUpdateInterval(int jobStatusUpdateInterval) {

        this.jobStatusUpdateInterval = jobStatusUpdateInterval;

    }

    public boolean isEnableOutputToDisk() {

        return enableOutputToDisk;

    }

    public void setEnableOutputToDisk(boolean enableOutputToDisk) {

        this.enableOutputToDisk = enableOutputToDisk;

    }

    public boolean isEnableOutputToRS() {

        return enableOutputToRS;

    }

    public void setEnableOutputToRS(boolean enableOutputToRS) {

        this.enableOutputToRS = enableOutputToRS;

    }

    public ValidatorManager getValidatorManager() {

        return validatorManager;

    }

    public void setValidatorManager(ValidatorManager validatorManager) {

        this.validatorManager = validatorManager;

    }

    public String getGroupBy() {

        return groupBy;

    }

    public void setGroupBy(String groupBy) {

        this.groupBy = groupBy;

    }

}