/*******************************************************************************
* 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 29-Oct-2004 by toa02
 *
 */
package uk.ac.ic.doc.kevlar.api.programs.framework.internal;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import uk.ac.ic.doc.kevlar.core.framework.internal.loaders.ExecuteHolder;
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 toa02
 *  
 */
class PipeManager implements IPipeManager, Runnable {

    //all the registered input and output groups
    final Set<IPipeGroup> allGroups;

    final static boolean DEBUG_STREAMS = Boolean.parseBoolean(System.getProperty("streamdebug","false"));
    
    //the stream to use
    final OutputStream debugStream;

    public PipeManager(OutputStream debugStream) {

        allGroups = new HashSet<IPipeGroup>();
        this.debugStream = debugStream;

        ExecuteHolder.getExecutor().execute(this);
        
    }

    public void waitTillAllDone() {
        
        synchronized ( allGroups ) {
            
            while ( !allGroups.isEmpty() ) {
                try {
                    allGroups.wait();
                } catch ( InterruptedException ie ) {
                    /* ignore it */
                }
            }
        }
        
    }
    
    public void managePipes(final InputStream out, final OutputStream in,
            final Map<String, IInputPipe> ins,
            final Map<String, IOutputPipe> outs, final InputStream debug, final Serializable initial,
			ClassLoader cl, Runnable fini) {

/*        //This fires off a new thread as it can block
        ExecuteHolder.getExecutor().execute(new Runnable() {

            public void run() { */
                try {
                    
                    if (debug != null) {
                        ExecuteHolder.getExecutor().execute(
                                new RepeatThread(new BufferedInputStream(debug)));
                    }
                    
                    IPipeGroup group = new PipeGroup(out, in, ins, outs, debug, initial, cl);

                    synchronized ( allGroups ) {
                        allGroups.add(group);
                        allGroups.notifyAll();
                    }
                    
                    ExecuteHolder.getExecutor().execute(
                            new HandlePipeGroupOutputs(group));
                    
                    
                } catch (IOException e2) {
                    /* Piped closed before we even started. */
                    e2.printStackTrace();
                    return;
                    
                }
        /*    }
        });*/

    }

    /**
     * @see java.lang.Runnable#run()
     */
    public void run() {

        /* TODO toa02 : find a way to fake the multiplexing in here.
         * The thread.yield should make things better for the moment, but is
         * not a solution.  This will come after a fair bit of thought and a
         * major re-structure to take advantage of the event model that we now have.
         * 
         * continually poll all inputs and pass along anything that needs to go
         * along from them. this works by just looping across them all. It isn't
         * very efficient, but there isn't any other way to do it I believe,
         * since we can't multiplex, and we don't want a thread for each one.
         * 
         * This also checks the output pipes to see if any have been closed. If
         * they have it passes that information across the line.
         */
        while (true) {

            synchronized (allGroups) {

                //if there is absolutely nothing to do, just wait until there
                // is.
                while (allGroups.isEmpty() ) {
                    try {
                        allGroups.wait();
                    } catch (InterruptedException e) {
                        /* propogate up the interrupt */
                        throw new RuntimeException(e);
                    }
                }
                
                //TODO: toa02 - see comment above. This is not a solution.
                Thread.yield();
                
                // The set of groups we need to remove because they have been
                // all closed.
                Set<IPipeGroup> groupsToRemove = new HashSet<IPipeGroup>();

                for (IPipeGroup currentGroup : allGroups) {

                    synchronized (currentGroup) {

                        ObjectOutputStream oos = currentGroup.getOutputStream();
                        boolean keepAlive = false;
                        boolean kill = false;

                        for (String inName : currentGroup.getInputPipeNames()) {

                            IInputPipe currIn = currentGroup
                                    .getInputPipe(inName);

                            if (currIn.closed()) {
                                
                                try {
                                    oos.writeObject(new AControlUnit(inName,
                                            AControlUnit.CLOSE_OUTPUT_PIPE));
                                    oos.flush();
                                    
                                    if( DEBUG_STREAMS ) {
                                        System.err.println("/" + System.currentTimeMillis() + "/ Closing Input Pipe: " + inName);
                                    }
                                    
                                } catch (IOException e) {
                                    if (DEBUG_STREAMS) {
                                        System.err
                                                .println("IO Exception on stream write A");
                                    }
                                    kill = true;
                                    keepAlive = true; //we dont' need to
                                    // evaluate the other
                                    // direction
                                    break;
                                }
                                currentGroup.registerClosedInput(inName);
                                continue;
                            }
                            
                            keepAlive = true;
                            
                            if (currIn.ready()) {

                                try {
                                    IType type = currIn.read();
                                    
                                    long l = currentGroup.getNextReciept();
                                    
                                    if( DEBUG_STREAMS ) {
                                        System.err.println("Writing out object, rpt: " + l );
                                    }
                                    
                                    oos.writeObject(new ADataUnit<IType>(type,
                                            inName, l));
                                    oos.flush();
                                } catch (InterruptedException e) {
                                    throw new AssertionError(
                                            "Precondition failure");
                                } catch (IOException e) {
                                    if (DEBUG_STREAMS) {
                                        System.err
                                                .println("IO Exception on stream write B");
                                    }
                                    kill = true;
                                    keepAlive = true; //we dont' need to
                                    // evaluate the other
                                    // direction
                                }
                            }

                        }

                        /*
                         * Now check if we should be closing output pipes if any
                         * of the output pipes are still open. If there arn't,
                         * we should remove this group and close its streams to
                         * save on resources.
                         */
                        for (String outName : currentGroup.getOutputPipeNames()) {
                            IOutputPipe currOut = currentGroup
                                    .getOutputPipe(outName);
                            if (currOut.closed()) {
                                try {
                                    oos.writeObject(new AControlUnit(outName,
                                            AControlUnit.CLOSE_INPUT_PIPE));
                                    oos.flush();
                                    if( DEBUG_STREAMS ) {
                                        System.err.println("/" + System.currentTimeMillis() + "/ Closing pipe: " + outName);
                                    }
                                    
                                } catch (IOException e) {
                                    if (DEBUG_STREAMS) {
                                        System.err
                                                .println("IO Exception on stream write C");
                                    }
                                    kill = true;
                                    keepAlive = true; //we dont' need to
                                    // evaluate the other
                                    // direction
                                    currentGroup.registerClosedOutput(outName);
                                    break;
                                }
                                currentGroup.registerClosedOutput(outName);
                                continue;
                            } else {
                                keepAlive = true;
                            }
                        }

                        if( currentGroup.missingReciepts() ) {
                            keepAlive = true;
                        }
                        
                        /* Ok: kill the whole damn lot if we need to */
                        if (!keepAlive || kill) {
                            groupsToRemove.add(currentGroup);
                            try {
                                currentGroup.getInputStream().close();
                            } catch (IOException e) {
                                if( DEBUG_STREAMS ) {
                                    e.printStackTrace();
                                }
                            }

                            try {
                                currentGroup.getOutputStream().close();
                            } catch (IOException e1) {
                                if ( DEBUG_STREAMS ) {
                                    e1.printStackTrace();
                                }
                            }

                            for (String s : currentGroup.getOutputPipeNames()) {
                                currentGroup.getOutputPipe(s).close();
                                currentGroup.registerClosedOutput(s);
                            }

                            for (String s : currentGroup.getInputPipeNames()) {
                                currentGroup.getInputPipe(s).close();
                                currentGroup.registerClosedInput(s);
                            }
                        }
                    }//synchronized on the current group
                }//for allGroups

                if ( allGroups.removeAll(groupsToRemove) ) {
                    allGroups.notifyAll();
                }
            }//synchronized

        }//while true

    } // method

    /**
     * Takes the output of
     * 
     * @author tora
     */
    class RepeatThread implements Runnable {

        private final InputStream source;

        RepeatThread(InputStream in) {
            source = in;
        }

        /**
         * @see java.lang.Runnable#run()
         */
        public void run() {
            int i;

            try {
                while (-1 != (i = source.read())) {
                    debugStream.write(i);
                }
            } catch (IOException ioe) {
                if( DEBUG_STREAMS ) {
                    ioe.printStackTrace();
                }
            }
        }

    }

    class HandlePipeGroupOutputs implements Runnable {

        private final IPipeGroup ipg;

        HandlePipeGroupOutputs(IPipeGroup ipg) {
            this.ipg = ipg;
            
        }

        /**
         * @see java.lang.Runnable#run()
         */
        public void run() {
            //Run the pipe group outputs thread

            boolean keepRunning = true;
            boolean error = false;

            while (keepRunning) {

                ObjectInputStream source = ipg.getInputStream();

                try {
                    IPacket packet = (IPacket) source.readObject();

                    if (packet instanceof ADataUnit) {
                        ADataUnit<IType> adu = (ADataUnit<IType>) packet;
                        
                        String name = packet.getPipeName();

                        synchronized (ipg) {
                            
                            if( DEBUG_STREAMS ) {
                                System.err.println("Writing Reciept: " + adu.getRecieptNumber());
                            }
                            
                            ipg.getOutputStream().writeObject(
                                    new AControlUnit(
                                            AControlUnit.RECIEPT,adu.getRecieptNumber()));
                            
                            
                            if( ipg.getOutputPipe(name) == null ) {
                                error = true;
                                break;
                            }
                            if (!ipg.getOutputPipe(name).closed()) {
                                ipg.getOutputPipe(name).write(adu.getData());
                            }
                            
                            
                            
                        }

                    } else if (packet instanceof AControlUnit) {
                        AControlUnit acu = (AControlUnit) packet;

                        if (acu.getControlValue() == AControlUnit.CLOSE_INPUT_PIPE) {
                            
                            if( DEBUG_STREAMS ) {
                                System.err.println("Closing Input Pipe:" + acu.getPipeName()) ; 
                            }
                            
                            synchronized (ipg) {
                                IInputPipe iip = ipg.getInputPipe(acu.getPipeName());
                                if( iip != null ) {
                                    iip.close();
                                    ipg.registerClosedInput(acu.getPipeName());
                                } else {
                                    error = true;
                                    ipg.registerClosedInput(acu.getPipeName());
                                    break;
                                }

                            }
                            
                        } else if ( acu.getControlValue() == AControlUnit.CLOSE_OUTPUT_PIPE) {
                            synchronized (ipg) {
                                IOutputPipe iop = ipg.getOutputPipe(acu.getPipeName());
                                
                                if( DEBUG_STREAMS ) {
                                    System.err.println("/" + System.currentTimeMillis() + "/ Closing Output Pipe: " + acu.getPipeName());
                                }
                                
                                if( iop != null ) {
                                    
                                    iop.close();
                                    ipg.registerClosedOutput(acu.getPipeName());
                                } else {
                                    error = true;
                                    ipg.registerClosedOutput(acu.getPipeName());
                                    break;
                                }

                            }
                            
                        } else if ( acu.getControlValue() == AControlUnit.RECIEPT ) {
                            if( DEBUG_STREAMS ) {
                                System.err.println("Got a reciept: " + Long.valueOf(acu.getRecieptNumber())  );
                            }
                            
                            synchronized ( ipg ) {
                                ipg.registerReciept( Long.valueOf(acu.getRecieptNumber()) );
                            }
                            
                            
                        } else {
                            if (DEBUG_STREAMS) {
                                System.err
                                        .println("Got an unknown packet Control Value type");
                            }
                            error = true;
                            break;
                        }

                    } else {
                        if (DEBUG_STREAMS) {
                            System.err.println("Got an unknown packet type");
                        }
                        error = true;
                        break;
                    }
                } catch (IOException e) {
                    if( DEBUG_STREAMS ) {
                        System.err.println("/" + System.currentTimeMillis() + "/ IO Exception");
                        e.printStackTrace();
                    }
                    error = true;
                    break;

                } catch (ClassNotFoundException e) {
                    e.printStackTrace();
                    error = true;
                    break;
                }
                
                synchronized ( allGroups ) {
	                synchronized (ipg) {
	             
	                    keepRunning = !error && ( ipg.getInputPipeNames().size() != 0
	                            || ipg.getOutputPipeNames().size() != 0 || ipg.missingReciepts()  )
	                            && allGroups.contains(ipg);
	                }
                }
            }

            if (error || !keepRunning) {
                //again, panick and kill everything
                synchronized (ipg) {
                    try {
                        ipg.getInputStream().close();
                    } catch (IOException e) {
                        if( DEBUG_STREAMS ) {
                            e.printStackTrace();
                        }
                    }
                    try {
                        ipg.getOutputStream().close();
                    } catch (IOException e1) {
                        if( DEBUG_STREAMS ) {
                            e1.printStackTrace();
                        }
                    }

                    for (String s : ipg.getOutputPipeNames()) {
                        ipg.getOutputPipe(s).close();
                        ipg.registerClosedOutput(s);
                    }

                    for (String s : ipg.getInputPipeNames()) {
                        ipg.getInputPipe(s).close();
                        ipg.registerClosedInput(s);
                    }
                    
                }
                
            }
            
            synchronized ( allGroups ) {
                allGroups.remove(ipg);
                allGroups.notifyAll();
            }
            
            if( DEBUG_STREAMS) {
                System.err.println("A Pipe Handler is exiting");
            }
            
        }

    }

}
