SearchServerSocket.java

  1. package gov.usgs.earthquake.indexer;

  3. import java.io.BufferedInputStream;
  4. import java.io.BufferedOutputStream;
  5. import java.io.InputStream;
  6. import java.net.ServerSocket;
  7. import java.net.Socket;
  8. import java.util.concurrent.Executors;
  9. import java.util.logging.Level;
  10. import java.util.logging.Logger;
  11. import java.util.zip.DeflaterOutputStream;
  12. import java.util.zip.InflaterInputStream;

  14. import gov.usgs.earthquake.distribution.ConfigurationException;
  15. import gov.usgs.util.Config;
  16. import gov.usgs.util.DefaultConfigurable;
  17. import gov.usgs.util.SocketAcceptor;
  18. import gov.usgs.util.SocketListenerInterface;
  19. import gov.usgs.util.StreamUtils;

  21. /**
  22.  * Server side of socket search interface.
  23.  */
  24. public class SearchServerSocket extends DefaultConfigurable implements
  25.         SocketListenerInterface {

  27.     /** Logging object. */
  28.     private static final Logger LOGGER = Logger
  29.             .getLogger(SearchServerSocket.class.getName());

  31.     /** The configuration property used for listen port. */
  32.     public static final String SEARCH_PORT_PROPERTY = "port";

  34.     /** The default listen port, as a string. */
  35.     public static final String DEFAULT_SEARCH_PORT = "11236";

  37.     /** The configuration property used for listen thread count. */
  38.     public static final String THREAD_POOL_SIZE_PROPERTY = "threads";

  40.     /** The default number of threads, as a string. */
  41.     public static final String DEFAULT_THREAD_POOL_SIZE = "10";

  43.     /** The configuration property used to reference a ProductIndex. */
  44.     public static final String PRODUCT_INDEXER_PROPERTY = "indexer";

  46.     /** The configuration property used to reference a URLProductStorage. */
  47.     public static final String PRODUCT_STORAGE_PROPERTY = "storage";

  49.     /** The port to bind. */
  50.     private int port = -1;

  52.     /** The number of threads to use. */
  53.     private int threads = -1;

  55.     /** The server socket accept thread. */
  56.     private SocketAcceptor acceptor;

  58.     /** The indexer that will be searched. */
  59.     private Indexer indexer;

  61.     /**
  62.      * Construct a new SearchServerSocket using defaults.
  63.      */
  64.     public SearchServerSocket() {
  65.         this.port = Integer.parseInt(DEFAULT_SEARCH_PORT);
  66.         this.threads = Integer.parseInt(DEFAULT_THREAD_POOL_SIZE);
  67.     }

  69.     /**
  70.      * Method to perform search.
  71.      *
  72.      * Calls Indexer.search(SearchRequest). Simplifies testing.
  73.      *
  74.      * @param request
  75.      *            the search to execute.
  76.      * @return the search response.
  77.      * @throws Exception if error occurs
  78.      */
  79.     protected SearchResponse search(final SearchRequest request)
  80.             throws Exception {
  81.         return indexer.search(request);
  82.     }

  84.     /**
  85.      * This method is called each time a SearchSocket connects.
  86.      */
  87.     @Override
  88.     public void onSocket(Socket socket) {
  89.         LOGGER.info("[" + getName() + "] accepted search connection "
  90.                 + socket.toString());

  92.         InputStream in = null;
  93.         DeflaterOutputStream out = null;

  95.         try {
  96.             in = socket.getInputStream();
  97.             in = new InflaterInputStream(new BufferedInputStream(
  98.                     new StreamUtils.UnclosableInputStream(in)));
  99.             // read request
  100.             SearchRequest request = SearchXML
  101.                     .parseRequest(new StreamUtils.UnclosableInputStream(in));

  103.             // do search
  104.             SearchResponse response = this.search(request);

  106.             // send response
  107.             out = new DeflaterOutputStream(new BufferedOutputStream(
  108.                     socket.getOutputStream()));
  109.             SearchXML.toXML(response, new StreamUtils.UnclosableOutputStream(
  110.                     out));

  112.             // finish compression
  113.             out.finish();
  114.             out.flush();
  115.         } catch (Exception ex) {
  116.             LOGGER.log(Level.WARNING, "[" + getName()
  117.                     + "] exception while processing search", ex);
  118.         } finally {
  119.             StreamUtils.closeStream(in);
  120.             StreamUtils.closeStream(out);

  122.             try {
  123.                 socket.shutdownInput();
  124.             } catch (Exception e) {
  125.                 // ignore
  126.             }
  127.             try {
  128.                 socket.shutdownOutput();
  129.             } catch (Exception e) {
  130.                 // ignore
  131.             }

  133.             try {
  134.                 socket.close();
  135.             } catch (Exception e) {
  136.                 // ignore
  137.             }
  138.         }

  140.         LOGGER.info("[" + getName() + "] closed search connection "
  141.                 + socket.toString());
  142.     }

  144.     @Override
  145.     public void configure(Config config) throws Exception {
  146.         port = Integer.parseInt(config.getProperty(SEARCH_PORT_PROPERTY,
  147.                 DEFAULT_SEARCH_PORT));
  148.         LOGGER.config("[" + getName() + "] search port is " + port);

  150.         threads = Integer.parseInt(config.getProperty(
  151.                 THREAD_POOL_SIZE_PROPERTY, DEFAULT_THREAD_POOL_SIZE));
  152.         LOGGER.config("[" + getName() + "] number of threads is " + threads);

  154.         String indexerName = config.getProperty(PRODUCT_INDEXER_PROPERTY);
  155.         if (indexerName == null) {
  156.             throw new ConfigurationException("[" + getName() + "] '"
  157.                     + PRODUCT_INDEXER_PROPERTY
  158.                     + "' is a required configuration property");
  159.         }
  160.         LOGGER.config("[" + getName() + "] loading indexer '" + indexerName
  161.                 + "'");
  162.         indexer = (Indexer) Config.getConfig().getObject(indexerName);
  163.         if (indexer == null) {
  164.             throw new ConfigurationException("[" + getName() + "] indexer '"
  165.                     + indexerName + "' is not configured properly");
  166.         }
  167.     }

  169.     @Override
  170.     public void shutdown() throws Exception {
  171.         // stop accepting connections
  172.         acceptor.stop();
  173.         acceptor = null;
  174.     }

  176.     @Override
  177.     public void startup() throws Exception {
  178.         ServerSocket socket = new ServerSocket(port);
  179.         socket.setReuseAddress(true);
  180.         acceptor = new SocketAcceptor(socket, this,
  181.                 Executors.newFixedThreadPool(threads));
  182.         // start accepting connections via socket
  183.         acceptor.start();
  184.     }

  186.     /** @return int port */
  187.     public int getPort() {
  188.         return port;
  189.     }

  191.     /** @param port int to set */
  192.     public void setPort(int port) {
  193.         this.port = port;
  194.     }

  196.     /** @return int threads */
  197.     public int getThreads() {
  198.         return threads;
  199.     }

  201.     /** @param threads into to set */
  202.     public void setThreads(int threads) {
  203.         this.threads = threads;
  204.     }

  206.     /** @return indexer */
  207.     public Indexer getIndexer() {
  208.         return indexer;
  209.     }

  211.     /** @param indexer to set */
  212.     public void setIndex(Indexer indexer) {
  213.         this.indexer = indexer;
  214.     }

  216. }
Created with JaCoCo 0.8.8.202204050719