libcom.java

SRI international 发布的OAA框架软件

        result = item.getTerm(1);
      }
      else if (Unifier.getInstance().unify(new IclStruct("status", new IclVar("_")),
                                           infoIn) != null) {
        result = item.getTerm(4);
      }
      else if (Unifier.getInstance().unify(new IclStruct("type", new IclVar("_")),
                                           infoIn) != null) {
        result = item.getTerm(2);
      }
      else {
        result = IclUtils.getMember(infoIn, item.getTerm(3));
      }
    }
    return result;
  }

    /**
     * purpose: Looks up a connection that contains a given information
     * returns: Null if fail, an IclTerm if success
     *
     * @param conStruct a com_connection_info struct
     */
    public String comGetConnectionFromInfo(IclTerm conStruct) {
        IclList result = new IclList();

        String cmd = "com_connection_info(Conn,Protocol,Type,InfoList,Status)";
        IclTerm info = null;
        info = IclTermCache.getNoAnon(cmd);
        String connectionName = null;

        if (commdb.db_Solve(info, new IclList(), result)) {
            for (int indx = 0; indx < result.size(); indx++) {
                //logger.info("comGetConnectionFromInfo: result = " + result);
                connectionName = (result.getTerm(indx).getTerm(0)).toString();
                IclTerm unified = Unifier.getInstance().unify(result.getTerm(indx), conStruct);

                if (unified != null) {
                    return connectionName;
                }
            }
        }

        // No connection was found.
        return null;
    }

    /**
     * purpose: Looks up a connection that contains the given information in
     * the info list.
     * returns: Null if fail, an IclTerm if success
     *
     * @param Info an info parameter struct (ex: other_address(..))
     */
    public String comGetConnectionId(IclTerm Info) {
        IclList result = new IclList();

        String cmd = "com_connection_info(Conn,Protocol,Type,InfoList,Status)";
        IclTerm info = null;
        info = IclTermCache.getNoAnon(cmd);
        String connectionName = null;

        if (commdb.db_Solve(info, new IclList(), result)) {
            for (int indx = 0; indx < result.size(); indx++) {
                connectionName = (result.getTerm(indx).getTerm(0)).toString();
                IclList params = (IclList)result.getTerm(indx).getTerm(3);
                if (IclUtils.termParamValue(Info.toIdentifyingString(),
                        Info.getTerm(0), params) != null) {
                    return connectionName;
                }
            }
        }

        // No connection was found.
        return null;
    }

  /**
   * Determines whether a given connection is actually valid.
   * <p>
   * @param connectionId	a Java String that specifies the name of the Facilitator connection.
   */
  public boolean comConnected(String connectionId) {
    // Make sure there's a valid connection
    try {
      if ((connectionId == null) || (comGetInfo(connectionId,
                                                IclTermCache.getNoAnon("status(connected)"))==null)) {
        return false;
      }
      return true;
    }
    catch(Exception e) {
        e.printStackTrace();
      RuntimeException re = new RuntimeException(e.toString());
      re.fillInStackTrace();
      throw re;
    }

  }

  /**
   * Returns an IclList containing all valid connections' ids.
   * <p>
   */

  public IclList getAllValidConnections() {
    String cmd = "com_connection_info(Id,Protocol,Type,InfoList,connected)";
    IclTerm info = null;
    info = IclTerm.fromString(true, cmd);

    // DEBUG
    // System.out.println("info " + info);
    IclList resultFromDb = new IclList();
    if (commdb.db_Solve(info, new IclList(), resultFromDb)) {
      // DEBUG
      //System.out.println("result " + resultFromDb);

      // Retreives all Id from the list of connected connections.
      // Not the best optimized way to to it (n^2 order), but lists of connections
      // are not going to be huge.
      IclTerm item = null;
      IclList result = new IclList();
      for (int index = 1; index <=resultFromDb.size() ; index ++) {
	item = resultFromDb.getTerm(index - 1);
	// DEBUG
	// System.out.println("item " + item);
	result.add(item.getTerm(0));
      }
      return result;
    }
    return null;
  }


  /**
   * Add NewInfo to the database entry for the client connection described by ConnectionId.
   * If NewInfo already exists in the database for this client, update its value instead.
   * <p>
   * @param connectionId	a Java String that specifies the name of the Facilitator connection.
   * @param newInfo  a Java String which contains one or more elements to be
   *				   stored for this client such as status(S), type(T), or protocol(P);
   *				   e.g., "status(connected), protocol(tcp)".
   * @return true if the entry was successfully added and false otherwise.
   * @see #comGetInfo
   * @see #comConnect
   */
  public boolean comAddInfo(String connectionId, IclTerm newInfo) {
    IclTerm info;
    IclList result = new IclList();
    boolean changed = false;

    // Make sure there's a valid connection id
    if (connectionId == null)
      return false;

    // Finds and extracts the information about connectionId within
    // the database
    String cmd = "com_connection_info('" + connectionId +
      "', Protocol, Type,InfoList,Status)";
    info = IclTermCache.get(cmd);

    if (commdb.db_Solve(info, new IclList(), result)) {
      commdb.db_Retract(info, new IclList());
      IclTerm item = result.getTerm(0);
      IclTerm protocol = null;
      IclTerm type = null;
      IclTerm status = null;
      IclTerm infolist = null;
      IclTerm ninfo = null;

      // Replace the protocol or keeps the current one
      if (Unifier.getInstance().unify(new IclStruct("protocol", new IclVar("_")),
			   newInfo)!=null) {
	protocol = newInfo.getTerm(0);
	changed = true;
      }
      else
	protocol = item.getTerm(1);

      // Replace the type or keeps the current one
      if (Unifier.getInstance().unify(new IclStruct("type", new IclVar("_")),
			   newInfo) != null) {
	type = newInfo.getTerm(0);
	changed = true;
      }
      else
	type = item.getTerm(2);

      // Replace the status or keeps the current one
      if (Unifier.getInstance().unify(new IclStruct("status", new IclVar("_")),
			   newInfo) != null) {
	status = newInfo.getTerm(0);
	changed = true;
      }
      else
	status = item.getTerm(4);

      // Build the infolist
      // If nothing has changed, the infolist has to be expanded by the
      // incoming term
      if (!changed) {
	if (newInfo.isList()) {
	  for (ListIterator infos = newInfo.listIterator();
	       infos.hasNext();) {
	    (item.getTerm(3)).add((IclTerm)infos.next());
	  }
	}
	else
	  (item.getTerm(3)).add(newInfo);
      }
      infolist = item.getTerm(3);

      // Build the new entry for the database
      ninfo = new IclStruct("com_connection_info", new IclStr(connectionId));
      ninfo.add(protocol);
      ninfo.add(type);
      ninfo.add(infolist);
      ninfo.add(status);

      // DEBUG
      // System.out.println("comAddInfo : " + ninfo);

      commdb.db_Assert(ninfo, new IclList());
      return true;
    }
    return false;
  }

  private int totalQueued = 0;
  
  private String[] connectionIds = {};
  
  private int connectionIdIndex = 0;
  
  /*
  public void wantMoreData(String connId)
  {
    ComSem lock = getConnLock(connId);
    lock.release();
  }

  public void holdOffData(String connId)
  {
    ComSem lock = getConnLock(connId);
    lock.releaseTo(0);
  }

  public ComSem getConnLock(String connId)
  {
    return (ComSem)connectionsBuffersLocks.get(connId);
  }
  */

  /**
   * Get next term from any connection, in round robin fashion
   */
  public synchronized IclTerm getNextTermRR()
  {
    int examined = 0;
    ArrayBlockingQueue buf;
    if(connectionIds.length > 0) {
      while(examined < connectionIds.length) {
        connectionIdIndex = connectionIdIndex % connectionIds.length; 
        buf = (ArrayBlockingQueue)connectionsBuffersTable.get(connectionIds[connectionIdIndex]);
        if(buf.size() > 0) {
          try {
            IclTerm toRet = (IclTerm)buf.take();
            --this.totalQueued;
            return toRet;
          }
          catch (InterruptedException e) {
            continue;
          }
        }
        ++connectionIdIndex;
        ++examined;
      }
    }
    return null;
  }
  
  /**
   * Called by LibComConnection each time a message is received.
   * Set free any thread waiting from a call to waitForIncomingMessage
   */
  static Logger logger = Logger.getLogger(LibCom.class.getName());

  public void processIncomingTerm(String connectionId, IclTerm t)
  {
    if(logger.isDebugEnabled()) {
      logger.debug("LibCom.processIncomingTerm got term " + t.toString());
    }
    if(logger.isDebugEnabled()) {
      logger.debug("LibCom.processIncomingTerm about to sync");
    }
    ArrayBlockingQueue buf;
    synchronized(this) {
      if(logger.isDebugEnabled()) {
        logger.debug("LibCom.processIncomingTerm got sync");
      }
      buf = (ArrayBlockingQueue)(connectionsBuffersTable.get(connectionId));
    }
    try {
      /*
        does nothing--why is this here?
      IclTerm number = null;
      try {
        number = t.getTerm(0).getTerm(0).getTerm(1).getTerm(0);
      }
      catch(IndexOutOfBoundsException ioobe) {
        number = new IclStr("unknown");
      }
      */
      t.add(0, new IclStruct("connection_id", new IclStr(connectionId)));
      if(buf != null) {
        buf.put(t);
      }
    }
    catch(InterruptedException ie) {
      System.err.println("LibCom.processIncomingTerm() interrupted");
    }
    synchronized(this) {
      ++totalQueued;
      notify();
    }
    if(logger.isDebugEnabled()) {
      logger.debug("LibCom.processIncomingTerm notified");
    }
  }

  public synchronized boolean haveTerms()
  {
    return totalQueued > 0;
  }

  public synchronized int availTerms()
  {
    return totalQueued;
  }
  
  public synchronized HashMap getBuffers() {
    return (HashMap)connectionsBuffersTable.clone();
  }

  public String[] getCmdLine() {
    return mCmdLine;
  }

  static void ComPrintError (String s) {
    System.out.println("LibCom Error --> " + s);
  }

  void ComPrintInfo (String message) {
    System.out.println("LibCom Info --> " + message);
  }
  //BOGUS, for Debugging only
  public void ComPrintDB(String inMessage)	{
    System.out.println("LibCom DB --> " + inMessage);
    commdb.db_PrintDB();
  }

    public void notifyConnectionShutDown(LibComConnection connection, ConnectionId connectionId, Throwable ex) {
        logger.error("LibCom: Connection has been shut down: " + connectionId, ex);
        String retractInfo = "com_connection_info('" + connectionId +
            "',_,client,_Info,connected)";
        IclTerm t = null;
        t = IclTerm.fromString(true, retractInfo);

        commdb.db_Retract(t, new IclList());
    }
}