AdminSocketServer.java

package gov.usgs.earthquake.distribution;

import gov.usgs.util.DefaultConfigurable;
import gov.usgs.util.SocketAcceptor;
import gov.usgs.util.SocketListenerInterface;
import gov.usgs.util.StreamUtils;

import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Telnet to this socket to get a "command prompt".
 *
 * @author jmfee
 */
public class AdminSocketServer extends DefaultConfigurable implements SocketListenerInterface {

  private static final Logger LOGGER = Logger.getLogger(AdminSocketServer.class.getName());

  /** Variable for default thread pool size */
  private static final int DEFAULT_THREAD_POOL_SIZE = 10;
  /** Variable for default admin port */
  private static final int DEFAULT_ADMIN_PORT = 11111;

  private int port = -1;
  private int threads = -1;
  private SocketAcceptor acceptor = null;

  /** the client this server is providing stats for. */
  private ProductClient client = null;

  /** Initializes socket with default thread pool size and port */
  public AdminSocketServer() {
    this(DEFAULT_ADMIN_PORT, DEFAULT_THREAD_POOL_SIZE, null);
  }

  /**
   * Initializes socket with custom port, threads, and client
   *
   * @param port    Admind port
   * @param threads Thread pool size
   * @param client  Product Client
   */
  public AdminSocketServer(final int port, final int threads, final ProductClient client) {
    this.port = port;
    this.threads = threads;
    this.client = client;
  }

  public void startup() throws Exception {
    // call DefaultNotificationReceiver startup first
    super.startup();

    ServerSocket socket = new ServerSocket(port);
    socket.setReuseAddress(true);
    acceptor = new SocketAcceptor(socket, this, Executors.newFixedThreadPool(threads));
    // start accepting connections via socket
    acceptor.start();
  }

  public void shutdown() throws Exception {
    // stop accepting connections
    try {
      acceptor.stop();
    } finally {
      // shutdown no matter what
      // call DefaultNotificationReceiver shutdown last
      super.shutdown();
    }
  }

  /**
   * Process a line of input.
   *
   * @param line input
   * @param out  write generated output to stream
   * @throws Exception if misconfigured or the client quits.
   */
  protected void processLine(final String line, final OutputStream out) throws Exception {
    if (client == null) {
      throw new Exception("No product client configured");
    }

    String s = line.trim();
    if (s.equals("status")) {
      out.write(getStatus().getBytes());
    } else if (s.startsWith("reprocess")) {
      out.write(("Reprocess not yet supported").getBytes());
      // reprocess(out, s.replace("reprocess", "").split(" "));
    } else if (s.startsWith("search")) {
      out.write(("Search not yet supported").getBytes());
      // search(out, s.replace("search", "").split(" "));
    } else if (s.equals("quit")) {
      throw new Exception("Bye");
    } else {
      out.write(("Help:\n" + "status - show server status\n" + "SOON search [source=SOURCE] [type=TYPE] [code=CODE]\n"
          + "SOON reprocess listener=LISTENER id=PRODUCTID").getBytes());
    }
  }

  private String getStatus() {
    StringBuffer buf = new StringBuffer();
    // receiver queue status
    Iterator<NotificationReceiver> iter = client.getReceivers().iterator();
    while (iter.hasNext()) {
      NotificationReceiver receiver = iter.next();
      if (receiver instanceof DefaultNotificationReceiver) {
        Map<String, Integer> status = ((DefaultNotificationReceiver) receiver).getQueueStatus();
        if (status != null) {
          Iterator<String> queues = status.keySet().iterator();
          while (queues.hasNext()) {
            String queue = queues.next();
            buf.append(queue).append(" = ").append(status.get(queue)).append("\n");
          }
        }
      }
    }

    String status = buf.toString();
    if (status.equals("")) {
      status = "No queues to show";
    }
    return status;
  }

  public void onSocket(Socket socket) {
    LOGGER.info("[" + getName() + "] accepted connection " + socket.toString());

    InputStream in = null;
    OutputStream out = null;

    try {
      in = socket.getInputStream();
      out = socket.getOutputStream();

      BufferedReader br = new BufferedReader(new InputStreamReader(in));
      String line = null;
      while ((line = br.readLine()) != null) {
        processLine(line, out);
      }
    } catch (Exception ex) {
      LOGGER.log(Level.WARNING, "[" + getName() + "] exception while processing socket", ex);
      // tell sender "exception"
      try {
        out.write(("Error receiving product '" + ex.getMessage() + "'").getBytes());
      } catch (Exception ex2) {
        LOGGER.log(Level.WARNING, "[" + getName() + "] unable to notify sender of exception", ex2);
      }
    } finally {
      StreamUtils.closeStream(in);
      StreamUtils.closeStream(out);

      try {
        socket.shutdownInput();
      } catch (Exception e) {
        // ignore
      }
      try {
        socket.shutdownOutput();
      } catch (Exception e) {
        // ignore
      }

      try {
        socket.close();
      } catch (Exception e) {
        // ignore
      }
    }

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

  /** @return port */
  public int getPort() {
    return port;
  }

  /** @param port port number */
  public void setPort(int port) {
    this.port = port;
  }

  /** @return threads */
  public int getThreads() {
    return threads;
  }

  /** @param threads set number of threads */
  public void setThreads(int threads) {
    this.threads = threads;
  }

  /** @return product client */
  public ProductClient getClient() {
    return client;
  }

  /** @param client set product client */
  public void setClient(ProductClient client) {
    this.client = client;
  }

}