adminproxy.java

一个类似于openJMS分布在ObjectWeb之下的JMS消息中间件。

/*
 * Copyright (C) 2001 - 2004 ScalAgent Distributed Technologies
 * Copyright (C) 1996 - 2000 BULL
 * Copyright (C) 1996 - 2000 INRIA
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA.
 */
package fr.dyade.aaa.agent;

import java.io.*;
import java.net.*;
import java.text.*;
import java.util.*;
import java.lang.reflect.*;

import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.util.Daemon;
import fr.dyade.aaa.agent.conf.A3CML;
import fr.dyade.aaa.agent.conf.A3CMLConfig;

/**
 * A <code>AdminProxy</code> service provides an interface to access
 * to administration functions in running agent servers.
 * <p>
 * The <code>AdminProxy</code> service can be configured by the way of
 * service argument:
 * the TCP port number, by default this port is 8091.
 * the number of monitor needed to handled requests.
 */
public class AdminProxy {

  static AdminProxy proxy = null;

  public static boolean debug = true;

  /** Property that define the TCP listen port */
  public final static String LISTENPORT = "fr.dyade.aaa.agent.AdminProxy.port";
  /** The TCP listen port */
  private static int port = 8091;
  /** The number of monitors.*/
  private static int nbm;
  /** Property that define the number of monitor */
  public final static String NBMONITOR = "fr.dyade.aaa.agent.AdminProxy.nbm";
  /** Hashtable that contain all <code>Process</code> of running AgentServer */
//   Hashtable ASP = null;

  AdminMonitor monitors[] = null;
  ServerSocket listen = null;

  static Logger xlogmon = null;

  /**
   * Initializes the package as a well known service.
   * <p>
   * Creates a <code>AdminProxy</code> proxy listen on .
   *
   * @param args	parameters from the configuration file
   * @param firstTime	<code>true</code> when service starts anew
   */
  public static void init(String args, boolean firstTime) throws Exception {
    try {
      if (args.length()!=0) {
	port = Integer.parseInt(args);
      } else {
        port = Integer.parseInt(AgentServer.getProperty(LISTENPORT, "8091"));
      }
    } catch (NumberFormatException exc) {
      port = 8091;
    }

    try {
      nbm = Integer.parseInt(AgentServer.getProperty(NBMONITOR, "1"));
    } catch (NumberFormatException exc) {
      nbm = 1;
    }

    // Get the logging monitor from current server MonologMonitorFactory
    xlogmon = Debug.getLogger(Debug.A3Service + ".AdminProxy" +
                               ".#" + AgentServer.getServerId());

    if (proxy != null) {
      xlogmon.log(BasicLevel.ERROR,
                  "AdminProxy#" + AgentServer.getServerId() +
                  ": already initialized.");
      throw new Exception("AdminProxy" + ".#" + AgentServer.getServerId() +
                          ": already initialized.");
    }

    try {
      proxy = new AdminProxy();
    } catch (IOException exc) {
      xlogmon.log(BasicLevel.ERROR,
                  "AdminProxy#" + AgentServer.getServerId() +
                  ", can't get listen port", exc);
      throw exc;
    }
    start();
  }

  /**
   * Creates an AdminProxy service.
   *
   * @param port  TCP listen port of this proxy
   */
  private AdminProxy() throws IOException {
    for (int i=0; ; i++) {
      try {
        listen = new ServerSocket(port);
        break;
      } catch (BindException exc) {
        if (i > 5) throw exc;
        try {
	  // Wait ~15s: n*(n+1)*500 ms with n=5
          Thread.sleep(i * 500);
        } catch (InterruptedException e) {}
      }
    }

//     ASP = new Hashtable();

    monitors = new AdminMonitor[nbm];
    for (int i=0; i<monitors.length; i++) {
      monitors[i] = new AdminMonitor("AdminProxy#" +
                                     AgentServer.getServerId() + '.' + i);
    }
  }

  public static void start() {
    for (int i=0; i<proxy.monitors.length; i++) {
      proxy.monitors[i].start();
    }
  }

  public static void stopService() {
    for (int i=0; i<proxy.monitors.length; i++) {
      if (proxy.monitors[i] != null) proxy.monitors[i].stop();
      proxy.monitors[i] = null;
    }
    proxy = null;
  }

  static final String HELP = "help";
  static final String NONE = "";

  // Server's administration commands
  public static final String STOP_SERVER = "halt";
  public static final String CRASH_SERVER = "crash";
  public static final String PING = "ping";
  public static final String CONFIG = "config";

  // Environment control
  static final String SET_VARIABLE = "set";
  static final String GET_VARIABLE = "get";

  // JVM's monitoring and control
  static final String GC = "gc";
  static final String THREADS = "threads";

  // Consumer's administration commands
  static final String LIST_MCONS = "consumers";
  static final String START_MCONS = "start";
  static final String STOP_MCONS = "stop";

  // Service's administration commands
  static final String LIST_SERVICE = "services";
  static final String ADD_SERVICE = "add";
  static final String REMOVE_SERVICE = "remove";

  // Debug's tool
  static final String DUMP = "dump";

  // update traces configuration
  public static final String LOG = "log";

  /**
   * Provides a string image for this object.
   */
  public String toString() {
    StringBuffer strBuf = new StringBuffer();

    strBuf.append("(").append(super.toString());
    strBuf.append(",port=").append(port);
    strBuf.append(",monitors=[");
    for (int i=0; i<monitors.length; i++) {
      strBuf.append(monitors[i].toString()).append(",");
    }
    strBuf.append("]");
    strBuf.append(")");

    return strBuf.toString();
  }

  class AdminMonitor extends Daemon {
    Socket socket = null;
    BufferedReader reader = null;
    PrintWriter writer = null;

    /**
     * Constructor.
     */
    protected AdminMonitor(String name) {
      // Get the logging monitor from AdminProxy (overload Daemon setup)
      super(name, AdminProxy.xlogmon);
      this.setThreadGroup(AgentServer.getThreadGroup());
    }

    /**
     * Provides a string image for this object.
     *
     * @return	printable image of this object
     */
    public String toString() {
      return "(" + super.toString() +
	",socket=" + socket + ")";
    }

    public void run() {
      try {
        while (running) {
          canStop = true;
          try {
            logmon.log(BasicLevel.DEBUG, getName() + ", waiting: " + listen);
            socket = listen.accept();
            logmon.log(BasicLevel.DEBUG, getName() + ", receiving.");
            canStop = false;
          } catch (IOException exc) {
            if (running)
              logmon.log(BasicLevel.ERROR,
                         getName() + ", error during accept", exc);
          }

          if (! running) break;

          try {
            // Get the streams
            reader = new BufferedReader(
              new InputStreamReader(socket.getInputStream()));
            writer = new PrintWriter(socket.getOutputStream(), true);
	  
            // Reads then parses the request
            doRequest(reader.readLine());

            writer.flush();
          } catch (Exception exc) {
            logmon.log(BasicLevel.ERROR,
                       getName() + ", error during connection", exc);
          } finally {
            // Closes the connection
            try {
              reader.close();
            } catch (Exception exc) {}
            reader = null;
            try {
              writer.close();
            } catch (Exception exc) {}
            writer = null;
            try {
              socket.close();
            } catch (Exception exc) {}
            socket = null;
          }
        }
      } finally {
        logmon.log(BasicLevel.DEBUG, getName() + ", finishing.");
        finish();
      }
    }

    protected void close() {
      try {
        logmon.log(BasicLevel.DEBUG, getName() + ", closing: " + listen);
	listen.close();
      } catch (Exception exc) {}
      listen = null;
    }

    protected void shutdown() {
      logmon.log(BasicLevel.DEBUG, getName() + ", close(): ");
      close();
    }

    public void doRequest(String request) {
      String cmd = null;

      logmon.log(BasicLevel.DEBUG, getName() + ", request=" + request);

      try {
	// Tokenizes the request to parse it.
	StringTokenizer st = new StringTokenizer(request);

	cmd = st.nextToken();
	if (cmd.equals(STOP_SERVER)) {
          // Stop the AgentServer
	  AgentServer.stop(false);
          logmon.log(BasicLevel.WARN, getName() + ", bye.");
	} else if (cmd.equals(CRASH_SERVER)) {
          // Kill the AgentServer
          logmon.log(BasicLevel.WARN, getName() + ", crash!");