EIDSInputWedge.java

  1. package gov.usgs.earthquake.eids;

  3. import java.io.File;
  4. import java.net.URL;
  5. import java.util.ArrayList;
  6. import java.util.Date;
  7. import java.util.HashMap;
  8. import java.util.Iterator;
  9. import java.util.List;
  10. import java.util.Map;
  11. import java.util.logging.Level;
  12. import java.util.logging.Logger;

  14. import javax.xml.bind.JAXBException;

  16. import org.xml.sax.SAXParseException;

  18. import gov.usgs.earthquake.distribution.Bootstrappable;
  19. import gov.usgs.earthquake.distribution.CLIProductBuilder;
  20. import gov.usgs.earthquake.distribution.ConfigurationException;
  21. import gov.usgs.earthquake.distribution.ProductBuilder;
  22. import gov.usgs.earthquake.distribution.ProductSender;
  23. import gov.usgs.earthquake.distribution.SocketProductSender;
  24. import gov.usgs.earthquake.product.Content;
  25. import gov.usgs.earthquake.product.FileContent;
  26. import gov.usgs.earthquake.product.Product;
  27. import gov.usgs.earthquake.product.ProductId;
  28. import gov.usgs.earthquake.quakeml.FileToQuakemlConverter;
  29. import gov.usgs.util.Config;
  30. import gov.usgs.util.CryptoUtils;
  31. import gov.usgs.util.DirectoryPoller;
  32. import gov.usgs.util.FileUtils;
  33. import gov.usgs.util.StringUtils;
  34. import gov.usgs.util.CryptoUtils.Version;

  36. /**
  37.  * Read messages from files or a poll directory, and push products into PDL.
  38.  *
  39.  * This is supports EIDS/QDDS style polling. The input messages are converted to
  40.  * Quakeml using the FileToQuakemlConverter interface, then sent as Quakeml
  41.  * based products.
  42.  *
  43.  * Much of the configuration can be supplied using either a configuration file,
  44.  * or command line arguments.
  45.  */
  46. public class EIDSInputWedge extends ProductBuilder implements Runnable,
  47.         Bootstrappable {

  49.     /**
  50.      * Global reference to arguments array, when EIDSInputWedge is run via
  51.      * Bootstrap.
  52.      */
  53.     public static String[] ARGS = null;

  55.     private static final Logger LOGGER = Logger.getLogger(EIDSInputWedge.class
  56.             .getName());

  58.     /** Property for parser class */
  59.     public static final String PARSER_CLASS_PROPERTY = "parserClass";
  60.     /** Default parser class */
  61.     public static final String DEFAULT_PARSER_CLASS = "gov.usgs.earthquake.event.QuakemlToQuakemlConverter";

  63.     /** Property for polldir */
  64.     public static final String POLLDIR_PROPERTY = "directory";
  65.     /** Default polldir */
  66.     public static final String DEFAULT_POLLDIR = "polldir";
  67.     private File polldir = new File(DEFAULT_POLLDIR);

  69.     /** Property for storage directory */
  70.     public static final String STORAGEDIR_PROPERTY = "oldinputdir";
  71.     /** Default storage directory */
  72.     public static final String DEFAULT_STORAGEDIR = "oldinput";
  73.     private File storagedir = new File(DEFAULT_STORAGEDIR);

  75.     /** Property for error directory */
  76.     public static final String ERRORDIR_PROPERTY = "errordir";
  77.     /** Default error directory */
  78.     public static final String DEFAULT_ERRORDIR = "errordir";
  79.     private File errordir = new File(DEFAULT_ERRORDIR);

  81.     /** Property for validate */
  82.     public static final String VALIDATE_PROPERTY = "validate";
  83.     /** Default status of validate */
  84.     public static final String DEFAULT_VALIDATE = "false";

  86.     /** Property for sendOriginWhenPhasesExist */
  87.     public static final String SEND_ORIGIN_WHEN_PHASES_EXIST_PROPERTY = "sendOriginWhenPhasesExist";
  88.     /** Default status of sendOrigin...  */
  89.     public static final String DEFAULT_SEND_ORIGIN_WHEN_PHASES_EXIST = "false";

  91.     /** Property for sendMechanismWhenPhasesExist */
  92.     public static final String SEND_MECHANISM_WHEN_PHASES_EXIST_PROPERTY = "sendMechanismWhenPhasesExist";
  93.     /** Default status of sendMechanism... */
  94.     public static final String DEFAULT_SEND_MECHANISM_WHEN_PHASES_EXIST = "false";

  96.     /** Convert parsed quakeml to a product. */
  97.     private ProductCreator productCreator = new QuakemlProductCreator();

  99.     /** Whether created products should be converted to internal types. */
  100.     public static final String CREATE_INTERNAL_PRODUCTS_PROPERTY = "createInternalProducts";
  101.     /** Default status of CREATE_INTERNAL_PRODUCTS */
  102.     public static final String DEFAULT_CREATE_INTERNAL_PRODUCTS = "false";
  103.     private boolean createInternalProducts = false;

  105.     /** Whether created products should be converted to scenario types. */
  106.     public static final String CREATE_SCENARIO_PRODUCTS_PROPERTY = "createScenarioProducts";
  107.     /** Default status of CREATE_SCENARIO_PRODUCTS */
  108.     public static final String DEFAULT_CREATE_SCENARIO_PRODUCTS = "false";
  109.     private boolean createScenarioProducts = false;

  111.     /** Directory polling object. */
  112.     private DirectoryPoller directoryPoller;

  114.     /** Poll interval property */
  115.     public static final String POLLINTERVAL_PROPERTY = "interval";
  116.     /** Default interval for POLLINTERVAL */
  117.     public static final String DEFAULT_POLLINTERVAL = "1000";
  118.     private long pollInterval = 1000L;

  120.     /** Property for pollCarefully */
  121.     public static final String POLL_CAREFULLY_PROPERTY = "pollCarefully";
  122.     /** Default status of POLL_CAREFULLY */
  123.     public static final String DEFAULT_POLL_CAREFULLY = "false";
  124.     private boolean pollCarefully = false;

  126.     /** Property for doBufferFix */
  127.     public static final String DO_BUFFER_FIX_PROPERTY = "doBufferFix";
  128.     /** Default status of DO_BUFFER_FIX property */
  129.     public static final String DEFAULT_DO_BUFFER_FIX = "true";
  130.     private boolean doBufferFix = true;

  132.     private Thread pollThread = null;

  134.     /**
  135.      * Empty constructor
  136.      * @throws Exception if error occurs
  137.      */
  138.     public EIDSInputWedge() throws Exception {
  139.     }

  141.     /**
  142.      * Gets products from file and iterates through each product
  143.      * During iteration, sets type to internal/scenario if createInternalProducts
  144.      * or createScenarioProducts is true. Attaches Content files to product,
  145.      * Sends product
  146.      * @param file File containing products
  147.      * @param attachContent Map of String and Content
  148.      * @return Map of product IDs and sent products
  149.      * @throws Exception if error occurs
  150.      */
  151.     public Map<ProductId, Map<ProductSender, Exception>> parseAndSend(
  152.             final File file, final Map<String, Content> attachContent)
  153.             throws Exception {

  155.         Map<ProductId, Map<ProductSender, Exception>> sendProductResults = new HashMap<ProductId, Map<ProductSender, Exception>>();

  157.         List<Product> products = productCreator.getProducts(file);

  159.         Iterator<Product> iter = products.iterator();
  160.         while (iter.hasNext()) {
  161.             Product product = iter.next();
  162.             ProductId id = product.getId();

  164.             if (createInternalProducts) {
  165.                 id.setType("internal-" + id.getType());
  166.             }
  167.             if (createScenarioProducts) {
  168.                 id.setType(id.getType() + "-scenario");
  169.             }

  171.             // attach files to generated product
  172.             if (attachContent != null && attachContent.size() > 0) {
  173.                 if (products.size() > 1) {
  174.                     throw new Exception("Trying to attach files,"
  175.                             + " generated more than 1 product");
  176.                 }
  177.                 product.getContents().putAll(attachContent);
  178.             }

  180.             // send product, save any exceptions
  181.             sendProductResults.put(product.getId(), sendProduct(product));
  182.         }

  184.         return sendProductResults;
  185.     }

  187.     /**
  188.      * Parses given file, looking for send exceptions and reports statistics
  189.      * @param file to parse and look for errors
  190.      */
  191.     public void onFile(File file) {
  192.         Date inputtime = new Date();
  193.         LOGGER.info("Reading file " + file.getName());

  195.         try {
  196.             Map<ProductId, Map<ProductSender, Exception>> sendExceptions = parseAndSend(
  197.                     file, null);

  199.             // check how send went
  200.             int numSenders = getProductSenders().size();
  201.             int total = sendExceptions.size();
  202.             int successful = 0;
  203.             int partialFailures = 0;
  204.             int totalFailures = 0;

  206.             Iterator<ProductId> sentIds = sendExceptions.keySet().iterator();
  207.             while (sentIds.hasNext()) {
  208.                 ProductId sentId = sentIds.next();
  209.                 if (sendExceptions.get(sentId).size() == numSenders) {
  210.                     totalFailures++;
  211.                     LOGGER.severe("Total failure sending product "
  212.                             + sentId.toString());
  213.                 } else {
  214.                     // output built product id because it was sent at least once
  215.                     System.out.println(sentId.toString());

  217.                     if (sendExceptions.get(sentId).size() == 0) {
  218.                         successful++;
  219.                     } else {
  220.                         partialFailures++;
  221.                         LOGGER.warning("Partial failure sending product "
  222.                                 + sentId.toString());
  223.                     }
  224.                 }
  225.             }

  227.             LOGGER.info("generated " + total + " products: " + successful
  228.                     + " sent, " + partialFailures + " partially sent, "
  229.                     + totalFailures + " failed to send");

  231.             // notify of failures using exit code
  232.             if (totalFailures > 0) {
  233.                 // consider this failure, event if some products sent
  234.                 throw new Exception();
  235.             }

  237.             // succeeded, at least somewhat
  238.             // move to oldinput directory
  239.             file.renameTo(new File(storagedir, file.getName() + "_"
  240.                     + inputtime.getTime()));
  241.         } catch (Exception e) {
  242.             if (e instanceof JAXBException
  243.                     && ((JAXBException) e).getLinkedException() instanceof SAXParseException) {
  244.                 SAXParseException spe = (SAXParseException) ((JAXBException) e)
  245.                         .getLinkedException();
  246.                 LOGGER.warning("Parse error: " + spe.getMessage() + "; line="
  247.                         + spe.getLineNumber() + ", column="
  248.                         + spe.getColumnNumber());
  249.             }

  251.             if (errordir != null) {
  252.                 if (!errordir.exists()) {
  253.                     errordir.mkdirs();
  254.                 }
  255.                 File errorfile = new File(errordir, file.getName() + "_"
  256.                         + new Date().getTime());

  258.                 // move to error directory
  259.                 try {
  260.                 LOGGER.log(Level.WARNING, "Errors processing file, moving to "
  261.                         + errorfile.getCanonicalPath(), e);
  262.                 } catch (Exception ignore) {
  263.                     //ignore
  264.                 }
  265.                 file.renameTo(errorfile);
  266.             } else {
  267.                 LOGGER.warning("Error processing file " + file.getName()
  268.                         + ", and no error directory configured");
  269.                 FileUtils.deleteTree(file);
  270.             }
  271.         }
  272.     }

  274.     @Override
  275.     public void configure(Config config) throws Exception {
  276.         super.configure(config);

  278.         String parserClassName = config.getProperty(PARSER_CLASS_PROPERTY);
  279.         if (parserClassName == null) {
  280.             LOGGER.config("Using QuakemlToProductConverter");
  281.         } else {
  282.             Object parserObj = Class.forName(parserClassName)
  283.                     .getConstructor().newInstance();
  284.             if (parserObj instanceof ProductCreator) {
  285.                 productCreator = (ProductCreator) parserObj;
  286.             } else if (parserObj instanceof FileToQuakemlConverter) {
  287.                 QuakemlProductCreator quakemlCreator = new QuakemlProductCreator(doBufferFix);
  288.                 quakemlCreator.setConverter((FileToQuakemlConverter) parserObj);
  289.                 productCreator = quakemlCreator;
  290.             } else {
  291.                 throw new ConfigurationException("configured parser class "
  292.                         + parserClassName + " does not implement "
  293.                         + FileToQuakemlConverter.class.getName());
  294.             }
  295.             LOGGER.config("Using parser class " + parserClassName);
  296.         }

  298.         boolean validate = Boolean.getBoolean(config.getProperty(
  299.                 VALIDATE_PROPERTY, DEFAULT_VALIDATE));
  300.         productCreator.setValidate(validate);
  301.         LOGGER.config("Validation " + (validate ? "enabled" : "disabled"));

  303.         boolean sendOriginWhenPhasesExist = Boolean.valueOf(config
  304.                 .getProperty(SEND_ORIGIN_WHEN_PHASES_EXIST_PROPERTY,
  305.                         DEFAULT_SEND_ORIGIN_WHEN_PHASES_EXIST));
  306.         if (productCreator instanceof EQMessageProductCreator) {
  307.             ((EQMessageProductCreator) productCreator)
  308.                     .setSendOriginWhenPhasesExist(sendOriginWhenPhasesExist);
  309.         } else {
  310.             ((QuakemlProductCreator) productCreator)
  311.                     .setSendOriginWhenPhasesExist(sendOriginWhenPhasesExist);
  312.         }
  313.         LOGGER.config("sendOriginWhenPhasesExist = "
  314.                 + sendOriginWhenPhasesExist);

  316.         boolean sendMechanismWhenPhasesExist = Boolean.valueOf(config
  317.                 .getProperty(SEND_MECHANISM_WHEN_PHASES_EXIST_PROPERTY,
  318.                         DEFAULT_SEND_MECHANISM_WHEN_PHASES_EXIST));
  319.         if (productCreator instanceof EQMessageProductCreator) {
  320.             if (sendMechanismWhenPhasesExist) {
  321.                 LOGGER.warning(SEND_MECHANISM_WHEN_PHASES_EXIST_PROPERTY
  322.                         + " is not supported for EQMessageProductCreator");
  323.             }
  324.         } else {
  325.             ((QuakemlProductCreator) productCreator)
  326.                     .setSendMechanismWhenPhasesExist(sendMechanismWhenPhasesExist);
  327.         }
  328.         LOGGER.config("sendMechanismWhenPhasesExist = "
  329.                 + sendMechanismWhenPhasesExist);

  331.         polldir = new File(
  332.                 config.getProperty(POLLDIR_PROPERTY, DEFAULT_POLLDIR));
  333.         LOGGER.config("Using poll directory " + polldir.getCanonicalPath());

  335.         pollInterval = Long.valueOf(config.getProperty(POLLINTERVAL_PROPERTY,
  336.                 DEFAULT_POLLINTERVAL));
  337.         LOGGER.config("Using poll interval " + pollInterval + "ms");

  339.         pollCarefully = Boolean.valueOf(config.getProperty(POLL_CAREFULLY_PROPERTY,
  340.                 DEFAULT_POLL_CAREFULLY));
  341.         LOGGER.config("Poll carefully = " + pollCarefully);

  343.         storagedir = new File(config.getProperty(STORAGEDIR_PROPERTY,
  344.                 DEFAULT_STORAGEDIR));
  345.         LOGGER.config("Using oldinput directory "
  346.                 + storagedir.getCanonicalPath());

  348.         errordir = new File(config.getProperty(ERRORDIR_PROPERTY,
  349.                 DEFAULT_ERRORDIR));
  350.         LOGGER.config("Using error directory " + errordir.getCanonicalPath());

  352.         createInternalProducts = Boolean.valueOf(config
  353.                 .getProperty(CREATE_INTERNAL_PRODUCTS_PROPERTY,
  354.                         DEFAULT_CREATE_INTERNAL_PRODUCTS));
  355.         LOGGER.config("createInternalProducts = " + createInternalProducts);

  357.         createScenarioProducts = Boolean.valueOf(config
  358.                 .getProperty(CREATE_SCENARIO_PRODUCTS_PROPERTY,
  359.                         DEFAULT_CREATE_SCENARIO_PRODUCTS));
  360.         LOGGER.config("createScenarioProducts = " + createScenarioProducts);

  362.         doBufferFix = Boolean.valueOf(config
  363.                 .getProperty(DO_BUFFER_FIX_PROPERTY,
  364.                                 DEFAULT_DO_BUFFER_FIX));
  365.         LOGGER.config("doBufferFix = " + doBufferFix);
  366.     }

  368.     @Override
  369.     public void shutdown() throws Exception {
  370.         if (pollThread != null) {
  371.             pollThread.interrupt();
  372.             pollThread = null;
  373.         }

  375.         super.shutdown();
  376.     }

  378.     @Override
  379.     public void startup() throws Exception {
  380.         super.startup();

  382.         if (pollThread == null) {
  383.             pollThread = new Thread(this);
  384.             pollThread.setName("poll thread");
  385.             pollThread.start();
  386.         }
  387.     }

  389.     /** @return polldir */
  390.     public File getPolldir() {
  391.         return polldir;
  392.     }

  394.     /** @param polldir File to set */
  395.     public void setPolldir(File polldir) {
  396.         this.polldir = polldir;
  397.     }

  399.     /** @return storagedir */
  400.     public File getStoragedir() {
  401.         return storagedir;
  402.     }

  404.     /** @param storagedir File to set */
  405.     public void setStoragedir(File storagedir) {
  406.         this.storagedir = storagedir;
  407.     }

  409.     /** @return errordir */
  410.     public File getErrordir() {
  411.         return errordir;
  412.     }

  414.     /** @param errordir File to send */
  415.     public void setErrordir(File errordir) {
  416.         this.errordir = errordir;
  417.     }

  419.     /** @return productCreator */
  420.     public ProductCreator getProductCreator() {
  421.         return productCreator;
  422.     }

  424.     /** @param productCreator to set */
  425.     public void setProductCreator(ProductCreator productCreator) {
  426.         this.productCreator = productCreator;
  427.     }

  429.     /** @return directoryPoller */
  430.     public DirectoryPoller getDirectoryPoller() {
  431.         return directoryPoller;
  432.     }

  434.     /** @param directoryPoller to set */
  435.     public void setDirectoryPoller(DirectoryPoller directoryPoller) {
  436.         this.directoryPoller = directoryPoller;
  437.     }

  439.     /** @return pollInterval long */
  440.     public long getPollInterval() {
  441.         return pollInterval;
  442.     }

  444.     /** @param pollInterval long to set */
  445.     public void setPollInterval(long pollInterval) {
  446.         this.pollInterval = pollInterval;
  447.     }

  449.     /** @return pollCarefully boolean */
  450.     public boolean isPollCarefully() {
  451.         return pollCarefully;
  452.     }

  454.     /** @param pollCarefully boolean to set */
  455.     public void setPollCarefully(boolean pollCarefully) {
  456.         this.pollCarefully = pollCarefully;
  457.     }

  459.     /**
  460.      * @return the createInternalProducts
  461.      */
  462.     public boolean isCreateInternalProducts() {
  463.         return createInternalProducts;
  464.     }

  466.     /**
  467.      * @param createInternalProducts
  468.      *            the createInternalProducts to set
  469.      */
  470.     public void setCreateInternalProducts(boolean createInternalProducts) {
  471.         this.createInternalProducts = createInternalProducts;
  472.     }

  474.     /**
  475.      * @return the createScenarioProducts
  476.      */
  477.     public boolean isCreateScenarioProducts() {
  478.         return createScenarioProducts;
  479.     }

  481.     /**
  482.      * @param createScenarioProducts
  483.      *            the createScenarioProducts to set
  484.      */
  485.     public void setCreateScenarioProducts(boolean createScenarioProducts) {
  486.         this.createScenarioProducts = createScenarioProducts;
  487.     }

  489.     /**
  490.      * Parses a string of servers into SocketProductSenders, all put into a list
  491.      * of product senders
  492.      * @param servers String of servers, split by commas
  493.      * @param connectTimeout int timeout
  494.      * @param binaryFormat boolean if binary format
  495.      * @param enableDeflate boolean if Deflate should be enabled
  496.      * @return List of product senders
  497.      */
  498.     public static List<ProductSender> parseServers(final String servers,
  499.             final Integer connectTimeout, final boolean binaryFormat,
  500.             final boolean enableDeflate) {
  501.         List<ProductSender> senders = new ArrayList<ProductSender>();

  503.         Iterator<String> iter = StringUtils.split(servers, ",").iterator();
  504.         while (iter.hasNext()) {
  505.             String server = iter.next();
  506.             String[] parts = server.split(":");
  507.             SocketProductSender sender = new SocketProductSender(parts[0],
  508.                     Integer.parseInt(parts[1]), connectTimeout);
  509.             sender.setBinaryFormat(binaryFormat);
  510.             sender.setEnableDeflate(enableDeflate);
  511.             senders.add(sender);
  512.         }

  514.         return senders;
  515.     }

  517.     /** Argument for help */
  518.     public static final String HELP_ARGUMENT = "--help";
  519.     /** Argument for poll */
  520.     public static final String POLL_ARGUMENT = "--poll";

  522.     /** Argument for poleCarefully */
  523.     public static final String POLL_CAREFULLY_ARGUMENT = "--pollCarefully";
  524.     /** Argument for polldir */
  525.     public static final String POLLDIR_ARGUMENT = "--polldir=";
  526.     /** Argument for errordir */
  527.     public static final String ERRORDIR_ARGUMENT = "--errordir=";
  528.     /** Argument for storagedir */
  529.     public static final String STORAGEDIR_ARGUMENT = "--oldinputdir=";
  530.     /** Argument for poll interval */
  531.     public static final String POLL_INTERVAL_ARGUMENT = "--pollInterval=";
  532.     /** Argument for tracker url */
  533.     public static final String TRACKER_URL_ARGUMENT = "--trackerURL=";

  535.     /** Argument for file */
  536.     public static final String FILE_ARGUMENT = "--file=";

  538.     /** Argument for parser */
  539.     public static final String PARSER_ARGUMENT = "--parser=";
  540.     /** Argument for validate */
  541.     public static final String VALIDATE_ARGUMENT = "--validate";
  542.     /** Argument for privateKey */
  543.     public static final String PRIVATE_KEY_ARGUMENT = "--privateKey=";
  544.     /** Argument for signatureVersion */
  545.     public static final String SIGNATURE_VERSION_ARGUMENT = "--signatureVersion=";

  547.     /** Argument for servers */
  548.     public static final String SERVERS_ARGUMENT = "--servers=";
  549.     /** Default server for server argument */
  550.     public static final String SERVERS_DEFAULT = "prod01-pdl01.cr.usgs.gov:11235,prod02-pdl01.cr.usgs.gov:11235";
  551.     /** Argument for connection timeout */
  552.     public static final String CONNECT_TIMEOUT_ARGUMENT = "--connectTimeout=";
  553.     /** Default timeout for connection */
  554.     public static final Integer DEFAULT_CONNECT_TIMEOUT = 15000;
  555.     /** Argument for binaryFormat */
  556.     public static final String BINARY_FORMAT_ARGUMENT = "--binaryFormat";
  557.     /** Argument for disableDeflate */
  558.     public static final String DISABLE_DEFLATE_ARGUMENT = "--disableDeflate";
  559.     /** Argument for attach */
  560.     public static final String ATTACH_ARGUMENT = "--attach=";
  561.     /** Argument for sending origin with phases */
  562.     public static final String SEND_ORIGINS_WITH_PHASES = "--sendOriginWhenPhasesExist";
  563.     /** Argument for sending mechanisms with phases */
  564.     public static final String SEND_MECHANISMS_WITH_PHASES = "--sendMechanismWhenPhasesExist";

  566.     /** Argument for creating internal products */
  567.     public static final String CREATE_INTERNAL_PRODUCTS = "--internal";
  568.     /** Argument for creating scenario products */
  569.     public static final String CREATE_SCENARIO_PRODUCTS = "--scenario";

  571.     /** Argument for testing */
  572.     public static final String TEST_ARGUMENT = "--test";

  574.     /**
  575.      * Bootstrappable interface.
  576.      */
  577.     @Override
  578.     public void run(final String[] args) throws Exception {
  579.         // save arguments in global for access by FileToQuakemlParser objects.
  580.         EIDSInputWedge.ARGS = args;

  582.         boolean test = false;
  583.         boolean help = false;
  584.         boolean poll = false;
  585.         boolean validate = this.productCreator.isValidate();
  586.         boolean sendOriginWhenPhasesExist = false;
  587.         boolean sendMechanismWhenPhasesExist = false;

  589.         // preserve any existing settings from config file
  590.         if (productCreator instanceof QuakemlProductCreator) {
  591.             sendOriginWhenPhasesExist = ((QuakemlProductCreator) productCreator)
  592.                     .isSendOriginWhenPhasesExist();
  593.             sendMechanismWhenPhasesExist = ((QuakemlProductCreator) productCreator)
  594.                     .isSendMechanismWhenPhasesExist();
  595.         } else if (productCreator instanceof EQMessageProductCreator) {
  596.             sendOriginWhenPhasesExist = ((EQMessageProductCreator) productCreator)
  597.                     .isSendOriginWhenPhasesExist();
  598.         }

  600.         File file = null;
  601.         // when sending 1 product, allow extra files to be attached.
  602.         HashMap<String, Content> attachContent = new HashMap<String, Content>();
  603.         Integer connectTimeout = DEFAULT_CONNECT_TIMEOUT;
  604.         boolean binaryFormat = false;
  605.         boolean enableDeflate = true;

  607.         StringBuffer arguments = new StringBuffer();
  608.         for (String arg : args) {
  609.             arguments.append(arg).append(" ");
  610.             if (arg.equals(HELP_ARGUMENT)) {
  611.                 help = true;
  612.             } else if (arg.equals(POLL_ARGUMENT)) {
  613.                 poll = true;
  614.             } else if (arg.equals(POLL_CAREFULLY_ARGUMENT)) {
  615.                 pollCarefully = true;
  616.             } else if (arg.equals(SEND_ORIGINS_WITH_PHASES)) {
  617.                 sendOriginWhenPhasesExist = true;
  618.             } else if (arg.equals(SEND_MECHANISMS_WITH_PHASES)) {
  619.                 sendMechanismWhenPhasesExist = true;
  620.             } else if (arg.startsWith(POLLDIR_ARGUMENT)) {
  621.                 setPolldir(new File(arg.replace(POLLDIR_ARGUMENT, "")));
  622.             } else if (arg.startsWith(ERRORDIR_ARGUMENT)) {
  623.                 setErrordir(new File(arg.replace(ERRORDIR_ARGUMENT, "")));
  624.             } else if (arg.startsWith(STORAGEDIR_ARGUMENT)) {
  625.                 setStoragedir(new File(arg.replace(STORAGEDIR_ARGUMENT, "")));
  626.             } else if (arg.startsWith(FILE_ARGUMENT)) {
  627.                 file = new File(arg.replace(FILE_ARGUMENT, ""));
  628.             } else if (arg.startsWith(ATTACH_ARGUMENT)) {
  629.                 File attach = new File(arg.replace(ATTACH_ARGUMENT, ""));
  630.                 if (attach.isDirectory()) {
  631.                     attachContent.putAll(FileContent
  632.                             .getDirectoryContents(attach));
  633.                 } else {
  634.                     attachContent
  635.                             .put(attach.getName(), new FileContent(attach));
  636.                 }
  637.             } else if (arg.startsWith(PARSER_ARGUMENT)) {
  638.                 Object parser = Class.forName(arg.replace(PARSER_ARGUMENT, ""))
  639.                         .getConstructor().newInstance();
  640.                 if (parser instanceof ProductCreator) {
  641.                     setProductCreator((ProductCreator) parser);
  642.                 } else {
  643.                     QuakemlProductCreator productCreator = new QuakemlProductCreator();
  644.                     productCreator
  645.                             .setConverter((FileToQuakemlConverter) parser);
  646.                     setProductCreator(productCreator);
  647.                 }
  648.             } else if (arg.startsWith(VALIDATE_ARGUMENT)) {
  649.                 validate = true;
  650.             } else if (arg.startsWith(SERVERS_ARGUMENT)) {
  651.                 // ignore servers argument when in test mode
  652.                 if (!test) {
  653.                     getProductSenders().clear();
  654.                     getProductSenders().addAll(
  655.                             parseServers(arg.replace(SERVERS_ARGUMENT, ""),
  656.                                     connectTimeout, binaryFormat, enableDeflate));
  657.                 }
  658.             } else if (arg.startsWith(TEST_ARGUMENT)) {
  659.                 test = true;
  660.                 getProductSenders().clear();
  661.                 getProductSenders().add(new DebugProductSender());
  662.             } else if (arg.startsWith(PRIVATE_KEY_ARGUMENT)) {
  663.                 setPrivateKey(CryptoUtils.readOpenSSHPrivateKey(FileUtils
  664.                         .readFile(new File(arg
  665.                                 .replace(PRIVATE_KEY_ARGUMENT, ""))), null));
  666.             } else if (arg.startsWith(SIGNATURE_VERSION_ARGUMENT)) {
  667.                 setSignatureVersion(Version.fromString(
  668.                         arg.replace(SIGNATURE_VERSION_ARGUMENT, "")));
  669.             } else if (arg.startsWith(CONNECT_TIMEOUT_ARGUMENT)) {
  670.                 connectTimeout = Integer.valueOf(arg.replace(
  671.                         CONNECT_TIMEOUT_ARGUMENT, ""));
  672.             } else if (arg.equals(BINARY_FORMAT_ARGUMENT)) {
  673.                 binaryFormat = true;
  674.             } else if (arg.equals(DISABLE_DEFLATE_ARGUMENT)) {
  675.                 enableDeflate = false;
  676.             } else if (arg.startsWith(POLL_INTERVAL_ARGUMENT)) {
  677.                 setPollInterval(Long.valueOf(arg.replace(
  678.                         POLL_INTERVAL_ARGUMENT, "")));
  679.             } else if (arg.startsWith(TRACKER_URL_ARGUMENT)) {
  680.                 this.setTrackerURL(new URL(arg
  681.                         .replace(TRACKER_URL_ARGUMENT, "")));
  682.             } else if (arg.equals(CREATE_INTERNAL_PRODUCTS)) {
  683.                 createInternalProducts = true;
  684.             } else if (arg.equals(CREATE_SCENARIO_PRODUCTS)) {
  685.                 createScenarioProducts = true;
  686.             } else if (arg.equals(CLIProductBuilder.DISABLE_PARALLEL_SEND)) {
  687.                 parallelSend = false;
  688.             } else if (arg.startsWith(CLIProductBuilder.PARALLEL_SEND_TIMEOUT_ARGUMENT)) {
  689.                 parallelSendTimeout = Long.valueOf(
  690.                         arg.replace(CLIProductBuilder.PARALLEL_SEND_TIMEOUT_ARGUMENT, ""));
  691.             }
  692.         }

  694.         ProductCreator creator = getProductCreator();
  695.         creator.setValidate(validate);
  696.         if (creator instanceof EQMessageProductCreator) {
  697.             ((EQMessageProductCreator) creator)
  698.                     .setSendOriginWhenPhasesExist(sendOriginWhenPhasesExist);
  699.         } else if (creator instanceof QuakemlProductCreator) {
  700.             QuakemlProductCreator quakemlCreator = ((QuakemlProductCreator) creator);
  701.             quakemlCreator
  702.                     .setSendOriginWhenPhasesExist(sendOriginWhenPhasesExist);
  703.             quakemlCreator
  704.                     .setSendMechanismWhenPhasesExist(sendMechanismWhenPhasesExist);
  705.         }

  707.         if (
  708.         // want usage, or didn't provide arguments
  709.         (help || args.length == 0)
  710.         // or didn't provide correct arguments
  711.                 || (!poll && file == null)) {
  712.             printUsage();
  713.         }

  715.         // run continuously
  716.         else if (poll) {
  717.             startup();
  718.         }

  720.         // send, then shutdown
  721.         else {
  722.             // file != null
  723.             try {
  724.                 // send
  725.                 Map<ProductId, Map<ProductSender, Exception>> sendExceptions = parseAndSend(
  726.                         file, attachContent);

  728.                 // check how send went
  729.                 int numSenders = getProductSenders().size();
  730.                 int total = sendExceptions.size();
  731.                 int successful = 0;
  732.                 int partialFailures = 0;
  733.                 int totalFailures = 0;

  735.                 Iterator<ProductId> sentIds = sendExceptions.keySet()
  736.                         .iterator();
  737.                 while (sentIds.hasNext()) {
  738.                     ProductId sentId = sentIds.next();
  739.                     if (sendExceptions.get(sentId).size() == numSenders) {
  740.                         totalFailures++;
  741.                         LOGGER.severe("Total failure sending product "
  742.                                 + sentId.toString());
  743.                     } else {
  744.                         // output built product id because it was sent at least
  745.                         // once
  746.                         System.out.println(sentId.toString());

  748.                         if (sendExceptions.get(sentId).size() == 0) {
  749.                             successful++;
  750.                         } else {
  751.                             partialFailures++;
  752.                             LOGGER.warning("Partial failure sending product "
  753.                                     + sentId.toString());
  754.                         }
  755.                     }
  756.                 }

  758.                 LOGGER.info("Generated " + total + " products: " + successful
  759.                         + " sent, " + partialFailures + " partially sent, "
  760.                         + totalFailures + " failed to send");

  762.                 // notify of failures using exit code
  763.                 if (totalFailures > 0) {
  764.                     System.exit(CLIProductBuilder.EXIT_UNABLE_TO_SEND);
  765.                 }
  766.                 if (partialFailures > 0) {
  767.                     System.exit(CLIProductBuilder.EXIT_PARTIALLY_SENT);
  768.                 }
  769.             } catch (Exception e) {
  770.                 if (e instanceof JAXBException
  771.                         && ((JAXBException) e).getLinkedException() instanceof SAXParseException) {
  772.                     SAXParseException spe = (SAXParseException) ((JAXBException) e)
  773.                             .getLinkedException();
  774.                     LOGGER.severe("Parse error: " + spe.getMessage()
  775.                             + "; line=" + spe.getLineNumber() + ", column="
  776.                             + spe.getColumnNumber());
  777.                 } else {
  778.                     LOGGER.log(Level.SEVERE, "Exception while sending", e);
  779.                 }
  780.                 System.exit(CLIProductBuilder.EXIT_UNABLE_TO_SEND);
  781.             }
  782.         }

  784.     }

  786.     /** Usage for interface */
  787.     public static void printUsage() {
  788.         System.err
  789.                 .println("\nUsage:\n\n"
  790.                         + "java -cp ProductClient.jar gov.usgs.earthquake.eids.EIDSInputWedge"
  791.                         + " ("
  792.                         + HELP_ARGUMENT
  793.                         + "|"
  794.                         + POLL_ARGUMENT
  795.                         + "|"
  796.                         + FILE_ARGUMENT
  797.                         + "FILE) ["
  798.                         + PRIVATE_KEY_ARGUMENT
  799.                         + "KEYFILE] ["
  800.                         + SIGNATURE_VERSION_ARGUMENT
  801.                         + "VERSION] ["
  802.                         + SERVERS_ARGUMENT
  803.                         + "SERVERS] ["
  804.                         + TEST_ARGUMENT
  805.                         + "] ["
  806.                         + CONNECT_TIMEOUT_ARGUMENT
  807.                         + "TIMEOUT] ["
  808.                         + PARSER_ARGUMENT
  809.                         + "PARSER] ["
  810.                         + POLLDIR_ARGUMENT
  811.                         + "POLLDIR] ["
  812.                         + POLL_INTERVAL_ARGUMENT
  813.                         + "INTERVAL] ["
  814.                         + STORAGEDIR_ARGUMENT
  815.                         + "STORAGEDIR] ["
  816.                         + ERRORDIR_ARGUMENT
  817.                         + "ERRORDIR] ["
  818.                         + ATTACH_ARGUMENT
  819.                         + "ATTACH] ["
  820.                         + SEND_ORIGINS_WITH_PHASES
  821.                         + "] ["
  822.                         + SEND_MECHANISMS_WITH_PHASES
  823.                         + "] ["
  824.                         + CREATE_INTERNAL_PRODUCTS
  825.                         + "] ["
  826.                         + CREATE_SCENARIO_PRODUCTS
  827.                         + "] ["
  828.                         + BINARY_FORMAT_ARGUMENT
  829.                         + "] ["
  830.                         + DISABLE_DEFLATE_ARGUMENT
  831.                         + "] ["
  832.                         + CLIProductBuilder.DISABLE_PARALLEL_SEND
  833.                         + "] ["
  834.                         + CLIProductBuilder.PARALLEL_SEND_TIMEOUT_ARGUMENT
  835.                         + "300]");

  837.         System.err.println();

  839.         System.err.println("\t" + HELP_ARGUMENT);
  840.         System.err.println("\t\tdisplay this message");
  841.         System.err.println("\t" + FILE_ARGUMENT + "FILE");
  842.         System.err.println("\t\tparse and send one file");
  843.         System.err.println("\t" + POLL_ARGUMENT);
  844.         System.err.println("\t\trun continuously, checking POLLDIR for files");

  846.         System.err.println();

  848.         System.err.println("\t" + PRIVATE_KEY_ARGUMENT + "KEYFILE");
  849.         System.err.println("\t\topenssh private key used to sign products");
  850.         System.err.println("\t" + SIGNATURE_VERSION_ARGUMENT + "VERSION");
  851.         System.err.println("\t\t'v1' is default, 'v2' is other option.");

  853.         System.err.println("\t" + CONNECT_TIMEOUT_ARGUMENT + "TIMEOUT");
  854.         System.err.println("\t\tmilliseconds before timeout while connecting");
  855.         System.err.println("\t\tdefault is \"" + DEFAULT_CONNECT_TIMEOUT
  856.                 + "\"ms");
  857.         System.err.println("\t\t(must appear before " + SERVERS_ARGUMENT + ")");

  859.         System.err.println("\t" + SERVERS_ARGUMENT + "SERVERS");
  860.         System.err
  861.                 .println("\t\tcomma delimited list of servers(host:port) where products are sent");
  862.         System.err.println("\t\tdefault is \"" + SERVERS_DEFAULT + "\"");
  863.         System.err.println("\t" + TEST_ARGUMENT);
  864.         System.err.println("\t\tPrint generated products to console for testing, ignores "
  865.                 + SERVERS_ARGUMENT);
  866.         System.err.println("\t" + PARSER_ARGUMENT + "PARSER");
  867.         System.err.println("\t\tclass that implements "
  868.                 + "gov.usgs.earthquake.quakeml.FileToQuakemlConverter");
  869.         System.err.println("\t\tdefault is \"" + DEFAULT_PARSER_CLASS + "\"");

  871.         System.err.println();

  873.         System.err.println("\t" + POLLDIR_ARGUMENT + "POLLDIR");
  874.         System.err.println("\t\tdirectory to poll for messages");

  876.         System.err.println("\t" + POLL_INTERVAL_ARGUMENT + "INTERVAL");
  877.         System.err.println("\t\tmilliseconds between polling");
  878.         System.err.println("\t\tdefault is \"" + DEFAULT_POLLINTERVAL + "\"ms");

  880.         System.err.println("\t" + STORAGEDIR_ARGUMENT + "STORAGEDIR");
  881.         System.err.println("\t\tdirectory for files that were processed");

  883.         System.err.println("\t" + ERRORDIR_ARGUMENT + "ERRORDIR");
  884.         System.err.println("\t\tdirectory for files that weren't processed");

  886.         System.err.println("\t" + ATTACH_ARGUMENT + "ATTACH");
  887.         System.err
  888.                 .println("\t\tattach a file or directory to one generated product, repeatable");
  889.         System.err
  890.                 .println("\t\tdirectory trees are preserved, each path must be unique");
  891.         System.err
  892.                 .println("\t\tif more than one product is generated, an exception will be thrown");
  893.         System.err.println("\t" + SEND_ORIGINS_WITH_PHASES);
  894.         System.err
  895.                 .println("\t\tWhen a phase-data product is generated, also send an origin product without the phase data");
  896.         System.err.println("\t" + SEND_MECHANISMS_WITH_PHASES);
  897.         System.err
  898.                 .println("\t\tWhen an phase-data product is generated, also send focal mechanism products without the phase data");
  899.         System.err.println();

  901.         System.err.println("\t" + CREATE_INTERNAL_PRODUCTS);
  902.         System.err
  903.                 .println("\t\tuse the product type prefix 'internal-' for all generated products");
  904.         System.err.println("\t" + CREATE_SCENARIO_PRODUCTS);
  905.         System.err
  906.                 .println("\t\tuse the product type suffix '-scenario' for all generated products");

  908.         System.err.println("\t" + BINARY_FORMAT_ARGUMENT);
  909.         System.err.println("\t\tsend to hub using binary format");

  911.         System.err.println("\t" + DISABLE_DEFLATE_ARGUMENT);
  912.         System.err.println("\t\tdisable deflate compression when sending to hubs");

  914.         System.err.println("\t" + CLIProductBuilder.DISABLE_PARALLEL_SEND);
  915.         System.err.println("\t\tsend to servers sequentially");

  917.         System.err.println("\t" + CLIProductBuilder.PARALLEL_SEND_TIMEOUT_ARGUMENT);
  918.         System.err.println("\t\ttimeout for parallel sends in seconds");

  920.         System.exit(1);
  921.     }

  923.     @Override
  924.     public void run() {
  925.         if (!polldir.exists()) {
  926.             polldir.mkdirs();
  927.         }
  928.         while (!Thread.currentThread().isInterrupted()) {
  929.             try {
  930.                 Date pollStart = new Date();

  932.                 String[] polldirFiles = polldir.list();
  933.                 if (polldirFiles.length > 0) {
  934.                     LOGGER.fine("Polldir contains " + polldirFiles.length
  935.                             + " files");
  936.                 }

  938.                 for (int i = 0, len = polldirFiles.length; i < len; i++) {
  939.                     File file = new File(polldir, polldirFiles[i]);
  940.                     try {
  941.                         if (pollCarefully) {
  942.                             // wait until file is at least pollInterval ms old,
  943.                             // in case it is still being written
  944.                             long age = new Date().getTime() - file.lastModified();
  945.                             if (age <= pollInterval) {
  946.                                 continue;
  947.                             }
  948.                         }

  950.                         onFile(file);

  952.                         if (storagedir != null) {
  953.                             if (!storagedir.exists()) {
  954.                                 storagedir.mkdirs();
  955.                             }
  956.                             file.renameTo(new File(storagedir, pollStart
  957.                                     .getTime() + "_" + file.getName()));
  958.                         } else {
  959.                             FileUtils.deleteTree(file);
  960.                         }
  961.                     } catch (Exception e) {
  962.                         if (errordir != null) {
  963.                             if (!errordir.exists()) {
  964.                                 errordir.mkdirs();
  965.                             }
  966.                             file.renameTo(new File(errordir, pollStart
  967.                                     .getTime() + "_" + file.getName()));
  968.                         } else {
  969.                             LOGGER.warning("Error processing file "
  970.                                     + file.getName()
  971.                                     + ", and no error directory configured");
  972.                             FileUtils.deleteTree(file);
  973.                         }
  974.                     }
  975.                 }

  977.                 Date pollEnd = new Date();
  978.                 Long pollTime = pollEnd.getTime() - pollStart.getTime();
  979.                 if (pollTime < pollInterval) {
  980.                     Thread.sleep(pollInterval - pollTime);
  981.                 }
  982.             } catch (InterruptedException ie) {
  983.                 // interrupted means shutdown
  984.                 return;
  985.             }
  986.         }

  988.     }

  990. }
Created with JaCoCo 0.8.8.202204050719