NATSStreamingNotificationReceiver.java

  1. package gov.usgs.earthquake.nats;

  3. import gov.usgs.earthquake.distribution.ConfigurationException;
  4. import gov.usgs.earthquake.distribution.DefaultNotificationReceiver;
  5. import gov.usgs.earthquake.distribution.URLNotification;
  6. import gov.usgs.earthquake.distribution.URLNotificationJSONConverter;
  7. import gov.usgs.util.Config;
  8. import gov.usgs.util.FileUtils;
  9. import io.nats.streaming.Message;
  10. import io.nats.streaming.MessageHandler;
  11. import io.nats.streaming.Subscription;
  12. import io.nats.streaming.SubscriptionOptions;

  14. import javax.json.Json;
  15. import javax.json.JsonObject;
  16. import javax.json.JsonReader;
  17. import java.io.ByteArrayInputStream;
  18. import java.io.File;
  19. import java.io.InputStream;
  20. import java.io.IOException;
  21. import java.util.concurrent.TimeoutException;
  22. import java.util.logging.Level;
  23. import java.util.logging.Logger;

  25. /**
  26.  * Connects directly to a NATS streaming server to receive notifications using a NATSClient
  27.  */
  28. public class NATSStreamingNotificationReceiver extends DefaultNotificationReceiver implements MessageHandler {

  30.   private static final Logger LOGGER = Logger
  31.           .getLogger(DefaultNotificationReceiver.class.getName());

  33.   /** Property for tracking file name */
  34.   public static String TRACKING_FILE_NAME_PROPERTY = "trackingFile";
  35.   /** Property on if update sequence should occur after exception */
  36.   public static String UPDATE_SEQUENCE_AFTER_EXCEPTION_PROPERTY = "updateSequenceAfterException";
  37.   /** Property for sequence */
  38.   public static String SEQUENCE_PROPERTY = "sequence";

  40.   /** Name of deafult tracking file */
  41.   public static String DEFAULT_TRACKING_FILE_NAME_PROPERTY = "data/STANReceiverInfo.json";
  42.   /** Default state of update after exception */
  43.   public static String DEFAULT_UPDATE_SEQUENCE_AFTER_EXCEPTION_PROPERTY = "true";

  45.   private NATSClient client = new NATSClient();
  46.   private Subscription subscription;

  48.   private String subject;
  49.   private long sequence = 0;
  50.   private String trackingFileName;
  51.   private boolean updateSequenceAfterException;
  52.   private boolean exceptionThrown = false;

  54.   /**
  55.    * Configures receiver based on included properties
  56.    *
  57.    * @param config
  58.    *            The user-defined configuration
  59.    *
  60.    * @throws Exception If required properties are ignored
  61.    */
  62.   @Override
  63.   public void configure(Config config) throws Exception {
  64.     super.configure(config);
  65.     client.configure(config);

  67.     subject = config.getProperty(NATSClient.SUBJECT_PROPERTY);
  68.     if (subject == null) {
  69.       throw new ConfigurationException(NATSClient.SUBJECT_PROPERTY + " is a required parameter");
  70.     }

  72.     trackingFileName = config.getProperty(TRACKING_FILE_NAME_PROPERTY, DEFAULT_TRACKING_FILE_NAME_PROPERTY);
  73.     updateSequenceAfterException = Boolean.parseBoolean(config.getProperty(
  74.       UPDATE_SEQUENCE_AFTER_EXCEPTION_PROPERTY,
  75.       DEFAULT_UPDATE_SEQUENCE_AFTER_EXCEPTION_PROPERTY));
  76.   }

  78.   /**
  79.    * Does initial tracking file management and subscribes to server
  80.    * With a tracking file, gets the last sequence
  81.    *
  82.    * @throws InterruptedException if interrupted
  83.    * @throws IOException if IO error occurs
  84.    */
  85.   @Override
  86.   public void startup() throws Exception {
  87.     super.startup();

  89.     //Start client
  90.     client.startup();

  92.     //Check properties if tracking file exists
  93.     JsonObject properties = readTrackingFile();
  94.     if (properties != null &&
  95.         properties.getString(NATSClient.SERVER_HOST_PROPERTY).equals(client.getServerHost()) &&
  96.         properties.getString(NATSClient.SERVER_PORT_PROPERTY).equals(client.getServerPort()) &&
  97.         properties.getString(NATSClient.CLUSTER_ID_PROPERTY).equals(client.getClusterId()) &&
  98.         properties.getString(NATSClient.CLIENT_ID_PROPERTY).equals(client.getClientId()) &&
  99.         properties.getString(NATSClient.SUBJECT_PROPERTY).equals(subject)) {
  100.       sequence = Long.parseLong(properties.get(SEQUENCE_PROPERTY).toString());
  101.     }

  103.     subscription = client.getConnection().subscribe(
  104.       subject,
  105.       this,
  106.       new SubscriptionOptions.Builder().startAtSequence(sequence).build());
  107.     // Always starts at stored sequence; initialized to 0 and overwritten by storage
  108.   }

  110.   /**
  111.    * Closes subscription/connection and writes state in tracking file
  112.    * Wraps each statement in a try/catch to ensure each step still happens
  113.    *
  114.    * @throws IOException if IO error occurs
  115.    * @throws InterruptedException if interrupted
  116.    * @throws TimeoutException if timeout
  117.    */
  118.   @Override
  119.   public void shutdown() throws Exception {
  120.     try {
  121.       writeTrackingFile();
  122.     } catch (Exception e) {
  123.       LOGGER.log(Level.WARNING, "[" + getName() + "] failed to write to tracking file");
  124.     }
  125.     try {
  126.       subscription.unsubscribe();
  127.     } catch (Exception e) {
  128.       LOGGER.log(Level.WARNING, "[" + getName() + "] failed to unsubscribe from NATS channel");
  129.     }
  130.     subscription = null;
  131.     client.shutdown();
  132.     super.shutdown();
  133.   }

  135.   /**
  136.    * Writes pertinent configuration information to tracking file
  137.    * @throws Exception if error occurs
  138.    */
  139.   public void writeTrackingFile() throws Exception {
  140.     JsonObject json = Json.createObjectBuilder()
  141.       .add(NATSClient.SERVER_HOST_PROPERTY,client.getServerHost())
  142.       .add(NATSClient.SERVER_PORT_PROPERTY,client.getServerPort())
  143.       .add(NATSClient.CLUSTER_ID_PROPERTY,client.getClusterId())
  144.       .add(NATSClient.CLIENT_ID_PROPERTY,client.getClientId())
  145.       .add(NATSClient.SUBJECT_PROPERTY,subject)
  146.       .add(SEQUENCE_PROPERTY,sequence)
  147.     .build();

  149.     FileUtils.writeFileThenMove(
  150.       new File(trackingFileName + "_tmp"),
  151.       new File(trackingFileName),
  152.       json.toString().getBytes());
  153.   }

  155.   /**
  156.    * Reads contents of tracking file
  157.    *
  158.    * @return JsonObject containing tracking file contents, or null if file doesn't exist
  159.    * @throws Exception if error occurs
  160.    */
  161.   public JsonObject readTrackingFile() throws Exception {
  162.     JsonObject json = null;

  164.     File trackingFile = new File(trackingFileName);
  165.     if (trackingFile.exists()) {
  166.       InputStream contents = new ByteArrayInputStream(FileUtils.readFile(trackingFile));
  167.       JsonReader jsonReader = Json.createReader(contents);
  168.       json = jsonReader.readObject();
  169.       jsonReader.close();
  170.     }
  171.     return json;
  172.   }

  174.   /**
  175.    * Defines behavior for message receipt. Attempts to process notifications, with configurable behavior
  176.    * for exception handling
  177.    *
  178.    * @param message
  179.    *            The message received from the STAN server
  180.    */
  181.   @Override
  182.   public void onMessage(Message message) {
  183.     try {
  184.       // parse message, send to listeners
  185.       URLNotification notification = URLNotificationJSONConverter.parseJSON(new ByteArrayInputStream(message.getData()));
  186.       receiveNotification(notification);
  187.       // update sequence and tracking file if exception not thrown or we still want to update sequence anyway
  188.       if (!exceptionThrown || updateSequenceAfterException) {
  189.         sequence = message.getSequence();
  190.         writeTrackingFile();
  191.       }
  192.     } catch (Exception e) {
  193.       exceptionThrown = true;
  194.       LOGGER.log(Level.WARNING,
  195.         "[" + getName() + "] exception handling NATSStreaming message." +
  196.         (!updateSequenceAfterException ? " Will no longer update sequence; restart PDL to reprocess.":"") +
  197.         " Stack Trace: " + e);
  198.       LOGGER.log(Level.FINE, "[" + getName() + "] Message: " + message.getData());
  199.     }
  200.   }

  202.   /** @return trackingFileName */
  203.   public String getTrackingFileName() {
  204.     return trackingFileName;
  205.   }

  207.   /** @param trackingFileName to set */
  208.   public void setTrackingFileName(String trackingFileName) {
  209.     this.trackingFileName = trackingFileName;
  210.   }

  212.   /** @return NATSClient */
  213.   public NATSClient getClient() {
  214.     return client;
  215.   }

  217.   /** @param client NATSClient to set */
  218.   public void setClient(NATSClient client) {
  219.     this.client = client;
  220.   }

  222.   /** @return subject */
  223.   public String getSubject() {
  224.     return subject;
  225.   }

  227.   /** @param subject to set */
  228.   public void setSubject(String subject) {
  229.     this.subject = subject;
  230.   }

  232. }
Created with JaCoCo 0.8.8.202204050719