/*
 * Copyright (c) 2009-2010 The Regents of the University of California.
 * All rights reserved.
 *
 * '$Author: crawl $'
 * '$Date: 2015-06-17 21:53:43 +0000 (Wed, 17 Jun 2015) $' 
 * '$Revision: 33482 $'
 * 
 * Permission is hereby granted, without written agreement and without
 * license or royalty fees, to use, copy, modify, and distribute this
 * software and its documentation for any purpose, provided that the above
 * copyright notice and the following two paragraphs appear in all copies
 * of this software.
 *
 * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY
 * FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES
 * ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF
 * THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE
 * PROVIDED HEREUNDER IS ON AN "AS IS" BASIS, AND THE UNIVERSITY OF
 * CALIFORNIA HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES,
 * ENHANCEMENTS, OR MODIFICATIONS.
 *
 */

package org.kepler.provenance;

import java.util.Arrays;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Vector;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import ptolemy.actor.IOPort;
import ptolemy.actor.IOPortEvent;
import ptolemy.actor.Receiver;
import ptolemy.kernel.util.IllegalActionException;

/** A utility class to map tokens from output to input ports.
 *
 * @author Daniel Crawl
 * @version $Id: PortConnector.java 33482 2015-06-17 21:53:43Z crawl $
 *
 */
public class PortConnector<Id>
{

    /** Construct a new PortConnector. */
    public PortConnector()
    {
        _inputPortQueueMap = new HashMap<IOPort,Queue<Id>[]>();
        _lastIdMap = new HashMap<IOPort, Vector<Id>>();
    }

    /** Clear all connections. */
    public void clear() {
        _inputPortQueueMap.clear();
        _lastIdMap.clear();
    }
    
    /** Create connections for a port. */
    public void createConnections(IOPort port) throws RecordingException
    {
        int width = -1;
        try
        {
            width = port.getWidth();
        }
        catch(IllegalActionException e)
        {
            throw new RecordingException("Error getting port width: ", e);
        }

        // see if port is already in table
        Queue<Id>[] queues = _inputPortQueueMap.get(port);
        Vector<Id> lastIds = _lastIdMap.get(port);
        if(queues != null)
        {
            
            // clear any values that were left by the previous execution.
            // this can happen, e.g., if the execution is stopped by the
            // user, an error occurred, etc.
            for(Queue<Id> queue: queues) {
                queue.clear();
            }
            
            for(int i = 0; i < lastIds.size(); i++) {
                lastIds.set(i, null);
            }
            
            // see if the allocated queue array is too small
            if(width > queues.length)
            {
                // increase the queue array to be the current width
                // of the port and copy the current queues to the
                // new array.
                Queue<Id>[] newQueues = new ConcurrentLinkedQueue[width];
                for(int i = 0; i < width; i++)
                {
                    if(i < queues.length)
                    {
                        newQueues[i] = queues[i];
                    }
                    else
                    {
                        newQueues[i] = new ConcurrentLinkedQueue<Id>();

                        // increase the vector length by one for the
                        // additional channel
                        lastIds.add(null);
                    }
                }

                _inputPortQueueMap.put(port, newQueues);
            }
        }
        // register a new port
        // NOTE: port may be output port in composite
        else if(port.isOpaque())
        {
            queues = new ConcurrentLinkedQueue[width];
            lastIds = new Vector<Id>();
            for(int i = 0; i < width; i++)
            {
                queues[i] = new ConcurrentLinkedQueue<Id>();
                //_debug("creating queue, channel " + i + ", for port " +
                // port.getFullName() + " = " + queue[i].hashCode());
                
                // increase the vector length by one so the total
                // size will be the port width.
                lastIds.add(null);
            }
            _inputPortQueueMap.put(port, queues);
            _lastIdMap.put(port, lastIds);
        }        
    }

    /** Get the next Id for an input port and channel. */
    public Id getNextId(IOPort port, int channel)
        throws RecordingException
    {
        
        //System.out.println("getNextId for " + port.getFullName() + " " + channel);
        
        Queue<Id> queue = _getQueueForPortAndChannel(port, channel);            
        Id retval = queue.poll();
        if(retval == null)
        {   
            throw new RecordingException("No write event for token " +
                "received by input port " + port.getFullName() +
                " on channel " + channel);
        }
        
        
        // save the Id for this port and channel in case it is refilled.
        Vector<Id> lastIds = _lastIdMap.get(port);
        lastIds.set(channel, retval);
        
        if(_isDebugging)
        {
            _log.debug("in port " + port.getFullName() + " pass = " + retval);
        }

        return retval;
    }

    /** Restore the previous Id to a port and channel. */
    public void refillId(IOPort port, int channel) throws RecordingException
    {
        Vector<Id> lastIds = _lastIdMap.get(port);
        if(lastIds == null)
        {
            throw new RecordingException("No last id map for port: " +
                port.getFullName());
        }
        
        if(channel >= lastIds.size())
        {
            throw new RecordingException("Invalid channel for port " + 
                port.getFullName() + ": " + channel);
        }
        
        if(lastIds.get(channel) == null)
        {
            throw new RecordingException("No previous data in port " +
                port.getFullName() + " channel " + channel);
        }
        
        // since we can't place the restored Id at the front of the
        // queue, create a new queue with the restored Id, and then
        // add all the existing queued elements at the end.
        // XXX this is inefficient
        Queue<Id> queue = _inputPortQueueMap.get(port)[channel];
        synchronized(queue)
        {
            Queue<Id> queueCopy = new ConcurrentLinkedQueue<Id>(queue);
            queue.clear();
            queue.add(lastIds.get(channel));
            queue.addAll(queueCopy);
        }
    }
    
    /** A port should receive an id on a specific channel. This method should
     *  be used when a token is manually placed inside a receiver instead of
     *  when sent with one of the IOPort.send methods.
     */
    public void receiveId(IOPort receiverPort, int channel, Id pass)
        throws RecordingException
    {
        Queue<Id> queue = _getQueueForPortAndChannel(receiverPort, channel);
        queue.add(pass);
    }
    
    /** Send an Id to an output port's connections on a channel.
     *  NOTE: the receivers must be created in the actors, otherwise
     *  this method may fail to find connected ports. 
     */
    public void sendIdToConnections(IOPort senderPort, int channel, Id pass)
        throws RecordingException
    {
        //System.out.println("send port " + senderPort.getFullName() +
                //" " + channel);// +
                //" pass = " + pass);

        try
        {
            Receiver[][] allReceivers;
           
            // see if this an output port
            if(senderPort.isOutput())
            {
                //_debug("is output");
                
                // FIXME: the javadoc for getRemoteReceivers() implies
                // that remote receivers will be created if missing, but
                // this does not always seem to be the case.
                allReceivers = senderPort.getRemoteReceivers();
            }
            // this is not an output port, so in order to generate
            // IOPortEvents that are writes, the port must be an
            // opaque input port on a non-atomic entity.
            else
            {
                //_debug("is input");
                allReceivers = senderPort.deepGetReceivers();
            }

            // see if the port is connected
            if(allReceivers.length > 0)
            {
                //_debug("has receivers, channel = " + channel);
                List<Receiver> receiverList = new LinkedList<Receiver>();

                // see if port write was broadcast
                if(channel == IOPortEvent.ALLCHANNELS)
                {
                    // add all receivers on all channels to our receiver list
                    for(Receiver[] receiverArray : allReceivers)
                    {
                        // make sure there are receivers on this channel
                        if(receiverArray != null)
                        {
                            receiverList.addAll(Arrays.asList(receiverArray));
                        }
                    }
                }
                else
                {
                    // we only care about a specific channel

                    // make sure there are receivers on this channel
                    if(allReceivers[channel] != null)
                    {
                        receiverList = Arrays.asList(allReceivers[channel]);
                    }
                }

                for(Receiver curReceiver : receiverList)
                {
                    //System.out.println("  receiver " + curReceiver);

                    // get the port and channel number of this receiver
                    IOPort recvPort = curReceiver.getContainer();

                    //System.out.println("port " + recvPort);

                    int recvChannel =
                        recvPort.getChannelForReceiver(curReceiver);

                    Queue<Id>[] queueArray = _inputPortQueueMap.get(recvPort);

                    // see if queue array was not created or is too small.
                    // this can happen if the port or channel was created
                    // after the provenance recorder records the workflow
                    // contents. for example the multi instance composite
                    // actor creates its clones in preinitialize().
                    if(queueArray == null || queueArray.length <= recvChannel)
                    { 
                        // create the queue array or make it bigger.
                        createConnections(recvPort);
                    }
                    
                    queueArray = _inputPortQueueMap.get(recvPort);

                    if(queueArray == null)
                    { 
                        throw new RecordingException(
                            "Unable to find queue array for " +
                            recvPort.getFullName());
                    }
                    else if(queueArray.length <= recvChannel)
                    {
                        throw new RecordingException(
                            "Invalid channel index for queue for " +
                            recvPort.getFullName() + ": " + recvChannel);
                    }

                    queueArray[recvChannel].add(pass);
        
                    if(_isDebugging)
                    {
                        _log.debug("    to rcvr " + recvPort.getFullName());
                    }

                    //_debug("added to queue for " + recvPort.getFullName()
                    // + " channel " + recvChannel + " q " +
                    // queueArray[recvChannel].hashCode());
                }
            }
        }
        catch(IllegalActionException e)
        {
            throw new RecordingException("Error getting receivers for " +
                senderPort.getFullName() + ": ", e);
        }
    }
    
    /** Return a queue for a port and channel. */
    private Queue<Id> _getQueueForPortAndChannel(IOPort port, int channel)
        throws RecordingException
    {
        Queue<Id>[] queue = _inputPortQueueMap.get(port);
        if(queue == null)
        {
            throw new RecordingException("No event queue for input port: " +
                port.getFullName());
        }   
                
        //_debug("reading queue for  " + port.getFullName() + " channel " +
        // channel + " polling q " + queue[channel].hashCode());

        if(channel >= queue.length)
        {
            throw new RecordingException("Invalid channel for port " +
                port.getFullName() + ": " + channel);
        }
        
        return queue[channel];
    }
    
    ///////////////////////////////////////////////////////////////////////
    //// private fields                                                ////

    /** A mapping of input port to an array of queues; the length of the
     *  array is the width of the port.
     */
    private Map<IOPort,Queue<Id>[]> _inputPortQueueMap;
    
    /** A mapping of port to a vector of ids; used to restore the last Id
     *  read from a port when a port refill event occurs. A vector of Ids
     *  is used instead of an array since java does not permit the creation
     *  of arrays of generics.
     */
    private Map<IOPort, Vector<Id>> _lastIdMap;
        
    private static final Log _log = LogFactory.getLog(PortConnector.class.getName());
        private static final boolean _isDebugging = _log.isDebugEnabled();
}