/*******************************************************************************
* 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 05-Nov-2004
 *
 */
package uk.ac.ic.doc.kevlar.api.programs.framework.internal.execute;

import java.util.HashSet;
import java.util.LinkedList;
import java.util.Set;

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.IReadyEventListener;
import uk.ac.ic.doc.kevlar.publicInterfaces.IType;

/**
 * Ok, So I've ripped off Steve's code to make my life easier.
 * 
 * @author Steve Zymler Hacked around a bit by Tris - mainly access modifiers
 *         getting dropped and the removal of an ID.
 * 
 */
public class ClientSidePipe {

    private static final boolean DEBUG_STREAMS = Boolean.parseBoolean(System
            .getProperty("streamdebug", "false"));

    final IInputPipe inputPipe;

    final IOutputPipe outputPipe;

    private boolean inputClosed;

    private boolean outputClosed;

    final LinkedList<IType> objectStream;

    private final Set<ICloseEventListener> inputListeners;

    private final Set<ICloseEventListener> outputListeners;

    private final Set<IReadyEventListener> inputReadyListeners;

    public ClientSidePipe() {
        this.objectStream = new LinkedList<IType>();
        this.inputPipe = new InputPipe();
        this.outputPipe = new OutputPipe();
        this.inputClosed = false;
        this.outputClosed = false;
        this.inputListeners = new HashSet<ICloseEventListener>();
        this.outputListeners = new HashSet<ICloseEventListener>();
        this.inputReadyListeners = new HashSet<IReadyEventListener>();
    }

    public IInputPipe getInputPipe() {
        return inputPipe;
    }

    public IOutputPipe getOutputPipe() {
        return outputPipe;
    }

    protected void addInputListener(ICloseEventListener icel) {
        synchronized (inputListeners) {
            inputListeners.add(icel);
        }
    }

    protected void addOutputListener(ICloseEventListener icel) {
        synchronized (outputListeners) {
            outputListeners.add(icel);
        }
    }

    protected void addInputReadyListener(IReadyEventListener irel) {
        synchronized (inputReadyListeners) {
            inputReadyListeners.add(irel);
        }
    }

    protected void closeInput() {
        synchronized (objectStream) {
            if (!inputClosed) {
                inputClosed = true;
                objectStream.notifyAll();

                tellInputsClosed();

                if (outputClosed()) {
                    tellOutputsClosed();
                }
            }
        }
    }

    /**
     * Tells the output listeners a pipe has been closed
     */
    private void tellOutputsClosed() {
        synchronized (outputListeners) {
            for (final ICloseEventListener icel : new HashSet<ICloseEventListener>(
                    outputListeners)) {
                icel.pipeClosed(outputPipe);
            }
        }
    }

    protected void closeOutput() {
        synchronized (objectStream) {
            if (!outputClosed) {
                outputClosed = true;
                objectStream.addLast(new EOS());
                objectStream.notifyAll();

                tellOutputsClosed();

                if (inputClosed()) {
                    tellInputsClosed();
                }
            }
        }
    }

    /**
     * Tells the input listeners a pipe has been closed.
     */
    private void tellInputsClosed() {
        synchronized (inputListeners) {
            for (final ICloseEventListener icel : new HashSet<ICloseEventListener>(
                    inputListeners)) {
                icel.pipeClosed(inputPipe);
            }
        }
    }

    protected boolean inputClosed() {
        synchronized (objectStream) {
            return inputClosed
                    || (outputClosed && (objectStream.getFirst() instanceof EOS));
        }
    }

    protected boolean outputClosed() {
        synchronized (objectStream) {
            return (inputClosed || outputClosed);
        }
    }

    protected void write(IType object) {
        synchronized (objectStream) {

            if (outputClosed()) {
                throw new IllegalStateException("IOutputPipe Closed on Write");
            }

            if (objectStream.isEmpty()) {
                synchronized (inputReadyListeners) {
                    for (final IReadyEventListener irel : new HashSet<IReadyEventListener>(
                            inputReadyListeners)) {
                        irel.pipeReadyEvent(inputPipe);
                    }
                }
            }

            if (DEBUG_STREAMS) {
                System.err.println("/" + System.currentTimeMillis()
                        + "/ Writing out " + object);
            }

            objectStream.addLast(object);
            objectStream.notifyAll();
        }
    }

    protected IType read() throws InterruptedException {
        synchronized (objectStream) {

            while (objectStream.isEmpty()) {
                if (inputClosed()) {
                    throw new IllegalStateException("IInputPipe Closed on Read");
                }
                objectStream.wait();
            }

            if (inputClosed()) {
                throw new IllegalStateException("IInputPipe Closed on Read");
            }

            objectStream.notifyAll();

            IType o = objectStream.removeFirst();

            if (inputClosed()) {
                closeInput();
            }

            if (DEBUG_STREAMS) {
                System.err.println("/" + System.currentTimeMillis()
                        + "/ Reading " + o);
            }

            return o;
        }
    }

    protected boolean ready() {
        synchronized (objectStream) {
            if (inputClosed()) {
                throw new IllegalStateException("IInputPipe Closed on Ready");
            }
            return !objectStream.isEmpty();
        }
    }

    private class InputPipe implements IInputPipe {
        public IType read() throws InterruptedException {
            synchronized (objectStream) {
                return ClientSidePipe.this.read();
            }
        }

        public boolean ready() {
            synchronized (objectStream) {
                return ClientSidePipe.this.ready();
            }
        }

        public void close() {
            synchronized (objectStream) {
                ClientSidePipe.this.closeInput();
            }
        }

        public boolean closed() {
            synchronized (objectStream) {
                return ClientSidePipe.this.inputClosed();
            }
        }

        public void addCloseEventListener(ICloseEventListener icel) {
            addInputListener(icel);
        }

        public void addReadyEventListener(IReadyEventListener irel) {
            addInputReadyListener(irel);
        }
    }

    private class OutputPipe implements IOutputPipe {
        public void write(IType obj) {
            synchronized (objectStream) {
                ClientSidePipe.this.write(obj);
            }
        }

        public void close() {
            synchronized (objectStream) {
                ClientSidePipe.this.closeOutput();
            }
        }

        public boolean closed() {
            synchronized (objectStream) {
                return ClientSidePipe.this.outputClosed();
            }
        }

        public void addCloseEventListener(ICloseEventListener icel) {
            addOutputListener(icel);
        }
    }

    private class EOS implements IType {
        /**
         * Comment for <code>serialVersionUID</code>
         */
        private static final long serialVersionUID = 1L;

        public Object clone() {
            return new EOS();
        }
    }
}
