View Javadoc
   /* ====================================================================
    * License: 
    * 
    * Redistribution and use in source and binary forms, with or without
    * modification, are permitted provided that the following conditions
    * are met:
    *
    * 1. Redistributions of source code must retain the above copyright
    *    notice, this list of conditions and the following disclaimer.
   *
   * 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 end-user documentation included with the redistribution,
   *    if any, must include the following acknowledgment:
   *       "This product includes software developed by 
   *        Robert Half International (http://www.rhi.com/)."
   *    Alternately, this acknowledgment may appear in the software itself,
   *    if and wherever such third-party acknowledgments normally appear.
   *
   * 4. The names "Parc", "RHI", and "Robert Half International" must
   *    not be used to endorse or promote products derived from this
   *    software without prior written permission. For written
   *    permission, please contact pete.mckinstry@rhi.com.
   *
   * 5. Products derived from this software may not be called "PARC",
   *    nor may "PARC" appear in their name, without prior written
   *    permission of Robert Half International.
   *
   * THIS SOFTWARE IS PROVIDED ``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 THE APACHE SOFTWARE FOUNDATION 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.
   * ====================================================================
   *
   */
  package com.rhi.architecture.batch;
  
  import com.rhi.architecture.logging.LogUtil;
  import com.rhi.architecture.logging.Logger;
  import com.rhi.architecture.parc.Record;
  import com.rhi.architecture.resource.InitializationException;
  import com.rhi.architecture.config.ConfigurationException;
  import com.rhi.architecture.db.DBUtil;
  import com.rhi.architecture.xa.DatabaseTransaction;
  import com.rhi.architecture.xa.Transaction;
  
  import java.sql.Connection;
  import java.sql.SQLException;
  import java.util.Collection;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.Properties;
  
  import javax.sql.DataSource;
  
  /***
   * Abstract Audit Agent providing common audit count functionality
   *
   * @author  Pete McKinstry
   * @copyright 2002, Robert Half Int'l., Inc. All rights reserved.
   *
   * @since 1.0
   */
  public abstract class AbstractAuditAgent extends AuditAgent {
  
     /***
      * Logger
      */
     private static Logger log = null;
  
     /***
      * key for audit agent insert sql statement. (in config)
      */
     public static final String INSERT_SQL_KEY =
        "AbstractAuditAgent.InsertRecord";
     /***
      * key for audit agent update sql statement. (in config)
      */
     public static final String UPDATE_SQL_KEY =
        "AbstractAuditAgent.UpdateRecord";
     /***
      * key for audit agent select sql statement. (in config)
      */
     public static final String SELECT_SQL_KEY =
        "AbstractAuditAgent.SelectRecord";
  
     /***
      * variable holding actual sql for insert.
     */
    private String insertRecordSQL;
    /***
     * variable holding actual sql for update.
     */
    private String updateRecordSQL;
    /***
     * variable holding actual sql for select.
     */
    private String selectRecordSQL;
 
    /***
     * Set (ignores duplicates) of all valid record keys. Used to audit 
     * number of records processed. 
     */
    private HashSet validSet;
    /***
     * Set (ignores duplicates) of all errored record keys. Used to audit 
     * number of records processed.
     */
    private HashSet errSet;
 
    /***
     * lock for concurrency checks.
     */
    private Object objLock = new Object();
 
    /***
     * Constructor.
     *
     * @since 1.0
     */
    public AbstractAuditAgent() {
       validSet = new HashSet();
       errSet = new HashSet();
    }
 
    /***
     * Initialization Logic
     * 
     * @param p configuration properties for the audit agent.
     * @throws InitializationException thrown when a problem is
     * encountered configuring the audit agent.
     * @since 1.0
     */
    public void init(Properties p) throws InitializationException {
       try {
          log = LogUtil.getLogger();
       }
       catch (ConfigurationException e) {
          throw new InitializationException("Unable to create Logger", e);
       }
 
       log.debug("init()");
 
       loadSQL(p);
 
       configureRecord(p);
    }
 
    /***
     * Allow concrete subclasses a hook to set values on their AuditRecord
     * class.
     * 
     * @param p configuration properties for the audit agent.
     * @throws InitializationException error initializing the audit agent
     * @since 1.0
     */
    public abstract void configureRecord(Properties p)
       throws InitializationException;
 
    /***
     * Helper method called by init() to load the required SQL entries from
     * the properties file.
     *
     * @param p configuration properties for the audit agent.
     * @throws InitializationException error initializing the audit agent
     * @since 1.0
     */
    protected void loadSQL(Properties p) throws InitializationException {
 
       // load the SQL for updating the audit counts.
       this.insertRecordSQL = p.getProperty(getInsertSQLKey(), "not found");
       if (insertRecordSQL.equals("not found")) {
          log.error(getInsertSQLKey() + " missing from config properties.");
          throw new InitializationException(
             getInsertSQLKey() + " missing from configuration properties.");
       }
       this.updateRecordSQL = p.getProperty(getUpdateSQLKey(), "not found");
       if (updateRecordSQL.equals("not found")) {
          log.error(getUpdateSQLKey() + " missing from config properties.");
          throw new InitializationException(
             getUpdateSQLKey() + " missing from configuration properties.");
       }
       this.selectRecordSQL = p.getProperty(getSelectSQLKey(), "not found");
       if (selectRecordSQL.equals("not found")) {
          log.error(getSelectSQLKey() + " missing from config properties.");
          throw new InitializationException(
             getSelectSQLKey() + " missing from configuration properties.");
       }
    }
 
    /***
     * Create an appropriate audit /control record. This
     * method should create a record w/ only a unprocessed
     * count > 0. The auditCollection method will be called
     * as records are passed through the pipeline & this
     * method will modify the counts; subtracting from the
     * unprocesed, and adding to the errored, valid, or
     * duplicate counts.
     *
     * @param c a collection of records that need to be audited.
     * @throws AuditException error auditing collection, usually fatal.
     * @since 1.0
     */
    public void createAudits(Collection c) throws AuditException {
       int size = c.size();
       log.debug("createAudits(), unprocessed count = " + size);
 
       Connection conn = null;
       try {
          DataSource ds = DBUtil.getDataSource();
          conn = ds.getConnection();
          if (loadRecord(conn) == false) {
             // set unprocessed count.
             getRecord().setUnprocessedCount(new Integer(size));
             insertRecord(conn);
          }
          else {
             // set unprocessed count.
             getRecord().setUnprocessedCount(new Integer(size));
             updateRecord(conn);
          }
       } // try
       catch (ConfigurationException e) {
          throw new AuditException("can't create audit record", e);
       }
       catch (SQLException e) {
          throw new AuditException("can't create audit record", e);
       }
       finally {
          try {
             if (conn != null && !conn.isClosed()) {
                conn.close();
             }
          }
          catch (SQLException e) {
             // ignore
          }
       } // finally
    }
 
    /***
     * Audit the given collection. This method should
     * subtract the given collection.size() from the un-
     * processed count for this audit set & add the
     * collection.size() in some manner to the processed
     * columns (valid,errored,duplicate,other)
     *
     * @param c - records to be audited.
     * @param t - the transaction to participate in.
     * @throws AuditException thrown when an error occurs auditing the given
     * collection.
     * @since 1.0
     */
    public void auditCollection(Collection c, Transaction t)
       throws AuditException {
       log.debug("auditing " + c.size() + " new records.");
 
       synchronized (objLock) {
          updateRecordCounts(c);
 
          // The connection is controlled by the transaction & should
          // not be closed w/in the audit agent.
          Connection conn = ((DatabaseTransaction) t).getConnection();
          updateRecord(conn);
       }
    }
 
    /***
     * Update record counts based on the new collection.
     *
     * @param c update counts for this collection
     * @throws AuditException thrown if error updating counts.
     * @since 1.0
     */
    protected void updateRecordCounts(Collection c) throws AuditException {
 
       // The newlyProcessed flag represents the unique new source
       // records that are counted (subtracted from the unprocessed
       // count) as part of this audit cycle. It they have previously
       // been counted, they are not re-counted.
       int newlyProcessed = 0;
 
       // extract info from collection.
       Iterator iter = c.iterator();
       while (iter.hasNext()) {
          // this is the pipeline record, not the audit record
          Record record = (Record) iter.next();
          Object key = record.getSourceKey();
          if (record.isValid()) {
             if (errSet.contains(key) == false) {
                if (validSet.add(key) == true) {
                   ++newlyProcessed;
                }
             }
          }
          else {
             if (validSet.contains(key) == true) {
                if (validSet.remove(key) == true) {
                   // the decrement is only here because we'll
                   // be incrementing it 2 statements later.
                   // We don't want to double count.
                   --newlyProcessed;
                }
             }
             if (errSet.add(key) == true) {
                ++newlyProcessed;
             }
          }
       }
       // calculate new values
       int processed = validSet.size();
       int errored = errSet.size();
       int unprocessed =
          getRecord().getUnprocessedCount().intValue() - (newlyProcessed);
       log.debug(
          "processed = "
             + processed
             + ", errored = "
             + errored
             + ", unprocessed = "
             + unprocessed);
       // Check that you don't have a negative unprocessed value.
       // This would only happen if you somehow counted the valid &
       // errored records incorrectly.
       if (unprocessed < 0) {
          log.error("control metric totals invalid. u<0");
          throw new AuditException("control metric totals invalid.");
       }
 
       // update control record.
       getRecord().setProcessedCount(new Integer(processed));
       getRecord().setErroredCount(new Integer(errored));
       getRecord().setUnprocessedCount(new Integer(unprocessed));
    }
 
    /***
     * Load record (check for existence.
     *
     * @param conn db connection used for audits. (maintaining a single
     * db transaction)
     * @return true if records exists, else false. 
     * @throws AuditException error retrieving audit record.
     * @since 1.0
     */
    protected abstract boolean loadRecord(Connection conn)
       throws AuditException;
 
    /***
     * Insert new audit record
     *
     * @param conn db connection used for audits in order to maintain all data
     * updates w/in a single transaction.
     * @throws AuditException error inserting a new audit record.
     * @since 1.0
     */
    protected abstract void insertRecord(Connection conn) 
       throws AuditException;
 
    /***
     * Update record
     *
     * @param conn db connection used for audits in order to maintain all data
     * updates w/in a single transaction.
     * @throws AuditException error updating a audit record.
     * @since 1.0
     */
    protected abstract void updateRecord(Connection conn) 
       throws AuditException;
 
    /***
     * No op. This method is not very useful now that a Transaction is used
     * for all OutputAdapter commit logic. Subclasses will probably not need
     * to override this method.
     *
     * @throws AuditException error commiting the audits. 
     * @since 1.0
     */
    public void commitAudits() throws AuditException {
       log.debug("commitAudits() - no op");
    }
 
    /***
     * Get concrete AuditRecord
     * Concrete AuditAgent will return the actual audit record being
     * used. Allow more flexibility in sub-classes.
     *
     * @return AuditRecord the audit record sub-class to be used. 
     * @since 1.0
     */
    public abstract AuditRecord getRecord();
 
    /***
     * get insert SQL
     *
     * @return String insert sql 
     * @since 1.0
     */
    protected String getInsertSQL() {
       return this.insertRecordSQL;
    }
 
    /***
     * get update SQL
     *
     * @return String update sql
     * @since 1.0
     */
    protected String getUpdateSQL() {
       return this.updateRecordSQL;
    }
 
    /***
     * get select SQL
     *
     * @return String select sql 
     * @since 1.0
     */
    protected String getSelectSQL() {
       return this.selectRecordSQL;
    }
 
    /***
     * Allow subclasses to override config key.
     *
     * @return String insert sql 
     * @since 1.0
     */
    protected String getInsertSQLKey() {
       return AbstractAuditAgent.INSERT_SQL_KEY;
    }
 
    /***
     * Allow subclasses to override config key.
     *
     * @return String update sql key
     * @since 1.0
     */
    protected String getUpdateSQLKey() {
       return AbstractAuditAgent.UPDATE_SQL_KEY;
    }
 
    /***
     * Allow subclasses to override config key.
     *
     * @return String select sql key
     * @since 1.0
     */
    protected String getSelectSQLKey() {
       return AbstractAuditAgent.SELECT_SQL_KEY;
    }
 
    /***
     * Get the Logger reference
     *
     * @return Logger log instance. 
     * @since 1.0
     */
    public Logger log() {
       return AbstractAuditAgent.log;
    }
 
 }