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.filter;
  
  import com.rhi.architecture.logging.LogUtil;
  import com.rhi.architecture.logging.Logger;
  import com.rhi.architecture.parc.Consumer;
  import com.rhi.architecture.parc.ExceptionHandler;
  import com.rhi.architecture.parc.Filter;
  import com.rhi.architecture.parc.PARCApplication;
  import com.rhi.architecture.parc.ProcessingException;
  import com.rhi.architecture.parc.Supplier;
  import com.rhi.architecture.resource.InitializationException;
  import com.rhi.architecture.config.ConfigurationException;
  
  import java.util.Collection;
  import java.util.Properties;
  
  /***
   * The AbstractFilter provides a partially implemented
   * Filter allowing simpler concrete Filter objects. It
   * deals w/ thread safe exit logic, and idle cycle
   * detection as well as providing the required hooks for
   * pre & post channel hookup. All concrete filter
   * processing is deferred to a doWork() abstract method.
   * <p/>
   * Note: a simple run() method is provided to support
   * the Runnable Interface.
   *
   * @author Pete McKinstry
   * @copyright 2002, Robert Half Int'l., Inc. All rights reserved.
   *
   * @since 1.0
   */
  public abstract class AbstractFilter implements Filter {
  
     // for normal logging activities, use the log category.
     private static Logger log = null;
  
     // for the few cycle statistics messages that are logged,
     // use the stats category.
     private static Logger stats = null;
  
     private Supplier supplier;
     private Consumer consumer;
     private Consumer errors;
  
     // for reporting fatal errors in a thread.
     private ExceptionHandler handler;
  
     private int maxRecords = 20; // default
     private int numThreads = 1; // default
  
     private String name = null;
  
    /*
        the shutdown flag needs to be volatile to ensure that
        each thread accesses the correct value. The flag is
        used to trigger a thread exit.
    */
    private volatile boolean shutdown = false;
 
    /***
     * constructor
     */
    protected AbstractFilter() {
        super();
    }
 
    /***
     * constructor
     *
     * @param max - greatest number of records that this
     * filter will accept from the channel in 1 batch. Used
     * in conjunction w/ the thread count to performance tune
     * the pipeline.
     *
     * @since 1.0
     */
    protected AbstractFilter(int max) {
       this.maxRecords = max;
    }
 
    /***
     * Initialize common Filter settings. Note: sub-classes may
     * override this method, but they must call super.init() in
     * addition to doing their own local initialization. If
     * any of these values are already set, they will be over-
     * written. If you don't want that to happen, do not provide
     * a property setting for these attributes.
     * </p>
     * Sets the following common Filter attributes.
     *      1) <ConcreteFilterClass>.max_records: the max number of
     *  records to be pulled from the channel during each work
     *      cycle.
     *      2) <ConcreteFilterClass>.num_threads: the number of
     *      threads to run for this filter.
     *  If no setting is found, the default value will be used.
     *  - For threads, the default is 1.
     *  - For work sets, the default size is 20.
     *
     * @param properties
     * @throws InitializationException
     *
     * @since 1.0
     */
    public void init(Properties properties)
       throws InitializationException {
       try {
          log = LogUtil.getLogger();
          stats = LogUtil.getLogger(PARCApplication.CYCLE_STATS);
       }
       catch (ConfigurationException e) {
          throw new InitializationException(e.toString());
       }
 
       try {
          String full_name = this.getClass().getName();
          this.name = full_name.substring(full_name.lastIndexOf('.') + 1);
          log.debug("Filter name = " + name);
 
          String sizeStr =
             properties.getProperty(this.name + ".max_records", "20");
          this.maxRecords = Integer.parseInt(sizeStr);
          log.debug("max_records = " + maxRecords);
 
          String threadsStr =
             properties.getProperty(this.name + ".num_threads", "1");
          this.numThreads = Integer.parseInt(threadsStr);
          log.debug("num_threads = " + numThreads);
       }
       catch (NumberFormatException nfe) {
          log.error(
             "configuration error. Filter settings "
                + "provided, but not valid.",
             nfe);
          throw new InitializationException("Filter settings invalid", nfe);
       }
    }
 
    /***
     * Thread execution method. Inherited from Runnable
     * All this method does is call process(). & catch
     * the ProcessingException.
     *
     * @since 1.0
     */
    public void run() {
       try {
          process();
       }
       catch (ProcessingException pe) {
          log.fatal(
             "Processing exception caught in Filter "
                + "<"
                + getName()
                + ">",
             pe);
 
          // run() is only called when a Thread is running the
          // Filter. In that case, the error cannot be propagated
          // up the call stack, so we report it using the
          // ExceptionHandler class. The Pipeline will watch the
          // ExceptionHandler class & shutdown if a fatal error
          // has been detected. If the pipeline is running w/in
          // a single thread, the run() method won't be called,
          // and the ProcessingException will roll up the call
          // stack as per normal execution.
          getExceptionHandler().reportException(pe);
 
          pe.printStackTrace();
       }
       catch (Throwable t) {
          log.fatal(
             "Non-PARC fatal exception caught in Filter "
                + "<"
                + getName()
                + ">",
             t);
 
          getExceptionHandler().reportException(new ProcessingException(t));
 
          t.printStackTrace();
       }
    }
 
    /***
     * Template method to facilitate simple Filter implementations.
     * <p/>
     * This method, just like process(), is part of the main thread
     * loop and therefore must be thread safe.
     *
     * @param in
     * @return
     * @throws ProcessingException
     */
    protected abstract Collection doWork(Collection in)
       throws ProcessingException;
 
    /***
     * Tell the filter to shutdown after detecting idle cycles.
     * markForDeath sets an 'exit' flag on the Filter. The
     * pipeline model assumes that a complete cycle leaves the
     * pipeline empty, therefore, the markForDeath() is _lazy_
     * in the sense that it only takes effect after a thread
     * finds no more work in the inbound channel.
     *
     * @since 1.0
     */
    public void markForDeath() {
       log.debug("Filter <" + getName() + "> marked for death....");
       this.shutdown = true;
 
       // notify any listeners that are waiting.
       synchronized (this) {
          notifyAll();
       }
    }
    /***
     * Reset the filter to ensure that it's ready to
     * process records again. This method must be called
     * after calling markForDeath() to reset the state
     * of the Filter.
     * <p/>
     * Note: Although it's thread safe due to the volatile
     * keyword, the reset() method is intended to be used
     * @ the beginning/end of a pipeline execution cycle
     * when no threads are actively executing the Filter.
     *
     * @since 1.1
     */
    public void reset() {
       log.debug("reset called on Filter");
       this.shutdown = false;
    }
    /***
     * Returns the shutdown flag
     * @return boolean
     */
    protected boolean isShutdown() {
       return shutdown;
    }
 
    /***
     * Set the inbound delivery mechanism.
     *
     * @param s
     *
     * @since 1.0
     */
    public void setInbound(Supplier s) {
       this.supplier = s;
    }
    /***
     * supplier configured on the inbound side of the filter.
     * @return Supplier
     */
    public Supplier getInbound() {
       return this.supplier;
    }
 
    /***
     * Set the outbound delivery mechanism.
     *
     * @param c
     *
     * @since 1.0
     */
    public void setOutbound(Consumer c) {
       this.consumer = c;
    }
    /***
     * consumer configured on the outbound side of this filter.
     * @return Consumer
     */
    public Consumer getOutbound() {
       return this.consumer;
    }
 
    /***
     * Set the error delivery mechanism.
     *
     * @param err
     *
     * @since 1.0
     */
    public void setErrorChannel(Consumer err) {
       this.errors = err;
    }
    /***
     * return error channel
     * @return Consumer
     */
    public Consumer getErrorChannel() {
       return this.errors;
    }
 
    /***
     * Set the exception handler mechanism.
     *
     * @param handler
     *
     * @since 1.0
     */
    public void setExceptionHandler(ExceptionHandler handler) {
       this.handler = handler;
    }
    /***
     * return exception handler
     * @return ExceptionHandler
     */
    public ExceptionHandler getExceptionHandler() {
       return this.handler;
    }
 
    /***
     * Easy implementation for sub-classes
     * @return String
     * @since 1.0
     */
    public String getName() {
       return this.name;
    }
 
    /***
     * Method setNumThreads.
     * @param val
     */
    public void setNumThreads(int val) {
       numThreads = val;
    }
    /***
     * Return the suggested number of threads to launch for this
     * Filter. Obviously this requires that the application uses
     * a Pipeline that implements multi-threading. The number of
     * threads is '1' by default, but can be altered by setting a
     * configuration property = <ConcreteFilterName>.num_threads
     * to be whatever value you wish.
     *
     * @return int the number of threads to launch for this Filter
     *
     * @since 1.1
     */
    public int numThreads() {
       return this.numThreads;
    }
 
    /***
     * Method setMaxRecords.
     * @param val
     */
    public void setMaxRecords(int val) {
       maxRecords = val;
    }
    /***
     * Returns the maxRecords.
     * @return int
     */
    public int getMaxRecords() {
       return maxRecords;
    }
 
    /***
     * Returns the Logger instance used for statistics.
     *
     * @return Logger
     */
    protected static Logger stats() {
       return stats;
    }
 
    /***
     * Returns the log.
     * @return Logger
     */
    public static Logger log() {
       return log;
    }
 
 }