incomingmessagehandler.java

SRI international 发布的OAA框架软件

package com.sri.oaa2.simplefac;

import com.sri.oaa2.icl.*;
import java.util.*;
import org.apache.log4j.Logger;
import org.apache.log4j.Priority;
import org.apache.log4j.NDC;

/**
 * Handles an incoming message
 */
public final class IncomingMessageHandler implements Runnable
{
  // Logger for this class
  static Logger logger = Logger.getLogger(IncomingMessageHandler.class.getName() + ".default");
  static Logger loggerRun = Logger.getLogger(IncomingMessageHandler.class.getName() + ".run");
  static Logger loggerSolve = Logger.getLogger(IncomingMessageHandler.class.getName() + ".handleSolve");
  static Logger loggerConnect = Logger.getLogger(IncomingMessageHandler.class.getName() + ".handleConnect");
  static Logger loggerRegisterSolvables = Logger.getLogger(IncomingMessageHandler.class.getName() + ".handleRegisterSolvables");
  static Logger loggerUpdateData = Logger.getLogger(IncomingMessageHandler.class.getName() + ".handleUpdateData");
  static Logger loggerV2 = Logger.getLogger(IncomingMessageHandler.class.getName() + ".V2");

  // The incoming message
  private IclTerm term;

  // Our connection
  private SimpleFacConnection conn;

  /**
   * Create handler with term
   */
  public IncomingMessageHandler(IclTerm t, SimpleFacConnection c)
  {
    setTerm(t);
    setConn(c);
  }

  /**
   * Set connection
   */
  protected final void setConn(SimpleFacConnection c)
  {
    conn = c;
  }

  /**
   * Get connection
   */
  protected final SimpleFacConnection getConn()
  {
    return conn;
  }

  /**
   * Set term to handle
   */
  protected final void setTerm(IclTerm t)
  {
    term = t;
  }

  /**
   * Get term to handle
   */
  protected final IclTerm getTerm()
  {
    return term;
  }

  /**
   * Get first term of first term
   */
  protected final IclTerm getEventContent(IclTerm t)
  {
    return t.getTerm(0).getTerm(0);
  }

  /**
   * Get second term of first term
   */
  protected final IclTerm getEventParams(IclTerm t)
  {
    return t.getTerm(0).getTerm(1);
  }

  public final void run()
  {
    IclTerm t = getTerm();
    if(getConn().getName() != null) {
      NDC.push(getConn().getName().toString());
    }
    if(loggerRun.isDebugEnabled()) {
      loggerRun.debug("IncomingMessageHandler.run() handling term " + t.toString());
    }
    try {
      if(!ToFunctor.getInstance().from(t).equals("term")) {
	NDC.pop();
        if(loggerRun.isDebugEnabled()) {
          loggerRun.debug("IncomingMessageHandler.run() functor not equal to 'term': " + ToFunctor.getInstance().from(t));
        }
        return;
      }
    }
    catch(UnsupportedOperationException uoe) {
      loggerRun.warn("IncomingMessageHandler.run() ignoring because I don't know what to do with it: [" + t.toString() + "]");
      NDC.pop();
      return;
    }
    if(t.size() != 1) {
      loggerRun.warn("IncomingMessageHandler.run() ignoring size is wrong: [" + t.toString() + "]");
      NDC.pop();
      return;
    }

    if(loggerRun.isDebugEnabled()) {
      loggerRun.debug("IncomingMessageHandler.run() functor is 'term'");
    }

    IclTerm firstChild = t.getTerm(0);
    if(loggerRun.isDebugEnabled()) {
      loggerRun.debug("IncomingMessageHandler.run() got first child of term");
    }
    try {
      if(((getConn().getVersion() != null) &&
          (getConn().getVersion().toString().compareTo("3.0") < 0)) ||
         !ToFunctor.getInstance().from(firstChild).equals("event")) {
        loggerRun.warn("IncomingMessageHandler.run() message appears to be a version 2.0 message: [" + firstChild.toString() + "]");
        handleVersion2(firstChild);
	NDC.pop();
        return;
      }
    }
    catch(UnsupportedOperationException uoe) {
      loggerRun.warn("IncomingMessageHandler.run() first term is not an IclStruct: [" + firstChild.toString() + "], in term [" + t.toString() + "]; ignoring");
      NDC.pop();
      return;
    }

    if(loggerRun.isDebugEnabled()) {
      loggerRun.debug("IncomingMessageHandler.run() checking size of first term");
    }

    if(t.getTerm(0).size() != 2) {
      loggerRun.warn("IncomingMessageHandler.run() ignoring size of first term is wrong: [" + t.toString() + "]");
      NDC.pop();
      return;
    }

    IclTerm content = getEventContent(t);
    IclTerm params = getEventParams(t);

    String name;
    try {
      if(loggerRun.isDebugEnabled()) {
        loggerRun.debug("IncomingMessageHandler.run() getting name of content");
      }

      name = ToFunctor.getInstance().from(content);
    }
    catch(UnsupportedOperationException uoe) {
      loggerRun.warn("IncomingMessageHandler.run() ignoring expecting IclStruct, got: [" + content + "]");
      NDC.pop();
      return;
    }
    StringBuffer b = new StringBuffer();
    b.append("IncomingMessageHandler:");
    b.append(getConn().getId().toString());
    b.append(":");
    b.append(name);
    Thread.currentThread().setName(b.toString());
    if(loggerRun.isDebugEnabled()) {
      loggerRun.debug("IncomingMessageHandler.run() matching name to handlers: " + name);
    }
    if(name.equals("ev_solve")) {
      handleSolve(content, params);
    }
    else if(name.equals("ev_update_data")) {
      handleUpdateData(content, params);
    }
    else if(name.equals("ev_ready")) {
      handleReady(content, params);
    }
    else if(name.equals("ev_register_solvables")) {
      handleRegisterSolvables(content, params);
    }
    else if(name.equals("ev_connect")) {
      if(loggerRun.isDebugEnabled()) {
        loggerRun.debug("IncomingMessageHandler.run() calling handleConnect: " + name);
      }
      handleConnect(content, params);
    }
    else if(name.equals("ev_query_formats")) {
      handleQueryFormats(content, params);
    }
    else {
      loggerRun.warn("Did not handle event: " + t);
    }
    NDC.pop();
  }

  /**
   * Handle a query about formats supported
   */
  protected final void handleQueryFormats(IclTerm ev, IclTerm params)
  {
    IclTerm connectParamsTerm = ev.getTerm(0);
    if(!connectParamsTerm.isList()) {
      logger.error("IncomingMessageHandler::handleConnect first arg not list");
      return;
    }
    IclList connectParams = (IclList)connectParamsTerm;

    IclTerm otherName = IclUtils.getParamValue("other_name", getConn().getId().toIclTerm(), connectParams);
    getConn().setName(otherName);

    IclTerm event = new IclStruct("event");
    IclTerm reply = new IclStruct("ev_supported_formats");
    reply.add(getConn().getFac().getSupportedFormatsList());
    event.add(reply);
    event.add(new IclList());
    getConn().sendTerm(event, true);
  }

  /**
   * Handle a connection request
   */
  protected final void handleConnect(IclTerm event, IclTerm params)
  {
    //term(event(ev_connect([other_name('TestOaaListener'),other_language(java),other_type(client),format(pure_text),other_version(3.2)]),[])).
    //term(event(ev_connected([oaa_address(addr(tcp('130.107.65.150',3378),1)),other_id(0),other_type(facilitator),other_name(root),other_language(prolog),other_version(3.2),other_dialect(sicstus),format(default)]),[])).
    //ev_connected([oaa_address(addr(tcp('127.0.0.1',3379),1)),other_id(0),other_type(facilitator),other_name(root),other_language(java),other_version(3.2),other_version(sicstus),format(default)])

    IclTerm connectParamsTerm = event.getTerm(0);
    if(!connectParamsTerm.isList()) {
      loggerConnect.error("IncomingMessageHandler::handleConnect first arg not list");
      return;
    }
    IclList connectParams = (IclList)connectParamsTerm;

    IclTerm queryFormats = IclUtils.getParamValue("query_formats", null, connectParams);
    if(queryFormats != null) {
      if(loggerConnect.isDebugEnabled()) {
        loggerConnect.debug("IncomingMessageHandler::handleConnect redirect to handleQueryFormats");
      }
      handleQueryFormats(event, params);
      return;
    }

    IclTerm otherName = IclUtils.getParamValue("other_name", getConn().getId().toIclTerm(), connectParams);
    getConn().setName(otherName);

    if(loggerConnect.isInfoEnabled()) {
      loggerConnect.info("IncomingMessageHandler for agent " +
                         getConn().getId().toString() +
                         "(" +
                         getConn().getName()+
                         ") connected");
    }

    IclTerm otherLang = IclUtils.getParamValue("other_language", null, connectParams);
    if(otherLang == null) {
        String errMessage = "IncomingMessageHandler for agent " + getConn().getId().toString() + " no language specified in connection request";
      loggerConnect.error(errMessage);
      getConn().shutdown(new Exception(errMessage));
      return;
    }
    getConn().setLang(otherLang);

    IclTerm otherType = IclUtils.getParamValue("other_type", null, connectParams);
    if(otherType == null) {
        String errMessage = "IncomingMessageHandler for agent " + getConn().getId().toString() + " no type specified in connection request";
      loggerConnect.error(errMessage);
      getConn().shutdown(new Exception(errMessage));
      return;
    }
    getConn().setType(otherType);

    IclTerm format = IclUtils.getParamValue("format", null, connectParams);
    if(format != null) {
      getConn().verifyFormat(format);
    }

    IclTerm version = IclUtils.getParamValue("other_version", null, connectParams);
    if(version == null) {
        String errMessage = "IncomingMessageHandler for agent " + getConn().getId().toString() + " no version specified in connection request";
      loggerConnect.error(errMessage);
      getConn().shutdown(new Exception(errMessage));
      return;
    }