/*******************************************************************************
* 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 04-Jan-2005 by toa02
 *
 */
package uk.ac.ic.doc.kevlar.core.programs.coreUtils;

import java.util.HashSet;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;

import uk.ac.ic.doc.kevlar.api.programs.arguments.IArgumentInstance;
import uk.ac.ic.doc.kevlar.publicInterfaces.AbstractProgram;
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 toa02
 *
 */
public class Multiplexer {

    public void run(IArgumentInstance args,
            final Map<String, IInputPipe> inputs,
            final Map<String, IOutputPipe> outputs) {

        final Queue<IInputPipe> ready = new LinkedList<IInputPipe>();

        if (inputs.isEmpty() || outputs.isEmpty()) {
           // System.err.println("**They are empty**");
            return;
        }

        class ReadyHandler implements IReadyEventListener {

            public ReadyHandler(IInputPipe source) {
                pipeReadyEvent(source);
            }

            public void pipeReadyEvent(IInputPipe source) {
                //System.err.println("***A Pipe is ready");
                synchronized (ready) {
                    ready.add(source);
                    ready.notifyAll();
                }
                //System.err.println("/A Pipe is ready***");
            }

        }

        class CloseListener implements ICloseEventListener {

            final String keyName;

            public CloseListener(String keyName) {
                this.keyName = keyName;
            }

            public void pipeClosed(IPipeEnd source) {
               // System.err.println("***source closed " + source);

                if (source instanceof IInputPipe) {
                    synchronized (ready) {
                        inputs.remove(keyName);
                        ready.notifyAll();
                    }

                } else if (source instanceof IOutputPipe) {
                    synchronized (ready) {
                        outputs.remove(keyName);
                        ready.notifyAll();
                    }

                } else {
                    throw new AssertionError("No idea");
                }
                //System.err.println(":***");
            }

        }

        // add events to inputs
        synchronized (ready) {
            for (String key : new HashSet<String>(inputs.keySet())) {
                IInputPipe pipe = inputs.get(key);
                pipe.addReadyEventListener(new ReadyHandler(inputs.get(key)));
                pipe.addCloseEventListener(new CloseListener(key));

                if (pipe.closed()) {
                    inputs.remove(key);
                } else if (pipe.ready()) {
                    ready.add(pipe);
                }
            }

            for (String key : new HashSet<String>(outputs.keySet())) {
                IOutputPipe pipe = outputs.get(key);
                pipe.addCloseEventListener(new CloseListener(key));

                if (pipe.closed()) {
                    outputs.remove(key);
                }
            }
        }

        // Main loop
        try {
            while (true) {

                if (inputs.isEmpty() || outputs.isEmpty()) {
                    System.err.println("[Junction] Returning");
                    return;
                }

                IInputPipe me;
                
                synchronized (ready) {

                    while (ready.isEmpty()) {
                        //System.err.println("[Junction] Waiting");
                        ready.wait();
                        //System.err.println("[Junction] Woken");
                        if (inputs.isEmpty() || outputs.isEmpty()) {
                            System.err.println("[Junction] Returning");
                            return;
                        }
                    }

                    me = ready.remove();

                }
                
                try {
                    writeFromPipe(outputs, me);
                } catch (IllegalStateException ise) {
                    me.close();
                }

            }
        } catch (InterruptedException e) {
            // TODO: tora Auto-generated catch block
            e.printStackTrace();
        }
    }

    void writeFromPipe(final Map<String, IOutputPipe> outputs, IInputPipe source) {
        while (source.ready()) {
            IType type = null;
            try {
                type = source.read();
            } catch (InterruptedException ie) {
                /* ignore it */
            }

            if (type == null) {
                return;
            }

            for (String key : outputs.keySet()) {
                try {
                    outputs.get(key).write(type);
                } catch (IllegalStateException ise) {
                    ise.printStackTrace();
                    /* ignore it */
                }
            }

        }
    }
}
