batchingrdbmsadapter.java

实现了Jms的服务器源码,支持多种适配器,DB,FTP,支持多种数据库

/**
 * 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 [e-mail].
 *
 * 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:[e-mail]">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);
    }