/*******************************************************************************
* Copyright (c)                                                                *
*           2005   Tristan Allwood,                                            *
*           2005   Daniel Burke,                                               *
*           2005   Marc Hull,                                                  *
*           2005   Ekaterina Itskova,                                          *
*           2005   Steve Zymler                                                *
*                                                                              *
* Permission is hereby granted, free of charge, to any person obtaining a      *
* copy of this software and associated documentation files (the                *
* "Software"), to deal in the Software without restriction, including          *
* without limitation the rights to use, copy, modify, merge, publish,          *
* distribute, sublicense, and/or sell copies of the Software, and to           *
* permit persons to whom the Software is furnished to do so, subject to        *
* the following conditions:                                                    *
*                                                                              *
* The above copyright notice and this permission notice shall be included      *
* in all copies or substantial portions of the Software.                       *
*                                                                              *
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS      *
* OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF                   *
* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.       *
* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY         *
* CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,         *
* TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE            *
* SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.                       *
*******************************************************************************/
/*
 * Created on Dec 24, 2004
 */
package uk.ac.ic.doc.kevlar.api.programs.framework.internal;

import java.io.IOException;

import uk.ac.ic.doc.kevlar.publicInterfaces.IInputPipe;
import uk.ac.ic.doc.kevlar.publicInterfaces.IOutputPipe;
import uk.ac.ic.doc.kevlar.publicInterfaces.IType;

/**
 * @author tora
 */
class InputProcessor implements Runnable {

    final IPipeGroup group;

    final Runnable fini;

    final PipeManager2 parent;

    public InputProcessor(IPipeGroup group, Runnable fini, PipeManager2 parent) {

        this.group = group;
        this.fini = fini;
        this.parent = parent;
    }

    public void run() {
        Thread.currentThread().setName("Input Processor");
        try {
            while (true) {
                IPacket ip = (IPacket) group.getInputStream().readObject();

                if (ip instanceof AControlUnit) {
                    handleControlUnit((AControlUnit) ip);

                } else if (ip instanceof ADataUnit) {
                    handleDataUnit((ADataUnit<IType>) ip);
                } else {
                    throw new AssertionError("Unknown Type");
                }

                parent.checkIfDone(group);
            }
        } catch (IOException ioe) {
            System.err.println("IO Exception in Input Processor");
            /* fall through */
        } catch (ClassNotFoundException cnfe) {
            /* fall through */
        } catch (AssertionError ae) {
            ae.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }

        parent.removeGroup(group);
        
        if (fini != null) {
            System.err.println("Running fini");
            fini.run();
        }

    }

    private void handleControlUnit(AControlUnit acu) {
        {
            int code = acu.getControlValue();

            if (code == AControlUnit.CLOSE_INPUT_PIPE) {

                synchronized (group) {
                    IInputPipe pipe = group.getInputPipe(acu.getPipeName());
                    if (pipe == null) {
                        throw new AssertionError("Unknown Pipe");
                    }
                    pipe.close();
                    group.registerClosedInput(acu.getPipeName());
                    group.notifyAll();
                }

            } else if (code == AControlUnit.CLOSE_OUTPUT_PIPE) {

                synchronized (group) {
                    IOutputPipe pipe = group.getOutputPipe(acu.getPipeName());
                    if (pipe == null) {
                        throw new AssertionError("Unknown Pipe");
                    }
                    pipe.close();
                    group.registerClosedOutput(acu.getPipeName());
                    group.notifyAll();
                }

            } else if (code == AControlUnit.RECIEPT) {

                synchronized (group) {
                    group.registerReciept(Long.valueOf(acu.getRecieptNumber()));
                    group.notifyAll();
                }

            } else {
                throw new AssertionError("Unknown Control Code");
            }
        }
    }

    private void handleDataUnit(ADataUnit<IType> adu) {

        IType data = adu.getData();
        String name = adu.getPipeName();

        parent.writeOut(group,new AControlUnit(
                        AControlUnit.RECIEPT, adu.getRecieptNumber()));

        synchronized (group) {

            if (group.getOutputPipe(name) == null) {
                throw new AssertionError("Unknown Pipe Name " + name);
            }

            if (!group.getOutputPipe(name).closed()) {
                group.getOutputPipe(name).write(data);
            }

            group.notifyAll();
        }

    }
}
