ReliableIndexerListener.java

  1. /*
  2.  * Reliable Indexer Listener
  3.  */
  4. package gov.usgs.earthquake.indexer;

  6. import java.util.List;
  7. import java.util.logging.Level;
  8. import java.util.logging.Logger;

  10. import gov.usgs.earthquake.distribution.ConfigurationException;
  11. import gov.usgs.util.Config;

  13. /**
  14.  * ReliableIndexerListener listens for product changes by the indexer, then handles the new products independently in a background thread.
  15.  *
  16.  * This class does little more than output logs for the products it has seen; it is designed to be extended.
  17.  *
  18.  * Several useful methods are availble to be overridden or otherwise used:
  19.  * <ul>
  20.  * <li>onBeforeProcessThreadStart</li>
  21.  * <li>onProcessException</li>
  22.  * <li>getNextProducts</li>
  23.  * <li>processProducts</li>
  24.  * </ul>
  25.  *
  26.  * This class accepts an index for querying in config:
  27.  *
  28.  * <dl>
  29.  * <dt>index</dt>
  30.  * <dd>(Required) The index to use for product querying.</dd>
  31.  * </dl>
  32.  */

  34. public class ReliableIndexerListener extends DefaultIndexerListener implements Runnable {

  36.   /** Logger object */
  37.   protected static final Logger LOGGER = Logger
  38.           .getLogger(ReliableIndexerListener.class.getName());

  40.   private static final int PRODUCTS_PER_QUERY = 1000;

  42.   private boolean stopThread = false;
  43.   private long lastIndexId = -1;
  44.   private final Object syncObject = new Object();

  46.   private Thread processThread;
  47.   /** Product Index */
  48.   protected ProductIndex productIndex;

  50.   /**
  51.    * Sets up an object on start
  52.    *
  53.    * @param config configuration
  54.    *
  55.    * @throws Exception if missing product index
  56.    */
  57.   public void configure(Config config) throws Exception {
  58.     super.configure(config);

  60.     //Getting indexer for queries
  61.     String indexName = config.getProperty(Indexer.INDEX_CONFIG_PROPERTY);
  62.     if (indexName != null) {
  63.       LOGGER.config("[" + getName() + "] loading ProductIndex '"
  64.         + indexName + "'");
  65.       productIndex = (ProductIndex) Config.getConfig().getObject(indexName);
  66.     }
  67.     if (productIndex == null) {
  68.       throw new ConfigurationException("[" + getName()
  69.           + "] ProductIndex is required");
  70.     }
  71.   }

  73.   /**
  74.    * Wakes thread when indexer makes changes
  75.    *
  76.    * @param delta Indexer Event - not used
  77.    *
  78.    * @throws Exception if something goes wrong
  79.    */
  80.   public void onIndexerEvent(IndexerEvent delta) throws Exception {
  81.     //Synchronized on the syncObject so we don't miss events
  82.     synchronized (syncObject) {
  83.       syncObject.notify();
  84.     }
  85.     LOGGER.log(Level.FINEST,"[" + getName() + "] done being notified");
  86.   }

  88.   /**
  89.    * Thread main body. Waits until notified, then tries to get the next products and process them.
  90.    */
  91.   @Override
  92.   public void run() {
  93.     //Run until we're told not to
  94.     while (!stopThread) {

  96.       List<ProductSummary> productList = null;

  98.       //Synchronized so we aren't notified of new products right before we wait
  99.       synchronized (syncObject) {
  100.         try {
  101.           productList = getNextProducts();
  102.         } catch (Exception e) {
  103.           try {
  104.             //Handle exception if we can
  105.             this.onProductGetException(e);
  106.           } catch (Exception e2) {
  107.               //Do nothing if we can't
  108.           }
  109.         }
  110.         if (productList == null || productList.size() == 0) {
  111.           try {
  112.             //Wait when there are no more products to process
  113.             syncObject.wait();
  114.           } catch (InterruptedException ignore) {
  115.             //Ignore because it's most likely we get interrupted by shutdown
  116.             LOGGER.log(Level.FINE,"[" + getName() + "] was told to stop, or something went wrong");
  117.           }
  118.           continue;
  119.         }
  120.       }

  122.       //Process the products we have
  123.       for(ProductSummary summary : productList) {
  124.         LOGGER.log(Level.FINEST,"[" + getName() + "] preparing to process product " + summary.getIndexId());
  125.         //Check for shutdown every iteration so we don't hog shutdown time
  126.         synchronized(syncObject) {
  127.           if (stopThread) {
  128.             break;
  129.           }
  130.         }
  131.         try {
  132.           //Process the product types we're told to in configuration
  133.           LOGGER.log(Level.FINEST,"[" + getName() + "] determining if we can process product " + summary.getIndexId());
  134.           if (accept(summary.getId())) {
  135.             LOGGER.log(Level.FINEST,"[" + getName() + "] attempting to process product " + summary.getIndexId());
  136.             this.processProduct(summary);
  137.           }
  138.           //Update internal storage so we don't reprocess products
  139.           this.setLastIndexId(summary.getIndexId());
  140.         } catch (Exception e) {
  141.           try {
  142.             //Handle exception if we can
  143.             this.onProcessException(summary,e);
  144.           } catch(Exception e2) {
  145.             //Give up if we can't
  146.             break;
  147.           }
  148.         }
  149.       }

  151.     }
  152.   }

  154.   /**
  155.    * Starts thread
  156.    *
  157.    * Calls onBeforeProcessThreadStart() in case subclasses want to add functionality
  158.    *
  159.    * @throws Exception if there's a thread issue
  160.    * @throws Exception if thread start fails
  161.    */
  162.   @Override
  163.   public void startup() throws Exception{
  164.     super.startup();
  165.     this.onBeforeProcessThreadStart();
  166.     synchronized(syncObject) {
  167.       stopThread = false;
  168.       this.processThread = new Thread(this);
  169.     }
  170.     this.processThread.start();
  171.   }

  173.   /**
  174.    * Closes thread
  175.    *
  176.    * @throws Exception if there's a thread issue
  177.    */
  178.   @Override
  179.   public void shutdown() throws Exception {
  180.     try {
  181.       LOGGER.log(Level.FINEST,"[" + getName() + "] trying to shut down...");
  182.       //When the thread is ready, tell it to stop
  183.       synchronized (syncObject) {
  184.         stopThread = true;
  185.         this.processThread.interrupt();
  186.       }
  187.       this.processThread.join();
  188.     } finally {
  189.       super.shutdown();
  190.     }
  191.   }

  193.   /** @return ProductIndex */
  194.   public ProductIndex getProductIndex() {
  195.     return this.productIndex;
  196.   }

  198.   /** @param productIndex to set */
  199.   public void setProductIndex(ProductIndex productIndex) {
  200.     this.productIndex = productIndex;
  201.   }


  204.   ////////////////////////
  205.   //Stubs for subclasses//
  206.   ////////////////////////

  208.   /**
  209.    * Gets index ID of last processed product
  210.    * @return lastIndexId
  211.    */
  212.   public long getLastIndexId() {
  213.     return lastIndexId;
  214.   }

  216.   /**
  217.    * Sets index ID of last processed product
  218.    * @param lastIndexId to set
  219.    */
  220.   public void setLastIndexId(final long lastIndexId) {
  221.     this.lastIndexId = lastIndexId;
  222.   }

  224.   /**
  225.    * Run before thread start.
  226.    *
  227.    * @throws Exception available for subclasses
  228.    */
  229.   protected void onBeforeProcessThreadStart() throws Exception {
  230.     //Do database call to update lastIndexId
  231.   }

  233.   /**
  234.    * Exception handling for product fetch
  235.    *
  236.    * @param e the caught exception
  237.    * @throws Exception in case we can't handle the first exception
  238.    */
  239.   protected void onProductGetException(Exception e) throws Exception {
  240.     LOGGER.log(Level.WARNING, "[" + getName() + "] Exception getting next products", e);
  241.   }

  243.   /**
  244.    * Exception handling for product processing.
  245.    *
  246.    * @param product the product that gave us the error
  247.    * @param e the caught exception
  248.    *
  249.    * @throws Exception in case we can't handle the first exception.
  250.    */
  251.   protected void onProcessException(ProductSummary product, Exception e) throws Exception {
  252.     LOGGER.log(Level.WARNING, "[" + getName() + "] Exception processing product " + product.getId(), e);
  253.   }

  255.   /**
  256.    * Gets the next products using the index provided in Config
  257.    *
  258.    * @return List of product summaries
  259.    * @throws Exception if we have a database issue
  260.    */
  261.   public List<ProductSummary> getNextProducts() throws Exception{
  262.     ProductIndexQuery query = new ProductIndexQuery();
  263.     query.setLimit(PRODUCTS_PER_QUERY);
  264.     query.setOrderBy(JDBCProductIndex.SUMMARY_PRODUCT_INDEX_ID); //Currently the only public field; should maybe change
  265.     query.setMinProductIndexId(this.getLastIndexId()+1);

  267.     return productIndex.getProducts(query);
  268.   }

  270.   /**
  271.    * Does a task with each product
  272.    *
  273.    * @param product ProductSummary  to process
  274.    * @throws Exception available for subclasses
  275.    */
  276.   public void processProduct(final ProductSummary product) throws Exception {
  277.     //Do stuff
  278.     LOGGER.log(Level.FINER,"[" + getName() + "] processing product " + product.getId());
  279.   }



  283. }
Created with JaCoCo 0.8.8.202204050719