/*
 * Copyright (c) 2007-2010 The Regents of the University of California.
 * All rights reserved.
 *
 * '$Author: crawl $'
 * '$Date: 2015-12-17 17:59:53 +0000 (Thu, 17 Dec 2015) $' 
 * '$Revision: 34345 $'
 * 
 * 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.sql;

import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.Collections;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.WeakHashMap;

import org.kepler.provenance.FireState;
import org.kepler.provenance.Recording;
import org.kepler.provenance.RecordingException;
import org.kepler.provenance.RecordingParameters;
import org.kepler.util.sql.DatabaseFactory;
import org.kepler.util.sql.DatabaseType;
import org.kepler.util.sql.Schema;

import ptolemy.actor.Actor;
import ptolemy.actor.Director;
import ptolemy.actor.FiringEvent;
import ptolemy.actor.IOPort;
import ptolemy.actor.IOPortEvent;
import ptolemy.actor.IORelation;
import ptolemy.actor.LocalClock;
import ptolemy.actor.TypedIOPort;
import ptolemy.actor.gui.ColorAttribute;
import ptolemy.actor.gui.SizeAttribute;
import ptolemy.actor.gui.WindowPropertiesAttribute;
import ptolemy.actor.gui.style.LineStyle;
import ptolemy.actor.gui.style.TextStyle;
import ptolemy.actor.parameters.ParameterPort;
import ptolemy.actor.parameters.PortParameter;
import ptolemy.actor.sched.Scheduler;
import ptolemy.data.StringToken;
import ptolemy.data.Token;
import ptolemy.data.expr.Parameter;
import ptolemy.domains.sdf.kernel.SDFDirector;
import ptolemy.kernel.undo.UndoStackAttribute;
import ptolemy.kernel.util.AbstractSettableAttribute;
import ptolemy.kernel.util.Attribute;
import ptolemy.kernel.util.ConfigurableAttribute;
import ptolemy.kernel.util.IllegalActionException;
import ptolemy.kernel.util.Location;
import ptolemy.kernel.util.NameDuplicationException;
import ptolemy.kernel.util.Nameable;
import ptolemy.kernel.util.NamedObj;
import ptolemy.vergil.basic.KeplerDocumentationAttribute;
import ptolemy.vergil.icon.BoxedValueIcon;
import ptolemy.vergil.icon.EditIconTableau;
import ptolemy.vergil.icon.EditorIcon;
import ptolemy.vergil.icon.TextIcon;
import ptolemy.vergil.icon.UpdatedValueIcon;
import ptolemy.vergil.icon.ValueIcon;
import ptolemy.vergil.icon.XMLIcon;


/** SQL Implementation of Recording using SDM SPA schema.
 *
 * @author Daniel Crawl
 * @version $Id: SQLRecording.java 34345 2015-12-17 17:59:53Z crawl $
 *
 */
    
public class SQLRecording extends Recording
{
    /** Construct a new SQLRecording. */
    public SQLRecording() throws RecordingException
    {
        super();

        _dbType = null;

        _schema = null;

        // NOTE: we use a LinkedHashMap instead of a HashMap for predictable
        // iteration order (the order in which keys are inserted into it).
        // The map is iterated at the end of workflow execution to record
        // the stop of rw-firings and we want a repeatable ordering for
        // tests.
        _fireStateTable = //Collections.synchronizedMap(
            new LinkedHashMap<Actor, FireState<Integer>>();//);

        _entityCacheTable = Collections.synchronizedMap(
            new WeakHashMap<Nameable, RegEntity>());

        _dbReset();
    }

    /** Stop recording. Called when the Recording type changes
     *  or ProvenanceListener removed from canvas. 
     *  NOTE: this is NOT called when the GUI exits or when the
     *  workflow runs from the command line.
     *  
     *  This is the last method called to a Recording instance.
     */
    @Override
    public void disconnect() throws RecordingException
    {
        _dbReset();
    } 
    
    // Specification
    
    /** Register an actor. */
    @Override
    public boolean regActor(Actor actor) throws RecordingException
    {
        //_debug("regActor(" + _getNameableFullName(actor) + ")");

        int id = _regActorDirector((NamedObj)actor, true);

        FireState<Integer> fs = new FireState<Integer>(actor, id);
        _fireStateTable.put(actor, fs);

        return true;
    }

    /** Register a director. */
    @Override
    public boolean regDirector(Director director) throws RecordingException
    {
        _regActorDirector(director, false);
        return true; 
    }

    /** Register a parameter. A parameter can be any <b>entity</b>
     * stored in the MoML that does not have its own
     * <code>regNNN()</code> method. This can be user-level 
     * parameters (e.g., Parameter, StringParameter, etc.) or
     * internal to Kepler (e.g., _location, semanticType000, etc.).
     * (A "parameter" corresponds to a property in the MoML).
     *
     */
    @Override
    public boolean regParameter(NamedObj parameter) throws RecordingException
    {
        boolean ignore = false;

        String name = parameter.getName();

        if(parameter instanceof WindowPropertiesAttribute ||
            parameter instanceof ConfigurableAttribute ||
            parameter instanceof Location ||
            parameter instanceof KeplerDocumentationAttribute ||
            parameter instanceof SizeAttribute ||
            parameter instanceof ColorAttribute ||
            parameter instanceof TextIcon ||
            parameter instanceof EditIconTableau.Factory ||
            parameter instanceof TextStyle ||
            parameter instanceof ValueIcon ||
            parameter instanceof XMLIcon ||
            parameter instanceof LineStyle ||
            parameter instanceof UndoStackAttribute ||
            parameter instanceof BoxedValueIcon ||
            parameter instanceof UpdatedValueIcon  ||
            parameter instanceof EditorIcon ||
            parameter instanceof org.kepler.moml.NamedObjIdReferralList ||
            parameter instanceof LocalClock ||
            parameter instanceof Scheduler ||
            ((parameter.getContainer() instanceof SDFDirector) &&
                        (name.equals(SDFDirector.AUTO_NAME) || name.equals(SDFDirector.UNBOUNDED_NAME))) ||
            ((parameter.getContainer() instanceof Director) &&
                        (name.equals("startTime") || name.equals("stopTime"))) ||                       
            name.equals("bold") ||
            name.equals("italic") ||
            name.equals("fontFamily") ||
            name.equals("textSize") ||
            name.equals("_notDraggable") ||
            name.indexOf("_vergil") == 0 ||
            name.indexOf("_hide") == 0 ||
            name.equals("_showName"))
        {
            ignore = true;
        }
        else
        {
            RegEntity re;
            boolean isPort = false;

            // if it's a PortParameter, we'll need to register both the
            // port and parameter.
            if(parameter instanceof PortParameter)
            {
                re = _checkEntity(parameter, RegEntity.EntityType.PortParameter);
                isPort = true;
            }
            else
            {
                re = _checkEntity(parameter, RegEntity.EntityType.Parameter);
            }

            if(re.isNew())
            {
                _regParameterReal(parameter, re);

                if(isPort)
                {
                    _regPortReal(((PortParameter)parameter).getPort(), re);
                }
            }
        }

        return (! ignore);
    }

    /** Register a link between two endpoints.  */
    @Override
    public boolean regLink(NamedObj endPoint1, NamedObj endPoint2)
        throws RecordingException
    {
      
        /*
         _debug("regLink(" + _getNameableFullName(endPoint1) + ", " + 
            _getNameableFullName(endPoint2) + ")");
        */

        // make sure both endpoints have been registered. 
        RegEntity.EntityType type = RegEntity.EntityType.getType(endPoint1);
        RegEntity re1 = _checkEntity(endPoint1, type);
        if(re1 == null)
        {
            throw new RecordingException("endPoint1 is not registered.");
        }
        
        type = RegEntity.EntityType.getType(endPoint2);
        RegEntity re2 = _checkEntity(endPoint2, type);
        if(re2 == null)
        {
            throw new RecordingException("endPoint2 is not registered.");
        }

        Link link = new Link(endPoint1, endPoint2);
        
        //_debug("linkname = " + linkName);

        RegEntity reLink = _checkEntity(link, RegEntity.EntityType.Link);
        if(reLink.isNew())
        {
            try
            {
                synchronized(_psLinkInsert)
                {
                    _psLinkInsert.setInt(1, reLink.getId());
                    _psLinkInsert.setInt(2, re1.getId());
                    _psLinkInsert.setInt(3, re2.getId());
                    _psLinkInsert.executeUpdate();
                }

                if(_debugWriter != null)
                {
                    _debugWrite("INSERT INTO LINK (" + reLink.getId() +
                        ", " + re1.getId() + ", " + re2.getId() + ")");
                }
            }
            catch(SQLException e)
            {
                throw new RecordingException(_getExceptionMessage(e));
            }
        }
        /*
        else
        {
            _debug("link already recorded: " + linkName);
        }
        */

        return true; 

    }

    /** Register a port or portparameter.  */
    @Override
    public boolean regPort(TypedIOPort port) throws RecordingException
    {
        //_debug("regPort(" + _getNameableFullName(port) + ")");
        //

        RegEntity re;
        boolean isParameter = false;
        
        // if it's a ParameterPort, we'll need to register both the
        // port and parameter.
        if(port instanceof ParameterPort)
        {
            re = _checkEntity(port, RegEntity.EntityType.PortParameter);
            isParameter = true;
        }
        else
        {
            re = _checkEntity(port, RegEntity.EntityType.Port);
        }

        if(re.isNew())
        {
            _regPortReal(port, re);

            if(isParameter)
            {
                _regParameterReal(((ParameterPort)port).getParameter(), re);
            }
        }
        
        return true;
    }
   
    /** Register a relation. */
    @Override
    public boolean regRelation(IORelation relation) throws RecordingException
    {
        //_debug("regRelation(" + _getNameableFullName(relation) + ")");

        RegEntity re = _checkEntity(relation, RegEntity.EntityType.Relation);
        if(re.isNew())
        {
            try
            {
                int width = relation.getWidth();
                synchronized(_psRelationInsert)
                {
                    _psRelationInsert.setInt(1, re.getId());
                    _psRelationInsert.setInt(2, width);
                    _psRelationInsert.executeUpdate();
                }
                if(_debugWriter != null)
                {
                    _debugWrite("INSERT INTO RELATION (id, " +
                        width + ")");
                }
            }
            catch(SQLException e)
            {
                throw new RecordingException(_getExceptionMessage(e));
            }
            catch(IllegalActionException e)
            {
                throw new RecordingException(_getExceptionMessage(e));
            }
        }
        
        return true; 
    }

    /** Called before registering workflow contents. */
    @Override
    public void specificationStart() throws RecordingException
    {
        super.specificationStart();

        // clear the entity cache table so that entity renames are handled
        // FIXME this will cause performance problems on large workflows
        _entityCacheTable.clear();
        
        // always clear the fire state table since the id of actors
        // can change between workflow executions due to the actor
        // being renamed.
        _fireStateTable.clear();
    }

  
    ////////////////////////////////////////////////////////////////////////
    //// Execution interface                                            ////
    
    /** Record the starting of workflow execution. */
    @Override
    public void executionStart(Date timestamp) throws RecordingException
    {
        if(_wfExecId != RegEntity.UNKNOWN_ID)
        {
            throw new RecordingException("Workflow already running.");
        }
            
        // NOTE: do not set this to false so that workflow contents
        // will be registered again. this handles cases where entities
        // are renamed between runs.
        //_needWorkflowContents(false);

        if(_wfUserStr == null)
        {
            throw new RecordingException("Need workflow user name");
        }
        
        // update the db
        try
        {
            synchronized(_psWorkflowExecStart)
            {
                _psWorkflowExecStart.setInt(1, _wfId);
                _psWorkflowExecStart.setString(2, _wfUserStr);
                _psWorkflowExecStart.setTimestamp(3, new Timestamp(timestamp.getTime()));
                _wfExecId = _dbType.insert(_psWorkflowExecStart,
                    "workflow_exec", "id");
            }

            if(_debugWriter != null)
            {
                _debugWrite("INSERT INTO WORKFLOW_EXEC (" + _wfId +
                    ", " + _wfUserStr + ", curTime)");
            }
        }
        catch(SQLException e)
        {
            throw new RecordingException(_getExceptionMessage(e));
        }
    }

    /** Record the stopping of workflow execution. */
    @Override
    public void executionStop(Date timestamp) throws RecordingException
    {
        if(_wfExecId == RegEntity.UNKNOWN_ID)
        {
            throw new RecordingException("Workflow not running.");
        }
        try
        {
            // stop any actors that may be rw firing
            for(FireState<Integer> fs : _fireStateTable.values())
            {
                synchronized(fs)
                {
                    Integer id = fs.getCurFireId(true);
                    if(id != null)
                    {
                        _recordFiringEvent(fs.getActor(), 
                            FiringEvent.AFTER_RW_FIRE, timestamp);
                        // reset the port state
                        // XXX what about port events between wf execs?
                        fs.setPortLastAccess(
                            FireState.PortAccessType.Uninitialized);
                    }

                    id = fs.getCurFireId(false);
                    if(id != null)
                    {
                        _recordFiringEvent(fs.getActor(),
                            fs.getNextStopFiringType(), timestamp);
                    }
                }
            }

            // 
            synchronized(_psWorkflowExecStop)
            {
                _psWorkflowExecStop.setTimestamp(1, new Timestamp(timestamp.getTime()));
                _psWorkflowExecStop.setInt(2, _wfExecId);
                _psWorkflowExecStop.executeUpdate();
            }

            if(_debugWriter != null)
            {
                _debugWrite("UPDATE WORKFLOW_EXEC SET end_time = " +
                    "curTime WHERE id = " + _wfExecId + ")");
            }

            _wfExecId = RegEntity.UNKNOWN_ID;
        }
        catch(SQLException e)
        {
            throw new RecordingException(_getExceptionMessage(e));
        }
    }
    
    /** An execution was imported. */
    @Override
    public void executionImported()
    {
        
    }
    
    /** Record an actor fire event. */
    @Override
    public void actorFire(FiringEvent event, Date timestamp) throws RecordingException
    {
        //_debug("actorFire: " + event);
        _recordFiringEvent(event.getActor(), event.getType(), timestamp);
    }

    /** Record a port event. */
    @Override
    public void portEvent(IOPortEvent event, Date timestamp) throws RecordingException
    {
        IOPort port = (TypedIOPort)event.getPort();

        int type = event.getEventType();
        boolean isRead = false;
        boolean recordEvent = false;

        switch(type)
        {
            case IOPortEvent.SEND_BEGIN:
                isRead = false;
                recordEvent = true;
                break;
            case IOPortEvent.GET_END:
                isRead = true;
                recordEvent = true;
                break;
            // ignore get_begin and send_end
            case IOPortEvent.GET_BEGIN:
            case IOPortEvent.SEND_END:
                break;
            // warn if received unknown type
            default:
                _warn("unhandled type of IOPortEvent: " + type);
                break;
        }

        Actor actor = (Actor)port.getContainer();
        FireState<Integer> fs = _fireStateTable.get(actor);

        // XXX check for null
        synchronized(fs)
        {
            // see if we need to generate rw fire event(s)
            FireState.PortAccessType lastAccess = fs.getPortLastAccess();
            if(lastAccess == FireState.PortAccessType.Uninitialized)
            {
                _recordFiringEvent(actor, FiringEvent.BEFORE_RW_FIRE, timestamp);
            }
            else if(lastAccess == FireState.PortAccessType.Write && isRead)
            {
                _recordFiringEvent(actor, FiringEvent.AFTER_RW_FIRE, timestamp);
                _recordFiringEvent(actor, FiringEvent.BEFORE_RW_FIRE, timestamp);
            }

            // record the type of last port access for this actor
            if(isRead)
            {
                fs.setPortLastAccess(FireState.PortAccessType.Read);
            }
            else
            {
                fs.setPortLastAccess(FireState.PortAccessType.Write);
            }

            if(recordEvent)
            {
                /*
                 _Debug("port " + _getNameableFullName(port) + 
                    (isRead ? " read " : " write ") +
                    (port.isOpaque() ? "opaque " : "transparent "));
                */
                //_Debug("type " + port.getType());

                Integer fireId = fs.getCurFireId(false);
                if(fireId == null)
                {
                    fireId = RegEntity.UNKNOWN_ID; 
                }
               
                // getCurFireId may return null
                int rwfireId = RegEntity.UNKNOWN_ID;
                Integer rwfireIdInteger = fs.getCurFireId(true);
                if(rwfireIdInteger != null)
                {
                    rwfireId = rwfireIdInteger; 
                }

                // an actor may read or write a token in a non-firing
                // method. e.g., SampleDelay writes a token in initilize().
                if(fireId == RegEntity.UNKNOWN_ID)
                {
                    /*
                    _warn("port event not in fire method for " +
                        _getNameableFullName(actor));
                    */

                    fireId = rwfireId;
                }
                
                if(event.getVectorLength() == IOPortEvent.SINGLETOKEN)
                {
                    _recordPortEvent(port, fireId, rwfireId, isRead, 
                        event.getChannel(), event.getToken(),
                        event.getReceiverPort(), timestamp);
                }
                else
                {
                    Token[] tokenArray = event.getTokenArray();
                    for(int i = 0; i < tokenArray.length; i++)
                    {
                        _recordPortEvent(port, fireId, rwfireId, isRead, 
                            event.getChannel(), tokenArray[i],
                            event.getReceiverPort(), timestamp);
                    }
                }
            }
        }
    }
    
    /** Add Parameters for ProvenanceListener. */
    @Override
    public RecordingParameters generateParameters(NamedObj no) 
        throws IllegalActionException, NameDuplicationException
    {
        _params = new SQLRecordingParameters(no);
        return _params;
    }
    
    /** React to a change in an attribute. */
    @Override
    public void attributeChanged(Attribute attribute)
        throws IllegalActionException
    {
        String name = attribute.getName();
        String val;
        
        //_debug("begin sql attr changed: " + attribute + " hash = " + _recorder.getContainer().hashCode() );

        DatabaseFactory.Parameter dbParameter =
            DatabaseFactory.Parameter.getType(name);
        
        if(dbParameter != null)
        {
            switch(dbParameter)
            {
            case USER:
                val = 
                    ((StringToken)((Parameter)attribute).getToken()).stringValue();

                String dbUserStr = _dbParams.get(DatabaseFactory.Parameter.USER.getName());
                if(dbUserStr == null || !val.equals(dbUserStr)) 
                {
                    _dbParams.put(DatabaseFactory.Parameter.USER.getName(), val);
                    _needReconnectDB = true;
                }

                break;
                
            case PASSWD:
                val = 
                    ((StringToken)((Parameter)attribute).getToken()).stringValue();

                String dbPasswdStr = _dbParams.get(DatabaseFactory.Parameter.PASSWD.getName());
                if(dbPasswdStr == null || !val.equals(dbPasswdStr)) 
                {
                    _dbParams.put(DatabaseFactory.Parameter.PASSWD.getName(), val); 
                    _needReconnectDB = true;
                }
                
                break;
                
            case HOST:
                val = 
                    ((StringToken)((Parameter)attribute).getToken()).stringValue();

                String dbHostStr = _dbParams.get(DatabaseFactory.Parameter.HOST.getName());
                if(dbHostStr == null || !val.equals(dbHostStr)) 
                {
                    _dbParams.put(DatabaseFactory.Parameter.HOST.getName(), val); 
                    _needReconnectDB = true;
                }

                break;

            case NAME:
                val = 
                    ((StringToken)((Parameter)attribute).getToken()).stringValue();

                String dbNameStr = _dbParams.get(DatabaseFactory.Parameter.NAME.getName());
                if(dbNameStr == null || !val.equals(dbNameStr)) 
                {
                    _dbParams.put(DatabaseFactory.Parameter.NAME.getName(), val); 
                    _needReconnectDB = true;
                }
 
                break;
            
            case TYPE:
                val = 
                    ((StringToken)((Parameter)attribute).getToken()).stringValue();

                if(_dbType == null || !val.equals(_dbType.getName()))
                {
                    if(_dbType != null)
                    {
                        try
                        {
                            _dbType.disconnect();
                        }
                        catch(SQLException e)
                        {
                            throw new IllegalActionException("Error " +
                                "disconnecting from database: " + e.getMessage());
                        }
                    }
       
                    _dbParams.put(DatabaseFactory.Parameter.TYPE.getName(), val);
                    _dbType = DatabaseFactory.getType(val);
                    if(_dbType == null)
                    {
                        throw new IllegalActionException(
                            "Invalid database type: " + val);
                    }
                    _needReconnectDB = true;
                }

                break;
            
            case PORT:
                val =
                    ((StringToken)((Parameter)attribute).getToken()).stringValue();
                
                String dbPortStr = _dbParams.get(DatabaseFactory.Parameter.PORT.getName());
                if(dbPortStr == null || !val.equals(dbPortStr)) 
                {
                    _dbParams.put(DatabaseFactory.Parameter.PORT.getName(), val); 
                    _needReconnectDB = true;
                }

                break;
                
            case TABLEPREFIX:
                val =
                    ((StringToken)((Parameter)attribute).getToken()).stringValue();
                
                String dbTablePrefixStr = _dbParams.get(DatabaseFactory.Parameter.TABLEPREFIX.getName());
                if(dbTablePrefixStr == null || !val.equals(dbTablePrefixStr)) 
                {
                    _dbParams.put(DatabaseFactory.Parameter.TABLEPREFIX.getName(), val); 
                    _needReconnectDB = true;
                }

                break;
            
            case JDBC_URL:                
                val =
                    ((StringToken)((Parameter)attribute).getToken()).stringValue();
                
                String dbJDBCURLStr = _dbParams.get(DatabaseFactory.Parameter.JDBC_URL.getName());
                if(dbJDBCURLStr == null || !val.equals(dbJDBCURLStr))
                {
                    _dbParams.put(DatabaseFactory.Parameter.JDBC_URL.getName(), val);
                    _needReconnectDB = true;
                }
                
                break;
                
            case CREATE_INDEXES:                
                val =
                    ((StringToken)((Parameter)attribute).getToken()).stringValue();
                
                String createIndexesStr = _dbParams.get(DatabaseFactory.Parameter.CREATE_INDEXES.getName());
                if(createIndexesStr == null || !val.equals(createIndexesStr))
                {
                    _dbParams.put(DatabaseFactory.Parameter.CREATE_INDEXES.getName(), val);
                    _needReconnectDB = true;
                }
                
                break;
                
            default:
                System.out.println("WARNING: unknown " +
                    "DatabaseFactory.Parameter type: " +
                    dbParameter.getName());
                break;
            }
        }
        else if(name.equals(SQLRecordingParameters.wfNameParamStr))
        {
            val = 
                ((StringToken)((Parameter)attribute).getToken()).stringValue();

            if(_wfNameStr == null || !val.equals(_wfNameStr))
            {
                _wfNameStr = val;
                _needReconnectWF = true;
                //_debug("wf name changed to " + _wfNameStr);
            }
        }
        else if(name.equals("User Name"))
        {
            val = 
                ((StringToken)((Parameter)attribute).getToken()).stringValue();
           
            // if empty, set default from system property
            if(val.length() == 0)
            {
                val = System.getProperty("user.name");
            }

            if(_wfUserStr == null || !val.equals(_wfUserStr))
            {
                _wfUserStr = val;
            }

            // for backwards compatibility
            String dbUserStr = _dbParams.get(DatabaseFactory.Parameter.USER.getName());
            if(dbUserStr == null || dbUserStr.length() == 0)
            {
                _dbParams.put(DatabaseFactory.Parameter.USER.getName(), val);
                _needReconnectDB = true;
            }
        }
        else
        {
            super.attributeChanged(attribute);
        }

        // get workflow contents if we need to reconnect to db
        // or workflow
        if(_needReconnectDB || _needReconnectWF)
        {
            _needWorkflowContents(true);

            // XXX
            if(_needReconnectWF)
            {
                _wfReset();
            }

            // XXX need dbReset?
        }
        
        //_debug("end sql attr changed: " + attribute);
    }
   
    ////////////////////////////////////////////////////////////////////////
    //// public variables                                               ////
    
    /** The token data is valid. */
    public static final int DATA_VALID              = 0;

    /** The token data was not recorded. */
    public static final int DATA_NONE               = 1;

    /** The token data was too large to completely store. */
    public static final int DATA_TRUNCATED          = 2;
 
    // port io types
    public enum PortDirection 
    {
        In,
        Out,
        InOut;

        static public PortDirection valueOf(int id)
        {
            if(id < 0 || id >= values().length)
            {
                return null;
            }
            else
            {
                return values()[id];
            }
        }
    };
            
    ////////////////////////////////////////////////////////////////////////
    //// protected methods                                              ////

    /** Check parameters before (re)connecting to database or workflow. */
    protected void _checkParameters() throws IllegalActionException,
        RecordingException
    {
        if(_needReconnectWF)
        {
            //_debug("checkParameters; need reconnectWF; name len = " +
            //  _wfNameStr.length() + " name = " + _wfNameStr);
            
            _checkWorkflowName();
        }

        if(_needReconnectDB)
        {
            if(_dbType == null)
            {
                throw new RecordingException("No database connection.");
            }

            // see if the jdbc url override is empty.
            // if so, need to check individual parameters.
            String dbJDBCURLStr = _dbParams.get(DatabaseFactory.Parameter.JDBC_URL.getName());
            if(dbJDBCURLStr == null || dbJDBCURLStr.length() == 0)
            {
                String dbUserStr = _dbParams.get(DatabaseFactory.Parameter.USER.getName());
                if(_dbType.needUserForConnect() && 
                    (dbUserStr == null || dbUserStr.length() == 0))
                {
                    throw new RecordingException("Need database user name."); 
                }
    
                String dbPasswdStr = _dbParams.get(DatabaseFactory.Parameter.PASSWD.getName());
                if(_dbType.needPasswordForConnect() && 
                    (dbPasswdStr == null || dbPasswdStr.length() == 0))
                {
                    throw new RecordingException("Need database password");
                }
    
                String dbHost = _dbParams.get(DatabaseFactory.Parameter.HOST.getName());
                if(_dbType.needHostForConnect() &&
                    (dbHost == null ||  dbHost.length() == 0))
                {
                    throw new RecordingException("Need database host name."); 
                }

                String dbNameStr = _dbParams.get(DatabaseFactory.Parameter.NAME.getName());
                if(dbNameStr == null || dbNameStr.length() == 0)
                {
                    throw new RecordingException("Need database name."); 
                }
            }
        }

        /*
        _debug("going to _connectProvStore: ");
        _debug("wf: " + _wfNameStr);
        _debug("user: " + _dbUserStr);
        _debug("dbhost: " + _dbHostStr);
        _debug("dbhost: " + _dbPortStr);
        _debug("passwd: " + _dbPasswdStr);
        */
    }

    /** Check validity of workflow name parameter. */
    protected void _checkWorkflowName() throws RecordingException
    {
        if(_wfNameStr == null || _wfNameStr.length() == 0)
        {
            throw new RecordingException("Need workflow name.");
        }
    }

    /** Create a Schema to reflect the v6 schema. */
    protected Schema _createSchema()
    {
        return Schemas.createSchemaV6();
    }

    /** Reconnect to the database. */
    protected void _reconnectDatabase(boolean resetDB) throws RecordingException
    {
        if(resetDB)
        {
            _dbReset();
        }
        
        try
        {
            _dbType = DatabaseFactory.getConnectedDatabaseType(_dbParams, "provenance");
            Schemas.checkVersion(_dbType, _schema);

            _maxDataSize = _getMaxTokenDataSize();
            _maxParameterValueSize = _getMaxParameterValueSize();

            //System.out.println("max data size = " + _maxDataSize);
            //System.out.println("max param value size = " + __maxParameterValueSize);
        }
        catch(SQLException e)
        {
            _errorReset();
            throw new RecordingException("Error connecting or " +
                "initializing database.\n" + _dbParams + "\n", e);
        }

        try
        {
            _nullPreparedStatements();
            _createPreparedStatements();
        }
        catch(SQLException e)
        {
            throw new RecordingException("Error creating prepared " +
                "statements: " + e.getMessage());
        }

        _needReconnectDB = false;
    }

    /** Get the maximum size of a parameter value that can be stored in the 
     *  database.
     */
    protected int _getMaxParameterValueSize() throws SQLException
    {
        return _dbType.getColumnSize("parameter", "value");
    }

    /** Get the maximum size of token data that can be stored in the 
     *  database.
     */
    protected int _getMaxTokenDataSize() throws SQLException
    {
        return _dbType.getColumnSize("token_flow", "data");
    }

    /** Re-acquire the current workflow id. */
    protected void _reconnectWorkflow() throws RecordingException
    {
        _wfReset();

        _needReconnectWF = false;

        int rc = _updateWorkflowID(false);
                
        if(rc == WORKFLOW_ALREADY_EXISTS)
        {
                        
        }
        else if(rc == WORKFLOW_DOES_NOT_EXIST)
        {
            //_debug("workflow does not exist; attempting to create " +
                //"new one.");

            //XXX need to check for return value
            _updateWorkflowID(true);
        }
    }

    /** Initialize the prepared statements. */
    protected void _createPreparedStatements() throws SQLException
    {
        // NOTE: for each PreparedStatement we check to see if
        // it has already been created by a subclass: it may
        // have a different columns for a table, or removed
        // the entire table.

        if(_psWorkflowInsert == null && _schema.containsTable("workflow"))
        {
            _psWorkflowInsert = _dbType.getSQLInsert("workflow", "id",
                "name, version, action_id", "?, ?, ?");
        }
       
        if(_psWorkflowQuery == null && _schema.containsTable("workflow"))
        {
            _psWorkflowQuery = _dbType.getSQLSelect("workflow", "id",
                "name = ?");
        }
   
        if(_psActionInsert == null && _schema.containsTable("action"))
        {
            _psActionInsert = _dbType.getSQLInsert("action", "id",
                "parent_id, user, time", "?, ?, ?");
        }

        if(_psEntityInsert == null && _schema.containsTable("entity"))
        {
            _psEntityInsert = _dbType.getSQLInsert("entity", "id",
                "container_id, workflow_id, type, name", "?, ?, ?, ?");
        }
            
        if(_psEntityQuery == null && _schema.containsTable("entity"))
        {
            _psEntityQuery = _dbType.getSQLSelect("entity", "id",
                "workflow_id = ? AND name = ? AND type = ?");
        }

        if(_psActorInsert == null && _schema.containsTable("actor"))
        {
            _psActorInsert = _dbType.getSQLInsert("actor", "id, class", "?, ?");
        }

        if(_psDirectorInsert == null && _schema.containsTable("director"))
        {
            _psDirectorInsert = _dbType.getSQLInsert("director", "id, class",
                "?, ?");
        }
    
        if(_psParameterInsert == null && _schema.containsTable("parameter"))
        {
            _psParameterInsert = _dbType.getSQLInsert("parameter",
                "id, type, value", "?, ?, ?");
        }

        if(_psLinkInsert == null && _schema.containsTable("link"))
        {
            _psLinkInsert = _dbType.getSQLInsert("link",
                "id, end_point_1, end_point_2", "?, ?, ?");
        }
    
        if(_psPortInsert == null && _schema.containsTable("port"))
        {
            _psPortInsert = _dbType.getSQLInsert("port",
                "id, type, direction, multiport", "?, ?, ?, ?");
        }
    
        if(_psRelationInsert == null && _schema.containsTable("relation"))
        {
            _psRelationInsert = _dbType.getSQLInsert("relation", "id, width",
                "?, ?");
        }
                
        if(_psWorkflowExecStart == null &&
            _schema.containsTable("workflow_exec"))
        {
            String defaultTimeStr = _dbType.getDefaultTimeStr();
            _psWorkflowExecStart = _dbType.getSQLInsert("workflow_exec", "id",
                "workflow_id, user, start_time, end_time", "?, ?, ?, " +
                defaultTimeStr);
        }

        if(_psWorkflowExecStop == null &&
            _schema.containsTable("workflow_exec"))
        {
            _psWorkflowExecStop = _dbType.getSQLUpdate("workflow_exec",
                "end_time = ?", "id = ?");
        }
        
        if(_psActorFireStart == null && _schema.containsTable("actor_fire"))
        {
            String defaultTimeStr = _dbType.getDefaultTimeStr();
            _psActorFireStart = _dbType.getSQLInsert("actor_fire", "id",
                "actor_id, wf_exec_id, start_time, type, end_time",
                "?, ?, ?, ?, " + defaultTimeStr);
        }
    
        if(_psActorFireStop == null && _schema.containsTable("actor_fire"))
        {
            _psActorFireStop = _dbType.getSQLUpdate("actor_fire",
                "end_time = ?", "id = ?");
        }

        if(_psTokenFlowInsert == null && _schema.containsTable("token_flow"))
        {
            _psTokenFlowInsert = _dbType.getSQLInsert("token_flow", "id",
                "port_id, time, data, channel, is_read, fire_id, rw_fire_id, " +
                "data_description", "?, ?, ?, ?, ?, ?, ?, ?");
        }
    }
    
    /** Set our prepared statements to null. */
    protected void _nullPreparedStatements()
    {
        _psWorkflowInsert = null;
        _psWorkflowQuery = null;
        _psActionInsert = null;
        _psEntityInsert = null;
        _psEntityQuery = null;
        _psActorInsert = null;
        _psDirectorInsert = null;
        _psParameterInsert = null;
        _psLinkInsert = null;
        _psPortInsert = null;
        _psRelationInsert = null;
        _psWorkflowExecStart = null;
        _psWorkflowExecStop = null;
        _psActorFireStart = null;
        _psActorFireStop = null;
        _psTokenFlowInsert = null;
    }

    /** Update workflow ID either by retrieving the id from an existing
     *  workflow in the database or creating a new one.
     *
     * @param isNew
     * @return
     */
    protected int _updateWorkflowID(boolean isNew) throws RecordingException
    {
        int retval = SUCCESS;

        // check to see if it already exists.
        int id = _getWorkflowId();

        // see if this is a new workflow
        if(isNew)
        {
            // see if it already exists
            if(id != RegEntity.UNKNOWN_ID)
            {
                retval = WORKFLOW_ALREADY_EXISTS;
            }
            else
            {
        
                if(_wfUserStr == null)
                {
                    throw new RecordingException("Need workflow user " +
                        "name");
                }
                    
                _addWorkflow();
            }
        }
        // see if workflow was not in db
        else if(id == RegEntity.UNKNOWN_ID)
        {
            retval = WORKFLOW_DOES_NOT_EXIST;
        }
        else
        {
            int oldWfId = _wfId;

            // workflow exists in db; get id from query.
            _wfId = id;

            // see if id is different
            if(oldWfId != _wfId)
            {
                _wfReset();
            }
        }

        return retval;
    }

    /** Get the internal workflow id. */
    protected int _getWorkflowId() throws RecordingException
    {
        // use the workflow name as the id string.
        return _getWorkflowId(_wfNameStr);
    }

    /** Get the internal workflow id given a id string. If no workflow has
     *  the id string, returns RegEntity.UNKNOWN_ID.
     */
    protected int _getWorkflowId(String idStr) throws RecordingException
    {
        int retval = RegEntity.UNKNOWN_ID;

        try
        {
            synchronized(_psWorkflowQuery)
            {
                _psWorkflowQuery.setString(1, idStr);
                ResultSet rs = _psWorkflowQuery.executeQuery();
                
                if(rs.next())
                {
                    retval = rs.getInt("id");
                }
                rs.close();
            }
        }
        catch(SQLException e)
        {
            throw new RecordingException("Error querying workflow.name: ", e);
        }

        //_debug("getWorkflowId idStr = " + idStr + " found id = " + retval);
        //_debug(_psWorkflowQuery);

        return retval;
    }

    /** Reset when we use a different db connection. */
    protected void _dbReset() throws RecordingException
    {
        _wfId = RegEntity.UNKNOWN_ID;
        _maxDataSize = -1;
        _needReconnectWF = true;
        _needReconnectDB = true;
       
        _wfReset();

        if(_dbType != null)
        {
            try
            {
                _dbType.disconnect();
            }
            catch(SQLException e)
            {
                throw new RecordingException("Error disconnection from " +
                    "database.", e);
            }
        }
    }

    /** Reset when we use a different workflow. */
    protected void _wfReset()
    {
        _wfExecId = RegEntity.UNKNOWN_ID;
        _entityCacheTable.clear();
        _fireStateTable.clear();
    }

    /** Reset when we encounter an error. */
    protected void _errorReset() throws RecordingException
    {
        _dbReset();
    }

    /**
     *  This method opens the database connection if necessary.
     *  @param allowReconnectWF - default true, only false if you want to force no _reconnectWF, e.g
     *  for non-write events like delete
     */ 
    protected void _checkConnection(boolean allowReconnectWF) throws RecordingException
    {
        // see if we need to (re)connect to DB
        if(_needReconnectDB || _needReconnectWF)
        {
            //_debug("in checkConnection: need to reconnectdb or wf");
            _updateContainerName();
            
            try
            {
                _checkParameters();

                if(_schema == null)
                {
                    _schema = _createSchema();
                }

                if(_needReconnectDB)
                {
                    _reconnectDatabase(true);
                }
        
                if(_needReconnectWF && allowReconnectWF)
                {
                    _reconnectWorkflow();
                }
            }
            catch(IllegalActionException e)
            {
                throw new RecordingException(_getExceptionMessage(e));
            }
        }

    }

    /** Modify a NamedObj's full name. */
    protected String _changeEntityFullName(String name)
    {
        return name;
    }

    /** Check if a NamedObj has been inserted into the entity table.
     * 
     * @param fullName fully qualified name of NamedObj.
     * @param displayName display name (NOTE: can be null)
     * @param type type of NamedObj. (NOTE: can be null)
     * @return
     */
    protected RegEntity _checkEntity(Nameable namedObj, RegEntity.EntityType type)
        throws RecordingException
    {
        RegEntity retval = null;
        
        String fullName = _getNameableFullName(namedObj);
        String displayName = namedObj.getDisplayName();
       
        _checkConnection(true);

        String changedName = _changeEntityFullName(fullName);
        
        String typeStr = type.toString();
        
        //_debug("getting from cache: " + cacheName);
        retval = _entityCacheTable.get(namedObj);
        
        // see if it's cached
        if(retval != null)
        {
            //_debug("found existing id for " + _getNameableFullName(namedObj) + " : " + retval.getId());
            
            // it was cached, so no longer new.
            retval.setOld();
        }
        else
        {
            try
            {
                // determine containing id
                int cntId = _getContainerId(namedObj);

                // see if name is in entity table
                synchronized(_psEntityQuery)
                {
                    _psEntityQuery.setInt(1, _wfId);
                    _psEntityQuery.setString(2, changedName);
                    _psEntityQuery.setString(3, typeStr);
                    ResultSet rs = _psEntityQuery.executeQuery();

                    boolean haveResult = rs.next();

                    if(!haveResult)
                    {
                        // not in entity table, so insert
                        retval = _addEntity(cntId, type, changedName, displayName,
                            RegEntity.UNKNOWN_ID);
                        
                        //_debug("added entity " + changedName + " id = " + retval.getId());
                    }
                    else if(haveResult)
                    {
                        retval = new RegEntity(rs.getInt("id"), false, cntId,
                            type);
                        
                        //_debug("queried entity " + changedName + " id = " + retval.getId());
                    }
                    
                    rs.close();
                }
            }
            catch(SQLException e)
            {
                throw new RecordingException("ERROR querying entity: ", e);
            }


            if(retval == null)
            {
                //throw new RecordingException("Entity " + name + 
                //    " not registered.");
            }
            else
            {
                //_debug("caching " + cacheName);
                _entityCacheTable.put(namedObj, retval);
            }
        }

        return retval;
    }

    /** Add a new row to the entity table. */
    protected RegEntity _addEntity(int containerId, RegEntity.EntityType type,
        String fullName, String displayName, int prevId)
        throws RecordingException, SQLException
    {

        int id = RegEntity.UNKNOWN_ID;
        
        synchronized(_psEntityInsert)
        {
            _psEntityInsert.setInt(1, containerId);
            _psEntityInsert.setInt(2, _wfId);
            _psEntityInsert.setString(3, type.toString());
            _psEntityInsert.setString(4, fullName);
            id = _dbType.insert(_psEntityInsert, "entity", "id");
        }
        
        RegEntity retval = new RegEntity(id, true, containerId, type);
        
        //_debug("inserted " + fullName + ", id = " + id);
        
        if(_debugWriter != null)
        {
            _debugWrite("INSERT INTO ENTITY(" + containerId + ", " + _wfId +
                ", " + type + ", " + fullName + ")");
        }

        return retval;
    }

    /** Add a new row to the workflow table. */
    protected void _addWorkflow() throws RecordingException
    {
        try
        {
            // insert the new action table row
            synchronized(_psActionInsert)
            {
                _psActionInsert.setInt(1, 0);
                _psActionInsert.setString(2, _wfUserStr);
                _psActionInsert.setTimestamp(3, new Timestamp(new Date().getTime()));
                int actionId = _dbType.insert(_psActionInsert, "action", "id");
                                
                if(_debugWriter != null)
                {
                    _debugWrite("INSERT INTO ACTION(0, " + _wfUserStr +
                        ", curTime)");
                }
                
                //XXX should psWorkflowInsert be synchronized?

                // insert the new workflow table row
                _psWorkflowInsert.setString(1, _wfNameStr);
                _psWorkflowInsert.setString(2, "v0");
                _psWorkflowInsert.setInt(3, actionId);
                _wfId = _dbType.insert(_psWorkflowInsert, "workflow","id");
                _wfReset();
                    
                if(_debugWriter != null)
                {
                    _debugWrite("INSERT INTO WORKFLOW(" + _wfNameStr +
                        ", v0, " + actionId + ")");
                }
            }
        }
        catch(SQLException e)
        {
            throw new RecordingException("Error adding row to workflow:", e);
        }
    }

    /** Find the container id of an entity. If entity has no container, i.e.,
     *  is the top-level object, returns 0. Otherwise if the container could
     *  not be found, returns RegEntity.UNKNOWN_ID. */
    protected int _getContainerId(Nameable namedObj) 
        throws RecordingException
    {
        //_debug("_getContainerId(" + fullName + ")");

        int retval = RegEntity.UNKNOWN_ID;
   
        NamedObj container = namedObj.getContainer();

        if(container == null)
        {
            retval = 0;
        }
        else
        {
            RegEntity.EntityType containerType =
                RegEntity.EntityType.getType(container);
            RegEntity re = _checkEntity(container, containerType);
            if(re == null)
            {
                //throw new RecordingException("Container for " + name +
                //    " is not in entity table.");
                retval = RegEntity.UNKNOWN_ID;
            }
            else
            {
                retval = re.getId();
            }
        }

        //_debug("getContainerId " + retval + " for " + name);

        return retval;
    }

    /** Convenience routine that regActor and regDirector can use since
     * the schema for actors and directors only differs by "actor" vs
     * "director".
     */
    protected int _regActorDirector(NamedObj namedObj, boolean actor) throws RecordingException
    {
        RegEntity.EntityType type = actor ? 
            RegEntity.EntityType.Actor : RegEntity.EntityType.Director;
        RegEntity re = _checkEntity(namedObj, type);
        if(re.isNew())
        {
            
            //_debug("is new actor " + namedObj.getFullName());
            
            PreparedStatement ps = null;

            if(actor)
            {
                ps = _psActorInsert;
            }
            else
            {
                ps = _psDirectorInsert;
            }
            
            try
            {
                String className = namedObj.getClassName();
                
                synchronized(ps)
                {
                    ps.setInt(1, re.getId());
                    ps.setString(2, className);
                    ps.executeUpdate();
                }

                if(_debugWriter != null)
                {
                    String table = actor ? "ACTOR" : "DIRECTOR";
                    _debugWrite("INSERT INTO " + table + " (" +
                        re.getId() + ", " + className + ")");
                }

            }
            catch(SQLException e)
            {
                String msg = "Error registering ";
                if(actor)
                {
                    msg += "actor: ";
                }
                else
                {
                    msg += "director: ";
                }

                if(_debugWriter != null)
                {
                    _debugWrite(msg, e);
                }

                throw new RecordingException(msg, e);
            }
        }

        return re.getId();
    }
    
    /** Add a port to the port table. */
    protected void _regPortReal(TypedIOPort port, RegEntity re)
        throws RecordingException
    {
        int multi = port.isMultiport() ? 1 : 0;

        int direction;
        if(port.isInput() && port.isOutput())
        {
            direction = PortDirection.InOut.ordinal();
        }
        else if(port.isInput())
        {
            direction = PortDirection.In.ordinal();
        }
        else
        {
            direction = PortDirection.Out.ordinal();
        }

        try
        {
            synchronized(_psPortInsert)
            {
                _psPortInsert.setInt(1, re.getId());
                _psPortInsert.setString(2, port.getType().toString());
                _psPortInsert.setInt(3, direction);
                _psPortInsert.setInt(4, multi);
                _psPortInsert.executeUpdate();
            }

            if(_debugWriter != null)
            {
                _debugWrite("INSERT INTO PORT (id, type, " + direction +
                    ", " + multi + ")");
            }

        }
        catch(SQLException e)
        {
            String msg = "Error inserting port " + _getNameableFullName(port) + ": ";
            _debugWrite(msg, e);
            throw new RecordingException(msg, e);
        }
    }

    /** Add an entity to the parameter table. */
    protected void _regParameterReal(NamedObj parameter, RegEntity re)
        throws RecordingException
    {
        String className = parameter.getClassName();
        String valueStr = "none";

        try
        {
            synchronized(_psParameterInsert)
            {
                //_debug("going to insert parameter " + _getNameableFullName(parameter) + " with id " + re.getId());

                _psParameterInsert.setInt(1, re.getId());
                _psParameterInsert.setString(2, className);

                if(parameter instanceof AbstractSettableAttribute)
                {
                    valueStr = ((AbstractSettableAttribute)parameter).
                        getValueAsString();
                }

                // replace null string with empty string.
                if(valueStr == null)
                {
                    valueStr = "";
                }
                
                //_debug(_getNameableFullName(parameter) + " value length = " + valueStr.length());

                // XXX need to set truncated bit somewhere
                if(valueStr.length() > _maxParameterValueSize)
                {
                    //_debug("TRUNCATING!");
                    valueStr = valueStr.substring(0, _maxParameterValueSize); 
                }

                _psParameterInsert.setString(3, valueStr);

                _psParameterInsert.executeUpdate();
            }
        
            if(_debugWriter != null)
            {
                if(className.equals("org.kepler.moml.NamedObjId"))
                {
                    _debugWrite("INSERT INTO PARAMETER (" + re.getId() +
                            ", " + className + ", lsid)");                    
                }   
                else
                {
                    _debugWrite("INSERT INTO PARAMETER (" + re.getId() +
                        ", " + className + ", " + valueStr + ")");
                }
            }
        }
        catch(SQLException e)
        {
            String msg = "Error registering parameter " + _getNameableFullName(parameter) + ": ";
            _debugWrite(msg, e);
            throw new RecordingException(msg, e);
        }
    }

    /** Record a specific type of firing for an actor. */
    protected void _recordFiringEvent(Actor actor, 
        FiringEvent.FiringEventType type, Date timestamp) throws RecordingException
    {
        //_debug("firing event: " + type + " for " + _getNameableFullName(actor));

        // see if it's a start or stop.
        FireState<Integer> fs = _fireStateTable.get(actor);

        if(fs == null)
        {
            throw new RecordingException(
                "Received actor fire event for unregistered actor: " +
                _getNameableFullName(actor));
        }

        synchronized(fs)
        {
            try
            {
                if(type.isStart())
                {
                    int fireId;

                    synchronized(_psActorFireStart)
                    {
                        _psActorFireStart.setInt(1, fs.getActorId());
                        _psActorFireStart.setInt(2, _wfExecId);
                        _psActorFireStart.setTimestamp(3,new Timestamp(timestamp.getTime()));
                        _psActorFireStart.setString(4, type.getTypeName());
                        fireId = _dbType.insert(_psActorFireStart, "actor_fire",
                            "id");
                    }

                    if(_debugWriter != null)
                    {
                        _debugWrite("INSERT INTO ACTOR_FIRE(" +
                            fs.getActorId() + ", " + _wfExecId + ", " +
                            "curTime, " + type.getTypeName() + ")");
                    }

                    fs.fireStart(type, fireId);
                }
                else
                {
                    Integer fireId = fs.fireStop(type);
                    
                    if(fireId == null)
                    {
                        String msg = "fire id is null for " + _getNameableFullName(actor);
                        _debugWrite(msg);
                        throw new RecordingException(msg);
                    }
                    
                    synchronized(_psActorFireStop)
                    {
                        _psActorFireStop.setTimestamp(1, new Timestamp(timestamp.getTime()));
                        _psActorFireStop.setInt(2, fireId);
                        _psActorFireStop.executeUpdate();
                    }

                    if(_debugWriter != null)
                    {
                        _debugWrite("UPDATE ACTOR_FIRE set end_time " +
                            "= curTime WHERE id = " + fireId + ")");
                    }
                }
            }
            catch(SQLException e)
            {
                throw new RecordingException(_getExceptionMessage(e));
            }
        }
    }

    /** Record a port read or write to the token_flow table. */
    protected void _recordPortEvent(IOPort port, int fireId, int rwfireId, 
        boolean isRead, int channel, Token token, IOPort destPort, Date timestamp)
        throws RecordingException
    {
    
        String dataStr;
        int dataDesc;

        String data = token.toString();

        // see if we have data
        /* FindBugs: data is known to be non-null
        if(data == null)
        {
            // we are not recording data
            dataStr = "";
            dataDesc = DATA_NONE;
        }
        // see if the data is too large
        else*/ if(data.length() > _maxDataSize)
        {
            // store as much data as possible
            dataStr = data.substring(0, _maxDataSize);
            dataDesc = DATA_TRUNCATED;
        }
        else
        {
            // data not too large
            dataStr = data;
            dataDesc = DATA_VALID;
        }
        
        RegEntity.EntityType type;
        if(port instanceof ParameterPort)
        {
            type = RegEntity.EntityType.PortParameter;
        }
        else
        {
            type = RegEntity.EntityType.Port;
        }

        RegEntity re = _checkEntity(port, type);
        if(re == null)
        {
            throw new RecordingException("Port has not been registered: " +
                _getNameableFullName(port));
        }

        int portId = re.getId();

        try
        {
            synchronized(_psTokenFlowInsert)
            {
                _psTokenFlowInsert.setInt(1, portId);
                _psTokenFlowInsert.setTimestamp(2, new Timestamp(timestamp.getTime()));
                _psTokenFlowInsert.setString(3, dataStr);
                _psTokenFlowInsert.setInt(4, channel);

                if(isRead)
                {
                    _psTokenFlowInsert.setInt(5, 1);
                }
                else
                {
                    _psTokenFlowInsert.setInt(5, 0);
                }

                _psTokenFlowInsert.setInt(6, fireId);
                _psTokenFlowInsert.setInt(7, rwfireId);
                _psTokenFlowInsert.setInt(8, dataDesc);
                _psTokenFlowInsert.executeUpdate();
            }

            if(_debugWriter != null)
            {
                _debugWrite("INSERT INTO TOKEN_FLOW(" + portId +
                    ", curTime, " + dataStr + ", " + channel + ", " +
                    isRead + ", " + fireId + ", " + rwfireId + ", " +
                    dataDesc + ")");
            }

        }
        catch(SQLException e)
        {
            _error("port = " + _getNameableFullName(port) + ", " +
                "portId = " + portId + ", " +
                "fireId = " + fireId + ", " +
                "rwFireId = " + rwfireId + ", " +
                "isRead = " + isRead + ", " +
                "channel = " + channel + ", " +
                "data = " + data);
            e.printStackTrace();
            _error("------------------");
            throw new RecordingException(_getExceptionMessage(e));
        }
    }
    
    ////////////////////////////////////////////////////////////////////////
    //// protected variables                                            ////

    /** The schema containing table and column definitions. */
    protected Schema _schema;

    /** The current workflow id. */
    protected int _wfId;

    /** The current workflow execution id. */
    protected int _wfExecId;

    /** The maximum size in the SQL table for data values. */
    protected int _maxDataSize;
    
    /** The maximum size in the SQL table for parameter values. */
    protected int _maxParameterValueSize;

    /** A table to map actor to its firing state object. */
    protected Map<Actor, FireState<Integer>> _fireStateTable = null;

    /** A table to cache RegEntity objects. */
    protected Map<Nameable,RegEntity> _entityCacheTable = null;

    protected PreparedStatement _psWorkflowInsert;
    protected PreparedStatement _psWorkflowQuery;
    protected PreparedStatement _psActionInsert;
    protected PreparedStatement _psEntityInsert;
    protected PreparedStatement _psEntityQuery;
    protected PreparedStatement _psActorInsert;
    protected PreparedStatement _psDirectorInsert;
    protected PreparedStatement _psParameterInsert;
    protected PreparedStatement _psLinkInsert;
    protected PreparedStatement _psPortInsert;
    protected PreparedStatement _psRelationInsert;
    protected PreparedStatement _psWorkflowExecStart;
    protected PreparedStatement _psWorkflowExecStop;
    protected PreparedStatement _psActorFireStart;
    protected PreparedStatement _psActorFireStop;
    protected PreparedStatement _psTokenFlowInsert;
    
    protected String _wfNameStr = null;
    protected String _wfUserStr = "unknown";

    /** The follow hold values from DatabaseFactory.Parameters. 
     *  NOTE: use a LinkedHashMap for predictable iteration order.
     */
    protected Map<String,String> _dbParams = new LinkedHashMap<String,String>();
    
    /** Connection to the database. */
    protected DatabaseType _dbType;

    // return values for connect
    protected static final int SUCCESS                 = 0;
    protected static final int WORKFLOW_ALREADY_EXISTS = 1;
    protected static final int WORKFLOW_DOES_NOT_EXIST = 2;

    /** The sql parameters object. */
    protected SQLRecordingParameters _params = null;

    /** True when we need to reconnect to the database. */
    protected boolean _needReconnectDB = true;

    /** True when we need to load workflow id from database. */
    protected boolean _needReconnectWF = true;
}