/*
 * $Id: ConversionProcessor.java,v 1.18 2007/08/16 11:10:57 marbobo Exp $
 *
 * Copyright (c) Siemens AG 2008 
 *           (c) Universität Paderborn 2008 
 * Licensed under EUPL 1.1. 
 *
 * $Log: ConversionProcessor.java,v $
 * Revision 1.18  2007/08/16 11:10:57  marbobo
 * + Bugfixes
 * + Refactorings
 *
 * Revision 1.17  2006/04/04 17:01:47  marbobo
 * + Einige Anpassungen bzgl. neu eingeführter Konvertierungsformate
 *
 */
package vkc.converter.server;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.net.Socket;
import java.net.SocketException;
import java.net.URL;
import java.sql.Timestamp;
import java.util.Collection;
import java.util.Iterator;

import vkc.converter.DocConverter;
import vkc.converter.common.ConversionRequest;
import vkc.converter.common.ConversionResponse;
import vkc.converter.common.Status;
import vkc.converter.common.StreamConversionRequest;
import vkc.converter.common.UrlConversionRequest;
import vkc.db.common.ConversionStatus;
import vkc.db.entities.VKCAdditionalDocumentFormat;
import vkc.exception.ConverterException;
import vkc.exception.DBException;
import vkc.misc.VKCLogManager;
import vkc.util.FileUtils;
import vkc.util.SocketUtils;
import vkc.util.VKCInputStream;

/**
 * This class implements a worker for the Converter Server
 * 
 * @created 22.11.2003
 * @author Andreas Koop
 */
public class ConversionProcessor implements Runnable {
    private Socket client;

    private ObjectInputStream is;

    private ObjectOutputStream os;

    private ConversionRequest request = null;

    private ConverterServer converterServer = null;

    private DocConverter converter = null;

    private ConversionStatus status = new ConversionStatus();

    private Object monitor = new Object();

    /**
     * Constructor
     * 
     * @param converterServer
     *            reference to the converter server
     * @param s
     *            the socket
     */
    public ConversionProcessor(ConverterServer converterServer, Socket s) {
        this.converterServer = converterServer;
        client = s;

        try {
            os = SocketUtils.getObjectOutputStream(client);
            os.flush();
            is = SocketUtils.getObjectInputStream(client);
        } catch (IOException e) {
        }
    }

    public String getHostnameOfClient() {
        return client.getInetAddress().getHostName();
    }

    /**
     * Sets the conversion status object for this conversion
     * 
     * @param status
     */
    public void setConversionStatus(ConversionStatus status) {
        this.status = status;
    }

    /**
     * Returns the associated outputstream
     * 
     * @return
     */
    public ObjectOutputStream getObjectOutputStream() {
        return this.os;
    }

    /**
     * Main thread loop
     */
    public void run() {
        if (is == null) {
            // Request war nur ein Ping vom Server.
            return;
        }

        converterServer.incJobsCount(1);
        File tmpFile = null;

        try {
            // get the converter request object
            request = (ConversionRequest) is.readObject();
            VKCLogManager.infoLog(this, "Got " + request);

            if (status.getDocumentId() != null) {
                VKCLogManager.infoLog(this, "for DOCID=" + status.getDocumentId());
            } else {
                VKCLogManager.infoLog(this, "for classification, analyse or standalone in synchron mode...");
            }

            // Creates an appropriate DocConverter
            if (request instanceof StreamConversionRequest) {
                StreamConversionRequest s = (StreamConversionRequest) request;

                // Stream temporär speichern
                // Könnte vermieden werden, aber die Kommunikation is damit
                // sicherer
                VKCInputStream vkcis = new VKCInputStream(new BufferedInputStream(is), s.getSize());

                tmpFile = File.createTempFile("converterFile", ".tmp");
                tmpFile.deleteOnExit();
                BufferedOutputStream fOut = new BufferedOutputStream(new FileOutputStream(tmpFile));
                FileUtils.write(vkcis, fOut);
                fOut.flush();
                fOut.close();

                // Konverter für tempdatei erzeugen
                converter = DocConverter.createConverter(new VKCInputStream(new BufferedInputStream(new FileInputStream(tmpFile)), s.getSize()), s.getDocTypeId(), s.isZipped(), s.getRootFilename());
            } else if (request instanceof UrlConversionRequest) {
                converter = DocConverter.createConverter(new URL(((UrlConversionRequest) request).getLink()));
            }

            // Generate and send response
            sendResponse();

            os.flush();
        } catch (Exception e) {
            VKCLogManager.errorLog(this, "", "Exception occured while processing " + (request == null ? "(ConversionRequest) is.readObject()" : request.toString()));

            if (status.getDocumentId() != null) {
                VKCLogManager.errorLog(this, "", "for DOCID=" + status.getDocumentId(), e);
            } else {
                VKCLogManager.errorLog(this, "", "for classification, analyse or standalone in synchron mode...", e);
            }

            if (!(e instanceof DBException)) {
                status.incReiterations();
                status.setFinished(new Timestamp(System.currentTimeMillis()));
                status.setStatus(Status.CONVERSION_SUSPENDED);
                status.setComment("Suspended due to: " + this.getShortStackStrace(e, 18));

                try {
                    status.makePersistent();
                } catch (DBException e1) {
                    VKCLogManager.errorLog(this, "", "Exception while making conversionstatus " + status + " persistent!", e1);
                }
            }

            if (!(e instanceof SocketException)) {
                // Der Rest des InputStreams muss immer gelesen werden,
                // sonst kann der Client ggfs. blockieren
                try {
                    FileUtils.write(is, null);
                    os.writeObject(new ConversionResponse());
                } catch (IOException e1) {
                    VKCLogManager.errorLog(this, "", e.getMessage(), e1);
                }
            }
        } finally {
            if (client != null) {
                SocketUtils.close(client);
            }

            converterServer.decJobsCount();

            if (tmpFile != null) {
                tmpFile.delete();
            }
        }
    }

    private String getShortStackStrace(Exception e, int length) {
        StringWriter sw = new StringWriter();
        e.printStackTrace(new PrintWriter(sw));

        String s = sw.toString();

        if (s.length() < 256 - length) {
            return s;
        }

        return s.substring(0, 250 - length) + "...";
    }

    /**
     * Writes the given stream to the outputstream
     * 
     * @param stream
     */
    private void out(InputStream stream) throws IOException {
        FileUtils.write(stream, os);
        stream.close();
    }

    /**
     * Sends the response according to the converter
     * 
     * @param converter
     * @throws IOException
     */
    private void sendResponse() throws Exception {
        if (converter == null) {
            throw new ConverterException("Converter not available for requested document!");
        }

        ConversionResponse response = new ConversionResponse();

        // do conversion
        ConversionProcess cProc = new ConversionProcess();
        new Thread(cProc).start();

        // Workaround
        Collection docFormats = null;

        while ((docFormats = cProc.getAdditionalDocumentFormats()) == null) {
            os.writeObject(new String("Dummy"));

            synchronized (monitor) {
                monitor.wait(100000);
            }
        }

        status.setFinished(new Timestamp(System.currentTimeMillis()));
        status.setStatus(Status.CONVERSION_DONE);
        status.setComment("Successfully converted by converter: " + this.converterServer.getConfiguration());
        status.makePersistent();

        response.setAdditionalDocFormats(docFormats);
        response.setTdfElements(converter.getElements());
        response.setTitle(converter.getTitle());
        response.setKeywords(converter.getKeyWords());
        response.setDescription(converter.getDescription());
        response.setStatus(Status.SUCCEEDED);
        response.setSendmail(status.isSendmail());
        response.setDocumentId(status.getDocumentId());

        os.writeObject(response);
        os.flush();

        VKCLogManager.infoLog(this, "Response object sent: " + response);

        // send resulting outputstreams
        for (Iterator it = docFormats.iterator(); it.hasNext();) {
            VKCAdditionalDocumentFormat f = (VKCAdditionalDocumentFormat) it.next();
            out(f.getInputStream());
        }

        VKCLogManager.infoLog(this, "Converted Streams sent to " + request.getAssocDbUrl());
    }

    /**
     * (non-Javadoc)
     * 
     * @see java.lang.Object#finalize()
     */
    protected void finalize() throws Throwable {
        this.is = null;
        this.os = null;
        super.finalize();
    }

    private class ConversionProcess implements Runnable {
        Collection addFormats = null;

        public void run() {
            addFormats = converter.getAdditionalDocumentFormats(request.getConversions());

            synchronized (monitor) {
                monitor.notify();
            }
        }

        public Collection getAdditionalDocumentFormats() {
            return addFormats;
        }
    }
}