agentserver.java

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

    try {
      if (args.length == 3)
      cid = (short) Integer.parseInt(args[2]);
    } catch (NumberFormatException exc) {}

    init(sid, path, null, cid);

    configController = new ConfigController();

    return 2;
  }

  public static void reset(boolean force) {
    if (force) {
      synchronized(status) {
        if (status.value != Status.STOPPED) {
          logmon.log(BasicLevel.WARN,
                     getName() + ", force status: " + status.value);
        }
        status.value = Status.STOPPED;
      }
    }
    reset();
  }

  /**
   *  Cleans an AgentServer configuration in order to restart it from
   * persistent storage.
   */
  public static void reset() {
    synchronized(status) {
      if (status.value != Status.STOPPED) {
        logmon.log(BasicLevel.WARN,
                   getName() + ", cannot reset, bad status: " + status.value);
        return;
      }
      status.value = Status.RESETING;
    }

    // Remove all consumers Mbean
    Enumeration e = getConsumers();
    if (e != null) {
      for (; e.hasMoreElements();) {
        MessageConsumer cons = (MessageConsumer) e.nextElement();
        try {
          MXWrapper.unregisterMBean(
            "AgentServer",
            "server=" + getName() + ",cons=" + cons.getName());
        } catch (Exception exc) {
          logmon.log(BasicLevel.DEBUG,
                     getName() + ", jmx failed: " +
                     "server=" + getName() + ",cons=" + cons.getName(), exc);
        }
      }
      consumers = null;
    }

    try {
      MXWrapper.unregisterMBean("AgentServer",
                                "server=" + getName() + ",cons=Transaction");
    } catch (Exception exc) {
      logmon.log(BasicLevel.DEBUG,
                 getName() + ", jmx failed: " +
                 "server=" + getName() + ",cons=Transaction", exc);
    }

    if (transaction != null) transaction.close();
    transaction = null;

    try {
      MXWrapper.unregisterMBean("AgentServer", "server=" + getName());
    } catch (Exception exc) {
      logmon.log(BasicLevel.DEBUG,
                 getName() + " jmx failed: "+ "server=" + getName(), exc);
    }
    
    a3config = null;

    synchronized(status) {
      status.value = Status.INSTALLED;
    }
  }

 /**
   * Initializes this agent server.
   * <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 sid		the server id
   * @param path        the persistency directory.
   * @param loggerFactory the monolog LoggerFactory;
   * 	
   *
   * @exception Exception
   *	unspecialized exception
   */
  public static void init(short sid,
                          String path,
                          LoggerFactory loggerFactory) throws Exception {
    init(sid, path, loggerFactory, NULL_ID);
  }

 /**
   * Initializes this agent server.
   * <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 sid		the server id
   * @param path        the persistency directory.
   * @param loggerFactory the monolog LoggerFactory;
   * @param cid		the cluster id
   * 	
   *
   * @exception Exception
   *	unspecialized exception
   */
  public static void init(short sid,
                          String path,
                          LoggerFactory loggerFactory,
                          short cid) throws Exception {
    name = new StringBuffer("AgentServer#").append(sid).toString();

    if (loggerFactory != null) Debug.setLoggerFactory(loggerFactory);
    logmon = Debug.getLogger(Debug.A3Debug + ".AgentServer.#" + sid);

    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, getName() + ", init()", new Exception());
    else
      logmon.log(BasicLevel.WARN, getName() + ", init()");

    synchronized(status) {
      if (status.value == Status.STOPPED) {
        logmon.log(BasicLevel.DEBUG, getName() + ", reset configuration");
        reset();
      }
      if (status.value != Status.INSTALLED)
        throw new Exception("cannot initialize, bad status: " + status.value);
      status.value = Status.INITIALIZING;
    }

    try {
      serverId = sid; 

      tgroup = new ThreadGroup(getName()) {
          public void uncaughtException(Thread t, Throwable e) {
            if (logmon.isLoggable(BasicLevel.WARN)) {
              logmon.log(BasicLevel.WARN,
                         "Abnormal termination for " +
                         t.getThreadGroup().getName() + "." + t.getName(),
                         e);
            }
          }
        };
   
      //  Try to get transaction type from disk, then initialize the rigth
      // transaction manager and get the configuration.
      File dir = new File(path);
      if (dir.exists() && dir.isDirectory()) {
        File tfc = new File(dir, "TFC");
        if (tfc.exists()) {
          DataInputStream dis = null;
          try {
            dis = new DataInputStream(new FileInputStream(tfc));
            String tname = dis.readUTF();
            Class tclass = Class.forName(tname);
            transaction = (Transaction) tclass.newInstance();
          } catch (Exception exc) {
            logmon.log(BasicLevel.FATAL,
                       getName() + ", can't instanciate transaction manager",
                       exc);
            throw new Exception("Can't instanciate transaction manager");
          } finally {
            if (dis != null) dis.close();
          }
          try {
            transaction.init(path);
          } catch (IOException exc) {
            logmon.log(BasicLevel.FATAL,
                       getName() + ", can't start transaction manager", exc);
            throw new Exception("Can't start transaction manager");
          }
        }
      }

      // Gets static configuration of agent servers from a file. This method
      // fills the object graph configuration in the <code>A3CMLConfig</code>
      // object, then the configure method really initializes the server.
      // There are two steps because the configuration step needs the
      // transaction components to be initialized.
      if (transaction != null) {
        // Try to read the serialiazed configuration (trough transaction)
        try {
          a3config = A3CMLConfig.load();
        } catch (Exception exc) {
          logmon.log(BasicLevel.WARN, getName() + ", config not found");
        }
      }

      if (a3config == null) {
        //  Try to load an initial configuration (serialized or XML), or
        // generates a default one in case of failure.
        try {
          a3config = A3CMLConfig.getConfig(DEFAULT_SER_CFG_FILE);
        } catch (Exception exc) {
          logmon.log(BasicLevel.WARN,
                     getName() + ", serialized a3cmlconfig not found");
        }

        if (a3config == null) {
          // Try to found XML configuration file, then parse it.
          try {
            a3config = A3CML.getXMLConfig();
          } catch (Exception exc) {
            logmon.log(BasicLevel.WARN,
                       getName() + ", XML configuration file not found");
          }
        }

        if (a3config == null) {
          // 3rd, Generate A3CMLConfig base.
          logmon.log(BasicLevel.WARN,
                     "Generate default configuration");
          A3CMLDomain d = new A3CMLDomain(ADMIN_DOMAIN,
                                          "fr.dyade.aaa.agent.SimpleNetwork");
          A3CMLServer s = new A3CMLServer((short) 0, ADMIN_SERVER, "localhost");
          s.networks.addElement(new A3CMLNetwork(ADMIN_DOMAIN, 27300));
          s.services.addElement(new A3CMLService("fr.dyade.aaa.agent.AgentAdmin",null));
          s.services.addElement(new A3CMLService("fr.dyade.aaa.agent.HttpDebug","20080"));
          d.addServer(s);
          a3config = new A3CMLConfig();
          a3config.addDomain(d);
          a3config.addServer(s);
        }
      }

      // if JGroups
      if (cid > NULL_ID) clusterId = cid;

      // set properties
      setProperties(serverId, clusterId);

      // Initializes the JMX Wrapper
      MXWrapper.init();

      if (transaction == null) {
        try {
          String tname = getProperty("Transaction",
                                     "fr.dyade.aaa.util.NTransaction");
          Class tclass = Class.forName(tname);
          transaction = (Transaction) Class.forName(tname).newInstance();
        } catch (Exception exc) {
          logmon.log(BasicLevel.FATAL,
                     getName() + ", can't instanciate transaction manager", exc);
          throw new Exception("Can't instanciate transaction manager");
        }

        try {
          transaction.init(path);
        } catch (IOException exc) {
          logmon.log(BasicLevel.FATAL,
                     getName() + ", can't start transaction manager", exc);
          throw new Exception("Can't start transaction manager");
        }
      }

      try {
        MXWrapper.registerMBean(transaction,
                                "AgentServer",
                                "server=" + getName() + ",cons=Transaction");
      } catch (Exception exc) {
        if (logmon == null)
          logmon = Debug.getLogger(Debug.A3Debug + ".AgentServer");
        logmon.log(BasicLevel.ERROR, getName() + " jmx failed", exc);
      }

      // save A3CMLConfig (May be we can omit it in some case).
      a3config.save();

      try {
        // Initialize AgentId class's variables.
        AgentId.init();
      } catch (ClassNotFoundException exc) {
        logmon.log(BasicLevel.FATAL,
                   getName() + ", can't initialize AgentId", exc);
        throw new Exception("Can't initialize AgentId, bad classpath");
      } catch (IOException exc) {
        logmon.log(BasicLevel.FATAL,
                   getName() + ", can't initialize AgentId", exc);
        throw new Exception("Can't initialize AgentId, storage problems");
      }

      try {
        // Configure the agent server.
        configure();
      } catch (Exception exc) {
        logmon.log(BasicLevel.FATAL, getName() + ", can't configure", exc);
        throw new Exception("Can't configure server");
      }

      try {
        // then restores all messages.
        String[] list = transaction.getList("@");
        for (int i=0; i<list.length; i++) {
          Message msg = Message.load(list[i]);

          if (msg.getSource() == serverId) {
            // The update has been locally generated, the message is ready to
            // deliver to its consumer (Engine or Network component). So we have
            // to insert it in the queue of this consumer.
            try {
              getServerDesc(msg.getDest()).domain.insert(msg);
            } catch (NullPointerException exc) {
              logmon.log(BasicLevel.ERROR,
                         getName() + ", discard message to unknown server id#" +
                         msg.getDest());
              msg.delete();
              msg.free();
              continue;
            } catch (ArrayIndexOutOfBoundsException exc) {
              logmon.log(BasicLevel.ERROR,
                         getName() + ", discard message to unknown server id#" +
                         msg.getDest());
              msg.delete();
              msg.free();
              continue;
            }
          } else {
            logmon.log(BasicLevel.ERROR,
                       getName() + ", discard undelivered message from server id#" +
                       msg.getDest());
            msg.delete();
            continue;
          }
        }
      } catch (ClassNotFoundException exc) {
        logmon.log(BasicLevel.FATAL,
                   getName() + ", can't restore messages", exc);
        throw new Exception("Can't restore messages, bad classpath");
      } catch (IOException exc) {
        logmon.log(BasicLevel.FATAL,
                   getName() + ", can't restore messages", exc);
        throw new Exception("Can't restore messages, storage problems");
      }

      // initializes channel before initializing fixed agents
      Channel.newInstance();    

      try {
        //  Initialize services.
        ServiceManager.init();

        logmon.log(BasicLevel.DEBUG,
                   getName() + ", ServiceManager initialized");

        /* Actually get Services from A3CML configuration file. */
        ServiceDesc services[] = AgentServer.getServices();
        if (services != null) {
          for (int i = 0; i < services.length; i ++) {
            ServiceManager.register(services[i].getClassName(),
                                    services[i].getArguments());
          }
        }
        ServiceManager.save();
      } catch (Exception exc) {
        logmon.log(BasicLevel.FATAL,
                   getName() + ", can't initialize services", exc);
        throw new Exception("Can't initialize services");