/***
    * Redistribution and use of this software and associated documentation
    * ("Software"), with or without modification, are permitted provided
    * that the following conditions are met:
    *
    * 1. Redistributions of source code must retain copyright
    *    statements and notices.  Redistributions must also contain a
    *    copy of this document.
    *
   * 2. Redistributions in binary form must reproduce the
   *    above copyright notice, this list of conditions and the
   *    following disclaimer in the documentation and/or other
   *    materials provided with the distribution.
   *
   * 3. The name "Exolab" must not be used to endorse or promote
   *    products derived from this Software without prior written
   *    permission of Exoffice Technologies.  For written permission,
   *    please contact info@exolab.org.
   *
   * 4. Products derived from this Software may not be called "Exolab"
   *    nor may "Exolab" appear in their names without prior written
   *    permission of Exoffice Technologies. Exolab is a registered
   *    trademark of Exoffice Technologies.
   *
   * 5. Due credit should be given to the Exolab Project
   *    (http://www.exolab.org/).
   *
   * THIS SOFTWARE IS PROVIDED BY EXOFFICE TECHNOLOGIES AND CONTRIBUTORS
   * ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT
   * NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
   * FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL
   * EXOFFICE TECHNOLOGIES OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
   * INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
   * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
   * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
   * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
   * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
   * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
   * OF THE POSSIBILITY OF SUCH DAMAGE.
   *
   * Copyright 2000-2003 (C) Exoffice Technologies Inc. All Rights Reserved.
   *
   * $Id: ObjectAdapter.java,v 1.30 2003/08/17 01:32:25 tanderson Exp $
   *
   * Date         Author  Changes
   * $Date	    jimm    Created
   */
  
  
  package org.exolab.jms.persistence;
  
  import java.io.Externalizable;
  import java.io.IOException;
  import java.io.ObjectInput;
  import java.io.ObjectOutput;
  import java.sql.Connection;
  import java.util.Enumeration;
  import java.util.HashMap;
  import java.util.Vector;
  
  import javax.jms.JMSException;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import org.exolab.core.database.recman.PMDHandle;
  import org.exolab.core.database.recman.PMDHashMap;
  import org.exolab.core.database.recman.PMDSessionManager;
  import org.exolab.core.database.recman.PMDVector;
  import org.exolab.core.database.recman.PageManagedDatabase;
  import org.exolab.core.foundation.DatabaseIOException;
  import org.exolab.core.foundation.FailedToAcquireLockException;
  import org.exolab.core.foundation.FailedToCreateCollectionException;
  import org.exolab.core.foundation.FailedToCreateDatabaseException;
  import org.exolab.core.foundation.FailedToCreateLockException;
  import org.exolab.core.foundation.FailedToCreateSessionException;
  import org.exolab.core.foundation.FailedToDestroySessionException;
  import org.exolab.core.foundation.HandleIfc;
  import org.exolab.core.foundation.ObjectNameExistsException;
  import org.exolab.core.foundation.PersistentObject;
  import org.exolab.core.foundation.SessionIfc;
  import org.exolab.core.foundation.TransactionException;
  import org.exolab.core.foundation.TransactionInProgressException;
  import org.exolab.core.foundation.TransactionNotInProgressException;
  import org.exolab.jms.authentication.User;
  import org.exolab.jms.client.JmsDestination;
  import org.exolab.jms.client.JmsQueue;
  import org.exolab.jms.client.JmsTopic;
  import org.exolab.jms.config.Configuration;
  import org.exolab.jms.config.ConfigurationManager;
  import org.exolab.jms.config.DatabaseConfiguration;
  import org.exolab.jms.events.BasicEventManager;
  import org.exolab.jms.events.Event;
  import org.exolab.jms.events.EventHandler;
  import org.exolab.jms.events.IllegalEventDefinedException;
  import org.exolab.jms.message.DestinationImpl;
  import org.exolab.jms.message.MessageImpl;
  import org.exolab.jms.messagemgr.PersistentMessageHandle;
  
 
 /***
  * This adapter is a wrapper class around the persistency mechanism.
  * It isolates the client from the working specifics of the database, by
  * providing a simple straight forward interface. Furure changes to
  * the database will only require changes to the adapter.
  *
  * <P>For this release the PageManager is used as the underlying database.
  *
  * @version     $Revision: 1.30 $ $Date: 2003/08/17 01:32:25 $
  * @author      <a href="mailto:mourikis@exolab.org">Jim Mourikis</a>
  * @see org.exolab.core.database.recman.PMDSessionManager
  * @see org.exolab.core.database.recman.PMDHashMap
  * @see	org.exolab.core.database.recman.PMDVector
  * @see org.exolab.core.foundation.PersistentObject
  */
 public class ObjectAdapter
     extends PersistenceAdapter
     implements EventHandler {
 
     // the database instance
     private PageManagedDatabase db_ = null;
 
     // The name of this database.
     private String dbname_ = null;
 
     // The max time to wait to aquire a lock
     static private int MAX_WAIT_TIME = 5000;
 
     // The root name for storing messages
     final static private String MESSAGES = "MESSAGES";
 
     // The root name for storing destinations
     final static private String DESTINATIONS = "DESTINATIONS";
 
     // All consumers or destinations have a handle
     final static private String HANDLES = "HANDLES";
 
     // All topics destinations are prefixed with
     final static private String TOPIC = "T_";
 
     // All queue destinations are prefixed with
     final static private String QUEUE = "Q_";
 
     //The root name of the max id storage location
     final static private String IDSTORAGE = "IDSTORAGE";
 
     //The database schema version root
     final static private String VERSIONID = "VERSIONID";
 
     // The schema version number. Note this must be incremented whenever
     // The schema changes.
     final static private long VERSIONNUM = 7;
 
     // The minimum persistent consumer list size
     private int minConsumerListSize_ = 10000;
 
     // The minimum persistent message list size
     private int minMessageListSize_ = 20000;
 
     /***
      * Reference to a dummy connection object, used to satisfy the fact
      * that the object adapter can look like an Connection object
      */
     private NullConnection _connection = new NullConnection();
 
     /***
      * This is the interval that the automatic garbage collector will
      * execute, if specified. It is specified in seconds.
      */
     private int _gcInterval = 600;
 
     /***
      * This is the block size that is used during purging.
      */
     private int _gcBlockSize = 500;
 
     /***
      * This is the thread priority for the GC Thread
      */
     private int _gcThreadPriority = Thread.NORM_PRIORITY;
 
     /***
      * This is the event that is fired to initiate garbage collection
      * in the database
      */
     private static final int COLLECT_DATABASE_GARBAGE_EVENT = 1;
 
     /***
      * The size of the CACHE used by the JDBM
      */
     private static final int MIN_CACHE_SIZE = 512;
 
     /***
      * The logger
      */
     private static final Log _log = LogFactory.getLog(ObjectAdapter.class);
 
 
     /***
      * Constructs a databse with the given name, if one does not exist,
      * otherwise opens the existing database.
      * The database name is checked to see if it already has a ".db" at the
      * end and stripped off if it does. The PageManagedDatabase always
      * prefixes ".db" to any name passed in.
      *
      * @throws PersistenceException for any database error
      *
      */
     public ObjectAdapter(String dbName, int minListSize, int minMsgSize,
                          int minCacheSize)
         throws PersistenceException {
 
         minConsumerListSize_ = Math.max(minConsumerListSize_, minListSize);
         minMessageListSize_ = Math.max(minMessageListSize_, minMsgSize);
 
         dbname_ = fixName(dbName);
         try {
             db_ = new PageManagedDatabase(dbname_);
 
             // set the cache object size
             //db_.setObjectCacheSize(Math.max(MIN_CACHE_SIZE, minCacheSize));
 
             // initialise the session manager with the created database and
             // then
             // check the version to ensure that it is compatible.
             PMDSessionManager.init(db_);
         } catch (FailedToCreateDatabaseException exception) {
             throw new PersistenceException(
                 "Failed to initialise database adapter", exception);
         }
 
         checkVersion();
         _log.debug("minConsumerListSize = " + minConsumerListSize_);
         _log.debug("minMessageListSize = " + minMessageListSize_);
 
         // create the roots if they do not already exist
         // jima - we really need to improve this part of the code
         try {
             createIdRoot();
         } catch (Exception exception) {
             throw new PersistenceException(
                 "Failed to create database root", exception);
         }
 
         // check whether we should initiate automatic garbage collection
         DatabaseConfiguration config =
             ConfigurationManager.getConfig().getDatabaseConfiguration();
         if (config.hasGarbageCollectionInterval()) {
             _gcInterval =
                 config.getGarbageCollectionInterval() * 1000;
             registerEvent();
         }
 
         if (config.hasGarbageCollectionBlockSize()) {
             _gcBlockSize =
                 config.getGarbageCollectionBlockSize();
         }
 
         if (config.hasGarbageCollectionThreadPriority()) {
             _gcThreadPriority =
                 config.getGarbageCollectionBlockSize();
             if (_gcThreadPriority < Thread.MIN_PRIORITY) {
                 _gcThreadPriority = Thread.MIN_PRIORITY;
             } else if (_gcThreadPriority > Thread.MAX_PRIORITY) {
                 _gcThreadPriority = Thread.MAX_PRIORITY;
             }
         }
     }
 
     /***
      * Fix up the database name, by striping off any ".xx" suffixes.
      *
      */
     private String fixName(String dbName) {
         String newDbName = dbName;
         newDbName.trim();
 
         int i = newDbName.lastIndexOf('.');
         if (i > 0 && i < (newDbName.length() - 1)) {
             newDbName = newDbName.substring(0, i);
         }
         return newDbName;
     }
 
     /***
      * Close the database if open.
      *
      */
     public void close() {
         if (db_ != null) {
             db_.close();
             dbname_ = null;
             db_ = null;
         }
     }
 
     /***
      * Check to see if the version is correct. if this is the first time the
      * database has been opened, then created the version root, and itialise
      * the version.
      *
      */
     private void checkVersion() {
         try {
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
             if ((vector = (PMDVector) session.lookup(VERSIONID)) == null) {
                 _log.debug("Setting Version Id");
                 vector =
                     (PMDVector) session.getCollectionManager().createVector();
                 session.createObject(vector);
                 session.bind(VERSIONID, vector);
                 session.getCurrentTransaction().commit();
 
                 session.getCurrentTransaction().begin();
                 vector = (PMDVector) session.lookup(VERSIONID);
                 session.acquireLock(vector, 10);
                 vector.addElement(new PMDLongInteger(VERSIONNUM));
                 session.updateObject(vector);
                 session.getCurrentTransaction().commit();
             } else {
                 if (vector.size() == 1) {
                     long ver;
 
                     if ((ver = ((PMDLongInteger) vector.get(0)).get())
                         != VERSIONNUM) {
                         _log.error(
                             "Incompatible Database version Schema\n" +
                             "Db Vesrsion = " + ver + "\tSchema Version = "
                             + VERSIONNUM + "\nExiting...");
                         System.exit(-1);
                     }
                 } else {
                     _log.error("Corrupted Db schema version ID:\n " +
                         "Exiting...");
                     System.exit(-1);
                 }
                 session.getCurrentTransaction().abort();
             }
             try {
                 PMDSessionManager.instance().destroySession();
             } catch (FailedToDestroySessionException exception) {
                 _log.error("Failed to destroy session", exception);
             }
         } catch (Exception exception) {
             _log.error("Error verifying DB schema, exiting", exception);
             System.exit(-1);
         }
 
     }
 
     /***
      * Check to see if the Id root has been created.
      * If not create it and initialise it. This is used as a sequencer
      * to number all Messages that are delivered to OpenJMS server.
      *
      */
     private void createIdRoot()
         throws FailedToCreateSessionException, TransactionInProgressException,
         FailedToCreateCollectionException, DatabaseIOException,
         TransactionNotInProgressException, ObjectNameExistsException,
         TransactionException, FailedToCreateLockException,
         FailedToAcquireLockException, FailedToDestroySessionException {
         PMDVector vector;
         PMDHashMap map;
         SessionIfc session = getSession();
 
         session.getCurrentTransaction().begin();
         if ((vector = (PMDVector) session.lookup(IDSTORAGE)) == null) {
             _log.debug("Creating ID Storage root");
             vector =
                 (PMDVector) session.getCollectionManager().createVector();
             session.createObject(vector);
             session.bind(IDSTORAGE, vector);
             session.getCurrentTransaction().commit();
 
             session.getCurrentTransaction().begin();
             vector = (PMDVector) session.lookup(IDSTORAGE);
             session.acquireLock(vector, MAX_WAIT_TIME);
             vector.addElement(new PMDLongInteger(0));
             session.updateObject(vector);
             session.getCurrentTransaction().commit();
         } else {
             session.getCurrentTransaction().abort();
         }
 
         session.getCurrentTransaction().begin();
         if ((map = (PMDHashMap) session.lookup(MESSAGES)) == null) {
             _log.debug("Creating Messages root");
             map =
                 (PMDHashMap) session.getCollectionManager().createHashMap();
             session.createObject(map);
             session.bind(MESSAGES, map);
             session.getCurrentTransaction().commit();
         } else {
             session.getCurrentTransaction().abort();
         }
 
         session.getCurrentTransaction().begin();
         if ((vector = (PMDVector) session.lookup(DESTINATIONS)) == null) {
             _log.debug("Creating Destinations root");
             vector =
                 (PMDVector) session.getCollectionManager().createVector();
             session.createObject(vector);
             session.bind(DESTINATIONS, vector);
             session.getCurrentTransaction().commit();
         } else {
             session.getCurrentTransaction().abort();
         }
         PMDSessionManager.instance().destroySession();
     }
 
     // implementation of PersistenceAdapter.getLastIds
     public long getLastId(Connection connection)
         throws PersistenceException {
         long lastId = -1;
 
         try {
             createIdRoot();
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
             if ((vector = (PMDVector) session.lookup(IDSTORAGE)) != null) {
                 if (vector.size() == 1) {
                     lastId = ((PMDLongInteger) vector.get(0)).get();
                 }
             }
             session.getCurrentTransaction().abort();
         } catch (Exception err) {
             _log.error("Failed to get Id: " + err.getMessage());
         }
         try {
             PMDSessionManager.instance().destroySession();
         } catch (FailedToDestroySessionException sessErr) {
             _log.error("Failed to destroy session: " +
                 sessErr.getMessage());
             sessErr.printStackTrace();
             // Session not destroyed.
         }
         return lastId;
     }
 
     // implementation of PersistenceAdapter.updateIds
     public void updateIds(Connection connection, long id)
         throws PersistenceException {
         try {
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
             if ((vector = (PMDVector) session.lookup(IDSTORAGE)) != null) {
                 session.acquireLock(vector, MAX_WAIT_TIME);
                 if (vector.size() == 1) {
                     vector.set(0, new PMDLongInteger(id));
                     session.updateObject(vector);
                 }
 
             }
             session.getCurrentTransaction().commit();
         } catch (Exception err) {
             throw new PersistenceException("Failed to get Id: " + err.getMessage());
         }
         try {
             PMDSessionManager.instance().destroySession();
         } catch (FailedToDestroySessionException sessErr) {
             _log.error("Failed to destroy session", sessErr);
         }
     }
 
     // implementation of PersistenceAdapter.addMessage
     public synchronized void addMessage(Connection connection,
                                         MessageImpl message)
         throws PersistenceException {
 
         try {
             DestinationImpl dest =
                 (DestinationImpl) message.getJMSDestination();
             String name = dest.getDestination();
             PMDHashMap messages;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             messages = messageTable(session);
 
             if (messages != null) {
                 try {
                     session.acquireLock(messages, MAX_WAIT_TIME);
 
                     // check to see that the message does not exist
                     String id = message.getMessageId().getId();
                     if (messages.get(id) == null) {
                         PersistentMessage pm = new PersistentMessage(message);
                         session.createObject(pm);
                         // when we initially add a message to the database it
                         // must go in the unprocessed list
                         messages.put(id, session.createHandle(pm));
                         session.updateObject(messages);
                     } else {
                         throw new PersistenceException("Mesage with this id: " +
                             message.getJMSMessageID() +
                             " already exists: NOT ADDING");
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in addMessage " +
                         err.toString());
                 }
             } else {
                 _log.error("Attempt to save message for non register" +
                     " queue/topic, name = " + name);
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException exception) {
             // rethrowe it
             throw exception;
         } catch (Exception err) {
             throw new PersistenceException("Error in addMessage " + err.toString());
         }
     }
 
     // implementation of PersistenceAdapter.updateMessage
     public synchronized void updateMessage(Connection connection,
                                            MessageImpl message)
         throws PersistenceException {
 
         try {
             DestinationImpl dest =
                 (DestinationImpl) message.getJMSDestination();
             String name = dest.getDestination();
             PMDHashMap messages;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             messages = messageTable(session);
 
             if (messages != null) {
                 try {
                     session.acquireLock(messages, MAX_WAIT_TIME);
 
                     // check to see that the message does not exist
                     String id = message.getMessageId().getId();
                     if (messages.get(id) != null) {
                         // remove the old message and update it
                         PMDHandle handle = (PMDHandle) messages.get(id);
                         PersistentMessage pm = (PersistentMessage) handle.resolve();
                         pm.setMessage(message);
                         pm.setProcessed(message.getProcessed());
                         session.updateObject(pm);
                     } else {
                         throw new PersistenceException("Mesage with this id: " +
                             message.getJMSMessageID() +
                             " doesn't exists: NOT UPDATING");
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in updateMessage " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Attempt to save message for non register" +
                     " queue/topic, name = " + name);
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in updateMessage " +
                 err.toString());
         }
     }
 
     // implementation of PersistenceAdapter.removeMessage
     public synchronized void removeMessage(Connection connection, String id)
         throws PersistenceException {
 
         try {
             PMDHashMap map;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             if ((map = messageTable(session)) != null) {
                 try {
                     session.acquireLock(map, MAX_WAIT_TIME);
                     PMDHandle handle = (PMDHandle) map.remove(id);
                     if (handle != null) {
                         session.deleteObject
                             ((PersistentMessage) handle.resolve());
                     }
 
                     session.updateObject(map);
                 } catch (Exception err) {
                     throw new PersistenceException("Error in removeMessage " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in removeMessage " +
                     "Cannot retrieve the message talbe ");
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in removeMessage " +
                 err.toString());
         }
     }
 
     // implementation of PersistenceAdapter.getMessage
     public synchronized MessageImpl getMessage(Connection connection, String id)
         throws PersistenceException {
         MessageImpl message = null;
 
         try {
             PMDHashMap map;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             if ((map = messageTable(session)) != null) {
                 try {
                     session.acquireLock(map, MAX_WAIT_TIME);
                     PMDHandle handle = (PMDHandle) map.get(id);
                     PersistentMessage m = null;
 
                     if (handle != null) {
                         m = (PersistentMessage) handle.resolve();
                     }
 
                     if (m != null) {
                         message = m.getMessage();
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in getMessage " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in getMessage " +
                     "Failed to retrieve the message table.");
             }
             session.getCurrentTransaction().abort();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in getMessage " +
                 err.toString());
         }
         return message;
     }
 
     // implementation of PersistenceAdapter.getUnprocessedMessages
     public synchronized Vector getUnprocessedMessages(Connection connection)
         throws PersistenceException {
         Vector result = new Vector();
 
         try {
             PMDHashMap map;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             if ((map = messageTable(session)) != null) {
                 try {
                     session.acquireLock(map, MAX_WAIT_TIME);
                     Enumeration iter = map.elements();
                     while (iter.hasMoreElements()) {
                         PMDHandle handle = (PMDHandle) iter.nextElement();
                         PersistentMessage m =
                             (PersistentMessage) handle.resolve();
 
                         if (!m.getProcessed()) {
                             result.add(m.getMessage());
                         }
                     }
                 } catch (Exception err) {
                     throw new PersistenceException(
                         "Error in getUnprocessedMessages " + err.toString());
                 }
             } else {
                 throw new PersistenceException(
                     "Error in getUnprocessedMessage. Failed to get message table");
             }
             session.getCurrentTransaction().abort();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in getUnprocessedMessages "
                 + err.toString());
         }
         return result;
     }
 
     // implementation of PersistenceAdapter.getMessages
     public synchronized Vector getMessages(Connection connection,
                                            PersistentMessageHandle handle)
         throws PersistenceException {
         Vector messages = new Vector();
 
         // for the jdbm only ever retrieve a single message. We could improve
         // this at a later date.
         MessageImpl message = getMessage(connection,
             handle.getMessageId().getId());
         if (message != null) {
             messages.add(message);
         }
 
         return messages;
     }
 
     // implementation of PersistenceAdapter.addMessageHandle
     public synchronized void addMessageHandle(Connection connection,
                                               PersistentMessageHandle handle)
         throws PersistenceException {
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
             String key = getHandlesRootName(handle.getDestination(),
                 handle.getConsumerName());
 
 
             session.getCurrentTransaction().begin();
 
             if ((vector = handleTable(key, session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
                     vector.addElement(handle);
                     session.updateObject(vector);
                 } catch (Exception err) {
                     throw new PersistenceException("Error in addMessageHandle " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in addMessageHandle " +
                     "Cannot get handle table for " + key);
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in addMessageHandle " +
                 err.toString());
         }
     }
 
     // implementation of PersistenceAdapter.updateMessageHandle
     public synchronized void updateMessageHandle(Connection connection,
                                                  PersistentMessageHandle handle)
         throws PersistenceException {
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
             String key = getHandlesRootName(handle.getDestination(),
                 handle.getConsumerName());
 
 
             session.getCurrentTransaction().begin();
 
             if ((vector = handleTable(key, session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
 
                     // linear search for the matching handle
                     Enumeration handles = vector.elements();
                     while (handles.hasMoreElements()) {
                         PersistentMessageHandle phdl =
                             (PersistentMessageHandle) handles.nextElement();
 
                         if (phdl.getMessageId().getId().equals(
                             handle.getMessageId().getId())) {
                             phdl.setDelivered(true);
                             break;
                         }
                     }
                     session.updateObject(vector);
                 } catch (Exception err) {
                     throw new PersistenceException("Error in addMessageHandle " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in updateMessageHandle " +
                     "Failed to get handle table for " + key);
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in updateMessageHandle " +
                 err.toString());
         }
     }
 
     // implementation of PersistenceAdapter.removeMessageHandle
     public synchronized void removeMessageHandle(Connection connection,
                                                  PersistentMessageHandle handle)
         throws PersistenceException {
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
             String key = getHandlesRootName(handle.getDestination(),
                 handle.getConsumerName());
 
 
             session.getCurrentTransaction().begin();
 
             if ((vector = handleTable(key, session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
                     vector.remove(handle);
                     session.updateObject(vector);
                 } catch (Exception err) {
                     throw new PersistenceException("Error in removeMessageHandle " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in removeMessageHandle " +
                     "Failed to get the handle table for " + key);
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in removeMessageHandle " +
                 err.toString());
         }
     }
 
     // implementation of PersistenceAdapter.getMessageHandles
     public synchronized Vector getMessageHandles(Connection connection,
                                                  JmsDestination destination, String name)
         throws PersistenceException {
         Vector result = new Vector();
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
             String key = getHandlesRootName(destination, name);
 
 
             session.getCurrentTransaction().begin();
 
             if ((vector = handleTable(key, session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
 
                     Enumeration handles = vector.elements();
                     while (handles.hasMoreElements()) {
                         PersistentMessageHandle handle =
                             (PersistentMessageHandle) handles.nextElement();
                         result.addElement(handle.clone());
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in getMessageHandles " +
                         err.toString());
                 }
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in getMessageHandles " +
                 err.toString());
         }
         return result;
     }
 
     // implementation of PersistenceAdapter.addDurableConsumer
     public synchronized void addDurableConsumer(Connection connection,
                                                 String topic, String consumer)
         throws PersistenceException {
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             if ((vector = destinationTable(session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
                     String target = "@" + consumer;
 
                     boolean found = false;
                     Enumeration entries = vector.elements();
                     while (entries.hasMoreElements()) {
                         PersistentString entry = (PersistentString) entries.nextElement();
                         if (entry.toString().endsWith(target)) {
                             found = true;
                             break;
                         }
                     }
 
                     String key = null;
                     if (!found) {
                         // add the target and also create a new table
                         // to hold handles for this consumer
                         vector.addElement(new PersistentString(TOPIC + topic + target));
                         session.updateObject(vector);
 
                         key = getHandlesRootName(consumer);
                         PMDVector handles =
                             (PMDVector) session.getCollectionManager().createVector();
                         session.createObject(handles);
                         session.bind(key, handles);
                     } else {
                         throw new PersistenceException("Error in addDurableConsumer " +
                             consumer + " already exists.");
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in addDurableConsumer " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in addDurableConsumer " +
                     "Failed to get the destination table.");
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in addDurableConsumer " +
                 err.toString());
         }
     }
 
     // implementation of PersistenceAdapter.removeDurableConsumer
     public synchronized void removeDurableConsumer(Connection connection,
                                                    String consumer)
         throws PersistenceException {
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             if ((vector = destinationTable(session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
                     String target = "@" + consumer;
 
                     boolean found = false;
                     PersistentString entry = null;
                     Enumeration entries = vector.elements();
                     while (entries.hasMoreElements()) {
                         entry = (PersistentString) entries.nextElement();
                         if (entry.toString().endsWith(target)) {
                             vector.remove(entry);
                             found = true;
                             break;
                         }
                     }
 
                     String key = null;
                     if (found) {
                         // add the target and also create a new table
                         // to hold handles for this consumer
                         vector.removeElement(entry);
                         session.updateObject(vector);
 
                         // unbind the table (i.e delete it)
                         key = getHandlesRootName(consumer);
                         session.unbind(key);
                     } else {
                         throw new PersistenceException("Error in removeDurableConsumer " +
                             "Cannot find consumer with name " + consumer);
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in removeDurableConsumer " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in removeDurableConsumer " +
                     "Cannot get access to the destination table");
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in removeDurableConsumer " +
                 err.toString());
         }
     }
 
     // implementation of PersistenceAdapter.durableConsumerExists
     public synchronized boolean durableConsumerExists(Connection connection,
                                                       String name)
         throws PersistenceException {
         boolean exists = false;
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             if ((vector = destinationTable(session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
                     String target = "@" + name;
 
                     Enumeration entries = vector.elements();
                     while (entries.hasMoreElements()) {
                         PersistentString entry = (PersistentString) entries.nextElement();
                         if (entry.toString().endsWith(target)) {
                             exists = true;
                             break;
                         }
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in durableConsumerExists " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in durableConsumerExists " +
                     "Cannot get access to the destinationtable.");
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in durableConsumerExists " +
                 err.toString());
         }
         return exists;
     }
 
     // implementation of PersistenceAdapter.getDurableConsumers
     public synchronized Enumeration getDurableConsumers(Connection connection,
                                                         String topic)
         throws PersistenceException {
         Vector consumers = new Vector();
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             if ((vector = destinationTable(session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
                     String target = TOPIC + topic + "@";
 
                     Enumeration entries = vector.elements();
                     while (entries.hasMoreElements()) {
                         PersistentString entry =
                             (PersistentString) entries.nextElement();
                         if (entry.toString().startsWith(target)) {
                             consumers.addElement(
                                 entry.toString().substring(target.length()));
                         }
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in getDurableConsumers " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in getDurableConsumers " +
                     "Failed to get access to the destination table.");
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in getDurableConsumers " +
                 err.toString());
         }
         return consumers.elements();
     }
 
     // implementation of PersistenceAdapter.getAllDurableConsumers
     public HashMap getAllDurableConsumers(Connection connection)
         throws PersistenceException {
         HashMap consumers = new HashMap();
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             if ((vector = destinationTable(session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
 
                     Enumeration entries = vector.elements();
                     while (entries.hasMoreElements()) {
                         PersistentString entry =
                             (PersistentString) entries.nextElement();
                         String temp = entry.toString().substring(TOPIC.length());
                         int index = temp.indexOf("@");
                         if (!(entry.toString().startsWith(TOPIC)) ||
                             (index == -1)) {
                             // this is not a durable consumer so continue
                             continue;
                         }
 
                         // we have a durable consumer
                         consumers.put(temp.substring(index + 1),
                             temp.substring(0, index));
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in getAllDurableConsumers " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in getAllDurableConsumers " +
                     "Failed to get access to the destination  table");
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in getAllDurableConsumers " +
                 err.toString());
         }
         return consumers;
     }
 
     // implementation of PersistenceAdapter.addDestination
     public synchronized void addDestination(Connection connection,
                                             String name, boolean queue)
         throws PersistenceException {
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             if ((vector = destinationTable(session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
                     String target;
                     if (queue) {
                         target = QUEUE + name;
                     } else {
                         target = TOPIC + name;
                     }
                     boolean found = false;
                     Enumeration entries = vector.elements();
                     while (entries.hasMoreElements()) {
                         PersistentString entry = (PersistentString) entries.nextElement();
                         if (entry.toString().equals(target)) {
                             found = true;
                             break;
                         }
                     }
 
                     String key = null;
                     if (!found) {
                         // add the target and also create a new table
                         // to hold handles for this consumer
                         vector.addElement(new PersistentString(target));
                         session.updateObject(vector);
 
                         // if it is a queue then we also need to construct
                         // a handles table
                         if (queue) {
                             key = getHandlesRootName(name);
                             PMDVector handles =
                                 (PMDVector) session.getCollectionManager().createVector();
                             session.createObject(handles);
                             session.bind(key, handles);
                         }
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in addDestination " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in addDestination " +
                     "Failed to get access to destination table");
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in addDestination " +
                 err.toString());
         }
     }
 
     // implementation of PersistenceAdapter.removeDestination
     public synchronized void removeDestination(Connection connection,
                                                String name)
         throws PersistenceException {
         try {
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             if ((vector = destinationTable(session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
 
                     // this will remove the destination and all registered
                     // consumers.
                     Vector to_delete = new Vector();
                     Enumeration entries = vector.elements();
                     while (entries.hasMoreElements()) {
                         PersistentString wrapped_entry =
                             (PersistentString) entries.nextElement();
                         String entry = wrapped_entry.toString();
 
                         // now the destination name must appear before
                         // the '.' separator
                         int name_idx = entry.indexOf("@");
                         int dest_idx = entry.indexOf(name);
 
                         if (entry.substring(TOPIC.length()).equals(name)) {
                             // we have found the actual destination entry
                             // Firstly mark it for deletion and then if it is a queue
                             // unbind the consumer state
                             to_delete.addElement(wrapped_entry);
                             if (entry.startsWith(QUEUE)) {
                                 session.unbind(getHandlesRootName(name));
                             }
                         } else if ((name_idx >= 0) &&
                             (dest_idx >= 0) &&
                             (dest_idx < name_idx)) {
                             // we have located an entry which describes a
                             // durable consumer for a topic
                             String consumer = entry.substring(name_idx + 1);
                             to_delete.addElement(wrapped_entry);
                             session.unbind(getHandlesRootName(consumer));
                         }
                     }
 
                     //  now we need to go through and process the to_delete
                     // entries
                     Enumeration elements = to_delete.elements();
                     while (elements.hasMoreElements()) {
                         vector.remove((PersistentString) elements.nextElement());
                     }
 
                     // update the vector object
                     session.updateObject(vector);
                 } catch (Exception err) {
                     throw new PersistenceException("Error in removeDestination " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in removeDestination " +
                     "Failed to get access to the destination table.");
             }
 
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in addDestination " +
                 err.toString());
         }
     }
 
     // implementation of PersistenceAdapter.checkDestination
     public boolean checkDestination(Connection connection, String name)
         throws PersistenceException {
         boolean success = false;
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             if ((vector = destinationTable(session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
 
                     // this will remove the destination and all registered
                     // consumers.
                     Vector to_delete = new Vector();
                     Enumeration entries = vector.elements();
                     while (entries.hasMoreElements()) {
                         PersistentString entry =
                             (PersistentString) entries.nextElement();
 
                         // check that it is the specified destination
                         if (entry.toString().substring(
                             TOPIC.length()).equals(name)) {
                             success = true;
                             break;
                         }
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in checkDestination " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in checkDestination " +
                     "Failed to get access to destination table.");
             }
 
             session.getCurrentTransaction().abort();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in checkDestination " +
                 err.toString());
         }
         return success;
     }
 
     // implementation of PersistenceAdapter.getAllDestinations
     public synchronized Enumeration getAllDestinations(Connection connection)
         throws PersistenceException {
         Vector destinations = new Vector();
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
 
             if ((vector = destinationTable(session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
 
                     // this will remove the destination and all registered
                     // consumers.
                     Vector to_delete = new Vector();
                     Enumeration entries = vector.elements();
                     while (entries.hasMoreElements()) {
                         PersistentString entry =
                             (PersistentString) entries.nextElement();
                         String dest = entry.toString();
                         // check that it is the specified destination
                         if (dest.indexOf("@") == -1) {
                             if (dest.startsWith(QUEUE)) {
                                 destinations.addElement(
                                     new JmsQueue(dest.substring(QUEUE.length())));
                             } else if (dest.startsWith(TOPIC)) {
                                 destinations.addElement(
                                     new JmsTopic(dest.substring(TOPIC.length())));
                             }
                         }
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in checkDestination " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in checkDestination " +
                     "Failed to get access to the destination table.");
             }
 
             session.getCurrentTransaction().abort();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in checkDestination " +
                 err.toString());
         }
 
         return destinations.elements();
     }
 
     // implementation of PersistenceAdapter.getQueueMessageCount
     public synchronized int getQueueMessageCount(Connection connection,
                                                  String queue)
         throws PersistenceException {
         int count = -1;
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
             String key = getHandlesRootName(queue);
 
 
             session.getCurrentTransaction().begin();
 
             if ((vector = handleTable(key, session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
                     count = vector.size();
                 } catch (Exception err) {
                     throw new PersistenceException("Error in getQueueMessageCount " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in getQueueMessageCount " +
                     "Failed to get access to queue " + queue);
             }
             session.getCurrentTransaction().abort();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in getQueueMessageCount " +
                 err.toString());
         }
 
         return count;
     }
 
     // implementation of PersistenceAdapter.getQueueMessageCount
     public synchronized int getDurableConsumerMessageCount(Connection connection,
                                                            String topic, String name)
         throws PersistenceException {
         int count = -1;
 
         try {
             PMDVector vector;
             SessionIfc session = getSession();
             String key = getHandlesRootName(name);
 
 
             session.getCurrentTransaction().begin();
 
             if ((vector = handleTable(key, session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
                     count = vector.size();
                 } catch (Exception err) {
                     throw new PersistenceException("Error in getDurableConsumerMessageCount " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in getDurableConsumerMessageCount " +
                     "Cannot access table for " + topic + " : " + name);
             }
 
             session.getCurrentTransaction().abort();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in getDurableConsumerMessageCount " +
                 err.toString());
         }
 
         return count;
     }
 
     // implementation of PersistenceAdapter.getQueueMessageCount
     public synchronized void removeExpiredMessages(Connection connection)
         throws PersistenceException {
 
         SessionIfc session = null;
         try {
             PMDVector vector;
             PMDHashMap map;
             session = getSession();
 
             session.getCurrentTransaction().begin();
             long now = System.currentTimeMillis();
 
             if ((map = messageTable(session)) != null) {
                 try {
                     session.acquireLock(map, MAX_WAIT_TIME);
                     Enumeration iter = map.elements();
                     Vector to_remove = new Vector();
 
                     // collect a list of messages that have expired so that
                     // they can be deleted later
                     while (iter.hasMoreElements()) {
                         PMDHandle handle = (PMDHandle) iter.nextElement();
                         PersistentMessage msg = (PersistentMessage) handle.resolve();
                         if ((msg.getExpiryTime() != 0) &&
                             (msg.getExpiryTime() <= now)) {
                             session.deleteObject(msg);
                             to_remove.add(msg.getMessage().getJMSMessageID());
                         }
                     }
 
                     // delete the expired messages
                     while (to_remove.size() > 0) {
                         map.remove(to_remove.remove(0));
                     }
                     session.updateObject(map);
                 } catch (Exception err) {
                     throw new PersistenceException("Error in removeExpiredMessages " +
                         err.toString());
                 }
 
                 // now we need to go through all the handle tables and
                 // remove expired messages. Very long and tedious
                 if ((vector = destinationTable(session)) != null) {
                     try {
                         session.acquireLock(vector, MAX_WAIT_TIME);
                         Enumeration entries = vector.elements();
                         Vector to_remove = new Vector();
 
                         while (entries.hasMoreElements()) {
                             to_remove.clear();
                             PersistentString entry = (PersistentString) entries.nextElement();
                             String name = getHandlesRootNameFromDestination(
                                 entry.toString());
                             if (name == null) {
                                 continue;
                             }
 
                             // retrieve the handle tbale based on the name
                             PMDVector handles_vector = handleTable(name, session);
                             if (handles_vector == null) {
                                 continue;
                             }
 
                             // we have the handle table. Now iterate over all the
                             // messages and removed expired messages
                             Enumeration handles = handles_vector.elements();
                             while (handles.hasMoreElements()) {
                                 PersistentMessageHandle handle =
                                     (PersistentMessageHandle) handles.nextElement();
                                 if ((handle.getExpiryTime() != 0) &&
                                     (handle.getExpiryTime() <= now)) {
                                     to_remove.add(handle);
                                 }
                             }
 
                             // now remove the handles
                             while (to_remove.size() > 0) {
                                 handles_vector.remove(
                                     (PersistentMessageHandle) to_remove.remove(0));
                             }
                             session.updateObject(handles_vector);
                         }
                     } catch (Exception err) {
                         throw new PersistenceException("Error in removeExpiredMessages " +
                             err.toString());
                     }
                 } else {
                     throw new PersistenceException("Error in removeExpiredMessages " +
                         "Failed to get the message table");
                 }
             }
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             try {
                 session.getCurrentTransaction().abort();
             } catch (Exception exception) {
                 // ignore
             }
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in removeExpiredMessages " +
                 err.toString());
         }
     }
 
     // implementation of PersistenceAdapter.removeExpiredMessageHandles
     public void removeExpiredMessageHandles(Connection connection,
                                             String consumer)
         throws PersistenceException {
 
         // no operation
     }
 
     // implementation of PersistenceAdapter.getQueueMessageCount
     public synchronized Vector getNonExpiredMessages(Connection connection,
                                                      JmsDestination destination)
         throws PersistenceException {
         Vector result = new Vector();
 
         SessionIfc session = null;
         try {
             PMDVector vector;
             session = getSession();
 
             session.getCurrentTransaction().begin();
             long now = System.currentTimeMillis();
 
             // now we need to go through all the handle tables and
             // retrieve messages which have not expired for the specified
             // destination
             if ((vector = destinationTable(session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
                     Enumeration entries = vector.elements();
                     while (entries.hasMoreElements()) {
                         PersistentString entry = (PersistentString) entries.nextElement();
 
                         // check to see if we are interested in this destination
                         if (entry.toString().indexOf(destination.getName()) == -1) {
                             continue;
                         }
 
                         // attempt to retrieve the handle name table for the specified
                         // destination,
                         String name = getHandlesRootNameFromDestination(
                             entry.toString());
                         if (name == null) {
                             continue;
                         }
 
                         // retrieve the handle tbale based on the name
                         PMDVector handles_vector = handleTable(name, session);
                         if (handles_vector == null) {
                             continue;
                         }
 
                         // we have the handle table. Now iterate over all the
                         // messages and removed expired messages
                         Enumeration handles = handles_vector.elements();
                         while (handles.hasMoreElements()) {
                             PersistentMessageHandle handle =
                                 (PersistentMessageHandle) handles.nextElement();
                             if ((handle.getExpiryTime() != 0) &&
                                 (handle.getExpiryTime() > now)) {
                                 result.add(handle);
                             }
                         }
                     }
                 } catch (Exception err) {
                     throw new PersistenceException("Error in removeExpiredMessages " +
                         err.toString());
                 }
             } else {
                 throw new PersistenceException("Error in removeExpiredMessages " +
                     "The destination table does not exist");
             }
             session.getCurrentTransaction().commit();
             PMDSessionManager.instance().destroySession();
         } catch (PersistenceException pe) {
             try {
                 session.getCurrentTransaction().abort();
             } catch (Exception exception) {
                 // ignore
             }
             throw pe;
         } catch (Exception err) {
             throw new PersistenceException("Error in removeExpiredMessages " +
                 err.toString());
         }
 
         return result;
     }
 
     // implementation of PersistenceAdapter.purgeMessages
     public synchronized int purgeMessages() {
         boolean errorSet = false;
         Vector keep = new Vector();
         int count = -1;
 
         try {
             PMDHashMap map;
             PMDVector vector;
             SessionIfc session = getSession();
 
             session.getCurrentTransaction().begin();
             long now = System.currentTimeMillis();
 
             // now we need to go through all the handle tables and
             // get a unique list of unacked messages
             if ((vector = destinationTable(session)) != null) {
                 try {
                     session.acquireLock(vector, MAX_WAIT_TIME);
                     Enumeration entries = vector.elements();
                     while (entries.hasMoreElements()) {
                         PersistentString entry = (PersistentString) entries.nextElement();
 
                         // attempt to retrieve the handle name table for the specified
                         // destination,
                         String name = getHandlesRootNameFromDestination(
                             entry.toString());
                         if (name == null) {
                             continue;
                         }
                         // retrieve the handle table based on the name
                         PMDVector handles_vector = handleTable(name, session);
                         if (handles_vector == null) {
                             continue;
                         }
 
                         // we have the handle table. Now iterate over all the
                         // messages and removed expired messages
                         Enumeration handles = handles_vector.elements();
                         while (handles.hasMoreElements()) {
                             PersistentMessageHandle handle =
                                 (PersistentMessageHandle) handles.nextElement();
                             String id = handle.getMessageId().getId();
                             if (!keep.contains(id)) {
                                 keep.add(id);
                             }
                         }
                     }
 
 
                     // now that we have a definitive list of messages we
                     // need to remove messages that are not in the list
                     if ((map = messageTable(session)) != null) {
                         try {
                             session.acquireLock(map, MAX_WAIT_TIME);
                             Enumeration iter = map.keys();
                             Vector to_remove = new Vector();
 
                             // collect a list of messages that have expired so that
                             // they can be deleted later
                             while (iter.hasMoreElements()) {
                                 String id = (String) iter.nextElement();
                                 if (!keep.contains(id)) {
                                     session.deleteObject((PersistentMessage)
                                         ((PMDHandle) map.get(id)).resolve());
                                     to_remove.add(id);
                                 }
                             }
 
                             // delete the expired messages
                             count = to_remove.size();
                             while (to_remove.size() > 0) {
                                 map.remove(to_remove.remove(0));
                             }
                             session.updateObject(map);
                             _log.info("GC removed " +
                                 count + " messages from database.");
                         } catch (Exception exception) {
                             errorSet = true;
                             _log.error("Error in purgeMessages", exception);
                         }
                     } else {
                         _log.error("Message table does not exist");
                         errorSet = true;
                     }
                 } catch (Exception err) {
                     errorSet = true;
                     _log.error("Error in getNonExpiredMessages", err);
                 }
             } else {
                 // topic does not exists, may have been removed already.
                 errorSet = true;
             }
 
             if (!errorSet) {
                 session.getCurrentTransaction().commit();
             } else {
                 session.getCurrentTransaction().abort();
             }
             PMDSessionManager.instance().destroySession();
         } catch (FailedToDestroySessionException sessErr) {
             // Session not destroyed but message saved.
             _log.error("Failed to destroy session", sessErr);
         } catch (Exception err) {
             _log.error(err, err);
         }
 
         return count;
     }
 
     public void addUser(Connection connection, User user)
         throws PersistenceException {
         // unsupported
     }
 
     public Enumeration getAllUsers(Connection connection)
         throws PersistenceException {
         // unsupported
         return new Vector().elements();
     }
 
     public User getUser(Connection connection, User user)
         throws PersistenceException {
         // unsupported
         return null;
     }
 
     public void removeUser(Connection connection, User user)
         throws PersistenceException {
         // unsupported
     }
 
     public void updateUser(Connection connection, User user)
         throws PersistenceException {
         // unsupported
     }
 
     // implementation of PersistenceAdapter.getConnection
     public Connection getConnection()
         throws PersistenceException {
         return _connection;
     }
 
     // implementation of EventHandler.getHandle
     public HandleIfc getHandle() {
         return null;
     }
 
     // implementation of EventHandler.handleEvent
     public void handleEvent(int event, Object callback, long time) {
         if (event == COLLECT_DATABASE_GARBAGE_EVENT) {
             // collect garbage now, but before doing so change the thread
             // priority to low.
             try {
                 Thread.currentThread().setPriority(_gcThreadPriority);
                 purgeMessages();
             } finally {
                 Thread.currentThread().setPriority(Thread.NORM_PRIORITY);
                 registerEvent();
             }
         }
     }
 
     /***
      * Get the session from the session manager.
      *
      * @return SessionIfc	The unique session for this thread.
      * @exception FailedToCreateSessionException internal error
      *
      */
     private SessionIfc getSession() throws FailedToCreateSessionException {
         return PMDSessionManager.instance().getSession();
     }
 
     /***
      * Return the map where messages are stored.
      *
      * @param session - the session to use
      * @return PMDHashMap
      */
     private PMDHashMap messageTable(SessionIfc session) {
         return (PMDHashMap) session.lookup(MESSAGES);
     }
 
     /***
      * Return the vector used to store destinations and also the
      * registered dursble consumers.
      *
      * @param session - the session to use
      * @return PMDHashMap
      */
     private PMDVector destinationTable(SessionIfc session) {
         return (PMDVector) session.lookup(DESTINATIONS);
     }
 
     /***
      * Return the message handle table for the specified destination and
      * and name.
      * <p>
      * Assume this is only called from within a transaction
      *
      * @param key - the key to the consumer
      * @param session - the session object.
      * @return PMDHashMap the root object if it exists or null.
      *
      */
     private PMDVector handleTable(String key, SessionIfc session) {
         return (PMDVector) session.lookup(key);
     }
 
     /***
      * Return a generated root name to store consumer handles using the
      * destination and name
      *
      * @param destination - the destination
      * @param name - the consumer name
      * @return String - the root name
      */
     private String getHandlesRootName(JmsDestination destination, String name) {
         if (destination instanceof JmsQueue) {
             return getHandlesRootName(destination.getName());
         } else {
             return getHandlesRootName(name);
         }
     }
 
     /***
      * Return the handles root name given an internal string, which is an
      * entry from the destination table.
      * <p>
      * The format of the entries are QUEUE_<queue_name> for queues
      * or TOPIC_<topic_name>.<consumer_name> for topics.
      *
      * @param entry - the destination table entry
      * @return String - the associated handle table name
      */
     private String getHandlesRootNameFromDestination(String entry) {
         String name = null;
 
         if (entry.startsWith(QUEUE)) {
             name = HANDLES + "@" + entry.substring(QUEUE.length());
         } else if (entry.startsWith(TOPIC)) {
             int index = entry.indexOf("@");
             if (index != -1) {
                 name = HANDLES + "@" + entry.substring(index + 1);
             }
         }
 
         return name;
     }
 
     /***
      * Return a generated root name to store consumer handles using the
      * destination and name. The destination is always the name of a queue
      * destination
      *
      * @param queue - the name of the queue destination
      * @return String - the root name
      */
     private String getHandlesRootName(String name) {
         return HANDLES + "@" + name;
     }
 
     /***
      * Strip the "ID:" prefix from the string, and increment the long
      * by one, then add the prefix again.
      *
      * @param st The string to increment.
      * @return String The string incremented by 1.
      *
      */
     private String increment(String st) {
         String newId = null;
 
         try {
             newId = "ID:" + (Long.parseLong(st.substring(3)) + 1);
         } catch (NumberFormatException err) {
             _log.error("Invalid id: " + st);
             System.exit(-1);
         }
         return newId;
     }
 
     /***
      * Register an event to collect and remove processed messages with the
      * {@link BasicEventManager}
      */
     private void registerEvent() {
         try {
             BasicEventManager.instance().registerEventRelative(
                 new Event(COLLECT_DATABASE_GARBAGE_EVENT, this, null),
                 _gcInterval);
         } catch (IllegalEventDefinedException exception) {
             _log.error("registerEvent failed", exception);
         }
     }
 }
 
 
 /***
  * A simple wrapper to store long values in the database.
  *
  */
 class PMDLongInteger
     extends PersistentObject
     implements Externalizable {
 
     /***
      * The actual value
      */
     private long value_;
 
 
     /***
      * Used for serialization
      */
     static final long serialVersionUID = 1;
 
     public PMDLongInteger() {
     }
 
     public PMDLongInteger(long value) {
         value_ = value;
     }
 
     public long get() {
         return value_;
     }
 
     // implementation of Externalizable.writeExternal
     public void writeExternal(ObjectOutput stream)
         throws IOException {
         stream.writeLong(serialVersionUID);
         stream.writeLong(value_);
         super.writeExternal(stream);
     }
 
     // implementation of Externalizable.writeExternal
     public void readExternal(ObjectInput stream)
         throws IOException, ClassNotFoundException {
         long version = stream.readLong();
         if (version == serialVersionUID) {
             value_ = stream.readLong();
             super.readExternal(stream);
         } else {
             throw new IOException("PMDLongInteger with version " +
                 version + " is not supported.");
         }
     }
 
 } //-- ObjectAdapter