agent.java

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

  private void initState(String name, boolean fixed, AgentId id) {
    if (name == null)
      this.name = nullName;
    else
      this.name = name;
    this.fixed = fixed;
    this.id = id;
    // Get the logging monitor from current server MonologLoggerFactory
    this.logmon = Debug.getLogger(getLogTopic());
  }

  /**
   * Constructor used to build Well Known Services agents.
   * <p>
   * System agents are created from the <code>agent</code> package.
   * WKS agents are similar to system agents, except that they may be
   * defined in separate packages, and they do not necessarily exist on all
   * agent servers. Their creation is controlled from the configuration file
   * of the agent server.<p>
   * This constructor takes the agent id as a parameter instead of building it.
   * Since the constructor has been made public, the consistency of agent ids
   * allocation must be enforced. This is done by the constructor checking
   * that the id stamp is comprised in the <code>AgentId.MinWKSIdStamp</code>
   * - <code>AgentId.MaxWKSIdStamp</code> interval.
   *
   * @param name	symbolic name
   * @param fixed	if <code>true</code> agent is pinned in memory
   * @param stamp	well known stamp
   */
  public Agent(String name, boolean fixed, int stamp) {
    if (stamp < AgentId.MinWKSIdStamp ||
	stamp > AgentId.MaxWKSIdStamp) {
      logmon = Debug.getLogger(fr.dyade.aaa.agent.Debug.A3Agent +
                               ".#" + AgentServer.getServerId());
      logmon.log(BasicLevel.ERROR,
                 AgentServer.getName() +
                 ", well known service stamp out of range: " + stamp);
      throw new IllegalArgumentException(
	"Well known service stamp out of range: " + stamp);
    }
    AgentId id = new AgentId(AgentServer.getServerId(),
                             AgentServer.getServerId(),
                             stamp);
    initState(name, fixed, id);
  }

  /**
   * Determines if the current <code>Agent</code> has already been deployed.
   */
  transient boolean deployed = false;

  /**
   * Returns if the currently <code>Agent</code> has already been deployed.
   */
  public boolean isDeployed() {
    return deployed;
  }

  /**
   * Deploys a new <i>agent</i>.
   * It works by sending a notification to a special agent, of class Factory,
   * running on the target agent server. The notification asks for a remote
   * creation of the agent. This solution presents the advantage of reusing
   * the standard communication mechanisms of the agent machine.<p>
   * The whole process involves then the following steps:
   * <ul>
   * <li>serializing the object state,
   * <li>building an <code>AgentCreateRequest</code> notification with the
   *     resulting bytes stream,
   * <li>sending it to the target Factory agent.
   * </ul>
   * In reaction, the factory agent builds the agent in the target server
   * from the serialized image, and saves it into operational storage.
   *
   * @exception IOException
   *	unspecialized exception
   */
  public final void deploy() throws IOException {
    deploy(null);
  }

  /**
   * Deploys a new <i>agent</i>.
   * It works as <a href="#deploy()">deploy()</a> method above; after the
   * agent creation, the Factory agent sends an <code>AgentCreateReply</code>
   * notification.
   *
   * @param reply	agent to reply to
   * @exception IOException
   *	unspecialized exception
   */
  public final void deploy(AgentId reply) throws IOException {
    if ((id == null) || id.isNullId()) {
      logmon.log(BasicLevel.ERROR,
                 AgentServer.getName() +
                 ", can't deploy " + this.toString() + ", id is null");
      throw new IOException("Can't deploy agent, id is null");
    }
    if (deployed) {
      logmon.log(BasicLevel.ERROR,
                 AgentServer.getName() +
                 ", can't deploy " + this.toString() + ", already deployed");
      throw new IOException("Can't deploy agent, already deployed");
    }

    //  If we use sendTo agent's method the from field is the agent id, and
    // on reception the from node (from.to) can be false.
    Channel.sendTo(AgentId.factoryId(id.getTo()),
		   new AgentCreateRequest(this, reply));
    deployed = true;

    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG, this.toString() + " deployed");
  }

  /**
   * Returns a string representation of this agent, including the agent's
   * class, name, global identication, and fixed property.
   *
   * @return	A string representation of this agent. 
   */
  public String toString() {
    StringBuffer strbuf = new StringBuffer();

    strbuf.append('(').append(super.toString());
    strbuf.append(",name=").append(name);
    strbuf.append(",id=").append(id.toString());
    strbuf.append(",fixed=").append(fixed);
    strbuf.append(')');

    return strbuf.toString();
  }

  /**
   * Returns String format of the global unique identifier of the agent.
   *
   * @return the global unique identifier of the agent.
   */
  public final String getAgentId() {
    return id.toString();
  }

  /**
   * Returns the global unique identifier of the agent. Each agent is
   * identified by a unique identifier allowing the agent to be found.
   * The identifiers format is detailed in <a href="AgentId.html">AgentId</a>
   * class.
   *
   * @return the global unique identifier of the agent.
   */
  public final AgentId getId() {
    return id;
  }

  /** 
   * Tests if the agent is pinned in memory.
   *
   * @return true if this agent is a pinned in memory; false otherwise.
   */
  public final  boolean isFixed() {
    return fixed;
  }

  /**
   * Gives this agent an opportunity to initialize after having been deployed,
   * and each time it is loaded into memory.
   * <p>
   * This function is first called by the factory agent, just after it deploys
   * the agent.
   * <p>
   * This function is used by agents with a <code>fixed</code> field set to
   * <code>true</code> to initialize their transient variables, as it is called
   * each time the agent server is restarted.
   * <p>
   * This function is not declared <code>final</code> so that derived classes
   * may change their reload policy. The implementation of this method provided
   * by the <code>Agent</code> class does nothing.
   *
   * @param firstTime		true when first called by the factory
   *
   * @exception Exception
   *	unspecialized exception
   */
  protected void agentInitialize(boolean firstTime) throws Exception {
    // Get the logging monitor from current server MonologLoggerFactory
    this.logmon = Debug.getLogger(getLogTopic());
    // Initializes the updated field to true:
    this.updated = true;

    try {
      MXWrapper.registerMBean(this, "AgentServer", getMBeanName());
    } catch (Exception exc) {
      logmon.log(BasicLevel.ERROR, getName() + " jmx failed", exc);
    }

    if (logmon.isLoggable(BasicLevel.DEBUG))
      logmon.log(BasicLevel.DEBUG,
                 "Agent" + id + " [" + name +
                 (firstTime?"] , first initialized":"] , initialized"));
  }

  private String getMBeanName() {
    return new StringBuffer()
      .append("server=").append(AgentServer.getName())
      .append(",cons=Engine#").append(getId().getTo())
      .append(",agent=").append((name == nullName)?getId().toString():name)
      .toString();
  }

  /**
   * This method sends a notification to the agent which id is given in
   * parameter. During an agent reaction alls notifications sent are buffered
   * until reaction commit.
   * <p>
   * Be careful if you use this method outside of an agent reaction,
   * its behavior is slightly different: each notification is immediatly
   * sent using a local transaction.
   *
   * @see Channel#sendTo
   *
   * @param to   the unique id. of destination <code>Agent</code>.
   * @param not  the notification to send.
   */
  protected final void
  sendTo(AgentId to, Notification not) {
//     try {
//       EngineThread thread = (EngineThread) Thread.currentThread();
//       // Use the engine's sendTo method that push message in temporary queue
//       // until the end of current reaction.
//       thread.engine.push(getId(), to, not);
//     } catch (ClassCastException exc) {
//       //  Be careful, the destination node use the from.to field to
//       // get the from node id.
//       Channel.channel.directSendTo(getId(), to, not);
//     }

//  if (Class.EngineThread.isAssignable(Thread.currentThread())) {
    if (Thread.currentThread() == AgentServer.engine.thread) {
      AgentServer.engine.push(getId(), to, not);
    } else {
      Channel.channel.directSendTo(getId(), to, not);
    }
  }

  /**
   * This method sends a notification to the agent which id is wrapped
   * in the specified role.
   *
   * @param role  the destination <code>Role</code>.
   * @param not   the notification to send.
   */
  protected final void sendTo(Role role, Notification not) {
    if (role == null) return;
    sendTo(role.getListener(), not);
  }
 
  /**
   * Sends a notification to all the agents registered in a role.
   *
   * @param role  the destination <code>MultiplRole</code>.
   * @param not   the notification to send.
   */
  protected final void
  sendTo(RoleMultiple role, Notification not) {
    if (role == null) return;
    Enumeration to = role.getListeners();
    if (to == null)
      return;
    while (to.hasMoreElements())
      sendTo((AgentId) to.nextElement(), not);
  }

  /**
   * Permits this agent to destroy it. If necessary, its method should be 
   * overloaded to work properly.
   */
  public void delete() {
    delete(null);
  }

  /**
   * Permits this agent to destroy it. If necessary, its method should be
   *overloaded to work properly. 
   *
   * @param agent	Id of agent to notify.
   */
  public void delete(AgentId agent) {
    if (deployed)
      sendTo(AgentId.factoryId(id.getTo()),
	     new AgentDeleteRequest(agent));
  }
 
  /**
   * Defines the reaction of the agent when receiving a notification. This
   * member function implements the common reactive behavior of an agent, it
   * is called by the execution engine (see <a href="Engine.html">Engine</a>
   * class).<p>
   * If there is no corresponding reaction, the agent send an
   * <code>UnknownNotification</code> notification to the sender.
   *
   * @param from	agent sending notification
   * @param not		notification to react to
   *
   * @exception Exception
   *	unspecialized exception
   */
  public void react(AgentId from, Notification not) throws Exception {
    if (not instanceof DeleteNot) {
      delete(((DeleteNot) not).reply);
    } else if ((not instanceof UnknownAgent) ||
               (not instanceof UnknownNotification) ||
               (not instanceof ExceptionNotification)) {
      logmon.log(BasicLevel.WARN,
                 this.toString() + ".react(" + from + ", " + not + ")");
     } else {
      logmon.log(BasicLevel.ERROR,
                 this.toString() + ".react(" + from + ", " + not + ")");
      sendTo(from, new UnknownNotification(id, not));
    }
  }

  /**
   * Called to inform this agent that it is garbaged and that it should free
   * any active ressources that it has allocated.
   * A subclass of <code>Agent</code> should override this method if it has
   * any operation that it wants to perform before it is garbaged. For example,
   * an agent with threads (a ProxyAgent for example) would use the initialize
   * method to create the threads and the <code>agentFinalize</code> method to
   * stop them.
   * The implementation of this method provided by the <code>Agent</code> class
   * does nothing.
   *
   * @param lastTime	true when last called by the factory on agent deletion.
   */
  public void agentFinalize(boolean lastTime) {
    try {
      MXWrapper.unregisterMBean("AgentServer", getMBeanName());
    } catch (Exception exc) {
      logmon.log(BasicLevel.ERROR, getName() + " jmx failed", exc);
    }
  }
}