/*******************************************************************************
* 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 java.io.InputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import static uk.ac.ic.doc.kevlar.api.programs.framework.internal.PipeManager2.PipeEventType.INPUT;
import static uk.ac.ic.doc.kevlar.api.programs.framework.internal.PipeManager2.PipeEventType.OUTPUT;
import uk.ac.ic.doc.kevlar.core.framework.internal.loaders.ExecuteHolder;
import uk.ac.ic.doc.kevlar.publicInterfaces.ICloseEventListener;
import uk.ac.ic.doc.kevlar.publicInterfaces.IInputPipe;
import uk.ac.ic.doc.kevlar.publicInterfaces.IOutputPipe;
import uk.ac.ic.doc.kevlar.publicInterfaces.IPipeEnd;
import uk.ac.ic.doc.kevlar.publicInterfaces.IReadyEventListener;
import uk.ac.ic.doc.kevlar.publicInterfaces.IType;

/**
 * @author tora
 */
class PipeManager2 implements IPipeManager, Runnable {

    final OutputStream out;

    final Set<IPipeGroup> allGroups;

    final BlockingQueue<PipeEvent> eventQueue = new LinkedBlockingQueue<PipeEvent>();

    final ToiletManager toiletManager;
    
    public PipeManager2(OutputStream out) {
        this.out = out;
        this.allGroups = new HashSet<IPipeGroup>();
        this.toiletManager = new ToiletManager(this);
        ExecuteHolder.getExecutor().execute(this);
    }

    public void managePipes(InputStream out, OutputStream in,
            Map<String, IInputPipe> ins, Map<String, IOutputPipe> outs,
            InputStream debug, Serializable initialData, ClassLoader cl,
            Runnable fini) {

        if (debug != null) {
            RepeatThreadProvider.makeRepeatThread(debug, this.out);
        }

        final IPipeGroup newGroup;
        try {
            newGroup = new PipeGroup(out, in, ins, outs, debug, initialData, cl);
        } catch (IOException ioe) {
            /* bad group */
            ioe.printStackTrace();
            return;
        }

        synchronized (allGroups) {
            allGroups.add(newGroup);
            allGroups.notifyAll();
        }

        toiletManager.manageNewGroup(newGroup);
        
        ExecuteHolder.getExecutor().execute(
                new InputProcessor(newGroup, fini, this));
        
        class CEL implements ICloseEventListener {

            final PipeEvent pe;

            CEL(PipeEvent pe) {
                this.pe = pe;
            }

            public void pipeClosed(IPipeEnd source) {
                eventQueue.add(pe);
            }

        }

        class IR implements IReadyEventListener {
            final PipeEvent pe;

            IR(PipeEvent pe) {
                this.pe = pe;
            }

            public void pipeReadyEvent(IInputPipe source) {
                eventQueue.add(pe);
            }
        }

        // now add all the pipes, and create a default event for each
        synchronized (newGroup) {
            for (String name : newGroup.getInputPipeNames()) {
                PipeEvent pe = new PipeEvent(name, newGroup,
                        PipeEventType.INPUT);
                newGroup.getInputPipe(name).addCloseEventListener(new CEL(pe));
                newGroup.getInputPipe(name).addReadyEventListener(new IR(pe));
                eventQueue.add(pe);
            }

            for (String name : newGroup.getOutputPipeNames()) {
                PipeEvent pe = new PipeEvent(name, newGroup,
                        PipeEventType.OUTPUT);
                newGroup.getOutputPipe(name).addCloseEventListener(new CEL(pe));
                eventQueue.add(pe);
            }
        }

        checkIfDone(newGroup);

    }

    public void checkIfDone(IPipeGroup group) {
        boolean cond;

        synchronized (group) {
            cond = (!group.getInputPipeNames().isEmpty()
                    || !group.getOutputPipeNames().isEmpty() || group
                    .missingReciepts());
        }

        if (!cond) {
            removeGroup(group);
        } else {
            //
        }
    }

    public void waitTillAllDone() {
        synchronized (allGroups) {
            try {
                while (!allGroups.isEmpty()) {
                    allGroups.wait();
                }
            } catch (InterruptedException ie) {
                /* ignore it */
            }
        }
    }

    public void run() {
        Thread.currentThread().setName("Pipe Manager 2");
        //Thread.currentThread().setPriority(Thread.NORM_PRIORITY-1);
        while (true) {

            try {
                PipeEvent pe = eventQueue.take();

                switch (pe.getType()) {

                case INPUT:
                    handleInputEvent(pe.getName(), pe.getGroup());
                    checkIfDone(pe.getGroup());
                    break;

                case OUTPUT:
                    handleOutputEvent(pe.getName(), pe.getGroup());
                    checkIfDone(pe.getGroup());
                    break;

                default:
                    throw new AssertionError("Bad event!");
                }

            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }

    }

    void writeOut(IPipeGroup group, IPacket data) {
        toiletManager.writeOut(group,data);
    }

    private void handleInputEvent(String name, IPipeGroup group) {

        synchronized (allGroups) {
            if (!allGroups.contains(group)) {
                return;
            }

            synchronized (group) {
                if (!group.getInputPipeNames().contains(name)) {
                    return;
                }

                IInputPipe pipe = group.getInputPipe(name);

                try {
                    if (pipe == null) {
                        removeGroup(group);
                        return;
                    } else if (pipe.closed()) {
                        if (group.getInputPipeNames().contains(name)) {
                            group.registerClosedInput(name);
                            writeOut(group,new AControlUnit(name,
                                            AControlUnit.CLOSE_OUTPUT_PIPE));
                        }

                    } else if (pipe.ready()) {
                        IType type = pipe.read();

                        long l = group.getNextReciept();

                        writeOut(group,new ADataUnit<IType>(type,
                                        name, l));

                        if (pipe.ready()) {
                            eventQueue.add(new PipeEvent(name, group,
                                    PipeEventType.INPUT));
                        }

                    } else {
                        /* but of a pointless event */
                    }
                } catch (IllegalStateException ise) {
                    //ise.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                group.notifyAll();
            }

        }

    }

    private void handleOutputEvent(String name, IPipeGroup group) {
        synchronized (allGroups) {
            if (!allGroups.contains(group)) {
                return;
            }

            synchronized (group) {
                if (!group.getOutputPipeNames().contains(name)) {
                    return;
                }

                IOutputPipe iop = group.getOutputPipe(name);

                if (iop.closed()) {
                    group.registerClosedOutput(name);
                    writeOut(group,new AControlUnit(name,
                            AControlUnit.CLOSE_INPUT_PIPE));
                } else {
                    /* default event */
                    // throw new AssertionError("How??");
                }

            }

        }
    }

    enum PipeEventType {
        INPUT, OUTPUT
    };

    static class PipeEvent {

        final PipeEventType type;

        final String name;

        final IPipeGroup group;


        public PipeEvent(String name, IPipeGroup group, PipeEventType type) {
            this.type = type;
            this.name = name;
            this.group = group;
        }

        public PipeEventType getType() {
            return type;
        }

        public String getName() {
            return name;
        }

        public IPipeGroup getGroup() {
            return group;
        }

    }

    /*
     * This must be the only way to remove a group from allGroups. It guarantees
     * that whenever a group is removed, both allGroups and the group are
     * notified.
     */
    void removeGroup(IPipeGroup group) {
        synchronized (allGroups) {
            if (!allGroups.contains(group)) {
                return;
            }
            synchronized (group) {

                try {
                    group.getInputStream().close();
                } catch (IOException ioe) {
                    /* ignore it */
                }

                try {
                    group.getOutputStream().close();
                } catch (IOException ioe) {
                    /* ignore it */
                }

                for (String name : group.getInputPipeNames()) {
                    group.getInputPipe(name).close();
                    group.registerClosedInput(name);
                }

                for (String name : group.getOutputPipeNames()) {
                    group.getOutputPipe(name).close();
                    group.registerClosedOutput(name);
                }

                group.notifyAll();
            }

            allGroups.remove(group);
            allGroups.notifyAll();
        }
        
        toiletManager.removeGroup(group);
    }
    
}
