WebSocketNotificationReceiver.java

  1. package gov.usgs.earthquake.distribution;

  3. import gov.usgs.util.Config;
  4. import gov.usgs.util.FileUtils;
  5. import gov.usgs.util.StreamUtils;

  7. import javax.json.Json;
  8. import javax.json.JsonObject;
  9. import javax.json.JsonReader;
  10. import javax.websocket.CloseReason;
  11. import javax.websocket.Session;
  12. import java.io.ByteArrayInputStream;
  13. import java.io.File;
  14. import java.io.InputStream;
  15. import java.net.URI;
  16. import java.util.logging.Level;
  17. import java.util.logging.Logger;

  19. /**
  20.  * Receives notifications from an arbitrary web socket.
  21.  */
  22. public class WebSocketNotificationReceiver extends DefaultNotificationReceiver implements WebSocketListener {

  24.   /** Logger for use in the file */
  25.   public static final Logger LOGGER = Logger
  26.           .getLogger(WebSocketNotificationReceiver.class.getName());

  28.   /** Property for serverHost */
  29.   public static final String SERVER_HOST_PROPERTY = "serverHost";
  30.   /** Property for serverPort */
  31.   public static final String SERVER_PORT_PROPERTY = "serverPort";
  32.   /** Property for serverPath */
  33.   public static final String SERVER_PATH_PROPERTY = "serverPath";
  34.   /** Property for sequence */
  35.   public static final String SEQUENCE_PROPERTY = "sequence";
  36.   /** Property for timestamp */
  37.   public static final String TIMESTAMP_PROPERTY = "timestamp";
  38.   /** Property for trackingFileName */
  39.   public static final String TRACKING_FILE_NAME_PROPERTY = "trackingFileName";
  40.   /** Property for connectAttempts */
  41.   public static final String CONNECT_ATTEMPTS_PROPERTY = "connectAttempts";
  42.   /** Property for connectTimeout */
  43.   public static final String CONNECT_TIMEOUT_PROPERTY = "connectTimeout";
  44.   /** Property for retryOnClose */
  45.   public static final String RETRY_ON_CLOSE_PROPERTY = "retryOnClose";

  47.   /** Default server host */
  48.   public static final String DEFAULT_SERVER_HOST = "http://www.google.com";
  49.   /** Default server port */
  50.   public static final String DEFAULT_SERVER_PORT = "4222";
  51.   /** Default server path */
  52.   public static final String DEFAULT_SERVER_PATH = "/sequence/";
  53.   /** Default tracking file */
  54.   public static final String DEFAULT_TRACKING_FILE_NAME = "data/WebSocketReceiverInfo";
  55.   /** Default number of connect attempts */
  56.   public static final String DEFAULT_CONNECT_ATTEMPTS = "5";
  57.   /** Default timeout in ms */
  58.   public static final String DEFAULT_CONNECT_TIMEOUT = "1000";
  59.   /** Default condiction for retry on close */
  60.   public static final String DEFAULT_RETRY_ON_CLOSE = "true";
  61.   /** attribute for data */
  62.   public static final String ATTRIBUTE_DATA = "data";

  64.   private String serverHost;
  65.   private String serverPort;
  66.   private String serverPath;
  67.   private String trackingFileName;
  68.   private int attempts;
  69.   private long timeout;

  71.   private WebSocketClient client;
  72.   private String sequence = "0";


  75.   @Override
  76.   public void configure(Config config) throws Exception {
  77.     super.configure(config);

  79.     serverHost = config.getProperty(SERVER_HOST_PROPERTY, DEFAULT_SERVER_HOST);
  80.     serverPort = config.getProperty(SERVER_PORT_PROPERTY, DEFAULT_SERVER_PORT);
  81.     serverPath = config.getProperty(SERVER_PATH_PROPERTY, DEFAULT_SERVER_PATH);
  82.     attempts = Integer.parseInt(config.getProperty(CONNECT_ATTEMPTS_PROPERTY, DEFAULT_CONNECT_ATTEMPTS));
  83.     timeout = Long.parseLong(config.getProperty(CONNECT_TIMEOUT_PROPERTY, DEFAULT_CONNECT_TIMEOUT));
  84.     trackingFileName = config.getProperty(TRACKING_FILE_NAME_PROPERTY, DEFAULT_TRACKING_FILE_NAME);
  85.   }

  87.   /**
  88.    * Reads a sequence from a tracking file if it exists. Otherwise, starting sequence is 0.
  89.    * Connects to web socket
  90.    * @throws Exception if error occurs
  91.    */
  92.   @Override
  93.   public void startup() throws Exception{
  94.     super.startup();

  96.     //read sequence from tracking file if other parameters agree
  97.     JsonObject json = readTrackingFile();
  98.     if (json != null &&
  99.             json.getString(SERVER_HOST_PROPERTY).equals(serverHost) &&
  100.             json.getString(SERVER_PORT_PROPERTY).equals(serverPort) &&
  101.             json.getString(SERVER_PATH_PROPERTY).equals(serverPath)) {
  102.       sequence = json.getString(SEQUENCE_PROPERTY);
  103.     }

  105.     //open websocket
  106.     client = new WebSocketClient(new URI(serverHost + ":" + serverPort + serverPath + sequence), this, attempts, timeout, true);
  107.   }

  109.   /**
  110.    * Closes web socket
  111.    * @throws Exception if error occurs
  112.    */
  113.   @Override
  114.   public void shutdown() throws Exception{
  115.     //close socket
  116.     client.shutdown();
  117.     super.shutdown();
  118.   }

  120.   /**
  121.    * Writes tracking file to disc, storing latest sequence
  122.    * @throws Exception if error occurs
  123.    */
  124.   public void writeTrackingFile() throws Exception {
  125.     JsonObject json = Json.createObjectBuilder()
  126.             .add(SERVER_HOST_PROPERTY,serverHost)
  127.             .add(SERVER_PATH_PROPERTY,serverPath)
  128.             .add(SERVER_PORT_PROPERTY,serverPort)
  129.             .add(SEQUENCE_PROPERTY,sequence)
  130.             .build();

  132.     FileUtils.writeFileThenMove(
  133.             new File(trackingFileName + "_tmp.json"),
  134.             new File(trackingFileName + ".json"),
  135.             json.toString().getBytes());
  136.   }

  138.   /**
  139.    * Reads tracking file from disc
  140.    * @return  JsonObject tracking file
  141.    * @throws Exception if error occurs
  142.    */
  143.   public JsonObject readTrackingFile() throws Exception {
  144.     JsonObject json = null;

  146.     File trackingFile = new File(trackingFileName + ".json");
  147.     if (trackingFile.exists()) {
  148.       InputStream contents = new ByteArrayInputStream(FileUtils.readFile(trackingFile));
  149.       JsonReader jsonReader = Json.createReader(contents);
  150.       json = jsonReader.readObject();
  151.       jsonReader.close();
  152.     }
  153.     return json;
  154.   }

  156.   @Override
  157.   public void onOpen(Session session) {
  158.     // do nothing
  159.   }

  161.   /**
  162.    * Message handler function passed to WebSocketClient
  163.    * Parses the message as JSON, receives the contained URL notification, and writes the tracking file.
  164.    * @param message String
  165.    */
  166.   @Override
  167.   public void onMessage(String message) {
  168.     JsonObject json;
  169.     try (InputStream in = StreamUtils.getInputStream(message); JsonReader reader = Json.createReader(in)) {
  170.       //parse input as json
  171.       json = reader.readObject();
  172.     } catch (Exception e) {
  173.       LOGGER.log(Level.WARNING, "[" + getName() + "] exception while receiving notification; is it encoded as JSON? ", e);
  174.       return;
  175.     }
  176.     try {
  177.       //convert to URLNotification and receive
  178.       JsonObject dataJson = json.getJsonObject(ATTRIBUTE_DATA);
  179.       URLNotification notification = URLNotificationJSONConverter.parseJSON(dataJson);
  180.       receiveNotification(notification);

  182.       //send heartbeat
  183.       HeartbeatListener.sendHeartbeatMessage(getName(), "nats notification timestamp", json.getString(TIMESTAMP_PROPERTY));

  185.       //write tracking file
  186.       sequence = json.getJsonNumber(SEQUENCE_PROPERTY).toString();
  187.       writeTrackingFile();
  188.     } catch (Exception e) {
  189.       LOGGER.log(Level.WARNING, "[" + getName() + "] exception while processing URLNotification ", e);
  190.     }
  191.   }

  193.   @Override
  194.   public void onClose(Session session, CloseReason closeReason) {
  195.     // do nothing
  196.   }

  198.   @Override
  199.   public void onConnectFail() {
  200.     // do nothing
  201.   }

  203.   @Override
  204.   public void onReconnectFail() {
  205.     // do nothing
  206.   }

  208.   /** @return serverHost */
  209.   public String getServerHost() {
  210.     return serverHost;
  211.   }

  213.   /** @param serverHost to set */
  214.   public void setServerHost(String serverHost) {
  215.     this.serverHost = serverHost;
  216.   }

  218.   /** @return serverPort */
  219.   public String getServerPort() {
  220.     return serverPort;
  221.   }

  223.   /** @param serverPort to set */
  224.   public void setServerPort(String serverPort) {
  225.     this.serverPort = serverPort;
  226.   }

  228.   /** @return serverPath */
  229.   public String getServerPath() {
  230.     return serverPath;
  231.   }

  233.   /** @param serverPath to set */
  234.   public void setServerPath(String serverPath) {
  235.     this.serverPath = serverPath;
  236.   }

  238.   /** @return trackingFileName */
  239.   public String getTrackingFileName() {
  240.     return trackingFileName;
  241.   }

  243.   /** @param trackingFileName to set */
  244.   public void setTrackingFileName(String trackingFileName) {
  245.     this.trackingFileName = trackingFileName;
  246.   }

  248.   /** @return sequence */
  249.   public String getSequence() {
  250.     return sequence;
  251.   }

  253.   /** @param sequence to set */
  254.   public void setSequence(String sequence) {
  255.     this.sequence = sequence;
  256.   }

  258.   /** @return attempts */
  259.   public int getAttempts() {
  260.     return attempts;
  261.   }

  263.   /** @param attempts to set */
  264.   public void setAttempts(int attempts) {
  265.     this.attempts = attempts;
  266.   }

  268.   /** @return timeout */
  269.   public long getTimeout() {
  270.     return timeout;
  271.   }

  273.   /** @param timeout to set */
  274.   public void setTimeout(long timeout) {
  275.     this.timeout = timeout;
  276.   }
  277. }
Created with JaCoCo 0.8.8.202204050719