/*
 * $Id: LookupDaemon.java,v 1.21 2007/10/30 13:47:06 marbobo Exp $
 *
 * Copyright (c) Siemens AG 2008 
 *           (c) Universität Paderborn 2008 
 * Licensed under EUPL 1.1. 
 *
 * $Log: LookupDaemon.java,v $
 * Revision 1.21  2007/10/30 13:47:06  marbobo
 * + removed unnecessary statements
 *
 * Revision 1.20  2007/08/16 11:10:57  marbobo
 * + Bugfixes
 * + Refactorings
 *
 * Revision 1.19  2006/08/10 12:32:02  marlin
 * Update of last change
 *
 */
package vkc.converter.server;

import java.io.IOException;
import java.net.Socket;
import java.net.UnknownHostException;
import java.sql.Timestamp;
import java.util.Iterator;

import vkc.converter.common.Status;
import vkc.db.PersistenceManager;
import vkc.db.common.ConversionStatus;
import vkc.db.common.ConverterInfo;
import vkc.db.common.ConverterTiming;
import vkc.db.common.VKCServerInfo;
import vkc.exception.DBException;
import vkc.misc.VKCLogManager;

/**
 * This class implements background daemon for looking up and initiating
 * outstanding conversion jobs.
 * 
 * @created 07.01.2004
 * @author Andreas Koop
 */
public class LookupDaemon implements Runnable {
    private boolean running = false;

    private ConverterServer server = null;

    private boolean shutdownRequested = false;

    private ConversionStatus conversionStatus = null;

    private LookupDaemon() {
    }

    public LookupDaemon(ConverterServer server) {
        this();
        this.server = server;
    }

    /**
     * Main processing loop.
     * <p>
     * After a specified period the method <code>lookupConverisons()</code> is
     * called iff we are in the converters timeframe.
     * 
     * @see java.lang.Runnable#run()
     */
    public void run() {
        VKCLogManager.infoLog(this, "Lookup daemon started...");

        try {
            // Beim Start abgebrochene Konvertierungen zurücksetzen
            ConversionStatus.checkAndResetStartedJobs((ConverterInfo) server.getConfiguration());
        } catch (DBException e1) {
            VKCLogManager.errorLog(this, "", "Fehler bei der Aktualisierung der Konvertierungsstatus!", e1);
        }

        running = true;
        ConverterTiming timing = null;

        try {
            while (running) {
                try {
                    for (Iterator it = PersistenceManager.getInstance().getDbUrls().iterator(); it.hasNext();) {
                        String dbUrl = (String) it.next();

                        timing = server.getTiming();
                        
                        if (timing.inTimeFrame()) {
                            lookupConversions(dbUrl);
                        } else {
                            VKCLogManager.infoLog(this, "Not in time frame: " + timing + ". Conversion lookup suspended!");
                        }

                        if (shutdownRequested) {
                            break;
                        }

                        try {
                            Thread.sleep(timing.getPollingPeriodMillis());
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                            break;
                        }
                    }

                    if (shutdownRequested) {
                        break;
                    }
                } catch (DBException e) {
                    VKCLogManager.errorLog(this, "", e.getMessage(), e);
                }
            }
        } finally {
            running = false;
        }
    }

    /**
     * Looks up outstanding conversions and initiates the conversion..
     */
    private void lookupConversions(String dbUrl) throws DBException {
        VKCServerInfo vkcServer;
        ConverterInfo conf;

        while (true) {
            conf = server.getConfiguration(dbUrl);

            // Falls der Status vom VKC-Server oder manuell
            // auf DOWN gesetzt worden ist, wird das hier wieder rückgängig
            // gemacht!
            if (conf.getStatus().equals(Status.SERVER_DOWN) && !shutdownRequested) {
                conf.setStatus(Status.SERVER_UP);
                conf.makeStatusPersistent();
            }

            // lookupConversions should be called only when
            // no other jobs (analyse and/or classify) are not to be processed
            int jobsCount = conf.getJobsInExecution();
            
            if (jobsCount > 0) {
                VKCLogManager.infoLog(this, "Conversion lookup suspended.\n" + "There are synchron jobs in execution (by analyse/classify), namely " + jobsCount + "!");
                
                return;
            }

            // Get next outstanding conversion
            synchronized (this) {
                conversionStatus = ConversionStatus.getOutstandingConversion(conf);
                
                if (conversionStatus == null) {
                    return;
                }
            }

            if (shutdownRequested) {
                resetConversionStatus("Shutdown requested!");
                
                return;
            }

            Socket socket = null;
            
            while (true) {
                vkcServer = VKCServerInfo.getMostUnderWorkedServer(dbUrl);
                
                if (vkcServer == null) {
                    VKCLogManager.infoLog(this, "Archive request cannot be sent.\n" + "There is no VKC-Server running on " + dbUrl);
                    resetConversionStatus("No running VKC-Server");
                    
                    return;
                }

                VKCLogManager.infoLog(this, "Connecting to " + vkcServer.getHostname() + ":" + vkcServer.getPort());
                
                try {
                    socket = vkcServer.openSocket(5000); // maximal 5 sec warten
                    break;
                } catch (UnknownHostException e1) {
                    VKCLogManager.errorLog(this, "", "Unknown host " + vkcServer.getHostname(), e1);
                } catch (IOException e1) {
                }

                vkcServer.setStatus(Status.SERVER_DOWN);
                vkcServer.setJobsRequestedCount(0);
                vkcServer.makeStatusPersistent();
                vkcServer.makeJobsCountPersistent();
            }

            // Initiate conversion
            try {
                ConversionProcessor cProc = new ConversionProcessor(server, socket);
                cProc.setConversionStatus(conversionStatus);

                // Request the stream for the given documentId
                cProc.getObjectOutputStream().writeObject(conversionStatus.getDocumentId());
                cProc.getObjectOutputStream().flush();
                cProc.run();
            } catch (IOException e) {
                VKCLogManager.errorLog(this, "", "I/O Error while communication with the VKCServer", e);
            } finally {
                synchronized (this) {
                    conversionStatus = null;
                }
            }
        }
    }

    private void resetConversionStatus(String cause) {
        synchronized (this) {
            if (conversionStatus == null) {
                return;
            }

            conversionStatus.setStatus(Status.CONVERSION_SUSPENDED);
            conversionStatus.setFinished(new Timestamp(System.currentTimeMillis()));
            conversionStatus.setComment("Suspended due to: " + cause);

            try {
                conversionStatus.makePersistent();
            } catch (DBException e) {
                VKCLogManager.errorLog(this, "", "Conversionstatus " + conversionStatus + " could not be updated!", e);
            }

            conversionStatus = null;
        }
    }

    public ConversionStatus getCurrentConversionStatus() {
        return this.conversionStatus;
    }

    public void requestShutdown() {
        this.shutdownRequested = true;
    }
}