DefaultNotificationListener.java

  1. /*
  2.  * DefaultNotificationListener
  3.  */
  4. package gov.usgs.earthquake.distribution;

  6. import gov.usgs.earthquake.product.AbstractListener;
  7. import gov.usgs.earthquake.product.Content;
  8. import gov.usgs.earthquake.product.Product;
  9. import gov.usgs.earthquake.product.ProductId;
  10. import gov.usgs.util.Config;
  11. import gov.usgs.util.ObjectLock;
  12. import gov.usgs.util.StringUtils;

  14. import java.util.ArrayList;
  15. import java.util.Iterator;
  16. import java.util.List;
  17. import java.util.Map;
  18. import java.util.Timer;
  19. import java.util.TimerTask;
  20. import java.util.logging.Level;
  21. import java.util.logging.Logger;

  23. /**
  24.  * A base implementation of a notification listener. Implements functionality
  25.  * that is useful for most notification listeners.
  26.  *
  27.  * Sub classes should override the onProduct(Product) method to add custom
  28.  * processing.
  29.  *
  30.  * The DefaultNotificationListener extends the AbstractListener and can use any
  31.  * of those configuration parameters.
  32.  *
  33.  * @see gov.usgs.earthquake.product.AbstractListener
  34.  */
  35. public class DefaultNotificationListener extends AbstractListener implements
  36.         NotificationListener, NotificationIndexCleanup.Listener {

  38.     /** Logging object. */
  39.     private static final Logger LOGGER = Logger
  40.             .getLogger(DefaultNotificationListener.class.getName());

  42.     /** Property referencing a notification index config section. */
  43.     public static final String NOTIFICATION_INDEX_PROPERTY = "listenerIndex";
  44.     /** Property for listener index file */
  45.     public static final String INDEX_FILE_PROPERTY = "listenerIndexFile";

  47.     /** How long to wait until checking for expired notifications/products. */
  48.     public static final String CLEANUP_INTERVAL_PROPERTY = "cleanupInterval";
  49.     /** Default time to wait for cleanup. 1h */
  50.     public static final String DEFAULT_CLEANUP_INTERVAL = "3600000";

  52.     /** Property for concurrentProducts */
  53.     public static final String CONCURRENT_PRODUCTS_PROPERTY = "concurrentProducts";
  54.     /** How many products to process at a time.  */
  55.     public static final String DEFAULT_CONCURRENT_PRODUCTS = "1";

  57.     /** Whether or not to process products more than once. */
  58.     public static final String PROCESS_DUPLICATES = "processDuplicates";
  59.     /** Default for process duplicates. False */
  60.     public static final String DEFAULT_PROCESS_DUPLICATES = "false";

  62.     /** Filter products based on content paths they contain. */
  63.     public static final String INCLUDE_PATHS_PROPERTY = "includePaths";
  64.     /** Property for exludePaths */
  65.     public static final String EXCLUDE_PATHS_PROPERTY = "excludePaths";

  67.     /** Optional notification index. */
  68.     private NotificationIndex notificationIndex = null;

  70.     /** How often to run cleanup task, in ms, <=0 = off. */
  71.     private Long cleanupInterval = 0L;

  73.     /** Timer that schedules sender cleanup task. */
  74.     private Timer cleanupTimer = null;

  76.     /** Notification index cleanup. */
  77.     private NotificationIndexCleanup notificationCleanup = null;

  79.     /** How many products to process at the same time. */
  80.     private int concurrentProducts = 1;

  82.     /** Whether or not to process products that have already been processed. */
  83.     private boolean processDuplicates = false;

  85.     /** Array of content paths to search. */
  86.     private final ArrayList<String> includePaths = new ArrayList<String>();

  88.     /** Array of content paths to search. */
  89.     private final ArrayList<String> excludePaths = new ArrayList<String>();

  91.     /**
  92.      * Locks used to keep concurrent listener from processing product
  93.      * more than once at the same time.
  94.      */
  95.     private ObjectLock<ProductId> storageLocks = new ObjectLock<ProductId>();

  97.     /**
  98.      * Implement the NotificationListener interface.
  99.      *
  100.      * This method calls accept, and if accept returns true, retrieves the
  101.      * product and calls onProduct.
  102.      */
  103.     public void onNotification(final NotificationEvent event) throws Exception {
  104.         Notification notification = event.getNotification();
  105.         ProductId id = notification.getProductId();
  106.         String productId = id.toString();

  108.         LOGGER.finest("[" + getName() + "] received notification for id="
  109.                 + productId);

  111.         if (!accept(id)) {
  112.             LOGGER.finest("[" + getName() + "] rejected notification for id="
  113.                     + productId);
  114.             return;
  115.         }

  117.         // only allow one thread to process a product
  118.         storageLocks.acquireLock(id);
  119.         try {
  120.             if (!onBeforeProcessNotification(notification)) {
  121.                 return;
  122.             }

  124.             LOGGER.finer("[" + getName() + "] processing notification for id="
  125.                     + productId);

  127.             Product product = event.getProduct();
  128.             if (product == null) {
  129.                 throw new ContinuableListenerException("retrieved product null,"
  130.                         + " notification id=" + productId);
  131.             }

  133.             if (!onBeforeProcessProduct(product)) {
  134.                 return;
  135.             }
  136.             LOGGER.finer("[" + getName() + "] processing product for id="
  137.                     + productId);
  138.             onProduct(product);

  140.             onAfterProcessNotification(notification);
  141.         } finally {
  142.             // be sure to release lock when done/error
  143.             storageLocks.releaseLock(id);
  144.         }
  145.     }

  147.     /**
  148.      * Called by onNotification when a product is retrieved.
  149.      *
  150.      * @param product
  151.      *            a product whose notification was accepted.
  152.      * @throws Exception if error occurs
  153.      */
  154.     public void onProduct(final Product product) throws Exception {
  155.         // subclasses do stuff here
  156.         ProductId id = product.getId();
  157.         StringBuffer b = new StringBuffer("[" + getName()
  158.                 + "] product processed source=" + id.getSource() + ", type="
  159.                 + id.getType() + ", code=" + id.getCode() + ", updateTime="
  160.                 + id.getUpdateTime().toString());

  162.         Map<String, String> properties = product.getProperties();
  163.         Iterator<String> iter = properties.keySet().iterator();
  164.         while (iter.hasNext()) {
  165.             String name = iter.next();
  166.             b.append(", ").append(name).append("=")
  167.                     .append(properties.get(name));
  168.         }

  170.         LOGGER.info(b.toString());
  171.         System.out.println(b.toString());
  172.     }

  174.     /**
  175.      * Called just before this listener processes a notification.
  176.      *
  177.      * @param notification
  178.      *            notification about to be processed.
  179.      * @return true to process the notification, false to skip
  180.      * @throws Exception if error occurs
  181.      */
  182.     protected boolean onBeforeProcessNotification(
  183.             final Notification notification) throws Exception {
  184.         // only check if we care
  185.         if (!processDuplicates && this.notificationIndex != null) {
  186.             List<Notification> notifications = this.notificationIndex
  187.                     .findNotifications(notification.getProductId());
  188.             if (notifications.size() > 0) {
  189.                 LOGGER.finer("[" + getName()
  190.                         + "] skipping existing product "
  191.                         + notification.getProductId().toString());
  192.                 return false;
  193.             }
  194.         }

  196.         return true;
  197.     }

  199.     /**
  200.      * Called after a product has been downloaded, but before onProduct is
  201.      * called.
  202.      *
  203.      * Sometimes a listener cannot tell whether it should process a product
  204.      * until its contents are available. This is where the "includePaths" and
  205.      * "excludePaths" are evaluated.
  206.      *
  207.      * @param product
  208.      *            product about to be processed.
  209.      * @return true to process the product, false to skip
  210.      * @throws Exception if error occurs
  211.      */
  212.     protected boolean onBeforeProcessProduct(final Product product)
  213.             throws Exception {
  214.         if (excludePaths.size() != 0) {
  215.             Map<String, Content> contents = product.getContents();
  216.             Iterator<String> pathIter = excludePaths.iterator();
  217.             while (pathIter.hasNext()) {
  218.                 String path = pathIter.next();
  219.                 if (contents.containsKey(path)) {
  220.                     // contains at least one matching include path
  221.                     LOGGER.fine("[" + getName() + "] skipping product "
  222.                             + product.getId().toString() + ", '" + path
  223.                             + "' matches excludePaths");
  224.                     return false;
  225.                 }
  226.             }
  227.         }

  229.         if (includePaths.size() != 0) {
  230.             boolean containsPath = false;
  231.             Map<String, Content> contents = product.getContents();
  232.             Iterator<String> pathIter = includePaths.iterator();
  233.             while (pathIter.hasNext()) {
  234.                 String path = pathIter.next();
  235.                 if (contents.containsKey(path)) {
  236.                     // contains at least one matching include path
  237.                     containsPath = true;
  238.                     break;
  239.                 }
  240.             }
  241.             if (!containsPath) {
  242.                 LOGGER.fine("[" + getName() + "] skipping product "
  243.                         + product.getId().toString()
  244.                         + ", does not match includePaths");
  245.                 return false;
  246.             }
  247.         }

  249.         return true;
  250.     }

  252.     /**
  253.      * Called when this listener has successfully processed a notification.
  254.      *
  255.      * @param notification
  256.      *            notification that was processed.
  257.      * @throws Exception if error occurs
  258.      */
  259.     protected void onAfterProcessNotification(final Notification notification)
  260.             throws Exception {
  261.         if (this.notificationIndex != null) {
  262.             this.notificationIndex.addNotification(notification);
  263.         }
  264.     }

  266.     /**
  267.      * Called when an expired notification is being removed from the index.
  268.      *
  269.      * @param notification to be removed
  270.      * @throws Exception if error occurs
  271.      */
  272.     @Override
  273.     public void onExpiredNotification(final Notification notification)
  274.             throws Exception {
  275.         // nothing to do
  276.     }

  278.         /**
  279.      * Periodic cleanup task.
  280.      *
  281.      * Called every cleanupInterval milliseconds.
  282.      */
  283.     public void cleanup() throws Exception {
  284.         LOGGER.finer("[" + getName() + "] running listener cleanup");
  285.         if (this.notificationCleanup == null) {
  286.             this.notificationCleanup = new NotificationIndexCleanup(this.notificationIndex, this);
  287.             this.notificationCleanup.startup();
  288.         } else {
  289.             this.notificationCleanup.wakeUp();
  290.         }
  291.     }

  293.     @Override
  294.     public void startup() throws Exception {
  295.         super.startup();
  296.         if (this.notificationIndex != null) {
  297.             this.notificationIndex.startup();
  298.         }

  300.         // only schedule cleanup if interval is non-zero
  301.         if (cleanupInterval > 0) {
  302.             cleanupTimer = new Timer();
  303.             cleanupTimer.scheduleAtFixedRate(new TimerTask() {
  304.                 public void run() {
  305.                     try {
  306.                         cleanup();
  307.                     } catch (Exception e) {
  308.                         LOGGER.log(Level.WARNING, "[" + getName()
  309.                                 + "] exception during cleanup", e);
  310.                     }
  311.                 }
  312.             }, 0, cleanupInterval);
  313.         }
  314.     }

  316.     @Override
  317.     public void shutdown() throws Exception {
  318.         super.shutdown();
  319.         if (this.notificationCleanup != null) {
  320.             try {
  321.                 this.notificationCleanup.shutdown();
  322.             } catch (Exception e) {
  323.                 LOGGER.log(Level.INFO, "[" + getName() + "] exception stopping notification cleanup", e);
  324.             } finally {
  325.                 this.notificationCleanup = null;
  326.             }
  327.         }
  328.         try {
  329.             this.notificationIndex.shutdown();
  330.         } catch (Exception e) {
  331.             // ignore
  332.         }
  333.         try {
  334.             this.cleanupTimer.cancel();
  335.         } catch (Exception e) {
  336.             // ignore
  337.         }
  338.     }

  340.     @Override
  341.     public void configure(final Config config) throws Exception {
  342.         super.configure(config);

  344.         String notificationIndexName = config
  345.                 .getProperty(NOTIFICATION_INDEX_PROPERTY);
  346.         String notificationIndexFile = config.getProperty(INDEX_FILE_PROPERTY);
  347.         if (notificationIndexName != null) {
  348.             LOGGER.config("[" + getName() + "] loading notification index '"
  349.                     + notificationIndexName + "'");
  350.             notificationIndex = (NotificationIndex) Config.getConfig()
  351.                     .getObject(notificationIndexName);
  352.             if (notificationIndex == null) {
  353.                 throw new ConfigurationException("[" + getName() + "] index '"
  354.                         + notificationIndexName
  355.                         + "' is not properly configured");
  356.             }
  357.         } else if (notificationIndexFile != null) {
  358.             LOGGER.config("[" + getName() + "] using notification index '"
  359.                     + notificationIndexFile + "'");
  360.             notificationIndex = new JDBCNotificationIndex(notificationIndexFile);
  361.         }

  363.         cleanupInterval = Long.parseLong(config.getProperty(
  364.                 CLEANUP_INTERVAL_PROPERTY, DEFAULT_CLEANUP_INTERVAL));
  365.         LOGGER.config("[" + getName() + "] cleanup interval = "
  366.                 + processDuplicates);

  368.         concurrentProducts = Integer.parseInt(config.getProperty(
  369.             CONCURRENT_PRODUCTS_PROPERTY, DEFAULT_CONCURRENT_PRODUCTS));
  370.         LOGGER.config("[" + getName() + "] concurrent products = "
  371.                 + concurrentProducts);

  373.         processDuplicates = Boolean.valueOf(config.getProperty(
  374.                 PROCESS_DUPLICATES, DEFAULT_PROCESS_DUPLICATES));
  375.         LOGGER.config("[" + getName() + "] process duplicates = "
  376.                 + processDuplicates);

  378.         includePaths.addAll(StringUtils.split(
  379.                 config.getProperty(INCLUDE_PATHS_PROPERTY), ","));
  380.         LOGGER.config("[" + getName() + "] include paths = " + includePaths);

  382.         excludePaths.addAll(StringUtils.split(
  383.                 config.getProperty(EXCLUDE_PATHS_PROPERTY), ","));
  384.         LOGGER.config("[" + getName() + "] exclude paths = " + excludePaths);
  385.     }

  387.     /** @return notificationIndex */
  388.     public NotificationIndex getNotificationIndex() {
  389.         return notificationIndex;
  390.     }

  392.     /** @param notificationIndex to set */
  393.     public void setNotificationIndex(NotificationIndex notificationIndex) {
  394.         this.notificationIndex = notificationIndex;
  395.     }

  397.     /** @return cleanupInterval */
  398.     public Long getCleanupInterval() {
  399.         return cleanupInterval;
  400.     }

  402.     /** @param cleanupInterval long to set */
  403.     public void setCleanupInterval(Long cleanupInterval) {
  404.         this.cleanupInterval = cleanupInterval;
  405.     }

  407.     /** @return concurrentProducts */
  408.     public int getConcurrentProducts() {
  409.         return concurrentProducts;
  410.     }

  412.     /** @param concurrentProducts int to set */
  413.     public void setConcurrentProducts(int concurrentProducts) {
  414.         this.concurrentProducts = concurrentProducts;
  415.     }

  417.     /** @return processDuplicates */
  418.     public boolean isProcessDuplicates() {
  419.         return processDuplicates;
  420.     }

  422.     /** @param processDuplicates boolean to set */
  423.     public void setProcessDuplicates(boolean processDuplicates) {
  424.         this.processDuplicates = processDuplicates;
  425.     }

  427.     /**
  428.      * @return the includePaths
  429.      */
  430.     public ArrayList<String> getIncludePaths() {
  431.         return includePaths;
  432.     }

  434.     /**
  435.      * @return the excludePaths
  436.      */
  437.     public ArrayList<String> getExcludePaths() {
  438.         return excludePaths;
  439.     }

  441. }
Created with JaCoCo 0.8.8.202204050719