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

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

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

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.batchserver.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.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.log4j.Logger;
import org.jmonks.batchserver.framework.ErrorCode;
import org.jmonks.batchserver.framework.JobStatistics;
import org.jmonks.batchserver.framework.config.ConfigurationException;
import org.jmonks.batchserver.framework.repository.Repository;
import org.jmonks.batchserver.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.batchserver.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>
 *      <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
{
    /**
     *  Map holds the jdbc configuration.
     */
    protected Map jdbcConfigMap=null;

    private static Logger logger=Logger.getLogger(JdbcRepository.class);
    
    /**
     *  <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.batchserver.framework.repository.Repository#clearDataTransferredToNextJob(String)
     */    
    public boolean clearDataTransferredToNextJob(String jobName)
    {
        logger.trace("Entering clearDataTransferredToNextJob");                
        logger.debug("clearDataTransferredToNextJob :: job name = " + jobName);
        
        if(jobName==null)
            throw new IllegalArgumentException("job name cannot be null.");
        
        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, 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 = " 
                        + jobName + ". Message = " + exception.getMessage(), exception);
                dataCleared=true;
            }
            finally
            {
                JdbcConnectionHelper.closeConnection(connection);
            }            
        }
        logger.trace("Exiting clearDataTransferredToNextJob");        
        return dataCleared;
    }

    /**
     * @see org.jmonks.batchserver.framework.repository Repository#sendDataToNextJob(String,String,String,Object)
     */    
    public boolean sendDataToNextJob(String dataIdentifier, String sourceJobName, String targetJobName, Object data)
    {
        logger.trace("Entering sendDataToNextJob);");
        logger.debug("data identifier = " + dataIdentifier + " source job name = " + sourceJobName + " target job name = " + targetJobName + " data = " + data);
        
        if(dataIdentifier==null || sourceJobName==null || targetJobName==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 target_job_name=? and job_data_identifier=?");
                deleteStatement.setString(1, sourceJobName);
                deleteStatement.setString(2, targetJobName);
                deleteStatement.setString(3, dataIdentifier);
                deleteStatement.executeUpdate();
                deleteStatement.close();
                logger.debug("Data has been cleared before loading = " + sourceJobName + " " + targetJobName + " " + dataIdentifier);

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

    /**
     * @see org.jmonks.batchserver.framework.repository.Repository#getDataFromPreviousJob(String,String,String)
     */
    public Object getDataFromPreviousJob(String dataIdentifier, String sourceJobName, String targetJobName)
    {
        logger.trace("Entering getDataFromPreviousJob");
        logger.debug("data identifier = " + dataIdentifier + " source job name = " + sourceJobName + " target job name = " + targetJobName);
        
        if(dataIdentifier==null || sourceJobName==null || targetJobName==null)
            throw new IllegalArgumentException("Data identifer, source job name and target 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 target_job_name=? and job_data_identifier=?");
                statement.setString(1, sourceJobName);
                statement.setString(2, targetJobName);
                statement.setString(3, dataIdentifier);
                ResultSet rs=statement.executeQuery();
                if(rs.next())
                {
                    InputStream inputStream=rs.getBinaryStream(4);
                    data=getObjectFromByteArray(inputStream);
                }
                else
                    data=null;
                rs.close();
                statement.close();
            }
            catch(SQLException exception)
            {
                exception.printStackTrace();
                logger.error("Exception while trying to send the data from " + sourceJobName + " to job = " 
                        + targetJobName + " with the identifier " + dataIdentifier + ". Message = " + exception.getMessage(), exception);
                data=null;
            }
            finally
            {
                JdbcConnectionHelper.closeConnection(connection);
            }            
        }
        logger.trace("Exiting getDataFromPreviousJob");
        return data;
    }
    
    /**
     *
     * @see org.jmonks.batchserver.framework.repository.Repository#logStatistics(org.jmonks.batchserver.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");
        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.setInt(6, statistics.getRecordsProcessed());
                statement.setInt(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.batchserver.framework.repository.Repository#getStatistics(String)
     */    
    public JobStatistics[] getStatistics(String jobName)
    {
        logger.trace("Entering getStatistics");                                        
        logger.debug("getStatistics :: job name = " + jobName);
        
        if(jobName==null)
            throw new IllegalArgumentException("job name cannot be null for querying the job statistics.");
        
        JobStatistics[] statistics=null;
        List jobStatisticsList=null;
        Connection connection=JdbcConnectionHelper.getConnection(this.jdbcConfigMap);
        if(connection==null)
            statistics=new JobStatistics[]{};
        else
        {
            try
            {
                PreparedStatement statement=connection.prepareStatement("select job_start_time,job_end_time,job_exit_code,job_exit_reason,records_processed,memory_usage from job_statistics where job_name=?");
                statement.setString(1, jobName);
                ResultSet rs=statement.executeQuery();
                jobStatisticsList=new ArrayList();
                while(rs.next())
                {
                    JobStatistics statistic=new JobStatistics(jobName);
                    statistic.setStartTime(new java.util.Date(rs.getTimestamp(1).getTime()));
                    statistic.setEndTime(new java.util.Date(rs.getTimestamp(2).getTime()));
                    int exitCode=rs.getInt(3);
                    String exitReason=rs.getString(4);
                    ErrorCode errorCode=ErrorCode.createErrorCode(exitCode, exitReason==null?"":exitReason);
                    statistic.setExitCode(errorCode);
                    statistic.setRecordsProcessed(rs.getInt(5));
                    statistic.setMaxMemeoryUsage(rs.getInt(6));
                    jobStatisticsList.add(statistic);
                }
                rs.close();
                statement.close();
            }
            catch(SQLException exception)
            {
                exception.printStackTrace();
                logger.error("Exception while trying to get the job statistics for the job = " + jobName + ". Message = " + exception.getMessage(), exception);
            }
            finally
            {
                JdbcConnectionHelper.closeConnection(connection);
                if(jobStatisticsList!=null)
                    statistics=(JobStatistics[])jobStatisticsList.toArray();
                else
                    statistics=new JobStatistics[]{};
            }            
        }
        logger.trace("Exiting getStatistics");
        return statistics;
    }

    /**
     * @see org.jmonks.batchserver.framework.repository.Repository#lookupJobMgmtMntrInfo(String)
     */    
    public Object lookupJobMgmtMntrInfo(String jobName)
    {
        logger.trace("Entering lookupJobMgmtMntrInfo");
        logger.debug("lookupJobMgmtMntrInfo :: job name = " + jobName);
        
        if(jobName==null)
            throw new IllegalArgumentException("jobName cannot be null to lookup the mgmt and mntr information.");
            
        Object registredInfo=null;
        Connection connection=JdbcConnectionHelper.getConnection(this.jdbcConfigMap);
        if(connection==null)
            registredInfo=null;
        else
        {
            try
            {
                PreparedStatement statement=connection.prepareStatement("select job_info from job_mgmt_mntr_info where job_name=?");
                statement.setString(1, jobName);
                ResultSet rs=statement.executeQuery();
                if(rs.next())
                {
                    InputStream inputStream=rs.getBinaryStream(1);
                    registredInfo=getObjectFromByteArray(inputStream);
                }
                else
                    registredInfo=null;
                rs.close();
                statement.close();
            }
            catch(SQLException exception)
            {
                exception.printStackTrace();
                logger.error("Exception while trying to lookup mgmt and mntr info = " + jobName);
                registredInfo=null;
            }
            finally
            {
                JdbcConnectionHelper.closeConnection(connection);
            }            
        }
        logger.trace("Exiting lookupJobMgmtMntrInfo");        
        return registredInfo;
    }

    /**
     * @see org.jmonks.batchserver.framework.repository.Repository#unregisterJobMgmtMntrInfo(String)
     */    
    public boolean unregisterJobMgmtMntrInfo(String jobName)
    {
        logger.trace("Entering unregisterJobMgmtMntrInfo");                
        logger.debug("unregisterJobMgmtMntrInfo :: job name = " + jobName);
        
        if(jobName==null)
            throw new IllegalArgumentException("jobName cannot be null to unregister the mgmt and mntr information.");

        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, 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 = " + jobName);
                unregistered=false;
            }
            finally
            {
                JdbcConnectionHelper.closeConnection(connection);
            }            
        }
        logger.trace("Exiting unregisterJobMgmtMntrInfo");                        
        return unregistered;
    }

    /**
     * @see org.jmonks.batchserver.framework.repository.Repository#registerJobMgmtMntrInfo(String,Object)
     */    
    public boolean registerJobMgmtMntrInfo(String jobName, Object registrationInfo)
    {
        logger.trace("Entering registerJobMgmtMntrInfo");                        
        logger.debug("job name = " + jobName + " registratinfo = " + registrationInfo);
        
        if(jobName==null)
            throw new IllegalArgumentException("Job name cannot be null to register the job in repository.");
        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, jobName);
                deleteStatement.executeUpdate();
                deleteStatement.close();
                logger.debug("Cleared the existing registration info = " + jobName);

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

    /**
     * @see org.jmonks.batchserver.framework.repository.Repository#getAllRegisteredMgmtMntrInfo()
     */        
    public Map getAllRegisteredMgmtMntrInfo()
    {
        logger.trace("Entering getAllRegisteredMgmtMntrInfo");
        
        Map registeredJobs=null;
        
        Connection connection=JdbcConnectionHelper.getConnection(this.jdbcConfigMap);
        if(connection==null)
            registeredJobs=new HashMap();
        else
        {
            try
            {
                PreparedStatement statement=connection.prepareStatement("select job_name, job_info from job_mgmt_mntr_info");
                ResultSet rs=statement.executeQuery();
                registeredJobs=new HashMap();
                while(rs.next())
                {
                    String jobName=rs.getString(1);
                    InputStream inputStream=rs.getBinaryStream(2);
                    Object registredInfo=getObjectFromByteArray(inputStream);
                    registeredJobs.put(jobName, registredInfo);
                }
                rs.close();
                statement.close();
            }
            catch(SQLException exception)
            {
                exception.printStackTrace();
                logger.error("Exception while trying to get all the registered jobs");
                registeredJobs=new HashMap();
            }
            finally
            {
                JdbcConnectionHelper.closeConnection(connection);
            }            
        }
        logger.trace("Exiting getAllRegisteredMgmtMntrInfo");        
        return registeredJobs;        
    }

    private byte[] getByteArrayFromObject(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;
        }
    }
    
    private Object getObjectFromByteArray(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;
        }        
    }
    
////////    Need to be implemented in the future.    
////////    public boolean releaseController(String jobName)
////////    {
////////        return false;
////////    }
////////
////////    public boolean saveController(String jobName, JobController controller)
////////    {
////////        return false;
////////    }
////////
////////    public JobController loadController(String jobName)
////////    {
////////        return null;
////////    }    
}

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

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