engine.java

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

    if (! AgentServer.getTransaction().isPersistent()) {
      NbMaxAgents = Integer.MAX_VALUE;
    }
    mq = new Queue();
 
    isRunning = false;
    canStop = false;
    thread = null;

    needToBeCommited = false;

    restore();
    if (modified) save();
  }

  void init() throws Exception {
    // Before any agent may be used, the environment, including the hash table,
    // must be initialized.
    agents = new Hashtable();
    try {
      // Creates or initializes AgentFactory, then loads and initializes
      // all fixed agents.
      fixedAgentIdList = (Vector) AgentServer.getTransaction().load(getName() + ".fixed");
      if (fixedAgentIdList == null) {
        // It's the first launching of this engine, in other case theres is
        // at least the factory in fixedAgentIdList.
        fixedAgentIdList = new Vector();
        // Creates factory
        AgentFactory factory = new AgentFactory(AgentId.factoryId);
        createAgent(AgentId.factoryId, factory);
        factory.save();
        logmon.log(BasicLevel.WARN, getName() + ", factory created");
      }

      // loads all fixed agents
      for (int i=0; i<fixedAgentIdList.size(); ) {
	try {
          if (logmon.isLoggable(BasicLevel.DEBUG))
            logmon.log(BasicLevel.DEBUG,
                       getName() + ", loads fixed agent" +
                       fixedAgentIdList.elementAt(i));
	  Agent ag = load((AgentId) fixedAgentIdList.elementAt(i));
          i += 1;
	} catch (Exception exc) {
          logmon.log(BasicLevel.ERROR,
                     getName() + ", can't restore fixed agent" + 
                     fixedAgentIdList.elementAt(i), exc);
          fixedAgentIdList.removeElementAt(i);
	}
      }
    } catch (IOException exc) {
      logmon.log(BasicLevel.ERROR, getName() + ", can't initialize");
      throw exc;
    }
    logmon.log(BasicLevel.DEBUG, getName() + ", initialized");
  }

  void terminate() {
    logmon.log(BasicLevel.DEBUG, getName() + ", ends");
    Agent[] ag = new Agent[agents.size()];
    int i = 0;
    for (Enumeration e = agents.elements() ; e.hasMoreElements() ;) {
      ag[i++] = (Agent) e.nextElement();
    }
    for (i--; i>=0; i--) {
      if (logmon.isLoggable(BasicLevel.DEBUG))
        logmon.log(BasicLevel.DEBUG,
                   "Agent" + ag[i].id + " [" + ag[i].name + "] garbaged");
      agents.remove(ag[i].id);
      ag[i].agentFinalize(false);
      ag[i] = null;
    }
  }

  /**
   * Creates and initializes an agent.
   *
   * @param agent	agent object to create
   *
   * @exception Exception
   *	unspecialized exception
   */
  final void createAgent(AgentId id, Agent agent) throws Exception {
    agent.id = id;
    agent.deployed = true;
    agent.agentInitialize(true);
    createAgent(agent);
  }

  /**
   * Creates and initializes an agent.
   *
   * @param agent	agent object to create
   *
   * @exception Exception
   *	unspecialized exception
   */
  final void createAgent(Agent agent) throws Exception {
    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, getName() + ", creates: " + agent);

    if (agent.isFixed()) {
      // Subscribe the agent in pre-loading list.
      addFixedAgentId(agent.getId());
    }
    if (agent.logmon == null)
      agent.logmon = Debug.getLogger(fr.dyade.aaa.agent.Debug.A3Agent +
                                     ".#" + AgentServer.getServerId());
    agent.save();

    // Memorize the agent creation and ...
    now += 1;
    garbage();
    
    agents.put(agent.getId(), agent);
  }

  /**
   * Deletes an agent.
   *
   * @param agent	agent to delete
   */
  void deleteAgent(AgentId from) throws Exception {
    Agent ag;
    try {
      ag = load(from);
      if (logmon.isLoggable(BasicLevel.DEBUG))
        logmon.log(BasicLevel.DEBUG,
                   getName() + ", delete Agent" + ag.id + " [" + ag.name + "]");
      AgentServer.getTransaction().delete(ag.id.toString());
    } catch (UnknownAgentException exc) {
      logmon.log(BasicLevel.ERROR,
                 getName() +
                 ", can't delete unknown Agent" + from);
      throw new Exception("Can't delete unknown Agent" + from);
    } catch (Exception exc) {
      logmon.log(BasicLevel.ERROR,
                 getName() + ", can't delete Agent" + from, exc);
      throw new Exception("Can't delete Agent" + from);
    }
    if (ag.isFixed())
      removeFixedAgentId(ag.id);
    agents.remove(ag.getId());
    ag.agentFinalize(true);
  }

  /**
   *  The <code>garbage</code> method should be called regularly , to swap out
   * from memory all the agents which have not been accessed for a time.
   */
  void garbage() {
    if (agents.size() < (NbMaxAgents + fixedAgentIdList.size()))
      return;

    if (logmon.isLoggable(BasicLevel.INFO))
      logmon.log(BasicLevel.INFO,
                 getName() + ", garbage: " + agents.size() +
                 '/' + NbMaxAgents + '+' + fixedAgentIdList.size() +
                 ' ' + now);
    long deadline = now - NbMaxAgents;
    Agent[] ag = new Agent[agents.size()];
    int i = 0;
    for (Enumeration e = agents.elements() ; e.hasMoreElements() ;) {
      ag[i++] = (Agent) e.nextElement();
    }
    for (i--; i>=0; i--) {
      if ((ag[i].last <= deadline) && (!ag[i].fixed)) {
        if (logmon.isLoggable(BasicLevel.DEBUG))
	  logmon.log(BasicLevel.DEBUG,
                     "Agent" + ag[i].id + " [" + ag[i].name + "] garbaged");
	agents.remove(ag[i].id);
        ag[i].agentFinalize(false);
        ag[i] = null;
      }
    }

    logmon.log(BasicLevel.INFO,
               getName() + ", garbage: " + agents.size());
  }

  /**
   * Removes an <code>AgentId</code> in the <code>fixedAgentIdList</code>
   * <code>Vector</code>.
   *
   * @param id	the <code>AgentId</code> of no more used fixed agent.
   */
  void removeFixedAgentId(AgentId id) throws IOException {
    fixedAgentIdList.removeElement(id);
    AgentServer.getTransaction().save(fixedAgentIdList, getName() + ".fixed");
  }

  /**
   * Adds an <code>AgentId</code> in the <code>fixedAgentIdList</code>
   * <code>Vector</code>.
   *
   * @param id	the <code>AgentId</code> of new fixed agent.
   */
  void addFixedAgentId(AgentId id) throws IOException {   
    fixedAgentIdList.addElement(id);
    AgentServer.getTransaction().save(fixedAgentIdList, getName() + ".fixed");
  }

  /**
   *   Method used for debug and monitoring. It returns an enumeration
   * of all agents loaded in memory.
   */
  AgentId[] getLoadedAgentIdlist() {
    AgentId list[] = new AgentId[agents.size()];
    int i = 0;
    for (Enumeration e = agents.elements(); e.hasMoreElements() ;)
      list[i++] = ((Agent) e.nextElement()).id;
    return list;
  }

  /**
   *  Returns a string representation of the specified agent.
   *
   * @param id	The string representation of the agent's unique identification.
   * @return	A string representation of the specified agent.
   * @see 	Engine#dumpAgent(AgentId)
   */
  public String dumpAgent(String id) throws Exception {
    return dumpAgent(AgentId.fromString(id));
  }

  /**
   *  Returns a string representation of the specified agent. If the agent
   * is not present it is loaded in memory, be careful it is not initialized
   * (agentInitialize) nor cached in agents vector.
   *
   * @param id	The agent's unique identification.
   * @return	A string representation of specified agent.
   */
  public String dumpAgent(AgentId id)
    throws IOException, ClassNotFoundException, Exception {
    Agent ag = (Agent) agents.get(id);
    if (ag == null) {
      ag = Agent.load(id);
      if (ag == null) {
        return id.toString() + " unknown";
      }
    }
    return ag.toString();
  } 

  /**
   *  The <code>load</code> method return the <code>Agent</code> object
   * designed by the <code>AgentId</code> parameter. If the <code>Agent</code>
   * object is not already present in the server memory, it is loaded from
   * the storage.
   *
   *  Be carefull, if the save method can be overloaded to optimize the save
   * processus, the load procedure used by engine is always load.
   *
   * @param	id		The agent identification.
   * @return			The corresponding agent.
   *
   * @exception	IOException
   *	If an I/O error occurs.
   * @exception	ClassNotFoundException
   *	Should never happen (the agent has already been loaded in deploy).
   * @exception	UnknownAgentException
   *	There is no correponding agent on secondary storage.
   * @exception Exception
   *	when executing class specific initialization
   */
  final Agent load(AgentId id)
    throws IOException, ClassNotFoundException, Exception {
    now += 1;

    Agent ag = (Agent) agents.get(id);
    if (ag == null)  {
      ag = reload(id);
      garbage();
    }
    ag.last = now;

    return ag;
  }

  /**
   * The <code>reload</code> method return the <code>Agent</code> object
   * loaded from the storage.
   *
   * @param	id		The agent identification.
   * @return			The corresponding agent.
   *
   * @exception IOException
   *	when accessing the stored image
   * @exception ClassNotFoundException
   *	if the stored image class may not be found
   * @exception Exception
   *	unspecialized exception
   */
  final Agent reload(AgentId id)
    throws IOException, ClassNotFoundException, Exception {
    Agent ag = null;
    if ((ag = Agent.load(id)) != null) {
      try {
        // Set current agent running in order to allow from field fixed
        // for sendTo during agentInitialize (We assume that only Engine
        // use this method).
        agent = ag;
        ag.agentInitialize(false);
      } catch (Throwable exc) {
        agent = null;
        // AF: May be we have to delete the agent or not to allow
        // reaction on it.
        logmon.log(BasicLevel.ERROR,
                   getName() + "Can't initialize Agent" + ag.id +
                   " [" + ag.name + "]",
                   exc);
        throw new Exception(getName() + "Can't initialize Agent" + ag.id);
      }
      if (ag.logmon == null)
        ag.logmon = Debug.getLogger(fr.dyade.aaa.agent.Debug.A3Agent +
                                    ".#" + AgentServer.getServerId());
      agents.put(ag.id, ag);
      if (logmon.isLoggable(BasicLevel.DEBUG))
        logmon.log(BasicLevel.DEBUG,
                   getName() + "Agent" + ag.id + " [" + ag.name + "] loaded");
    } else {
      throw new UnknownAgentException();
    }

    return ag;
  }

  /**
   * Insert a message in the <code>MessageQueue</code>.
   * This method is used during initialisation to restore the component
   * state from persistent storage.
   *
   * @param msg		the message
   */
  public void insert(Message msg) {
    qin.insert(msg);
  }

  /**
   * Validates all messages pushed in queue during transaction session.
   */
  public void validate() {
    qin.validate();
  }

  /**