[Batchserver-cvs] batchserver/src/org/jmonks/batch/framework/repository/jdbc JdbcRepository.java,

SourceForge Headquarters 225 Broadway Suite 1600 San Diego, CA 92101 +1 (858) 422-6466

Update of /cvsroot/batchserver/batchserver/src/org/jmonks/batch/framework/repository/jdbc
In directory sc8-pr-cvs3.sourceforge.net:/tmp/cvs-serv20339

Added Files:
	JdbcRepository.java 
Log Message:
no message

--- NEW FILE: JdbcRepository.java ---
/*
 * JdbcRepository.java
 *
 * Created on September 9, 2006, 10:34 AM
 *
 * To change this template, choose Tools | Options and locate the template under
 * the Source Creation and Management node. Right-click the template and choose
 * Open. You can then make changes to the template in the Source Editor.
 */

package org.jmonks.batch.framework.repository.jdbc;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.HashMap;
import java.util.Map;
import org.apache.log4j.Logger;
import org.jmonks.batch.framework.JobStatistics;
import org.jmonks.batch.framework.config.ConfigurationException;
import org.jmonks.batch.framework.Repository;
import org.jmonks.batch.framework.util.JdbcConnectionHelper;

/**
 * <p>
 * JdbcRepository implementation of repository uses any database that can be accessed by using
 * the JDBC technology. This looks for the properties "jdbc-driver-class-name","jdbc-url",
 * "username" and "password" in repository config in framework configuration contains the
 * values to establish the connection to implement the Repository. 
 * <br>
 *  Following is the example of repository configuration.
 * <br><br>
 * <pre>
 *    &lt;repository-config repository-class-name="org.jmonks.batch.framework.repository.jdbc.JdbcRepository"&gt;
 *       &lt;property key="jdbc-driver-class-name"&gt;oracle.jdbc.driver.OracleDriver&lt;/property&gt;
 *       &lt;property key="jdbc-url"&gt;jdbc:oracle:thin:@hostname:1521:instancename&lt;/property&gt;
 *       &lt;property key="username"&gt;scott&lt;/property&gt;
 *       &lt;property key="password"&gt;tiger&lt;/property&gt;
 *    &lt;/repository-config&gt;
 * </pre>
 * <br><br>
 *  The database user specified in the configuration should have update, insert & delete 
 *  privileges on the following objects in the database.
 *  <table border="1">
 *      <tr><td>JOB_DATA_TRANSFER</td></tr>
 *      <tr><td>JOB_STATISTICS</td></tr>
 *      <tr><td>JOB_MGMT_MNTR_INFO</td></tr>
 *  </table>
 * </p>
 *
 * @author  Suresh Pragada
 * @version 1.0
 * @since   1.0
 */
public class JdbcRepository extends Repository
{
    private static Logger logger=Logger.getLogger(JdbcRepository.class);
    
    /**
     *  Map holds the jdbc configuration.
     */
    protected Map jdbcConfigMap=null;
    
    /**
     *  <p>
     *   Receives the jdbc configuration in a map from the factory method and
     *   validates the given configuration is valid by making a connection to the database.
     *  </p>
     *
     *  @param  configProps Map contains the jdbc configuration.
     *
     *  @throws ConfigurationException  If not able to get the connection by using the properties 
     *                                      given in the map.
     */
    protected void init(Map configProps)
    {
        logger.trace("Entering init");
        this.jdbcConfigMap=new HashMap(configProps);
        /**
         *  Do the basic validation to make sure we can get a connection using the given configuration.
         */
        Connection connection=JdbcConnectionHelper.getConnection(this.jdbcConfigMap);
        if(connection==null)
            throw new ConfigurationException(ConfigurationException.REPOSITORY_CONFIG, "Unable to get the jdbc " +
                            "connection with the properties " + this.jdbcConfigMap +".");
        else
            JdbcConnectionHelper.closeConnection(connection);
        logger.trace("Exiting init");        
    }

    /**
     * @see org.jmonks.batch.framework.repository.Repository#clearDataTransferredToNextJob()
     */    
    public boolean clearDataTransferredFromThisJob()
    {
        logger.trace("Entering clearDataTransferredToNextJob");                
        
        boolean dataCleared=false;
        Connection connection=JdbcConnectionHelper.getConnection(this.jdbcConfigMap);
        if(connection==null)
            dataCleared=false;
        else
        {
            try
            {
                PreparedStatement statement=connection.prepareStatement("delete from job_data_transfer where source_job_name=?");
                statement.setString(1, this.jobName);
                statement.executeUpdate();
                statement.close();
                connection.commit();                
                dataCleared=true;
            }
            catch(SQLException exception)
            {
                exception.printStackTrace();
                logger.error("Exception while trying to clear the data for requested job = " 
                        + this.jobName + ". Message = " + exception.getMessage(), exception);
                dataCleared=false;
            }
            finally
            {
                JdbcConnectionHelper.closeConnection(connection);
            }            
        }
        logger.trace("Exiting clearDataTransferredToNextJob");        
        return dataCleared;
    }

    /**
     *  @see org.jmonks.batch.framework.repository Repository#sendDataToNextJob(String,String,Object)
     */    
    public boolean sendDataToNextJob(String dataIdentifier, String nextJobName, final Object data)
    {
        logger.trace("Entering sendDataToNextJob);");
        logger.debug("data identifier = " + dataIdentifier + " next job name = " + nextJobName + " data = " + data);
        
        if(dataIdentifier==null || nextJobName==null || data==null)
            throw new IllegalArgumentException("No arguments cannot be null to the sendDataToNextJob method.");

        boolean dataSaved=false;
        Connection connection=JdbcConnectionHelper.getConnection(this.jdbcConfigMap);
        if(connection==null)
            dataSaved=false;
        else
        {
            try
            {
                PreparedStatement deleteStatement=connection.prepareStatement("delete from job_data_transfer where source_job_name=? and destination_job_name=? and job_data_identifier=?");
                deleteStatement.setString(1, this.jobName);
                deleteStatement.setString(2, nextJobName);
                deleteStatement.setString(3, dataIdentifier);
                deleteStatement.executeUpdate();
                deleteStatement.close();
                logger.debug("Data has been cleared before loading = " + this.jobName + " " + nextJobName + " " + dataIdentifier);

                PreparedStatement insertStatement=connection.prepareStatement("insert into job_data_transfer (source_job_name,destination_job_name,job_data_identifier,job_data) values (?,?,?,?)");
                insertStatement.setString(1, this.jobName);
                insertStatement.setString(2, nextJobName);
                insertStatement.setString(3, dataIdentifier);
                byte[] objectByteArray=this.serializeObjectIntoByteArray(data);
                insertStatement.setBinaryStream(4, new ByteArrayInputStream(objectByteArray), objectByteArray.length);
                insertStatement.executeUpdate();
                insertStatement.close();
                logger.debug("Data has been loaded = " + this.jobName + " " + nextJobName + " " + dataIdentifier);
                
                connection.commit();                
                dataSaved=true;
            }
            catch(SQLException exception)
            {
                exception.printStackTrace();
                logger.error("Exception while trying to send the data from " + this.jobName + " to job = " 
                        + nextJobName + " with the identifier " + dataIdentifier + ". Message = " + exception.getMessage(), exception);
                dataSaved=false;
            }
            finally
            {
                JdbcConnectionHelper.closeConnection(connection);
            }            
        }
        logger.trace("Exiting sendDataToNextJob");        
        return dataSaved;
    }

    /**
     * @see org.jmonks.batch.framework.repository.Repository#getDataFromPreviousJob(String,String)
     */
    public Object getDataFromPreviousJob(String dataIdentifier, String previousJobName)
    {
        logger.trace("Entering getDataFromPreviousJob");
        logger.debug("data identifier = " + dataIdentifier + " previous job name = " + previousJobName);
        
        if(dataIdentifier==null || previousJobName==null)
            throw new IllegalArgumentException("Data identifer and previous job name cannot be null " +
                                                    "to get the data from previous job.");
        
        Object data=null;
        Connection connection=JdbcConnectionHelper.getConnection(this.jdbcConfigMap);
        if(connection==null)
            data=null;
        else
        {
            try
            {
                PreparedStatement statement=connection.prepareStatement("select job_data from job_data_transfer where source_job_name=? and destination_job_name=? and job_data_identifier=?");
                statement.setString(1, previousJobName);
                statement.setString(2, this.jobName);
                statement.setString(3, dataIdentifier);
                ResultSet rs=statement.executeQuery();
                if(rs.next())
                {
                    InputStream inputStream=rs.getBinaryStream(1);
                    data=deserializeObjectFromInputStream(inputStream);
                }
                else
                    data=null;
                rs.close();
                statement.close();
            }
            catch(SQLException exception)
            {
                exception.printStackTrace();
                logger.error("Exception while trying to send the data from " + previousJobName + " to job = " 
                        + this.jobName + " with the identifier " + dataIdentifier + ". Message = " + exception.getMessage(), exception);
                data=null;
            }
            finally
            {
                JdbcConnectionHelper.closeConnection(connection);
            }            
        }
        logger.trace("Exiting getDataFromPreviousJob");
        return data;
    }
    
    /**
     * @see org.jmonks.batch.framework.repository.Repository#logStatistics(org.jmonks.batch.framework.JobStatistics)
     */
    public boolean logStatistics(final JobStatistics statistics)
    {
        logger.trace("Entering logStatistics");                                

        if(statistics==null)
            throw new IllegalArgumentException("Null statistic objects will not be saved in repository");
        if(!statistics.getJobname().equals(this.jobName))
            throw new IllegalArgumentException("Statistics object is not related to the job configured for the repository.");
        
        logger.debug("Logging = " + statistics.toString());
        boolean logged=false;
        Connection connection=JdbcConnectionHelper.getConnection(this.jdbcConfigMap);
        if(connection==null)
            logged=false;
        else
        {
            try
            {
                PreparedStatement statement=connection.prepareStatement("insert into job_statistics (job_name,job_start_time,job_end_time," +
                        "job_exit_code,job_exit_reason,records_processed,memory_usage) values(?,?,?,?,?,?,?)");
                statement.setString(1, statistics.getJobname());
                statement.setTimestamp(2, new Timestamp(statistics.getStartTime().getTime()));
                statement.setTimestamp(3, new Timestamp(statistics.getEndTime().getTime()));
                statement.setInt(4, statistics.getExitCode().getCode());
                if(statistics.getExitCode().getMessage()!=null)
                    statement.setString(5, statistics.getExitCode().getMessage());
                else
                    statement.setNull(5,Types.VARCHAR);
                statement.setLong(6, statistics.getRecordsProcessed());
                statement.setLong(7, statistics.getMaxMemoryUsage());
                statement.executeUpdate();
                statement.close();
                connection.commit();
                logged=true;
            }
            catch(SQLException exception)
            {
                exception.printStackTrace();
                logger.error("Exception while logging the job statistics =  " + statistics.toString() + ". Message = " + exception.getMessage(), exception);
                logged=false;
            }
            finally
            {
                JdbcConnectionHelper.closeConnection(connection);
            }            
        }
        logger.trace("Exiting logStatistics");                                        
        return logged;
    }

    /**
     * @see org.jmonks.batch.framework.repository.Repository#unregisterJobMgmtMntrInfo()
     */    
    public boolean unregisterJobMgmtMntrInfo()
    {
        logger.trace("Entering unregisterJobMgmtMntrInfo");                
        boolean unregistered=false;
        
        Connection connection=JdbcConnectionHelper.getConnection(this.jdbcConfigMap);
        if(connection==null)
            unregistered=false;
        else
        {
            try
            {
                PreparedStatement statement=connection.prepareStatement("delete from job_mgmt_mntr_info where job_name=?");
                statement.setString(1, this.jobName);
                statement.executeUpdate();
                statement.close();
                connection.commit();
                unregistered=true;
            }
            catch(SQLException exception)
            {
                exception.printStackTrace();
                logger.error("Exception while trying to lookup mgmt and mntr info = " + this.jobName);
                unregistered=false;
            }
            finally
            {
                JdbcConnectionHelper.closeConnection(connection);
            }            
        }
        logger.trace("Exiting unregisterJobMgmtMntrInfo");                        
        return unregistered;
    }

    /**
     * @see org.jmonks.batch.framework.repository.Repository#registerJobMgmtMntrInfo(Object)
     */    
    public boolean registerJobMgmtMntrInfo(final Object registrationInfo)
    {
        logger.trace("Entering registerJobMgmtMntrInfo");                        
        logger.debug(" registratinfo = " + registrationInfo);
        
        if(registrationInfo==null)
            throw new IllegalArgumentException("Job registration information cannot be null to register the job in repository.");
        
        boolean registered=true;
        
        Connection connection=JdbcConnectionHelper.getConnection(this.jdbcConfigMap);
        if(connection==null)
            registered=false;
        else
        {
            try
            {
                PreparedStatement deleteStatement=connection.prepareStatement("delete from job_mgmt_mntr_info where job_name=?");
                deleteStatement.setString(1, this.jobName);
                deleteStatement.executeUpdate();
                deleteStatement.close();
                logger.debug("Cleared the existing registration info = " + this.jobName);

                PreparedStatement insertStatement=connection.prepareStatement("insert into job_mgmt_mntr_info (job_name,job_info) values (?,?)");
                insertStatement.setString(1, this.jobName);
                byte[] objectByteArray=this.serializeObjectIntoByteArray(registrationInfo);
                insertStatement.setBinaryStream(2, new ByteArrayInputStream(objectByteArray), objectByteArray.length);
                insertStatement.executeUpdate();
                insertStatement.close();
                logger.debug("Registered mgmt mntr info = " + this.jobName);
                
                connection.commit();                
                registered=true;
            }
            catch(SQLException exception)
            {
                exception.printStackTrace();
                logger.error("Exception while trying to register the mgmt mntr info =  " + this.jobName+ ". Message = " + exception.getMessage(), exception);
                registered=false;
            }
            finally
            {
                JdbcConnectionHelper.closeConnection(connection);
            }            
        }
        logger.trace("Exiting registerJobMgmtMntrInfo");
        return registered;
    }    

    /**
     *  Serializes the given object and return that serialized data as byte array.
     *
     *  @param  object  Object to be serialized.
     *
     *  @return Returns the serialized data as byte array.
     */
    protected byte[] serializeObjectIntoByteArray(Object object)
    {
        try
        {
            ByteArrayOutputStream bos=new ByteArrayOutputStream();
            ObjectOutputStream oos=new ObjectOutputStream(bos);
            oos.writeObject(object);
            oos.close();
            bos.flush();
            bos.close();
            return bos.toByteArray();        
        }
        catch(IOException exception)
        {
            exception.printStackTrace();
            logger.error("Error while serializing the object. Message = " + exception.getMessage(), exception);
            return null;
        }
    }

    /**
     *  Deserializes the given inputstream and return the object.
     *
     *  @param  inputStream  InputStream to be deserialized.
     *
     *  @return Returns the deserialized object.
     */    
    protected Object deserializeObjectFromInputStream(InputStream inputStream)
    {
        try
        {
            ObjectInputStream ois=new ObjectInputStream(inputStream);
            Object data=ois.readObject();
            ois.close();
            return data;
        }
        catch(ClassNotFoundException exception)
        {
            exception.printStackTrace();
            logger.error("Error while deserializing the object. Message = " + exception.getMessage(), exception);
            return null;
        }        
        catch(IOException exception)
        {
            exception.printStackTrace();
            logger.error("Error while deserializing the object. Message = " + exception.getMessage(), exception);
            return null;
        }        
    }    
}

[Batchserver-cvs] batchserver/src/org/jmonks/batch/framework/repository/jdbc JdbcRepository.java,

[Batchserver-cvs] batchserver/src/org/jmonks/batch/framework/repository/jdbc JdbcRepository.java, NONE, 1.1