agentserver.java

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

    servers.put(desc);
  }

  static ServerDesc removeServerDesc(short sid) throws Exception {
    return servers.remove(sid);
  }

  public static Enumeration elementsServerDesc() {
    return servers.elements();
  }

  public static Enumeration getServersIds() {
    return servers.keys();
  }

  /**
   * Gets the number of server known on the current server.
   *
   * @return	the number of server.
   */
  final static int getServerNb() {
    return servers.size();
  }

  /**
   * Gets the characteristics of the corresponding server.
   *
   * @param sid	agent server id.
   * @return	the server's descriptor.
   */
  final static ServerDesc getServerDesc(short sid) throws UnknownServerException {
    ServerDesc serverDesc = servers.get(sid);
    if (serverDesc == null)
      throw new UnknownServerException("Unknow server id. #" + sid);
    return serverDesc;
  }

  /**
   * Gets the message consumer for the corresponding server.
   *
   * @param sid	agent server id.
   * @return	the corresponding message consumer.
   */
  final static MessageConsumer getConsumer(short sid) throws UnknownServerException {
    return getServerDesc(sid).domain;
  }

  /**
   * Get the host name of an agent server.
   *
   * @param sid		agent server id
   * @return		server host name as declared in configuration file
   */
  public final static String getHostname(short sid) throws UnknownServerException {
    return getServerDesc(sid).getHostname();
  }

  /**
   * Get the description of all services of the current agent server.
   *
   * @return		server host name as declared in configuration file
   */
  final static ServiceDesc[] getServices() throws UnknownServerException {
    return getServerDesc(getServerId()).services;
  }

  /**
   * Get the argument strings for a particular service.
   * The information provides from the A3 configuration file, so it's
   * only available if this file contains service's informations for all
   * nodes.
   *
   * @see A3CMLConfig#getServiceArgs(short,String)
   *
   * @param sid		agent server id
   * @param classname	the service class name
   * @return		the arguments as declared in configuration file
   * @exception	UnknownServerException
   *	The specified server does not exist.
   * @exception UnknownServiceException
   *	The specified service is not declared on this server. 
   * @exception Exception
   *	Probably there is no configuration defined.
   */
  public final static
  String getServiceArgs(short sid,
			String classname) throws Exception {
    return getConfig().getServiceArgs(sid, classname);
  }

  /**
   * Get the argument strings for a particular service running on a server
   * identified by its host.
   * The information provides from the A3 configuration file, so it's
   * only available if this file contains service's informations for all
   * nodes.
   *
   * @see A3CMLConfig#getServiceArgs(String, String)
   *
   * @param hostname	hostname
   * @param classname	the service class name
   * @return		the arguments as declared in configuration file
   * @exception UnknownServiceException
   *	The specified service is not declared on this server. 
   * @exception Exception
   *	Probably there is no configuration defined.
   */
  public final static
  String getServiceArgs(String hostname,
			String classname) throws Exception {
    return getConfig().getServiceArgsHost(hostname, classname);
  }

  /**
   *  The second step of initialization. It needs the Transaction component
   * be up, then it initializes all <code>AgentServer</code> structures from
   * the <code>A3CMLConfig</code> ones. In particular the servers array is
   * initialized.
   */
  private static void configure() throws Exception {
    A3CMLServer root = getConfig().getServer(serverId, clusterId);
    //Allocates the temporary descriptors hashtable for each server.
    servers = new ServersHT();
    // Initialized the descriptor of current server in order to permit
    // Channel and Engine initialization.
    ServerDesc local = new ServerDesc(root.sid, root.name, root.hostname, -1);
    servers.put(local);

    // Parse configuration in order to fix route related to the
    // current server
    getConfig().configure(root);

    // Creates all the local MessageConsumer.
    createConsumers(root);
    
    for (Enumeration s = getConfig().servers.elements();
         s.hasMoreElements();) {
      A3CMLServer server = (A3CMLServer) s.nextElement();
      if (server.sid == root.sid) continue;

      ServerDesc desc = createServerDesc(server);
      addServerDesc(desc);
    }

    // for clusters
    for (Enumeration c = getConfig().clusters.elements();
         c.hasMoreElements();) {
      A3CMLCluster cluster = (A3CMLCluster) c.nextElement();

      for (Enumeration s = cluster.servers.elements();
           s.hasMoreElements();) {
        A3CMLServer server = (A3CMLServer) s.nextElement();
        if (server.sid == root.sid) continue;

        ServerDesc desc = servers.get(server.sid);
        if (desc == null) {
          desc = createServerDesc(server);
          addServerDesc(desc);
        } else {
          desc.addSockAddr(server.hostname, server.port);
        }
      }
    }

    initServices(root, local);
    local.domain = engine;

//     if (logmon.isLoggable(BasicLevel.DEBUG)) {
//       for (int i=0; i<servers.length; i++) {
//         logmon.log(BasicLevel.DEBUG,
//                    getName() + ", servers[" + i + "]=(" + 
//                    "sid=" + servers[i].sid + 
//                    ", name=" + servers[i].name + 
//                    ", gateway=" + servers[i].gateway + 
//                    ", domain=" + servers[i].domain + ")");
//       }
//     }
    
    return;
  }

  private static void createConsumers(A3CMLServer root) throws Exception {
    consumers = new Hashtable();

    // Creates the local MessageConsumer: the Engine.
    engine = Engine.newInstance();
    addConsumer("local", engine);

    // if JGroups
    if (clusterId > NULL_ID) {
      jgroups = new JGroups();
      if (engine instanceof HAEngine) {
        jgroups.setEngine((HAEngine) engine);
        ((HAEngine) engine).setJGroups(jgroups);
      } else
        logmon.log(BasicLevel.ERROR, getName() + ", createConsumers(" + root + ")\n" +
                   "engine [" + engine + "] is not a HAEngine");
    }

    // Search alls directly accessible domains.
    for (Enumeration n = root.networks.elements();
	 n.hasMoreElements();) {
      A3CMLNetwork network = (A3CMLNetwork) n.nextElement();

      A3CMLDomain domain = getConfig().getDomain(network.domain);
      // Creates the corresponding MessageConsumer.
      try {
        Network consumer = (Network) Class.forName(domain.network).newInstance();
        // Initializes it with domain description. Be careful, this array
        // is kept in consumer, don't reuse it!!
        consumer.init(domain.name, network.port, domain.getServersId());
        if (consumer instanceof SimpleNetwork &&
            jgroups != null) {//NTA modify to SimpleHANetwork
          ((SimpleNetwork) consumer).setJGroups(jgroups);
          jgroups.setNetWork((SimpleNetwork) consumer);
        }
//         domain.consumer = consumer;
        addConsumer(network.domain, consumer);
      } catch (ClassNotFoundException exc) {
        throw exc;
      } catch (InstantiationException exc) {
        throw exc;
      } catch (IllegalAccessException exc) {
        throw exc;
      }
    }
  }

  static void initServerDesc(ServerDesc desc,
                             A3CMLServer server) throws Exception {
    desc.gateway = server.gateway;
    // For each server set the gateway to the real next destination of
    // messages; if the server is directly accessible: itself.
    if ((desc.gateway == -1) || (desc.gateway == server.sid)) {
      desc.gateway = server.sid;
      desc.updateSockAddr(desc.getHostname(), server.port);   
      A3CMLServer current = getConfig().getServer(getServerId(),getClusterId());
      if (current.containsNat(server.sid)) {
        A3CMLNat nat = current.getNat(server.sid);
        desc.updateSockAddr(nat.host, nat.port);
        if (logmon.isLoggable(BasicLevel.DEBUG))
          logmon.log(BasicLevel.DEBUG, getName() + " : NAT sDesc = " + desc);
      }
    }
    desc.domain = getConsumer(server.domain);
  }   

  private static ServerDesc
      createServerDesc(A3CMLServer server) throws Exception {
    if (! server.visited)
      throw new Exception(server + " inaccessible");
    
    ServerDesc desc = new ServerDesc(server.sid, 
                                     server.name, 
                                     server.hostname,
                                     -1);

    initServerDesc(desc, server);
    initServices(server, desc);

    return desc;
  }

  private static void initServices(A3CMLServer server, ServerDesc desc) {
    if (server.services != null) {
      ServiceDesc services[]  = new ServiceDesc[server.services.size()];
      int idx = 0;
      for (Enumeration x = server.services.elements();
           x.hasMoreElements();) {
        A3CMLService service = (A3CMLService) x.nextElement();
        services[idx++] = new ServiceDesc(service.classname, service.args);
      }
      desc.services = services;
    }
  }
  
  private static void setProperties(short sid, short cid) throws Exception {
    if (a3config == null) return;

    // add global properties
    if (a3config.properties != null) {
      for (Enumeration e = a3config.properties.elements(); e.hasMoreElements();) {
        A3CMLProperty p = (A3CMLProperty) e.nextElement();
        System.getProperties().put(p.name,p.value);
      }
    }

    A3CMLServer server = null;
    if (cid != NULL_ID) {
      A3CMLCluster cluster = null;
      cluster = a3config.getCluster(sid);

      // add cluster properties
      if (cluster != null 
          && cluster.properties != null 
          && cluster.properties.size() > 0) {
        Enumeration e = cluster.properties.elements();
        do {
          A3CMLProperty p = (A3CMLProperty) e.nextElement();
          System.getProperties().put(p.name,p.value);
        } while (e.hasMoreElements());
      }
      server = cluster.getServer(cid);
    } else {
      server = a3config.getServer(sid);
    }

    // add server properties
    if (server != null && server.properties != null) {
      Enumeration e = server.properties.elements();
      do {
        A3CMLProperty p = (A3CMLProperty) e.nextElement();
        System.getProperties().put(p.name,p.value);
      } while (e.hasMoreElements());
    }
  }
  
  static class Status {
    public static final int INSTALLED = 0;
    public static final int INITIALIZING = 0x1;
    public static final int INITIALIZED = 0x2;
    public static final int STARTING = 0x3;
    public static final int STARTED = 0x4;
    public static final int STOPPING = 0x5;
    public static final int STOPPED = 0x6;
    public static final int RESETING = 0x7;

    private int value = INSTALLED;

    public static String[] info = {"installed",
                                   "initializing", "initialized",
                                   "starting", "started",
                                   "stopping", "stopped",
                                   "reseting"};
  }

  static Status status = new Status();

  public static int getStatus() {
    return status.value;
  }

  public static String getStatusInfo() {
    return Status.info[status.value];
  }

  /**
   * Parses agent server arguments, then initializes this agent server. The
   * <code>start</code> function is then called to start this agent server
   * execution. Between the <code>init</code> and </code>start</code> calls,
   * agents may be created and deployed, and notifications may be sent using
   * the <code>Channel</code> <code>sendTo</code> function.
   *
   * @param args	lauching arguments, the first one is the server id
   *			and the second one the persistency directory.
   * @return		number of arguments consumed in args
   *
   * @exception Exception
   *	unspecialized exception
   */
  public static int init(String args[]) throws Exception {
    if (args.length < 2)
      throw new Exception("usage: java <main> sid storage");
    short sid = NULL_ID;
    try {
      sid = (short) Integer.parseInt(args[0]);
    } catch (NumberFormatException exc) {
      throw new Exception("usage: java <main> sid storage");
    }
    String path = args[1];
    short cid = NULL_ID;