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

import java.io.BufferedOutputStream;
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.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;

import uk.ac.ic.doc.kevlar.publicInterfaces.IInputPipe;
import uk.ac.ic.doc.kevlar.publicInterfaces.IOutputPipe;
import uk.ac.ic.doc.kevlar.publicUtil.system.ObjectInputStreamWithACustomClassLoader;

/**
 * Implementation of the IPipeGroup interface.
 * 
 * @author toa02
 *  
 */
final class PipeGroup implements IPipeGroup {

    private static final Random rndGen = new Random();
    
    private final int hashCode = rndGen.nextInt();
    
    private final ObjectInputStream ois;
    private final ObjectOutputStream oos;

    private final Set<String> inputNames;
    private final Set<String> outputNames;
    
    private final Map<String,IInputPipe> ins;
    private final Map<String,IOutputPipe> outs;
    
    private final InputStream debug;
    
    private final AtomicLong nextReciept;
    private final Set<Long> missingReciepts;
    
    PipeGroup( InputStream out, OutputStream in,
            Map<String, IInputPipe> ins,  Map<String, IOutputPipe> outs,
            InputStream debug, Serializable initialData, ClassLoader cl) throws IOException {
    
        this.nextReciept = new AtomicLong(0);
        this.missingReciepts = new HashSet<Long>();
        
        this.ins = ins;
        this.outs = outs;
        
        this.inputNames = Collections.synchronizedSet(this.ins.keySet());
        this.outputNames = Collections.synchronizedSet(this.outs.keySet());
        
        this.debug = debug;
        
        if( in instanceof ObjectOutputStream ) {
            oos = (ObjectOutputStream) in;
        } else {
            oos = new ObjectOutputStream( new BufferedOutputStream( in ) );
            oos.flush();
        }
        
        if( initialData != null ) {       
            oos.writeObject(initialData);
            oos.flush();
        }
        
        if( out instanceof ObjectInputStream ) {
            ois = (ObjectInputStream) out;
        } else if ( cl == null ){
            //ois = new ObjectInputStream( new BufferedInputStream( out) );
            ois = new ObjectInputStream( out );
        } else {
            //ois = new ObjectInputStreamWithACustomClassLoader( new BufferedInputStream( out), cl );
            ois = new ObjectInputStreamWithACustomClassLoader( out, cl );
        }
    }


    
    /**
     * @see uk.ac.ic.doc.kevlar.api.programs.framework.internal.IPipeGroup#getInputStream()
     */
    public ObjectInputStream getInputStream() {
        return ois;
    }

    
    /**
     * @see uk.ac.ic.doc.kevlar.api.programs.framework.internal.IPipeGroup#getOutputStream()
     */
    public ObjectOutputStream getOutputStream() {
        return oos;
    }

    /**
     * @see uk.ac.ic.doc.kevlar.api.programs.framework.internal.IPipeGroup#getInputPipeNames()
     */
    public Set<String> getInputPipeNames() {
        return new HashSet<String>(inputNames);
    }

    /**
     * @see uk.ac.ic.doc.kevlar.api.programs.framework.internal.IPipeGroup#getOutputPipeNames()
     */
    public Set<String> getOutputPipeNames() {
        return new HashSet<String>(outputNames);
    }

    /**
     * @see uk.ac.ic.doc.kevlar.api.programs.framework.internal.IPipeGroup#getInputPipe(java.lang.String)
     */
    public IInputPipe getInputPipe(String name) {
        return ins.get(name);
    }

    /**
     * @see uk.ac.ic.doc.kevlar.api.programs.framework.internal.IPipeGroup#getOutputPipe(java.lang.String)
     */
    public IOutputPipe getOutputPipe(String name) {
        return outs.get(name);
    }

    /**
     * @see uk.ac.ic.doc.kevlar.api.programs.framework.internal.IPipeGroup#getDebugStream()
     */
    public InputStream getDebugStream() {
        return debug;
    }


    /**
     * @see uk.ac.ic.doc.kevlar.api.programs.framework.internal.IPipeGroup#registerClosedInput(java.lang.String)
     */
    public void registerClosedInput(String name) {
        inputNames.remove(name);
    }
    
    public long getNextReciept() {
        synchronized ( missingReciepts ) {
            long l = nextReciept.incrementAndGet();
            missingReciepts.add(Long.valueOf(l));
            return l;
        }
    }

    public void registerReciept( Long value ) {
        synchronized ( missingReciepts ) {
            missingReciepts.remove(value);    
        }
        
    }
    
    public boolean missingReciepts() {
        synchronized ( missingReciepts ) {
            return !missingReciepts.isEmpty();
        }
    }

    /**
     * @see uk.ac.ic.doc.kevlar.api.programs.framework.internal.IPipeGroup#registerClosedOutput(java.lang.String)
     */
    public void registerClosedOutput(String name) {
        outputNames.remove(name);
    }
    
    public int hashCode() {
        return hashCode;
    }

}
