icltermbinaryreceiver.java

SRI international 发布的OAA框架软件

package com.sri.oaa2.simplefac;

import com.sri.oaa2.icl.*;
import java.net.*;
import java.io.*;
import org.apache.log4j.Logger;
import org.apache.log4j.NDC;

/**
 *
 * This class is not intended for public consumption.
 *  
 * @author agno
 *
 */
public class IclTermBinaryReceiver implements IclTermReceiver
{
  // Logger
  static Logger logger = Logger.getLogger(IclTermBinaryReceiver.class.getName());

  // Socket connected to client agent
  private Socket clientSocket = null;

  // InputStream associated with socket
  private  DataInputStream clientReader = null;

  // Are we connected?
  private boolean connected = true;

  // The object that does the work
  BinaryReceiverParser parser;

  // The connection
  SimpleFacConnection conn;

  /**
   * Construct with the given socket
   */
  public IclTermBinaryReceiver(Socket s, SimpleFacConnection c)
  {
    connected = true;
    setSocket(s);
    setConn(c);
    try {
      clientReader = new DataInputStream(new BufferedInputStream(getSocket().getInputStream()));
    }
    catch(IOException ioe) {
      throw new RuntimeException("IclTermBinaryReceiver could not create DataInputStream: " + ioe.toString());
    }
    parser = new BinaryReceiverParser(clientReader, getConn());
    Thread t = new Thread(parser);
    StringBuffer b = new StringBuffer();
    b.append("BinaryReceiverParser:");
    b.append(getConn().getId().toString());
    t.setName(b.toString());
    t.start();
  }

  protected void setSocket(Socket s)
  {
    clientSocket = s;
  }

  protected Socket getSocket()
  {
    return clientSocket;
  }

  protected void setConn(SimpleFacConnection c)
  {
    conn = c;
  }

  protected SimpleFacConnection getConn()
  {
    return conn;
  }

  /**
   * Disconnect request
   */
  public synchronized void disconnect()
  {
    connected = false;
  }

  /**
   * Check if disconnect requested
   */
  public synchronized boolean isConnected()
  {
    return connected;
  }

    public Throwable getCloseException() {
        return parser.closeException;
    }

    /**
   * Get next term; may return null if not connected
   */
  public IclTerm getNextTerm()
  {
    IclTerm t = parser.getTerm();
    if(!parser.isConnected()) {
      disconnect();
      return null;
    }
    if(logger.isDebugEnabled()) {
      logger.debug("IclTermBinaryReceiver.getNextTerm() got term: [" + t.toString() + "]");
    }
    return t;
  }

  /**
   * Check if there is a next term
   */
  public boolean haveNextTerm()
  {
    return parser.hasTerm();
  }
}

/**
 * Class to parse incoming string IclTerms
 */
class BinaryReceiverParser extends AbstractReceiverParser
{
  // Logger
  static Logger logger = Logger.getLogger(BinaryReceiverParser.class.getName());
  static Logger runLogger = Logger.getLogger(BinaryReceiverParser.class.getName() + ".run");
  static Logger loggerIclInt = Logger.getLogger(BinaryReceiverParser.class.getName() + ".iclint");

  BinaryFormatReader reader;
  DataInputStream inputStream;

  /**
   * Construct using given reader.
   */
  public BinaryReceiverParser(DataInputStream r, SimpleFacConnection c)
  {
    this.reader = new BinaryFormatReader(r);
    this.inputStream = r;
    setConn(c);
  }

  /**
   * Keep trying to parse the reader, as long as we are connected
   */
  public void run()
  {
    boolean pushed = false;
    IclTerm t;
    if(runLogger.isDebugEnabled()) {
      runLogger.debug("BinaryReceiverParser.run() started");
    }
    GETTING_TERMS:
    while(true) {
      if(!pushed) {
        if(getConn().getName() != null) {
          NDC.push(getConn().getName().toString());
          pushed = true;
        }
      }
      if(!isConnected()) {
        try {
          runLogger.warn("BinaryReceiverParser.run() not connected; closing reader");
          this.inputStream.close();
        }
        catch(IOException ioe) {
          if(pushed) {
            NDC.pop();
          }
          throw new RuntimeException("BinaryReceiverParser could not close reader");
        }
        setTerm(null);
        if(pushed) {
          NDC.pop();
        }
        return;
      }
      try {
        if(runLogger.isDebugEnabled()) {
          runLogger.debug("BinaryReceiverParser.run() parsing stream");
        }
        t = this.reader.read();
        if(t == null) {
          this.disconnect();
          continue GETTING_TERMS;
        }
        if(runLogger.isDebugEnabled()) {
          runLogger.debug("BinaryReceiverParser.run() parsed stream; got term");
        }
        setTerm(t);
        if(runLogger.isDebugEnabled()) {
          runLogger.debug("BinaryReceiverParser.run() parsed stream; set term");
        }
      }
      catch(InterruptedIOException iio) {
        if(runLogger.isDebugEnabled()) {
          runLogger.debug("BinaryReceiverParser.run() parsing stream timed out");
        }
        continue GETTING_TERMS;
      }
      catch(EOFException eofe) {
        closeException = eofe;
        if(runLogger.isDebugEnabled()) {
          StringWriter sw = new StringWriter();
          PrintWriter pw = new PrintWriter(sw);
          eofe.printStackTrace(pw);
          pw.flush();
          runLogger.debug("BinaryReceiverParser.disconnect() called: " + sw.toString());
        }
        disconnect();
      }
      catch(IOException ioe) {
        closeException = ioe;
        if(runLogger.isDebugEnabled()) {
          StringWriter sw = new StringWriter();
          PrintWriter pw = new PrintWriter(sw);
          ioe.printStackTrace(pw);
          pw.flush();
          runLogger.debug("BinaryReceiverParser.disconnect() called: " + sw.toString());
        }
        disconnect();
        if(runLogger.isInfoEnabled()) {
          runLogger.info("BinaryReceiverParser IOException [" + ioe.toString() + "]; closing");
        }
        continue GETTING_TERMS;
      }
    }
  }

}