/**
 * JOnAS: Java(TM) Open Application Server
 * Copyright (C) 1999-2010 Bull S.A.
 * Contact: jonas-team@ow2.org
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA
 *
 * --------------------------------------------------------------------------
 * $Id: JManagedConnection.java 20650 2010-12-20 10:29:33Z alitokmen $
 * --------------------------------------------------------------------------
 */


package org.ow2.jonas.dbm.internal.cm;

import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import javax.sql.StatementEventListener;
import javax.sql.StatementEvent;
import javax.sql.ConnectionEvent;
import javax.sql.ConnectionEventListener;
import javax.sql.XAConnection;
import javax.transaction.Synchronization;
import javax.transaction.Transaction;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;

import org.ow2.jonas.lib.util.Log;
import org.ow2.jonas.tm.Enlistable;

import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

/**
 * This class represents the physical connection, managed by the pool.
 * It implements the PooledConnection interface, in fact via the derived
 * XAConnection interface.
 * @author durieuxp
 */
public class JManagedConnection implements Comparable, XAConnection, XAResource, Synchronization, Enlistable {

    static private Logger logger = Log.getLogger(Log.JONAS_DBM_PREFIX);
    static private Logger loggerxa = Log.getLogger(Log.JONAS_DBM_PREFIX + ".xa");
    static private Logger loggerps = Log.getLogger(Log.JONAS_DBM_PREFIX + ".ps");

    /**
     * The JConnection object (= wrapper on the logical Connection)
     */
    IConnection implConn = null;

    /**
     * The physical Connection behind.
     */
    Connection actConn = null;

    /**
     * The ConnectionManager managing this Connection.
     */
    ConnectionManager ds;

    // Most of the time, only 1 element, but must keep a Vector here.
    private Vector eventListeners = new Vector();

    // Listener for statements (No generics so that it will work on JDK5 as StatementEventListener is not available on this JDK)
    private Vector statementEventListeners = new Vector();

    // Not really used.
    private int timeout;

    /**
     * datasource name
     */
    private String rmid = null;

    /**
     * Use count. A positive value means that the connection has been opened
     * by someone.
     */
    private int open;

    /**
     * transaction the connection is involved with
     */
    private Transaction tx;

    /**
     * current transaction in case of late enlistment
     * Not really used, just a hint.
     */
    private Transaction enlistedInTx;

    /**
     * true if registered as Enlistable
     */
    private boolean rme = false;

    /**
     * When it is expected to die (for closed connections)
     */
    private long deathTime;

    /**
     * When it is expected to be closed (for opened connections)
     */
    private long closeTime;

    /**
     * List of Thread infos of clients that opened this connection.
     */
    private List openerThreadInfos = new ArrayList();

    /**
     * When the connection has been created.
     */
    private long creationTime;

    private static final long DEFAULT_OPENING_TIME = -1;
    /**
     * When that connection was used for an application (with open == 0).
     */
    private long openingTime = DEFAULT_OPENING_TIME;
    
    

    /**
     * Used internally for the PrepareStatement cache.
     */
    private static int objcount = 10;   // to have a minimum of prepare statements.
    private int ident;
    private int hitCount = 0;

    /**
     * Size of the PrepareStement cache for this Connection.
     * Notice that not all Connections have the same limit.
     * (optimization of the ps cache)
     */
    private int pstmtmax;

    /**
     * List of PreparedStatement in the pool.
     * This list is synchronized.
     */
    private Map psList = null;

    /**
     * List of Thread infos of clients that released this connection.
     */
    private List closerThreadInfos = new ArrayList();
    
    //  -----------------------------------------------------------------
    //  Constructors
    //  -----------------------------------------------------------------

    public JManagedConnection(Connection conn, ConnectionManager ds) {

        logger.log(BasicLevel.DEBUG, "constructor");

        this.actConn = conn;
        this.ds = ds;

        // An XAConnection holds 2 objects: 1 Connection + 1 XAResource
        this.implConn = (IConnection) Proxy.newProxyInstance(IConnection.class.getClassLoader(), new Class[] {IConnection.class},
                new JConnection(this, conn));


        this.rmid = ds.getDatasourceName();

        open = 0;
        creationTime = System.currentTimeMillis();
        deathTime = creationTime + ds.getMaxAgeMilli();

        // getPstmtMax() = 0 -> no pool for PreparedStatement's
        // Not all connections have the same number of prepare statement.
        ident = objcount++;
        if (ds.getPstmtMax() > 0) {
            int pstmtmin = 1 + ds.getPstmtMax() / 4;
            pstmtmax = ident % ds.getPstmtMax();
            if (pstmtmax < pstmtmin) {
                pstmtmax = pstmtmin;
            }
            psList = Collections.synchronizedMap(new HashMap(pstmtmax));
        }
    }

    /**
     * @return The ident of this JManagedConnection
     */
    public int getIdent() {
        return ident;
    }

    /**
     * Dynamically change the prepared statement pool size
     */
    public void setPstmtMax(int max) {
        pstmtmax = max;
        if (psList == null) {
            psList = Collections.synchronizedMap(new HashMap(pstmtmax));
        }
    }

    //  -----------------------------------------------------------------
    //  XAResource implementation
    //  Important Notice: This is a pseudo XA driver. Not full XA is
    //  supported. For example, prepare returns always OK.
    //  To use a real XA driver, you should use the RA-jdbc instead of DBM.
    //  -----------------------------------------------------------------

    /**
     * Commit the global transaction specified by xid.
     * @param xid transaction xid
     * @param onePhase true if one phase commit
     * @throws XAException XA protocol error
     */
    public void commit(Xid xid, boolean onePhase) throws XAException {
        if (loggerxa.isLoggable(BasicLevel.DEBUG)) {
            loggerxa.log(BasicLevel.DEBUG, "XA-COMMIT for " + xid);
        }

        // Avoids a NPE.
        // This should not occur because a close could not be called
        // on a connection implied inside a transaction.
        if (actConn == null) {
            loggerxa.log(BasicLevel.ERROR, "commit on a closed connection");
            return;
        }

        // Make sure that we are not in AutoCommit mode
        try {
            if (implConn.getAutoCommit() == true) {
                loggerxa.log(BasicLevel.ERROR, "Commit called on XAResource with AutoCommit set");
                throw (new XAException(XAException.XA_HEURCOM));
            }
        } catch (SQLException e) {
            loggerxa.log(BasicLevel.ERROR, "Cannot getAutoCommit:" + e);
            notifyError(e);
            throw(new XAException("Error on getAutoCommit"));
        }

        // Commit the transaction
        SQLException commitException = null;
        try {
            actConn.commit();
        } catch (SQLException e) {
        	// Only store the exception at this point since the
        	// connection cannot be removed until the auto-commit
        	// mode has been reset
        	commitException = e;
            throw(new XAException("Error on commit"));
        } finally {
        	// Reset the auto-commit flag
            try {
                implConn.setAutoCommit(true);
            } catch (Exception exc) {
                loggerxa.log(BasicLevel.DEBUG, "Unable to set autoCommit to true:", exc);
            }
            if (commitException != null) {
                loggerxa.log(BasicLevel.ERROR, "Cannot commit transaction:" + commitException);
                // Notify the listeners of the error
                notifyError(commitException);
            }
        }
    }

    /**
     * Ends the work performed on behalf of a transaction branch.
     * @param xid transaction xid
     * @param flags currently unused
     * @throws XAException XA protocol error
     */
    public void end(Xid xid, int flags) throws XAException {
        if (loggerxa.isLoggable(BasicLevel.DEBUG)) {
            loggerxa.log(BasicLevel.DEBUG, "XA-END for " + xid);
        }
    }

    /**
     * Tell the resource manager to forget about a heuristically
     * completed transaction branch.
     * @param xid transaction xid
     * @throws XAException XA protocol error
     */
    public void forget(Xid xid) throws XAException {
        // not implemented.
        if (loggerxa.isLoggable(BasicLevel.DEBUG)) {
            loggerxa.log(BasicLevel.DEBUG, "XA-FORGET for " + xid);
        }
    }

    /**
     * Obtain the current transaction timeout value set for this
     * XAResource instance.
     * @return the current transaction timeout in seconds
     * @throws XAException XA protocol error
     */
    public int getTransactionTimeout() throws XAException {
        if (logger.isLoggable(BasicLevel.DEBUG)) {
            logger.log(BasicLevel.DEBUG, "getTransactionTimeout for " + this);
        }
        return timeout;
    }

    /** Determine if the resource manager instance represented by the
     * target object is the same as the resource manager instance
     * represented by the parameter xares
     * @param xares An XAResource object
     * @return True if same RM instance, otherwise false.
     * @throws XAException XA protocol error
     */
    public boolean isSameRM(XAResource xares) throws XAException {


        // In this pseudo-driver, we must return true only if
        // both objects refer to the same XAResource, and not
        // the same Resource Manager, because actually, we must
        // send commit/rollback on each XAResource involved in
        // the transaction.
        if (xares.equals(this)) {
            if (loggerxa.isLoggable(BasicLevel.DEBUG)) {
                loggerxa.log(BasicLevel.DEBUG, "isSameRM = true " + this);
            }
            return true;
        }
        if (loggerxa.isLoggable(BasicLevel.DEBUG)) {
            loggerxa.log(BasicLevel.DEBUG, "isSameRM = false " + this);
        }
        return false;
    }


    /**
     * Ask the resource manager to prepare for a transaction commit
     * of the transaction specified in xid.
     * @param xid transaction xid
     * @throws XAException XA protocol error
     */
    public int prepare(Xid xid) throws XAException {

        if (loggerxa.isLoggable(BasicLevel.DEBUG)) {
            loggerxa.log(BasicLevel.DEBUG, "XA-PREPARE for " + xid);
        }
        // No 2PC on standard JDBC drivers
        return XA_OK;
    }

    /**
     * Obtain a list of prepared transaction branches from a resource
     * manager.
     * @return an array of transaction Xids
     * @throws XAException XA protocol error
     */
    public Xid[] recover(int flag) throws XAException {

        if (loggerxa.isLoggable(BasicLevel.DEBUG)) {
            loggerxa.log(BasicLevel.DEBUG, "XA-RECOVER for " + this);
        }
        // Not implemented
        return null;
    }

    /**
     * Inform the resource manager to roll back work done on behalf
     * of a transaction branch
     * @param xid transaction xid
     * @throws XAException XA protocol error
     */
    public void rollback(Xid xid) throws XAException {

        if (loggerxa.isLoggable(BasicLevel.DEBUG)) {
            loggerxa.log(BasicLevel.DEBUG, "XA-ROLLBACK for " + xid);
        }

        // Avoids a NPE.
        // This should not occur because a close could not be called
        // on a connection implied inside a transaction.
        if (actConn == null) {
            loggerxa.log(BasicLevel.ERROR, "rollback on a closed connection");
            return;
        }

        // Make sure that we are not in AutoCommit mode
        try {
            if (implConn.getAutoCommit() == true) {
                loggerxa.log(BasicLevel.ERROR, "Rollback called on XAResource with AutoCommit set");
                throw (new XAException(XAException.XA_HEURCOM));
            }
        } catch (SQLException e) {
            loggerxa.log(BasicLevel.ERROR, "Cannot getAutoCommit:" + e);
            notifyError(e);
            throw(new XAException("Error on getAutoCommit"));
        }

        // Rollback the transaction
        SQLException rbException = null;
        try {
            actConn.rollback();
        } catch (SQLException e) {
        	// Only store the exception at this point since the
        	// connection cannot be removed until the auto-commit
        	// mode has been reset
        	rbException = e;
            throw(new XAException("Error on rollback"));
        } finally {
        	// Reset the auto-commit flag
            try {
                implConn.setAutoCommit(true);
            } catch (Exception exc) {
                loggerxa.log(BasicLevel.DEBUG, "Unable to set autoCommit to true:", exc);
            }
            if (rbException != null) {
                loggerxa.log(BasicLevel.ERROR, "Cannot rollback transaction:" + rbException);
                // Notify the listeners of the error
                notifyError(rbException);
            }
        }
    }

    /**
     * Set the current transaction timeout value for this XAResource
     * instance.
     * @param seconds timeout value, in seconds.
     * @return always true
     * @throws XAException XA protocol error
     */
    public boolean setTransactionTimeout(int seconds) throws XAException {

        if (logger.isLoggable(BasicLevel.DEBUG)) {
            logger.log(BasicLevel.DEBUG, "setTransactionTimeout " + this);
        }
        timeout = seconds;
        return true;
    }

    /**
     * Start work on behalf of a transaction branch specified in xid
     * @param xid transaction xid
     * @throws XAException XA protocol error
     */
    public void start(Xid xid, int flags) throws XAException {

        if (loggerxa.isLoggable(BasicLevel.DEBUG)) {
            loggerxa.log(BasicLevel.DEBUG, "XA-START for " + xid);
        }
        // Unset AutoCommit mode on Connection
        try {
            implConn.setAutoCommit(false);
        } catch (Exception ex) {
            loggerxa.log(BasicLevel.ERROR, "Unable to set autoCommit to false:" + ex.getMessage());
            throw new XAException("XAResourceImpl.start : Cannot unset autoCommit");
        }
    }

    //  -----------------------------------------------------------------
    //  XAConnection implementation
    //  -----------------------------------------------------------------

    /**
     * Return an XA resource to the caller.
     * @return The XAResource
     * @exception SQLException - if a database-access error occurs
     */
    public XAResource getXAResource() throws SQLException {
        return this;
    }

    //  -----------------------------------------------------------------
    // Comparable implementation
    //  -----------------------------------------------------------------

    /**
     * Compare is based on the number of PrepareStatement cached in this Connection.
     */
    public int compareTo(Object o) {
        JManagedConnection other = (JManagedConnection) o;
        int diff = psNumber() - other.psNumber();
        if (diff == 0) {
            return getIdent() - other.getIdent();
        } else {
            return diff;
        }
    }

    public int psNumber() {
        return hitCount;
    }

    //  -----------------------------------------------------------------
    //  PooledConnection implementation
    //  -----------------------------------------------------------------

    /**
     * Create an object handle for a database connection.
     * @exception SQLException - if a database-access error occurs
     */
    public Connection getConnection() throws SQLException {
        // Just return the already created object.
        return implConn;
    }

    /**
     * Close the database connection.
     * This will physically close the Connection.
     * @exception SQLException - if a database-access error occurs
     */
    public void close() throws SQLException {
        logger.log(BasicLevel.DEBUG, "Close Physical Connection");

        // Close the actual Connection here.
        if (actConn != null) {
            actConn.close();
        } else {
            logger.log(BasicLevel.INFO, "Connection already closed");
        }
        actConn = null;
        implConn = null;
    }

    /**
     * Add an event listener.
     *
     * @param listener event listener
     */
    public void addConnectionEventListener(ConnectionEventListener listener) {
        logger.log(BasicLevel.DEBUG, "");
        eventListeners.addElement(listener);
    }

    /**
     * Remove an event listener.
     *
     * @param listener event listener
     */
    public void removeConnectionEventListener(ConnectionEventListener listener) {
        logger.log(BasicLevel.DEBUG, "");
        eventListeners.removeElement(listener);
    }

    /**
     * Enlist Connection in the Transaction
     */
    public void enlistConnection(Transaction transaction) throws javax.transaction.SystemException  {
        try {
            if (rme) {
                if (implConn == null) {
                    loggerxa.log(BasicLevel.WARN, "Cannot enlist a closed connection");
                    return;
                }
                // the tx value of the pool item does not match with the enlisted value !?
                enlistedInTx = transaction;
                // enlist the resource
                if (loggerxa.isLoggable(BasicLevel.DEBUG)) {
                    loggerxa.log(BasicLevel.DEBUG, "enlist XAResource on " + transaction);
                }
                transaction.enlistResource(getXAResource());
            }
        } catch (javax.transaction.RollbackException e) {
            javax.transaction.SystemException se = new javax.transaction.SystemException("Unexpected RollbackException exception");
            se.initCause(e);
            throw se;
        } catch (java.sql.SQLException e) {
            javax.transaction.SystemException se = new javax.transaction.SystemException("Unexpected SQL exception");
            se.initCause(e);
            throw se;
        }
    }

    /**
     * delistConnection: used for Stateful beans at postInvoke.
     */
    public void delistConnection(Transaction transaction) {
        try {
            if (rme) {
                if (loggerxa.isLoggable(BasicLevel.DEBUG)) {
                    loggerxa.log(BasicLevel.DEBUG, "delist XAResource on " + transaction);
                }
                transaction.delistResource(getXAResource(), XAResource.TMSUCCESS);
                enlistedInTx = null;
            }
        } catch (Exception e) {
            loggerxa.log(BasicLevel.WARN, "Cannot delist Resource:" + e);
        }
    }

    /**
     *  synchronization implementation
     */
    public void beforeCompletion() {
        // nothing to do
    }

    /**
     *  synchronization implementation
     */
    public void afterCompletion(int status) {
        if (tx == null ) {
            loggerxa.log(BasicLevel.ERROR, "NO TX!");
        }
        ds.freeConnections(tx != null ? tx : enlistedInTx);
    }

    /**
     * @return true if connection max age has expired
     */
    public boolean isAged() {
        return (deathTime < System.currentTimeMillis());
    }

    /**
     * @return true if connection is still open
     */
    public boolean isOpen() {
        return (open > 0);
    }

    /**
     * @return open count
     */
    public int getOpenCount() {
        return open;
    }

    /**
     * Check if the connection has been unused for too long time.
     * This occurs usually when the caller forgot to call close().
     * @return true if open time has been reached, and not involved in a tx.
     */
    public boolean inactive() {
        return (open > 0 && tx == null && enlistedInTx == null && closeTime < System.currentTimeMillis());
    }

    /**
     * @return true if connection is closed
     */
    public boolean isClosed() {
        return (open <= 0);
    }

    /**
     * @return the time when that connection was used for the first time in a sequence.
     */
    public long getOpeningTime() {
        return this.openingTime;
    }

    /**
     * @return the time when that connection was created (new instance).
     */
    public long getCreationTime() {
        return this.creationTime;
    }

    /**
     * @return the list of stack traces of clients that have opened this connection.
     */
    public List getOpenerThreadInfos() {
        return this.openerThreadInfos;
    }

    /**
     * @return the list of stack traces of clients that have closed this connection.
     */
    public List getCloserThreadInfos() {
        return this.closerThreadInfos;
    }

    /**
     * Notify as opened
     */
    public void hold() {
        if (ds.isObservable()) {
            // Store the first opening time (if connection is shared)
            if (open == 0) {
                this.openingTime = System.currentTimeMillis();
            }
            // Store the opener's stack
            openerThreadInfos.add(getThreadInfos());
        }
        open++;
        closeTime = System.currentTimeMillis() + ds.getMaxOpenTimeMilli();
    }

    /**
     * notify as closed
     * @return true if normal close.
     */
    public boolean release() {
        open--;
        if (ds.isObservable()) {
            if (open == 0) {
                // Reset the openingTime counter
                this.openingTime = DEFAULT_OPENING_TIME;
                // Reset the thread infos
                this.openerThreadInfos.clear();
                this.closerThreadInfos.clear();
            } else {
                // Store the closer's stack
                closerThreadInfos.add(getThreadInfos());
            }
        }        
        if (open < 0) {
            logger.log(BasicLevel.INFO, "connection was already closed");
            open = 0;
            return false;
        }
        if (tx == null && open > 0) {
            logger.log(BasicLevel.ERROR, "connection-open counter overflow");
            open = 0;
        }

        // Close all PreparedStatement not already closed
        // When a Connection has been closed, no PreparedStatement should
        // remain open. This can avoids lack of cursor on some databases.
        if (open == 0 && pstmtmax > 0) {
            synchronized(psList) {
                JStatement jst = null;
                Iterator i = psList.values().iterator();
                while (i.hasNext()) {
                    jst = (JStatement) i.next();
                    jst.forceClose();
                }
            }
        }

        return true;
    }

    /**
     * Compute Thread related informations.
     * @return a list containing: thread name, time of the action
     *         and a stack trace (as a string).
     */
    private Map getThreadInfos() {
        Map infos = new HashMap();
        // Name
        infos.put("thread.name", Thread.currentThread().getName());
        // Now
        infos.put("thread.time", new Long(System.currentTimeMillis()));
        // Stack trace (JDK 1.4 compatible)
        StringWriter str = new StringWriter();
        PrintWriter writer = new PrintWriter(str);
        new Throwable().printStackTrace(writer);
        infos.put("thread.stack", str.getBuffer().toString());

        return infos;
    }
    
    /**
     * Set the associated transaction
     * @param tx Transaction
     */
    public void setTx(Transaction tx) {
        this.tx = tx;
    }

    /**
     * @return the Transaction
     */
    public Transaction getTx() {
        return tx;
    }

    /**
     * @return true if registered as RME
     */
    public boolean isRME() {
        return rme;
    }

    /**
     * set/unset as RME
     */
    public void setRME(boolean rme) {
        this.rme = rme;
    }

    /**
     * remove this item, ignoring exception on close.
     */
    public void remove() {
        // Make sure that the Connection is logically closed
        release();
        // Close the physical connection
        try {
            close();
        } catch (java.sql.SQLException ign) {
            logger.log(BasicLevel.ERROR, "Could not close Connection: ", ign);
        }

        // remove all references (for GC)
        tx = null;
        enlistedInTx = null;
    }

    //  -----------------------------------------------------------------
    //  Other methods
    //  -----------------------------------------------------------------

    /**
     * Try to find a PreparedStatement in the pool
     */
    public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency)
        throws SQLException {

        if (actConn == null) {
            loggerps.log(BasicLevel.WARN, "A close has been forced on this Connection.");
            SQLException sqle = new SQLException("A close has been forced on this Connection.");
            sqle.printStackTrace();
            throw sqle;
        }
        
        loggerps.log(BasicLevel.DEBUG, sql);
        if (pstmtmax == 0) {
            return actConn.prepareStatement(sql, resultSetType, resultSetConcurrency);
        }
        JStatement ps = null;
        synchronized (psList) {
            ps = (JStatement) psList.get(sql);
            if (ps != null) {
                if (! ps.isClosed()) {
                    // Dont reuse it if already in use. (issue JONAS-109)
                    // Do not use the cache in that case.
                    return actConn.prepareStatement(sql, resultSetType, resultSetConcurrency);
                }
                ps.reuse();
                hitCount++;
            } else {
                // Not found in cache. Create a new one.
                PreparedStatement aps = actConn.prepareStatement(sql, resultSetType, resultSetConcurrency);
                ps = new JStatement(aps, this, sql);
                psList.put(sql, ps);
            }
        }
        return ps;
    }

    /**
     * Try to find a PreparedStatement in the pool
     */
    public PreparedStatement prepareStatement(String sql) throws SQLException {
        return prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
    }

    /**
     * A PreparedStatement has been logically closed.
     * @return
     */
    public void notifyPsClose(JStatement ps) {
        loggerps.log(BasicLevel.DEBUG, ps.getSql());

        // Notify event to listeners
        /*
        for (Object listener : statementEventListeners) {
            ((StatementEventListener) listener).statementClosed(new StatementEvent(this, ps));
        }*/

        if (pstmtmax == 0) {
            return;
        }
        synchronized (psList) {
            if (psList.size() >= pstmtmax) {
                // Choose a closed element to remove.
                // TODO: Should be the LRU.
                JStatement lru = null;
                Iterator i = psList.values().iterator();
                while (i.hasNext()) {
                    lru = (JStatement) i.next();
                    if (lru.isClosed()) {
                        // actually, remove the first closed element.
                        i.remove();
                        lru.forget();
                        break;
                    }
                }
            }
        }
    }

    /*
     * get the RMID
     * @return the rmid.
     */
    public String getRMID() {
        return rmid;
    }


    /*
     * Notify a Close error statement
     */
    public synchronized void notifyStatementError(PreparedStatement statement, SQLException e) {
        logger.log(BasicLevel.DEBUG, "");

        // Notify event to listeners
        /*
        for (Object listener : statementEventListeners) {
            ((StatementEventListener) listener).statementErrorOccurred(new StatementEvent(this, statement, e));
        }
        */
    }

    /*
     * Notify a Close event on Connection
     */
    public void notifyClose() {
        logger.log(BasicLevel.DEBUG, "");

        // Notify event to listeners
        logger.log(BasicLevel.DEBUG, "");
        for (int i = 0; i < eventListeners.size(); i++) {
            ConnectionEventListener l = (ConnectionEventListener) eventListeners.elementAt(i);
            l.connectionClosed(new ConnectionEvent(this));
        }
    }

    /*
     * Notify an Error event on Connection
     */
    public void notifyError(SQLException ex) {
        logger.log(BasicLevel.DEBUG, "");

        // Notify event to listeners
        for (int i = 0; i < eventListeners.size(); i++) {
            ConnectionEventListener l = (ConnectionEventListener) eventListeners.elementAt(i);
            l.connectionErrorOccurred(new ConnectionEvent(this, ex));
        }
    }

    /**
     * Registers a <code>StatementEventListener</code> with this <code>PooledConnection</code> object. Components that wish to be
     * notified when <code>PreparedStatement</code>s created by the connection are closed or are detected to be invalid may use
     * this method to register a <code>StatementEventListener</code> with this <code>PooledConnection</code> object.
     * <p>
     * @param listener an component which implements the <code>StatementEventListener</code> interface that is to be registered
     * with this <code>PooledConnection</code> object
     * <p>
     * @since 1.6
     */
    public void addStatementEventListener(StatementEventListener listener) {
        statementEventListeners.add(listener);
    }

    /**
     * Removes the specified <code>StatementEventListener</code> from the list of components that will be notified when the driver
     * detects that a <code>PreparedStatement</code> has been closed or is invalid.
     * <p>
     * @param listener the component which implements the <code>StatementEventListener</code> interface that was previously
     * registered with this <code>PooledConnection</code> object
     * <p>
     * @since 1.6
     */
    public void removeStatementEventListener(StatementEventListener listener) {
        statementEventListeners.remove(listener);
    }
}
