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.parc;
  
  import com.rhi.architecture.batch.AuditAgent;
  import com.rhi.architecture.batch.AuditException;
  import com.rhi.architecture.logging.LogUtil;
  import com.rhi.architecture.logging.Logger;
  import com.rhi.architecture.util.*;
  import com.rhi.architecture.threads.ThreadPool;
  
  import java.util.ArrayList;
  import java.util.Collection;
  import java.util.Properties;
  
  
  /***
   * The OutputAdapter is responsible for taking a completed
   * work set and storing it. File or DB persistence
   * mechanisms will be the only initially supported output
   * adapter types.
   *
   * @author Pete McKinstry
   * @copyright 2002, Robert Half Int'l., Inc. All rights reserved.
   *
   * @since 1.0
   */
  public abstract class AbstractOutputAdapter 
                                          implements OutputAdapter, Monitor {
  
     private static Logger log = null;
  
     private static final String MAX_CNT_TAG = "oa_max_rec_count";
     private static final String DEFAULT_RECORD_COUNT = "500";
     private static final String MAX_THREAD_CNT_TAG = "oa_max_threads";
     private static final String DEFAULT_THREAD_COUNT = "1";
  
  	private AuditAgent auditAgent;
  	private String txTypeName;
  
     // number of records to persist @ a time.
     private int COMMIT_SIZE;
  
     // Thread pool for the output adapter
     private ThreadPool myThreadPool = null;
     // maximum number of threads in the thread pool
     private int MAX_THREADS;
  
     private Channel outputChannel;
     private Channel errorChannel;
     private ExceptionHandler handler; 
     
     private static final String MAX_SLEEP_TAG = 
                                               "AbstractOutputAdapter.MaxSleep";
     private static final String DEFAULT_MAX_SLEEP = "2500"; // 2.5 seconds
    private int sleepTime;
    
    private volatile boolean done; 
    
    
 	/***
 	 * Default constructor.
 	 *
 	 * @since 1.0
 	 */
 	public AbstractOutputAdapter() {
 	}
 
 
 	/***
 	 * Set method to tell the OutputAdapter what audit
 	 * agent to use. The output Adapter is responsible for
 	 * setting the audit counts when it persists the records.
 	 *
 	 * @param AuditAgent
 	 *
 	 * @since 1.0
 	 */
 	public final void setAuditAgent(AuditAgent agent) {
 		this.auditAgent = agent;
 	}
 	/***
 	 * get audit agent
 	 * @return AuditAgent
 	 *
 	 * @since 1.0
 	 */
 	public final AuditAgent getAuditAgent() {
 		 return auditAgent;
 	}
 
 
 	/***
 	 * Load error logfile & create writer.
 	 *
 	 * @since 1.0
 	 */
 	public void init( Properties props )
 								throws InitializationException {
 		try {
 			log = LogUtil.getLogger();
 		}
 		catch (ConfigurationException e) {
 			throw new InitializationException(e.toString(), e);
 		}
 
 		log.debug("init");
 		try {
 			String rec_count = props.getProperty(MAX_CNT_TAG,
                                                 DEFAULT_RECORD_COUNT);
 			COMMIT_SIZE = Integer.parseInt( rec_count );
 			// transaction type
 			txTypeName = props.getProperty( Transaction.TYPE_KEY );
 			if (txTypeName == null) {
 				throw new InitializationException( Transaction.TYPE_KEY +
 					 			" missing from configuration 
 properties.");
 			}
 		}
 		catch ( NumberFormatException nfe ) {
 			throw new InitializationException(
         			"Invalid number for config value = " + MAX_CNT_TAG, nfe );
 		}
 
 		// Get maximum number of threads for the output adapter
 		try {
 			String thread_count = props.getProperty(MAX_THREAD_CNT_TAG,
                                                 DEFAULT_THREAD_COUNT);
 			MAX_THREADS = Integer.parseInt( thread_count );
 		}
 		catch ( NumberFormatException nfe ) {
 			throw new InitializationException(
         			"Invalid number for config value = " + MAX_THREAD_CNT_TAG, nfe );
 		}
 
       try { 
          String sleepValue = props.getProperty(MAX_SLEEP_TAG, 
                                                 DEFAULT_MAX_SLEEP);
          this.sleepTime = Integer.parseInt( sleepValue );
       }
       catch (NumberFormatException e) { 
          throw new InitializationException(
                "Invalid number for config value = " + MAX_SLEEP_TAG, e );
       }
 
 		// If the object implements SingleThreadModel then
 		//     default max threads to 1
 		if ( this instanceof SingleThreadModel ) {
 			MAX_THREADS = 1;
 		}
 		myThreadPool = new ThreadPool(1,MAX_THREADS);
 	}
 
 
 	/***
 	 * cleanup() is called by the BatchApplication during shutdown.
 	 * It closes the errorLog writer and threadpool.
 	 *
 	 * @since 1.0
 	 */
 	public void cleanup() {
 		if (log!=null) {
             log.debug("running cleanup");
         }
 
         if (myThreadPool != null) {
         	myThreadPool.close();
         }
 	}
 
 
 	/***
     * Do any required processing prior to completing the batch 
     * cycle. The flush() method is called by the strategy for
     * each execution cycle. This differs from the cleanup 
     * method, which is called only once upon application 
     * shutdown.
     * <p/>
     * flush() is called by the run() method during each cycle.
     * 
     * @throws ProcessingException Signals a fatal error. The 
     *    application will shutdown if this exception is thrown.
     *
 	 * @since 1.0
 	 */
 	public void flush() throws ProcessingException {
 		try {
 			auditAgent.commitAudits();
 		}
 		catch ( AuditException ae ) {
 			log.error("AuditException thrown by commitAudits()", ae);
 			throw new ProcessingException( ae.toString(), ae );
 		}
 	}
 
    /***
     * @see java.lang.Runnable#run()
     */
    public void run() {
       try {
          // OA should be notified if records are dumped in either channel.
          this.outputChannel.registerMonitor(this);
          this.errorChannel.registerMonitor(this);
       }
       catch (Throwable t) {
          log().fatal("Throwable caught in AbstractOutputAdapter." + 
                      "e = " + t, t);
          this.handler.reportException(new ProcessingException(t));
          t.printStackTrace();
          return;
       }
       
       // 'localDone' variable is used to ensure that write() is 
       // called once after the this.done variable is set to true. 
       // This will force all the records in the output & error 
       // channels to be flushed prior to exit. Otherwise you have
       // a race condition where records could be added to the 
       // channels after write() is called, but before the loop 
       // resets. 
       boolean localDone;
       do { 
          localDone = this.done;
          try { 
             write(this.outputChannel, this.errorChannel);
             flush();
             
             // if pipeline not complete, wait for notification of more records.
             if (!this.done) { 
                 try {
                     log.trace("Empty cycle detected, calling wait()");
                     synchronized (this) {
                         wait();
                     }
                 }
                 catch (InterruptedException e) {
                     // ignore
                 }
             }
          }
          catch (ProcessingException e) { 
             log().fatal("ProcessingException thrown in OutputAdapter. " +
                         "e = " + e, e);
             this.handler.reportException(e);
             e.printStackTrace();
          }
          catch (Throwable t) { 
             log().fatal("Throwable caught in AbstractOutputAdapter." + 
                         "e = " + t, t);
             this.handler.reportException(new ProcessingException(t));
             t.printStackTrace();
          }
       }
       while ((!localDone) && (handler.hasError()==false));
    }
    
 	/***
     * Persist valid and errored records to the appropriate data
     * store. Depending on the concrete interface, the output
     * channel may or may not have errored records in it. 
     * Output adapters should process the valid records first so
     * that any errors can be added to the error collection.
     * 
     * @param Channel holding bin for valid records waiting to be 
     *    persisted.
     * @param Channel holding bin for errored records 
     *    waiting to be persisted. 
     * 
     * @exception ProcessingException error occured 
     *    during write. Signals a fatal error. The application 
     *    will shutdown if this exception is thrown.
     * 
     * @since 1.0
 	 */
 	public void write( Channel ch, Channel err )
 												throws 
 Processin
 gExceptio
 n {
 
 		//1. write valid records
 		writeValidRecords(ch);
 
 		//2. write error records
 		writeErrorRecords(err);
 
 		// 3. do end of batch processing ( non record level )
 		Transaction transaction = createTransaction();;
 		try {
 			transaction.begin();
 			completeBatch( transaction );
 			transaction.commit();
 		}
 		catch (ProcessingException e) {
 			log.error("ProcessingException", e);
 			try {
 				transaction.rollback();
 			}
 			catch ( TransactionException te ) {
 				// ignore
 			}
 			throw e;
 		}
 		catch (TransactionException e) {
 			log.error("TransactionException", e);
 			try {
 				transaction.rollback();
 			}
 			catch ( TransactionException te ) {
 				// ignore
 			}
 			throw new ProcessingException( e.toString(), e );
 		}
 	}
 
 	/***
 	 *
 	 * To Improve Performance, a thread pool is created and used.
 	 * From valid channel, oa_max_rec_count number of records are
 	 * retrieved and handed over to each thread. At any point of time
 	 * at maximum oa_max_threads number of thread will be running.
 	 * Once, all threads are handling work, main thread waits for all of
 	 * them to finish and then only processes next set of records.
 	 *
 	 * @param Channel valid records.
 	 *
 	 * @since 1.0
 	 */
 	public void writeValidRecords(Channel ch)
 								throws ProcessingException	{
 
         int vSize = 0;
 		int threadCount = 0;
 		ExceptionHandler handler = new ExceptionHandler();
 		ValidRecordsTask prototype =
 				new ValidRecordsTask(this, MAX_THREADS, handler);
 
 		do {
 			Collection valid = null;
             valid = extractValidRecords( ch, COMMIT_SIZE );
             vSize = valid.size();
             if ( vSize > 0 ) {
             	log.info("Thread Id - "+threadCount+
             		":Processing a batch of "+ vSize + " valid records.");
             	ValidRecordsTask validTask =
             		(ValidRecordsTask) prototype.clone();
             	validTask.setRecords(valid);
             	myThreadPool.execute(validTask);
             	threadCount++;
             	if ( threadCount == MAX_THREADS ) 	{
             		log.debug("Valid: Waiting for join");
 			try {
             			myThreadPool.join();
 			}
 			catch (InterruptedException e) {
 			
 	throw new ProcessingException( e.toString(), e );
 			
 }
             		log.debug("Valid: Join finished");
             		threadCount = 0;
 					if (handler.hasError()) {
 						handler.rethrowException();
 					}
             	}
             }
         } while ( vSize > 0 );
 
 	    if ( threadCount > 0 ) {
             	log.debug("AfterValid: waiting for threads to join.");
 		try {
             		myThreadPool.join();
 		}
 		catch (InterruptedException e) {
 			throw new 
 ProcessingException( e.toString(), e );
 		}
             	
 log.debug("AfterValid: Finished waiting for threads to join.");
 				if (handler.hasError()) {
 					handler.rethrowException();
 				}
 	    }
 	    else {
 			log.debug("No threads waiting to be joined.");
 		}
 	}
 
 	/***
 	 *
 	 * To Improve Performance, a thread pool is created and used.
 	 * From error channel, oa_max_rec_count number of records are
 	 * retrieved and handed over to each thread. At any point of time
 	 * at maximin oa_max_threads number of thread will be running.
 	 * Once, all threads are handling work, main thread waits for all of
 	 * them to finish and then only processes next set of records.
 	 *
 	 * @param Channel error records.
 	 *
 	 * @since 1.0
 	 */
 	public void writeErrorRecords(Channel err)
 								throws ProcessingException	{
 		int eSize = 0;
 		int threadCount = 0;
 		ExceptionHandler handler = new ExceptionHandler();
 		ErrorRecordsTask prototype =
 				new ErrorRecordsTask(this, MAX_THREADS, handler);
 
  		do {
 			Collection errors = null;
 			errors = extractErrorRecords( err, COMMIT_SIZE );
 			eSize = errors.size();
             if ( eSize > 0 ) {
                	log.info("Thread Id - "+threadCount+
             		":Processing a batch of "+ eSize + " error records.");
             	ErrorRecordsTask errorTask =
             		(ErrorRecordsTask) prototype.clone();
             	errorTask.setRecords(errors);
             	myThreadPool.execute(errorTask);
             	threadCount++;
             	if ( threadCount == MAX_THREADS ) {
             		log.debug("Error: Waiting for join");
 			try {
             			myThreadPool.join();
 			}
 			catch (InterruptedException e) {
 			
 	throw new ProcessingException( e.toString(), e );
 			
 }
             		log.debug("Error: Join finished");
 					if (handler.hasError()) {
 						handler.rethrowException();
 					}
             		threadCount = 0;
             	}
             }
         } while (eSize > 0 );
 
 	    if ( threadCount > 0 ) {
             	log.debug("AfterError: waiting for threads to join.");
 		try {
             		myThreadPool.join();
 		}
 		catch (InterruptedException e) {
 			throw new 
 ProcessingException( e.toString(), e );
 		}
             	
 log.debug("AfterError: Finished waiting for threads to join.");
 				if (handler.hasError()) {
 					handler.rethrowException();
 				}
 	    }
 	    else {
 			log.debug("No threads waiting to be joined.");
 		}
 	}
 
 	/***
 	 * Helper method for converting the error channel into a collection
 	 *
 	 * @param Channel errored records.
 	 * @return Collection errored records as a collection.
 	 *
 	 * @since 1.0
 	 */
 	private Collection extractValidRecords( Channel ch, int batchSize )
 												throws 
 Processin
 gExceptio
 n {
 		Collection records = new ArrayList();
 		if ( ch == null ) {
 			log.error("valid channel null. ");
 			throw new ProcessingException("Valid channel null in " +
                                             "AbstractOutputAdapter");
 		}
 
 		Collection c = ch.pull( batchSize );
         records.addAll( c );
 		log.debug( "found " + records.size() + " records in valid channel");
 
 		return records;
 	}
 
 
 	/***
 	 * Defer handling of valid records to sub-classes..
 	 *
 	 * @param Collection valid records.
      * @param Transaction transaction under which these records should
      * be committed.
      * @return Collection
      * @throws ProcessingException
 	 *
 	 * @since 1.0
 	 */
 	protected abstract Collection handleValidRecords( Collection c,
                                                         Transaction t )
 												throws 
 Processin
 gExceptio
 n;
 
 
 	/***
 	 * Helper method for converting the error channel into a collection
 	 *
 	 * @param Channel errored records.
 	 * @return Collection errored records as a collection.
 	 *
 	 * @since 1.0
 	 */
 	private Collection extractErrorRecords( Channel err, int batchSize )
 												throws 
 Processin
 gExceptio
 n {
 		Collection errors = new ArrayList();
 		if ( err == null ) {
             log.error("error channel null. ");
             throw new ProcessingException("Error channel null in " +
                                             "AbstractOutputAdapter");
 		}
 
 		Collection c = err.pull( batchSize );
 		errors.addAll( c );
 
 		log.debug( "found " + errors.size() + " records in error channel");
 
 		return errors;
 	}
 
 
     /***
      * Defer handling of valid records to sub-classes..
      *
      * @param Collection valid records.
      * @param Transaction transaction under which these records should
      * be committed.
      * @return Collection
      * @throws ProcessingException
      *
      * @since 1.0
      */
     protected abstract void handleErrorRecords( Collection c, Transaction t )
                                                 throws ProcessingException;
 
 	/***
 	 * Do any non-record level processing required to finish this
 	 * batch cycle. An example of this is incrementing the file
 	 * sequence number after each batch (write()) is finished.
 	 * <p/>
 	 * Since this type of processing doesn't depend on records,
 	 * it should be called outside of the handle* methods.
 	 *
 	 * @param transaction
 	 * @throws ProcessingException
 	 *
 	 * @since 1.1
 	 */
 	protected void completeBatch( Transaction transaction )
 										throws 
 ProcessingException {
 		// optional method. no-op in base class.
 	}
 
 	/***
 	 * No op by default.
      *
      * @see com.rhi.architecture.parc.OutputAdapter#close()
      *
      * @since 1.1
      */
    public void close() throws ProcessingException {
       // no op. If you need it, override it.
    }
 
    /***
     * @see com.rhi.architecture.parc.OutputAdapter#setErrorChannel(Channel)
     */
    public void setErrorChannel(Channel ch) {
       this.errorChannel = ch;
    }
 
    /***
     * @see com.rhi.architecture.parc.OutputAdapter#setValidChannel(Channel)
     */
    public void setValidChannel(Channel ch) {
       this.outputChannel = ch;
    }
 
    /***
     * @see com.rhi.architecture.parc.OutputAdapter#setExceptionHandler(ExceptionHandler)
     */
    public void setExceptionHandler(ExceptionHandler h) {
       this.handler = h;
    }
    
    /***
     * @see com.rhi.architecture.parc.OutputAdapter#markComplete()
     */
    public void markComplete() {
       this.done = true;
       
       synchronized(this) {
         notifyAll();
       }
    }
    
     /***
      * Simple implementation of Monitor interface based on Thread
      * wait/notify mechanism.
      * 
      * @see com.rhi.architecture.parc.Monitor#notify(Event)
      */
     public void notify(Event e) {
         log.trace("Notification received. " + e);
         synchronized(this) { 
             notifyAll();
         }
     }
     
     /***
 	 * Create a transaction of the appropriate type.
 	 *
 	 * @return Transaction
 	 *
 	 * @since 1.0
 	 */
 	public Transaction createTransaction() throws ProcessingException {
 		try {
 			Class txClass = Class.forName(txTypeName);
 			Transaction tx = (Transaction)txClass.newInstance();
 			return tx;
 		}
 		catch (ClassNotFoundException cnfe) {
     		throw new ProcessingException("ClassNotFoundException", cnfe);
 		}
 		catch (InstantiationException ie) {
 			throw new ProcessingException("InstantiationException", ie);
 		}
 		catch (IllegalAccessException iae) {
 			throw new ProcessingException("IllegalAccessException", iae);
 		}
 	}
 
 	/***
 	 * Returns the Logger
      *
 	 * @return Logger
 	 */
 	public Logger log() {
 		return log;
 	}
 
 }