JsonNotificationIndex.java

  1. package gov.usgs.earthquake.aws;

  3. import java.io.ByteArrayInputStream;
  4. import java.net.URL;
  5. import java.sql.PreparedStatement;
  6. import java.sql.ResultSet;
  7. import java.sql.SQLException;
  8. import java.sql.Statement;
  9. import java.sql.Types;
  10. import java.time.Instant;
  11. import java.util.ArrayList;
  12. import java.util.Collections;
  13. import java.util.Date;
  14. import java.util.List;
  15. import java.util.logging.Logger;
  16. import java.util.logging.Level;

  18. import javax.json.Json;


  21. import gov.usgs.earthquake.distribution.DefaultNotification;
  22. import gov.usgs.earthquake.distribution.Notification;
  23. import gov.usgs.earthquake.distribution.NotificationIndex;
  24. import gov.usgs.earthquake.distribution.URLNotification;
  25. import gov.usgs.earthquake.product.Product;
  26. import gov.usgs.earthquake.product.ProductId;
  27. import gov.usgs.earthquake.product.io.JsonProduct;
  28. import gov.usgs.earthquake.util.JDBCConnection;
  29. import gov.usgs.util.Config;
  30. import gov.usgs.util.StringUtils;

  32. /**
  33.  * Store Notifications in a database.
  34.  *
  35.  * Only SQLITE or local development should rely on createSchema.
  36.  * Products (data column) have exceeded 64kb, plan accordingly.
  37.  *
  38.  * Mysql Schema Example:<br>
  39.  * <pre>
  40.  * CREATE TABLE IF NOT EXISTS indexer_receiver_index
  41.  * (id INTEGER PRIMARY KEY AUTO_INCREMENT
  42.  * , created VARCHAR(255)
  43.  * , expires VARCHAR(255)
  44.  * , source VARCHAR(255)
  45.  * , type VARCHAR(255)
  46.  * , code VARCHAR(255)
  47.  * , updatetime BIGINT
  48.  * , url TEXT
  49.  * , data LONGTEXT
  50.  * , KEY source_index (source)
  51.  * , KEY type_index (type)
  52.  * , KEY code_index (code)
  53.  * , KEY expires_index (expires)
  54.  * ) ENGINE=innodb CHARSET=utf8;
  55.  * </pre>
  56.  */
  57. public class JsonNotificationIndex
  58.     extends JDBCConnection
  59.     implements NotificationIndex {

  61.   private static final Logger LOGGER = Logger.getLogger(
  62.       JsonNotificationIndex.class.getName());

  64.   /** Variable for the default driver */
  65.   public static final String DEFAULT_DRIVER = "org.sqlite.JDBC";
  66.   /** Variable for the default table */
  67.   public static final String DEFAULT_TABLE = "notification";
  68.   /** Variable for the default URL */
  69.   public static final String DEFAULT_URL =
  70.       "jdbc:sqlite:json_notification_index.db";

  72.   /** Database table name. */
  73.   private String table;

  75.   /**
  76.    * Construct a JsonNotification using defaults.
  77.    */
  78.   public JsonNotificationIndex() {
  79.     this(DEFAULT_DRIVER, DEFAULT_URL);
  80.   }

  82.   /**
  83.    * Construct a JsonNotificationIndex with the default table.
  84.    * @param driver Driver to use
  85.    * @param url URL to use
  86.    */
  87.   public JsonNotificationIndex(final String driver, final String url) {
  88.     this(driver, url, DEFAULT_TABLE);
  89.   }

  91.   /**
  92.    * Construct a JsonNotificationIndex with custom driver, url, and table.
  93.    * @param driver Driver to use
  94.    * @param url URL to use
  95.    * @param table Table to use
  96.    */
  97.   public JsonNotificationIndex(
  98.       final String driver, final String url, final String table) {
  99.     super(driver, url);
  100.     this.table = table;
  101.   }

  103.   /** @return table */
  104.   public String getTable() { return this.table; }
  105.   /** @param table Table to set */
  106.   public void setTable(final String table) { this.table = table; }

  108.   @Override
  109.   public void configure(final Config config) throws Exception {
  110.     super.configure(config);
  111.     if (getDriver() == null) { setDriver(DEFAULT_DRIVER); }
  112.     if (getUrl() == null) { setUrl(DEFAULT_URL); }

  114.     setTable(config.getProperty("table", DEFAULT_TABLE));
  115.     LOGGER.config("[" + getName() + "] driver=" + getDriver());
  116.     LOGGER.config("[" + getName() + "] table=" + getTable());
  117.     // do not log url, it may contain user/pass
  118.   }

  120.   /**
  121.    * After normal startup, check whether schema exists and attempt to create.
  122.    * @throws Exception if error occurs
  123.    */
  124.   @Override
  125.   public void startup() throws Exception {
  126.     super.startup();
  127.     // make sure schema exists
  128.     if (!schemaExists()) {
  129.       LOGGER.warning("[" + getName() + "] schema not found, creating");
  130.       createSchema();
  131.     }
  132.   }

  134.   /**
  135.    * Check whether schema exists.
  136.    *
  137.    * @return boolean
  138.    * @throws Exception if error occurs
  139.    */
  140.   public boolean schemaExists() throws Exception {
  141.     final String sql = "select * from " + this.table + " limit 1";
  142.     beginTransaction();
  143.     try (final PreparedStatement test = getConnection().prepareStatement(sql)) {
  144.       // should throw exception if table does not exist
  145.       test.setQueryTimeout(60);
  146.       try (final ResultSet rs = test.executeQuery()) {
  147.         rs.next();
  148.       }
  149.       commitTransaction();
  150.       // schema exists
  151.       return true;
  152.     } catch (Exception e) {
  153.       rollbackTransaction();
  154.       return false;
  155.     }
  156.   }

  158.   /**
  159.    * Attempt to create schema.
  160.    *
  161.    * Only supports sqlite or mysql.  When not using sqlite, relying on this
  162.    * method is only recommended for local development.
  163.    *
  164.    * @throws Exception if error occurs
  165.    */
  166.   public void createSchema() throws Exception {
  167.     // create schema
  168.     beginTransaction();
  169.     try (final Statement statement = getConnection().createStatement()) {
  170.       String autoIncrement = "";
  171.       String engine = "";
  172.       if (getDriver().contains("mysql")) {
  173.         autoIncrement = " AUTO_INCREMENT";
  174.         engine = " ENGINE=innodb CHARSET=utf8";
  175.       }
  176.       statement.executeUpdate(
  177.           "CREATE TABLE " + this.table
  178.           + " (id INTEGER PRIMARY KEY" + autoIncrement
  179.           + ", created VARCHAR(255)"
  180.           + ", expires VARCHAR(255)"
  181.           + ", source VARCHAR(255)"
  182.           + ", type VARCHAR(255)"
  183.           + ", code VARCHAR(255)"
  184.           + ", updatetime BIGINT"
  185.           + ", url TEXT"
  186.           + ", data TEXT"
  187.           + ")" + engine);
  188.       statement.executeUpdate(
  189.           "CREATE INDEX source_index ON " + this.table + " (source)");
  190.       statement.executeUpdate(
  191.           "CREATE INDEX type_index ON " + this.table + " (type)");
  192.       statement.executeUpdate(
  193.           "CREATE INDEX code_index ON " + this.table + " (code)");
  194.       statement.executeUpdate(
  195.           "CREATE INDEX expires_index ON " + this.table + " (expires)");
  196.       commitTransaction();
  197.     } catch (Exception e) {
  198.       rollbackTransaction();
  199.       throw e;
  200.     }
  201.   }

  203.   /**
  204.    * Add a notification to the index.
  205.    *
  206.    * TrackerURLs are ignored.
  207.    * @param notification To be added to index
  208.    * @throws Exception if error occurs
  209.    */
  210.   @Override
  211.   public synchronized void addNotification(Notification notification)
  212.       throws Exception {
  213.     // all notifications
  214.     Instant expires = notification.getExpirationDate().toInstant();
  215.     ProductId id = notification.getProductId();
  216.     // json only
  217.     Instant created = null;
  218.     Product product = null;
  219.     // url only
  220.     URL url = null;
  221.     if (notification instanceof JsonNotification) {
  222.       JsonNotification jsonNotification = (JsonNotification) notification;
  223.       created = jsonNotification.created;
  224.       product = jsonNotification.product;
  225.     } else if (notification instanceof URLNotification) {
  226.       url = ((URLNotification) notification).getProductURL();
  227.     }
  228.     // prepare statement
  229.     beginTransaction();
  230.     try (
  231.       final PreparedStatement statement = getConnection().prepareStatement(
  232.           "INSERT INTO " + this.table
  233.           + " (created, expires, source, type, code, updatetime, url, data)"
  234.           + " VALUES (?, ?, ?, ?, ?, ?, ?, ?)")
  235.     ) {
  236.       try {
  237.         statement.setQueryTimeout(60);
  238.         // set parameters
  239.         statement.setString(1, created != null ? created.toString() : "");
  240.         statement.setString(2, expires.toString());
  241.         statement.setString(3, id.getSource());
  242.         statement.setString(4, id.getType());
  243.         statement.setString(5, id.getCode());
  244.         statement.setLong(6, id.getUpdateTime().getTime());
  245.         statement.setString(7, url != null ? url.toString() : "");
  246.         if (product == null) {
  247.           statement.setNull(8, Types.VARCHAR);
  248.         } else {
  249.           statement.setString(8,
  250.               new JsonProduct().getJsonObject(product).toString());
  251.         }
  252.         // execute
  253.         statement.executeUpdate();
  254.         commitTransaction();
  255.       } catch (SQLException e) {
  256.         LOGGER.log(Level.WARNING, "Exception adding notification", e);
  257.         try {
  258.           // otherwise roll back
  259.           rollbackTransaction();
  260.         } catch (SQLException e2) {
  261.           // ignore
  262.         }
  263.       }
  264.     }
  265.   }

  267.   /**
  268.    * Remove notification from index.
  269.    *
  270.    * Tracker URLs are ignored.
  271.    * @param notification to be removed from index
  272.    * @throws Exception if error occurs
  273.    */
  274.   @Override
  275.   public synchronized void removeNotification(Notification notification) throws Exception {
  276.     final List<Notification> notifications = new ArrayList<>();
  277.     notifications.add(notification);
  278.     this.removeNotifications(notifications);
  279.   }

  281.   /**
  282.    * Remove notifications from index.
  283.    *
  284.    * Tracker URLs are ignored.
  285.    * @param notifications
  286.    *     notifications to be removed from index
  287.    * @throws Exception if error occurs
  288.    */
  289.   @Override
  290.   public synchronized void removeNotifications(List<Notification> notifications) throws Exception {
  291.     // prepare statement
  292.     final String sql = "DELETE FROM " + this.table
  293.           + " WHERE created=? AND expires=? AND source=? AND type=? AND code=?"
  294.           + " AND updatetime=? AND url=?";
  295.     beginTransaction();
  296.     try (final PreparedStatement statement = getConnection().prepareStatement(sql)) {
  297.       try {
  298.         statement.setQueryTimeout(60);

  300.         for (Notification notification : notifications) {
  301.           // all notifications
  302.           Instant expires = notification.getExpirationDate().toInstant();
  303.           ProductId id = notification.getProductId();
  304.           // json only
  305.           Instant created = null;
  306.           // url only
  307.           URL url = null;
  308.           if (notification instanceof JsonNotification) {
  309.             JsonNotification jsonNotification = (JsonNotification) notification;
  310.             created = jsonNotification.created;
  311.           } else if (notification instanceof URLNotification) {
  312.             url = ((URLNotification) notification).getProductURL();
  313.           }

  315.           // set parameters
  316.           statement.setString(1, created != null ? created.toString() : "");
  317.           statement.setString(2, expires.toString());
  318.           statement.setString(3, id.getSource());
  319.           statement.setString(4, id.getType());
  320.           statement.setString(5, id.getCode());
  321.           statement.setLong(6, id.getUpdateTime().getTime());
  322.           statement.setString(7, url != null ? url.toString() : "");
  323.           statement.addBatch();
  324.         }
  325.         // execute
  326.         statement.executeBatch();
  327.         commitTransaction();
  328.       } catch (SQLException e) {
  329.         LOGGER.log(Level.WARNING, "Exception removing notification", e);
  330.         try {
  331.           // otherwise roll back
  332.           rollbackTransaction();
  333.         } catch (SQLException e2) {
  334.           // ignore
  335.         }
  336.       }
  337.     }
  338.   }

  340.   /**
  341.    * Search index for notifications.
  342.    *
  343.    * @param source
  344.    *     source, or null for all sources.
  345.    * @param type
  346.    *     type, or null for all types.
  347.    * @param code
  348.    *     code, or null for all codes.
  349.    * @return list with matching notifications, empty if not found.
  350.    * @throws Exception if error occurs
  351.    */
  352.   @Override
  353.   public synchronized List<Notification> findNotifications(
  354.       String source, String type, String code) throws Exception {
  355.     final ArrayList<Object> where = new ArrayList<Object>();
  356.     final ArrayList<String> values = new ArrayList<String>();
  357.     if (source != null) {
  358.       where.add("source=?");
  359.       values.add(source);
  360.     }
  361.     if (type != null) {
  362.       where.add("type=?");
  363.       values.add(type);
  364.     }
  365.     if (code != null) {
  366.       where.add("code=?");
  367.       values.add(code);
  368.     }
  369.     String sql = "SELECT * FROM " + this.table;
  370.     if (where.size() > 0) {
  371.       sql += " WHERE " + StringUtils.join(where, " AND ");
  372.     }
  373.     // prepare statement
  374.     beginTransaction();
  375.     try (final PreparedStatement statement = getConnection().prepareStatement(sql)) {
  376.       try {
  377.         statement.setQueryTimeout(1800);

  379.         // set parameters
  380.         for (int i = 0, len=values.size(); i < len; i++) {
  381.           statement.setString(i+1, values.get(i));
  382.         }

  384.         // execute
  385.         final List<Notification> notifications = getNotifications(statement);
  386.         commitTransaction();
  387.         return notifications;
  388.       } catch (SQLException e) {
  389.         LOGGER.log(Level.WARNING, "Exception finding notifications", e);
  390.         try {
  391.           // otherwise roll back
  392.           rollbackTransaction();
  393.         } catch (SQLException e2) {
  394.           // ignore
  395.         }
  396.       }
  397.     }
  398.     return new ArrayList<Notification>();
  399.   }

  401.   /**
  402.    * Search index for notifications.
  403.    *
  404.    * @param sources
  405.    *     sources, or null for all sources.
  406.    * @param types
  407.    *     types, or null for all types.
  408.    * @param codes
  409.    *     codes, or null for all codes.
  410.    * @return list with matching notifications, empty if not found.
  411.    * @throws Exception if error occurs
  412.    */
  413.   @Override
  414.   public synchronized List<Notification> findNotifications(
  415.       List<String> sources, List<String> types, List<String> codes)
  416.       throws Exception {
  417.     final ArrayList<Object> where = new ArrayList<Object>();
  418.     final ArrayList<String> values = new ArrayList<String>();
  419.     if (sources != null && sources.size() > 0) {
  420.       where.add("source IN (" +
  421.           StringUtils.join(
  422.               Collections.nCopies(sources.size(), (Object)"?"),
  423.               ",")
  424.           + ")");
  425.       values.addAll(sources);
  426.     }
  427.     if (types != null && types.size() > 0) {
  428.       where.add("type IN (" +
  429.           StringUtils.join(
  430.               Collections.nCopies(types.size(), (Object)"?"),
  431.               ",")
  432.           + ")");
  433.       values.addAll(types);
  434.     }
  435.     if (codes != null && codes.size() > 0) {
  436.       where.add("code IN (" +
  437.           StringUtils.join(
  438.               Collections.nCopies(codes.size(), (Object)"?"),
  439.               ",")
  440.           + ")");
  441.       values.addAll(codes);
  442.     }
  443.     String sql = "SELECT * FROM " + this.table;
  444.     if (where.size() > 0) {
  445.       sql += " WHERE " + StringUtils.join(where, " AND ");
  446.     } else {
  447.       // searching for all notifications

  449.       // this is typically done to requeue a notification index.
  450.       // run query in a way that returns list of default notifications,
  451.       // (by returning empty created, data, and url)
  452.       // since full details are not needed during requeue
  453.       sql = "SELECT DISTINCT"
  454.           + " '' as created, expires, source, type, code, updateTime"
  455.           + ", '' as url, null as data"
  456.           + " FROM " + this.table
  457.           + " WHERE expires > ?";
  458.       values.add(Instant.now().toString());
  459.     }
  460.     // prepare statement
  461.     beginTransaction();
  462.     try (final PreparedStatement statement = getConnection().prepareStatement(sql)) {
  463.       try {
  464.         statement.setQueryTimeout(1800);

  466.         // set parameters
  467.         for (int i = 0, len=values.size(); i < len; i++) {
  468.           statement.setString(i+1, values.get(i));
  469.         }

  471.         // execute
  472.         final List<Notification> notifications = getNotifications(statement);
  473.         commitTransaction();
  474.         return notifications;
  475.       } catch (SQLException e) {
  476.         LOGGER.log(Level.WARNING, "Exception finding notifications", e);
  477.         try {
  478.           // otherwise roll back
  479.           rollbackTransaction();
  480.         } catch (SQLException e2) {
  481.           // ignore
  482.         }
  483.       }
  484.     }
  485.     return new ArrayList<Notification>();
  486.   }

  488.   /**
  489.    * Find notifications with expires time before or equal to current time.
  490.    *
  491.    * @return list with matching notifications, empty if not found.
  492.    * @throws Exception if error occurs
  493.    */
  494.   @Override
  495.   public synchronized List<Notification> findExpiredNotifications() throws Exception {
  496.     final String sql = "SELECT * FROM " + this.table + " WHERE expires <= ? LIMIT 5000";
  497.     // prepare statement
  498.     beginTransaction();
  499.     try (final PreparedStatement statement = getConnection().prepareStatement(sql)) {
  500.       try {
  501.         statement.setQueryTimeout(1800);

  503.         // set parameters
  504.         statement.setString(1, Instant.now().toString());

  506.         // execute
  507.         final List<Notification> notifications = getNotifications(statement);
  508.         commitTransaction();
  509.         return notifications;
  510.       } catch (SQLException e) {
  511.         LOGGER.log(Level.WARNING, "Exception finding notifications", e);
  512.         try {
  513.           // otherwise roll back
  514.           rollbackTransaction();
  515.         } catch (SQLException e2) {
  516.           // ignore
  517.         }
  518.       }
  519.     }
  520.     return new ArrayList<Notification>();

  522.   }

  524.   /**
  525.    * Search index for notifications for a specific product.
  526.    *
  527.    * @param id
  528.    *     the product id to search.
  529.    * @return list with matching notifications, empty if not found.
  530.    * @throws Exception if error occurs
  531.    */
  532.   @Override
  533.   public synchronized List<Notification> findNotifications(ProductId id) throws Exception {
  534.     final String sql = "SELECT * FROM " + this.table
  535.         + " WHERE source=? AND type=? AND code=? AND updatetime=?";
  536.     // prepare statement
  537.     beginTransaction();
  538.     try (final PreparedStatement statement = getConnection().prepareStatement(sql)) {
  539.       try {
  540.         statement.setQueryTimeout(30);
  541.         // set parameters
  542.         statement.setString(1, id.getSource());
  543.         statement.setString(2, id.getType());
  544.         statement.setString(3, id.getCode());
  545.         statement.setLong(4, id.getUpdateTime().getTime());

  547.         // executes and commit ifsuccessful
  548.         final List<Notification> notifications = getNotifications(statement);
  549.         commitTransaction();
  550.         return notifications;
  551.       } catch (SQLException e) {
  552.         LOGGER.log(Level.WARNING, "Exception finding notifications", e);
  553.         try {
  554.           // otherwise roll back
  555.           rollbackTransaction();
  556.         } catch (SQLException e2) {
  557.           // ignore
  558.         }
  559.       }
  560.     }
  561.     return new ArrayList<Notification>();
  562.   }

  564.   /**
  565.    * This method is used to find notifications present in this index
  566.    * but not present in another JsonNotificationIndex table in the same
  567.    * database.
  568.    *
  569.    * This is used to optimize the queuing process at startup and returns
  570.    * DefaultNotifications.  The receiver process will look up the actual
  571.    * notification object during processing.
  572.    *
  573.    * @param otherTable
  574.    *     name of table in same database.
  575.    * @return
  576.    *     list of notifications found in this indexes table, but not found in the
  577.    *     other table.
  578.    * @throws Exception if error occurs
  579.    */
  580.   public synchronized List<Notification> getMissingNotifications(
  581.       final String otherTable) throws Exception {
  582.     // this is used to requeue a notification index.
  583.     // run query in a way that returns list of default notifications,
  584.     // (by returning empty created, data, and url)
  585.     // since full details are not needed during requeue
  586.     final String sql = "SELECT DISTINCT"
  587.         + " '' as created, t.expires, t.source, t.type, t.code, t.updateTime"
  588.         + ", '' as url, null as data"
  589.         + " FROM " + this.table + " t"
  590.         // only missing if not expired
  591.         + " WHERE t.expires > ?"
  592.         + " AND NOT EXISTS ("
  593.           + "SELECT * FROM " + otherTable
  594.             + " WHERE source=t.source AND type=t.type"
  595.             + " AND code=t.code AND updatetime=t.updateTime"
  596.         + ")";
  597.     // prepare statement
  598.     beginTransaction();
  599.     try (final PreparedStatement statement = getConnection().prepareStatement(sql)) {
  600.       try {
  601.         statement.setQueryTimeout(1800);

  603.         // set parameters
  604.         statement.setString(1, Instant.now().toString());

  606.         // execute and commit if successful
  607.         final List<Notification> notifications = getNotifications(statement);
  608.         commitTransaction();
  609.         return notifications;
  610.       } catch (SQLException e) {
  611.         LOGGER.log(Level.WARNING, "Exception finding notifications", e);
  612.         try {
  613.           // otherwise roll back
  614.           rollbackTransaction();
  615.         } catch (SQLException e2) {
  616.           // ignore
  617.         }
  618.       }
  619.     }
  620.     return new ArrayList<Notification>();
  621.   }

  623.   /**
  624.    * Parse notifications from a statement ready to be executed.
  625.    * @param ps PreparedStatement to be parsed
  626.    * @return List of notifications
  627.    * @throws Exception if error occurs
  628.    */
  629.   protected synchronized List<Notification> getNotifications(PreparedStatement ps)
  630.       throws Exception {
  631.     final List<Notification> n = new ArrayList<Notification>();
  632.     try (final ResultSet rs = ps.executeQuery()) {
  633.       while (rs.next()) {
  634.         n.add(parseNotification(
  635.             rs.getString("created"),
  636.             rs.getString("expires"),
  637.             rs.getString("source"),
  638.             rs.getString("type"),
  639.             rs.getString("code"),
  640.             rs.getLong("updatetime"),
  641.             rs.getString("url"),
  642.             rs.getString("data")));
  643.       }
  644.     }
  645.     return n;
  646.   }

  648.   /**
  649.    * Creates and returns a <code>Notification</code> based on the provided data.
  650.    *
  651.    * <ul>
  652.    * <li>Return a JSONNotification if <code>created</code> and <code>data</code>
  653.    * are set
  654.    * <li>Return a URLNotification if <code>url</code> is set
  655.    * <li>Otherwise, return a DefaultNotification
  656.    * </ul>
  657.    * @param created When created
  658.    * @param expires When notification expires
  659.    * @param source sources
  660.    * @param type types
  661.    * @param code codes
  662.    * @param updateTime updateTime
  663.    * @param url URL
  664.    * @param data data
  665.    * @return Notification, JSONNotification, URLNotification, or DefaultNotification
  666.    * @throws Exception if error occurs
  667.    */
  668.   protected Notification parseNotification(
  669.       final String created,
  670.       final String expires,
  671.       final String source,
  672.       final String type,
  673.       final String code,
  674.       final Long updateTime,
  675.       final String url,
  676.       final String data) throws Exception {
  677.     final Notification n;
  678.     final ProductId id = new ProductId(source, type, code, new Date(updateTime));
  679.     final Date expiresDate = Date.from(Instant.parse(expires));
  680.     if (!"".equals(created) && data != null) {
  681.       Product product = new JsonProduct().getProduct(
  682.           Json.createReader(
  683.             new ByteArrayInputStream(data.getBytes())
  684.           ).readObject());
  685.       n = new JsonNotification(Instant.parse(created), product);
  686.     } else if (!"".equals(url)) {
  687.       n = new URLNotification(id, expiresDate, null, new URL(url));
  688.     } else {
  689.       n = new DefaultNotification(id, expiresDate, null);
  690.     }
  691.     return n;
  692.   }

  694. }
Created with JaCoCo 0.8.8.202204050719