/***
    * 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 2002-2004 (C) Exoffice Technologies Inc. All Rights Reserved.
   *
   * $Id: BatchingRdbmsAdapter.java,v 1.11 2004/01/08 05:55:05 tanderson Exp $
   *
   * Date         Author  Changes
   * $Date	    jimm    Created
   */
  
  
  package org.exolab.jms.persistence;
  
  
  import java.sql.Connection;
  import java.util.Enumeration;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.LinkedList;
  import java.util.Vector;
  
  import javax.jms.JMSException;
  import javax.sql.ConnectionPoolDataSource;
  import javax.transaction.Transaction;
  import javax.transaction.TransactionManager;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  
  import org.exolab.core.foundation.HandleIfc;
  import org.exolab.jms.authentication.User;
  import org.exolab.jms.client.JmsDestination;
  import org.exolab.jms.config.Configuration;
  import org.exolab.jms.config.DatabaseConfiguration;
  import org.exolab.jms.events.EventHandler;
  import org.exolab.jms.message.MessageId;
  import org.exolab.jms.message.MessageImpl;
  import org.exolab.jms.messagemgr.PersistentMessageHandle;
  
  
  /***
   * The batching RDBMS adapter is used to improve the performance of persistent
   * messages by batching more instructions into a single transaction.
   * <p>
   * It encpasulates an {@link RDBMSAdapter} and delegates all the behaviour to
   * it. In the interim it will only batch 'insert', 'update' and 'delete'
   * statements. If it receives a 'select' or any query it will first commit the
   * batched statements before satisfying the query.
   * <p>
   * This piece of code is still under development and could effect the integrity
   * of your system if used. It could also improve the throughput of persistent
   * messages if it works correctly.
   *
   * @version     $Revision: 1.11 $ $Date: 2004/01/08 05:55:05 $
   * @author      <a href="mailto:jima@intalio.org">Jim Alateras</a>
   */
  public class BatchingRdbmsAdapter
      extends PersistenceAdapter
      implements EventHandler {
  
      /***
      * This is the maximum number of statements that can be batched before
      * actually committing the work to the database. This value defaults to
      * 100 but it can be changed at runtime.
      */
     private int _maxStatementsToBatch = 500;
 
     /***
      * The directory where the log files are stored. This can be set by the
      * client
      */
     private String _logDirectory = ".";
 
     /***
      *  Holds a reference to the RDBMSAdapter, which it delegates all the
      * work too.
      */
     private RDBMSAdapter _rdbms = null;
 
     /***
      * Holds the current batch of trnasactional objects.
      */
     private LinkedList _batch = new LinkedList();
 
     /***
      * Holds a reference to message handles that have been batched up.
      */
     private HashMap _handles = new HashMap();
 
     /***
      * Holds a reference to messages that have been batched up
      */
     private HashMap _messages = new HashMap();
 
     /***
      * The logger
      */
     private static final Log _log =
         LogFactory.getLog(BatchingRdbmsAdapter.class);
 
 
     /***
      * Connects to the given db.
      *
      * @param driver - the rdbms driver to use
      * @param url - the url to the database
      * @param userName - a valid user name for a connection to the database
      * @param password - the password for the connection to the database
      * @param batchSize - the number of requests it should batch
      * @throws PersistenceException for any database error
      */
     BatchingRdbmsAdapter(String driver, String url, String userName,
                          String password, int batchSize)
         throws PersistenceException {
         // create the rdbms adapter
         _rdbms = new RDBMSAdapter(driver, url, userName, password);
         _maxStatementsToBatch = batchSize;
     }
 
     /***
      * Close the database if open.
      *
      */
     public void close() {
         if (_rdbms != null) {
             try {
                 flush();
             } catch (PersistenceException exception) {
                 _log.error("Failed to flush statements", exception);
             }
             _rdbms.close();
         }
     }
 
     /***
      * Set the maximum number of SQL statements to batch
      *
      * @param count - number of statements to batch
      */
     public void setMaxStatementsToBatch(int max) {
         _maxStatementsToBatch = max;
     }
 
     /***
      * Return the maximum number of statements to batch
      *
      * @return int
      */
     public int getMaxStatementsToBatch() {
         return _maxStatementsToBatch;
     }
 
     // implementation of PersistenceAdapter.getLastId
     public long getLastId(Connection connection)
         throws PersistenceException {
         return _rdbms.getLastId(connection);
     }
 
     // implementation of PersistenceAdapter.updateIds
     public void updateIds(Connection connection, long id)
         throws PersistenceException {
         _rdbms.updateIds(connection, id);
     }
 
     // implementation of PersistenceMessage.addMessage
     public void addMessage(Connection connection, MessageImpl message)
         throws PersistenceException {
         addToBatch(TransactionalObjectWrapper.ADD_MESSAGE, message);
     }
 
     // implementation of PersistenceMessage.addMessage
     public void updateMessage(Connection connection, MessageImpl message)
         throws PersistenceException {
         addToBatch(TransactionalObjectWrapper.UPDATE_MESSAGE, message);
     }
 
     // implementation of PersistenceAdapter.getUnprocessedMessages
     public Vector getUnprocessedMessages(Connection connection)
         throws PersistenceException {
         flush();
         return _rdbms.getUnprocessedMessages(connection);
     }
 
 
     // implementation of PersistenceAdapter.removeMessage
     public void removeMessage(Connection connection, String id)
         throws PersistenceException {
         addToBatch(TransactionalObjectWrapper.DELETE_MESSAGE, id);
     }
 
     // implementation of PersistenceAdapter.getMessage
     public MessageImpl getMessage(Connection connection, String id)
         throws PersistenceException {
         flush();
         return _rdbms.getMessage(connection, id);
     }
 
     // implementation of PersistenceAdapter.getMessages
     public Vector getMessages(Connection connection,
                               PersistentMessageHandle handle)
         throws PersistenceException {
         flush();
         return _rdbms.getMessages(connection, handle);
     }
 
     // implementation of PersistenceAdapter.addMessageHandle
     public void addMessageHandle(Connection connection,
                                  PersistentMessageHandle handle)
         throws PersistenceException {
         addToBatch(TransactionalObjectWrapper.ADD_HANDLE, handle);
     }
 
     // implementation of PersistenceAdapter.updateMessageHandle
     public void updateMessageHandle(Connection connection,
                                     PersistentMessageHandle handle)
         throws PersistenceException {
         addToBatch(TransactionalObjectWrapper.UPDATE_HANDLE, handle);
     }
 
     // implementation of PersistenceAdapter.removeMessageHandle
     public void removeMessageHandle(Connection connection,
                                     PersistentMessageHandle handle)
         throws PersistenceException {
         addToBatch(TransactionalObjectWrapper.DELETE_HANDLE, handle);
     }
 
     // implementation of PersistenceAdapter.getMessageHandles
     public Vector getMessageHandles(Connection connection,
                                     JmsDestination destination, String name)
         throws PersistenceException {
         flush();
         return _rdbms.getMessageHandles(connection, destination, name);
     }
 
     // implementation of PersistenceAdapter.addDurableConsumer
     public void addDurableConsumer(Connection connection, String topic,
                                    String consumer)
         throws PersistenceException {
         flush();
         _rdbms.addDurableConsumer(connection, topic, consumer);
     }
 
     // implementation of PersistenceAdapter.removeDurableConsumer
     public void removeDurableConsumer(Connection connection, String consumer)
         throws PersistenceException {
         flush();
         _rdbms.removeDurableConsumer(connection, consumer);
     }
 
     // implementation of PersistenceAdapter.getDurableConsumers
     public Enumeration getDurableConsumers(Connection connection,
                                            String topic)
         throws PersistenceException {
         flush();
         return _rdbms.getDurableConsumers(connection, topic);
     }
 
     // implementation of PersistenceAdapter.getAllDurableConsumers
     public HashMap getAllDurableConsumers(Connection connection)
         throws PersistenceException {
         flush();
         return _rdbms.getAllDurableConsumers(connection);
     }
 
     // implementation of PersistenceAdapter.durableConsumerExists
     public boolean durableConsumerExists(Connection connection, String name)
         throws PersistenceException {
         flush();
         return _rdbms.durableConsumerExists(connection, name);
     }
 
     // implementation of PersistenceAdapter.addDestination
     public void addDestination(Connection connection, String name,
                                boolean queue)
         throws PersistenceException {
         flush();
         _rdbms.addDestination(connection, name, queue);
     }
 
     // implementation of PersistenceAdapter.removeDestination
     public void removeDestination(Connection connection, String name)
         throws PersistenceException {
         flush();
         _rdbms.removeDestination(connection, name);
     }
 
     // implementation of PersistenceAdapter.getAllDestinations
     public Enumeration getAllDestinations(Connection connection)
         throws PersistenceException {
         flush();
         return _rdbms.getAllDestinations(connection);
     }
 
     // implementation of PersistenceAdapter.checkDestination
     public boolean checkDestination(Connection connection, String name)
         throws PersistenceException {
         flush();
         return _rdbms.checkDestination(connection, name);
     }
 
     // implementation of getQueueMessageCount
     public int getQueueMessageCount(Connection connection, String name)
         throws PersistenceException {
         flush();
         return _rdbms.getQueueMessageCount(connection, name);
     }
 
     // implementation of PersistenceAdapter.getQueueMessageCount
     public int getDurableConsumerMessageCount(Connection connection,
                                               String destination, String name)
         throws PersistenceException {
         flush();
         return _rdbms.getDurableConsumerMessageCount(connection, destination,
             name);
     }
 
     // implementation of PersistenceAdapter.getQueueMessageCount
     public void removeExpiredMessages(Connection connection)
         throws PersistenceException {
         flush();
         _rdbms.removeExpiredMessages(connection);
     }
 
     // implementation of PersistenceAdapter.removeExpiredMessageHandles
     public void removeExpiredMessageHandles(Connection connection,
                                             String consumer)
         throws PersistenceException {
         flush();
         _rdbms.removeExpiredMessageHandles(connection, consumer);
     }
 
     // implementation of PersistenceAdapter.getQueueMessageCount
     public Vector getNonExpiredMessages(Connection connection,
                                         JmsDestination destination)
         throws PersistenceException {
         flush();
         return _rdbms.getNonExpiredMessages(connection, destination);
     }
 
     /***
      * Return a connection to the database from the pool of connections. It
      * will throw an PersistenceException if it cannot retrieve a connection.
      * The client should close the connection normally, since the pool is a
      * connection event listener.
      *
      * @return Connection - a pooled connection or null
      * @exception PersistenceException - if it cannot retrieve a connection
      */
     public Connection getConnection()
         throws PersistenceException {
 
         return _rdbms.getConnection();
     }
 
     /***
      * Purge all processed messages from the database.
      *
      * @return int - the number of messages deleted
      */
     public synchronized int purgeMessages() {
         try {
             flush();
         } catch (PersistenceException exception) {
             _log.error("Error in purgeMessages " + exception);
         }
 
         return _rdbms.purgeMessages();
     }
 
     // implementation of EventHandler.handleEvent
     public void handleEvent(int event, Object callback, long time) {
         _rdbms.handleEvent(event, callback, time);
     }
 
     // implementation of EventHandler.getHandle
     public HandleIfc getHandle() {
         return null;
     }
 
     /***
      * Close the current piece of work and commit it to the database. This is
      * called before a query or fetch is executed on the data.
      * <p>
      * It will grab a connection and commit the transactional objects before
      * returning. If there is any problem it will throw a PersistenceException
      * excpetion
      *
      * @throws PersistenceException
      */
     private synchronized void flush() throws PersistenceException {
         if (_batch.size() == 0) {
             return;
         }
 
         // need to do this in a separate thread since the current thread is
         // already associated with a Connection object
         Thread thread = new Thread(new Runnable() {
 
             public void run() {
                 Connection connection = null;
                 try {
                     connection = _rdbms.getConnection();
 
                     Iterator iter = _batch.iterator();
                     while (iter.hasNext()) {
                         TransactionalObjectWrapper wrapper =
                             (TransactionalObjectWrapper) iter.next();
                         switch (wrapper._action) {
                             case TransactionalObjectWrapper.ADD_MESSAGE:
                                 _rdbms.addMessage(connection,
                                     (MessageImpl) wrapper._object);
                                 break;
 
                             case TransactionalObjectWrapper.UPDATE_MESSAGE:
                                 _rdbms.updateMessage(connection,
                                     (MessageImpl) wrapper._object);
                                 break;
 
                             case TransactionalObjectWrapper.DELETE_MESSAGE:
                                 _rdbms.removeMessage(connection,
                                     (String) wrapper._object);
                                 break;
 
                             case TransactionalObjectWrapper.ADD_HANDLE:
                                 _rdbms.addMessageHandle(connection,
                                     (PersistentMessageHandle) wrapper._object);
                                 break;
 
                             case TransactionalObjectWrapper.UPDATE_HANDLE:
                                 _rdbms.updateMessageHandle(connection,
                                     (PersistentMessageHandle) wrapper._object);
                                 break;
 
                             case TransactionalObjectWrapper.DELETE_HANDLE:
                                 _rdbms.removeMessageHandle(connection,
                                     (PersistentMessageHandle) wrapper._object);
                                 break;
                         }
                     }
                     connection.commit();
 
                     // if the commit has worked then flush the batch list
                     _batch.clear();
                     _messages.clear();
                     _handles.clear();
                 } catch (PersistenceException exception) {
                     SQLHelper.rollback(connection);
                     _log.error("Failure in flush()", exception);
                 } catch (Exception exception) {
                     _log.error("Failure in flush()", exception);
                 } finally {
                     if (connection != null) {
                         try {
                             connection.close();
                         } catch (Exception nested) {
                             _log.error("Failure in flush()", nested);
                         }
                     }
                 }
             }
         });
 
         // start the thread.
         thread.start();
 
         // wait for the thread to finish...sort of defeat the purpose here.
         try {
             thread.join();
         } catch (InterruptedException exception) {
             // ignore the error
         }
     }
 
     /***
      * Add this transactional object along with the associated action to the
      * current batch job.
      *
      * @param action - the action to take
      * @param object - the transactional object
      * @throws PersistenceException
      */
     private synchronized void addToBatch(int action, Object object)
         throws PersistenceException {
         if (_batch.size() >= _maxStatementsToBatch) {
             flush();
         }
 
         switch (action) {
             case TransactionalObjectWrapper.ADD_MESSAGE:
                 {
                     TransactionalObjectWrapper txobj =
                         new TransactionalObjectWrapper(action, object);
                     MessageImpl message = (MessageImpl) object;
                     MessageId id = message.getMessageId();
                     if (_messages.containsKey(id)) {
                         throw new PersistenceException("Inconsistency in cache " +
                             id + " is present when it shouldn't be.");
                     }
                     _messages.put(id, txobj);
                     _batch.addLast(txobj);
                     break;
                 }
 
             case TransactionalObjectWrapper.UPDATE_MESSAGE:
                 {
                     MessageImpl message = (MessageImpl) object;
                     MessageId id = message.getMessageId();
                     TransactionalObjectWrapper txobj =
                         (TransactionalObjectWrapper) _messages.get(id);
                     TransactionalObjectWrapper newtxobj =
                         new TransactionalObjectWrapper(action, object);
 
                     if (txobj != null) {
                         // if the batch already contains a entry for this
                         // message then remove it and add the new version
                         // with an
                         _batch.remove(txobj);
                         if (txobj._action == TransactionalObjectWrapper.ADD_MESSAGE) {
                             newtxobj._action = TransactionalObjectWrapper.ADD_MESSAGE;
                             _batch.addLast(newtxobj);
                         } else if (txobj._action == TransactionalObjectWrapper.UPDATE_MESSAGE) {
                             _batch.addLast(newtxobj);
                         } else {
                             // could only be a delete and should never happen
                             throw new PersistenceException("Inconsistency in cache." +
                                 " Cannot update a deleted message.");
                         }
                     } else {
                         _batch.addLast(newtxobj);
                     }
                     _messages.put(id, newtxobj);
                     break;
                 }
 
             case TransactionalObjectWrapper.DELETE_MESSAGE:
                 {
                     MessageImpl message = (MessageImpl) object;
                     MessageId id = message.getMessageId();
                     TransactionalObjectWrapper txobj =
                         (TransactionalObjectWrapper) _messages.get(id);
                     TransactionalObjectWrapper newtxobj =
                         new TransactionalObjectWrapper(action, object);
 
                     if (txobj != null) {
                         // if the batch already contains a entry for this
                         // message then remove it and add the new version
                         // with an
                         _batch.remove(txobj);
                         if (txobj._action == TransactionalObjectWrapper.ADD_MESSAGE) {
                             // if an add and the delete happened in the same batch then
                             // we don't have to commit either of the transactions.
                         } else if (txobj._action == TransactionalObjectWrapper.UPDATE_MESSAGE) {
                             // if the update and the delete happened in the same batch then
                             // we need add the transaction
                             _batch.addLast(newtxobj);
                         } else {
                             // if the delete already exists then simple ignore it.
                         }
                     } else {
                         _batch.addLast(newtxobj);
                     }
                     _messages.put(id, newtxobj);
                     break;
                 }
 
             case TransactionalObjectWrapper.ADD_HANDLE:
                 {
                     TransactionalObjectWrapper txobj =
                         new TransactionalObjectWrapper(action, object);
                     PersistentMessageHandle handle = (PersistentMessageHandle) object;
                     if (_handles.containsKey(handle)) {
                         throw new PersistenceException("Inconsistency in cache " +
                             handle + " is present when it shouldn't be.");
                     }
                     _handles.put(handle, txobj);
                     _batch.addLast(txobj);
                     break;
                 }
 
             case TransactionalObjectWrapper.UPDATE_HANDLE:
                 {
                     PersistentMessageHandle handle = (PersistentMessageHandle) object;
                     TransactionalObjectWrapper txobj =
                         (TransactionalObjectWrapper) _handles.get(handle);
                     TransactionalObjectWrapper newtxobj =
                         new TransactionalObjectWrapper(action, object);
 
                     if (txobj != null) {
                         // if the batch already contains a entry for this
                         // handle then remove it and add the new version.
                         _batch.remove(txobj);
                         if (txobj._action == TransactionalObjectWrapper.ADD_HANDLE) {
                             newtxobj._action = TransactionalObjectWrapper.ADD_HANDLE;
                             _batch.addLast(newtxobj);
                         } else if (txobj._action == TransactionalObjectWrapper.UPDATE_HANDLE) {
                             _batch.addLast(newtxobj);
                         } else {
                             // could only be a delete and should never happen
                             throw new PersistenceException("Inconsistency in cache." +
                                 " Cannot update a deleted handle.");
                         }
                     } else {
                         _batch.addLast(newtxobj);
                     }
                     _handles.put(handle, newtxobj);
                     break;
                 }
 
             case TransactionalObjectWrapper.DELETE_HANDLE:
                 {
                     PersistentMessageHandle handle = (PersistentMessageHandle) object;
                     TransactionalObjectWrapper txobj =
                         (TransactionalObjectWrapper) _handles.get(handle);
                     TransactionalObjectWrapper newtxobj =
                         new TransactionalObjectWrapper(action, object);
 
                     if (txobj != null) {
                         // if the batch already contains a entry for this
                         // message then remove it and add the new version
                         // with an
                         _batch.remove(txobj);
                         if (txobj._action == TransactionalObjectWrapper.ADD_HANDLE) {
                             // if an add and the delete happenedd in the same batch then
                             // we don't have to commit either of the transactions.
                         } else if (txobj._action == TransactionalObjectWrapper.UPDATE_HANDLE) {
                             // if the update and the delete happened in the same batch then
                             // we need add the transaction
                             _batch.addLast(newtxobj);
                         } else {
                             // if the delete already exists then simple ignore it.
                         }
                     } else {
                         _batch.addLast(newtxobj);
                     }
                     _handles.put(handle, newtxobj);
                     break;
                 }
         }
 
     }
 
     public void addUser(Connection connection, User user)
         throws PersistenceException {
     }
 
     public Enumeration getAllUsers(Connection connection)
         throws PersistenceException {
         return null;
     }
 
     public User getUser(Connection connection, User user)
         throws PersistenceException {
         return null;
     }
 
     public void removeUser(Connection connection, User user)
         throws PersistenceException {
     }
 
     public void updateUser(Connection connection, User user)
         throws PersistenceException {
     }
 
     /***
      * Inner class that wraps up transactional objects so that they can
      * correctly be processed in a batch transaction
      */
     private class TransactionalObjectWrapper {
 
         /***
          * This is the list of actions that the wrapper supports.
          */
         public final static int ADD_MESSAGE = 1;
         public final static int UPDATE_MESSAGE = 2;
         public final static int DELETE_MESSAGE = 3;
         public final static int ADD_HANDLE = 4;
         public final static int UPDATE_HANDLE = 5;
         public final static int DELETE_HANDLE = 6;
 
         /***
          * Action associated with the transactional object
          */
         public int _action;
 
         /***
          * The transactional object
          */
         public Object _object;
 
         /***
          * Constructor using the action and object
          */
         public TransactionalObjectWrapper(int action, Object object) {
             _action = action;
             _object = object;
         }
     }
 }