libcomtcpconnection.java

SRI international 发布的OAA框架软件

/*
 * @(#)LibComTcpConnection.java	08/1999
 *
 * The contents of this file are subject to the OAA Community Research
 * License Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License. You may obtain a copy of the License
 * at http://www.ai.sri.com/~oaa/.  Software distributed under the License
 * is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, either
 * express or implied. See the License for the specific language governing
 * rights and limitations under the License.  Portions of the software are
 * Copyright (c) SRI International, 1999.  All rights reserved.
 * "OAA" is a registered trademark, and "Open Agent Architecture" is a
 * trademark, of SRI International, a California nonprofit public benefit
 * corporation.
 */

/*
* file:           LibComTcp.java
* purpose:        Low level socket functions for OAA V2.0
*
*   Copyright (C) 1998, SRI International.  All rights reserved.
*/

package com.sri.oaa2.com;

import com.sri.oaa2.icl.IclInt;
import com.sri.oaa2.icl.IclList;
import com.sri.oaa2.icl.IclStr;
import com.sri.oaa2.icl.IclStruct;
import com.sri.oaa2.icl.IclTerm;
import com.sri.oaa2.icl.IclTermCache;
import com.sri.oaa2.icl.IclUtils;
import com.sri.oaa2.icl.ToInt;
import com.sri.oaa2.icl.Unifier;
import com.sri.oaa2.lib.LibOaa;
import com.sri.oaa2.simplefac.ConnectionId;
import com.sri.oaa2.simplefac.IclTermBinaryReceiver;
import com.sri.oaa2.simplefac.IclTermStringReceiver;
import com.sri.oaa2.simplefac.OutgoingMessageHandler;
import com.sri.oaa2.simplefac.SimpleFacConnection;
import com.sri.oaa2.simplefac.StringConnectionId;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.StringWriter;
import java.io.PrintWriter;
import java.net.ConnectException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;

/**
 * Provides low-level communications methods for the agentlib
 */
public class LibComTcpConnection implements LibComConnection {
    /**
     * Copyright and version information.
     *
     */
    public final static String copyright = "Copyright (c) 2002, SRI International.  All Rights Reserved.";
    public final static String rcsId = "$Id: LibComTcpConnection.java,v 1.38 2005/03/27 20:26:17 agno Exp $";

    static Logger logger = Logger.getLogger(LibComTcpConnection.class.getName());

    protected boolean connected = false;
    protected LibComClient mClient;
    protected String mConnectionId;
    protected String[] mCmdLine;

    // The socket needed for the TCP client connection
    // If server mode, client socket bound to incoming client
    // If client mode, main socket.
    Socket clientSocket;

    OutputStream clientSocketOutputStream;
    OutputStreamWriter clientSocketOutputWriter = null;
    ObjectOutputStream clientSocketObjectOutputStream = null;

    // The socket needed for the TCP server connection
    ServerSocket serverSocket;
    ServerSocket binaryServerSocket;
    LibComServerListener serverListener;

    // Listener thread
    Thread listenerThread;
    RealConnection thread_listener;
    //InitialConnection queryConnect;
    LibComTcpServerThread serverThread;
    LibComTcpServerThread binaryServerThread;

    // Resolved addresses for client or listener
    IclTerm clientAddress, listenAddress;

    LibComListener listener;

    /**
     * Constructor
     */
    public LibComTcpConnection(LibComClient client, String connectionId, String[] cmdLine) {
        mClient = client;
        mConnectionId = connectionId;
        mCmdLine = cmdLine;
        connected = false;
    }

    public LibComTcpConnection(LibComClient inClient, String inConnectionId) {
        this(inClient, inConnectionId, null);
    }

    public boolean cnx_IsConnected() {
        return connected;
    }

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

    /**
     * Returns address tcp(IPAddr,Port) to connected socket.
     * Must be connected first, for this to work!
     * Currently, assumes client connection (should change!)
     */
    public IclTerm cnx_CanonicalAddress(IclTerm address) {
        IclTerm addressStruct = null;
        addressStruct = IclTermCache.getNoAnon("tcp(Host,Port)");

        IclTerm connectionInfo = Unifier.getInstance().unify(addressStruct, address);
        IclTerm host = null;
        IclTerm port = null;

        if (cnx_IsConnected() && (connectionInfo != null)) {
            String inet = clientSocket.getInetAddress().toString();
            // inet = "test.ai.sri.com/11.11.11.11", we want only IP num
            inet = inet.substring(inet.indexOf("/") + 1);
            host = new IclStr(inet);
            // HACK
            //port = connectionInfo.getTerm(1);
            port = new IclInt(clientSocket.getPort());
            return new IclStruct("addr", new IclStruct("tcp", host, port));
        } else {
            TcpPrintError("cnx_CanonicalAddress() can only be called on CONNECTED socket!");
            return null;
        }
    }

    /**
     * Opens a Socket connection on a Host at Port
     */
    public boolean cnx_Connect(IclTerm address, IclTerm params) 
    {
        InitialConnection queryConnect;
        address = LibComUtils.resolveAddress(address, mCmdLine, params, logger);

        if ((!cnx_IsConnected()) && (address != null)) {
            IclTerm host = address.getTerm(0);
            IclTerm port = address.getTerm(1);
            // By now, host and port should be known
            // System.out.println("Port " + port);
            // System.out.println("Host " + host);
            if (!host.isVar() && !port.isVar()) {
                // Creates, stores and uses the new connection
                try {
                    logger.info("Connecting to : " + host.toIdentifyingString() + " at " + ToInt.getInstance().from(port, -1));
                    clientSocket = new Socket(host.toIdentifyingString(), ToInt.getInstance().from(port, -1));
                    //logger.info("Created clientSocket");
                    // Send the format query message, and check for the fastest supported format
                    queryConnect = new InitialConnection(new StringConnectionId(mConnectionId), clientSocket);
                    queryConnect.setTermReceiver(new IclTermStringReceiver(clientSocket, queryConnect));
                    queryConnect.setFormat(new IclStr("default"));
                    /*
                    boolean usingPassword = false;
                    IclTerm usePassword = LibComUtils.oaaResolveVariable("use_password", this.getCmdLine());
                    if(usePassword != null) {
                      usingPassword = ((IclStr)usePassword).equals(new IclStr("true"));
                    }
                    String tempString;
                    if(usingPassword) {
                      String password = "_";
                      IclTerm clientPassword = LibComUtils.oaaResolveVariable("client_password", this.getCmdLine());
                      if(clientPassword != null) {
                        password = clientPassword.toString();
                        if(password.trim().equals("")) {
                          password = "_";
                        }
                      }
                      tempString =
                        "event(ev_connect([other_name('LibComTcpConnection_formatRequest_shouldDisconnect'),other_language(java),other_type(client),other_version(" + LibOaa.getLibraryVersion().toString() + "),query_formats(_),password(" + password + ")]),[])";
                    }
                    else {
                      tempString =
                        "event(ev_connect([other_name('LibComTcpConnection_formatRequest_shouldDisconnect'),other_language(java),other_type(client),other_version(" + LibOaa.getLibraryVersion().toString() + "),query_formats(_)]),[])";
                    }
                    */
                    String tempString =
                      "event(ev_connect([other_name('LibComTcpConnection_formatRequest_shouldDisconnect'),other_language(java),other_type(client),other_version(" + LibOaa.getLibraryVersion().toString() + "),query_formats(_)]),[])";
                    IclTerm whatFormats = IclTermCache.getNoAnon(tempString);
                    //logger.info("About to send whatFormats");
                    queryConnect.sendTerm(whatFormats);
                    //logger.info("sent whatFormats");
                    IclTerm reply = queryConnect.getNextTerm();
                    //logger.info("got reply to whatFormants");
                    connected = false;
                    // We'll take care of cleaning this connection up later

                    // reply is term(event(ev_supported_formats([<list of supported formats>])),[])
                    // or term(event(ev_connected([<connection params>])),[])
                    IclTerm returnedEvent = reply.getTerm(0).getTerm(0);
                    // we could also unify against ev_connected(exception(X)), but this should be faster
                    if(returnedEvent.getTerm(0).isStruct()) {
                      IclStruct res = (IclStruct)returnedEvent.getTerm(0);
                      if(res.getFunctor().compareTo("exception") == 0) {
                        logger.error("LibComTcpConnect.cnx_connect got exception on connect: " + res.getTerm(0).toString());
                        return false;
                      }
                    }
                    IclList supported = (IclList) reply.getTerm(0).getTerm(0).getTerm(0);
                    port = IclUtils.getParamValue("binary", null, supported);
                      if (port != null) {
                        try {
                          logger.info("LibComTcpConnection.cnx_connect got binary port: " + ToInt.getInstance().from(port, -1));
                          clientSocket = new Socket(host.toIdentifyingString(), ToInt.getInstance().from(port, -1));
                          clientSocket.setTcpNoDelay(true);
                          thread_listener = new RealConnection(new StringConnectionId(mConnectionId), clientSocket, mClient, this);
                          thread_listener.setTermReceiver(new IclTermBinaryReceiver(clientSocket, thread_listener));
                          thread_listener.setFormat(new IclStr("binary"));
                          logger.info("Starting binary receiver");
                          listenerThread = new Thread(thread_listener);
                          listenerThread.start();
                          connected = true;
                        } catch (IOException e) {
                          logger.error("Failed to connect to binary server port: " + port, e);
                        }
                      } 
                      else {
                        try {
                          port = IclUtils.getParamValue("default", null, supported);
                          if (port != null) {
                            logger.debug("LibComTcpConnection.cnx_connect got default port: " + port);
                            clientSocket = new Socket(host.toIdentifyingString(), ToInt.getInstance().from(port, -1));
                            thread_listener = new RealConnection(new StringConnectionId(mConnectionId), clientSocket, mClient, this);
                            thread_listener.setTermReceiver(new IclTermStringReceiver(clientSocket, thread_listener));
                            thread_listener.setFormat(new IclStr("default"));
                            listenerThread = new Thread(thread_listener);
                            listenerThread.start();
                            connected = true;
                          }
                        } catch (IOException e) {
                          logger.error("Failed to connect to default server port: " + port, e);
                        }
                      }

                    if (!connected) {
                        logger.info("LibComTcpConnection.cnx_connect will assume default port ");
                        clientSocket = new Socket(host.toIdentifyingString(), clientSocket.getPort());
                        thread_listener = new RealConnection(new StringConnectionId(mConnectionId), clientSocket, mClient, this);
                        thread_listener.setTermReceiver(new IclTermStringReceiver(clientSocket, thread_listener));
                        thread_listener.setFormat(new IclStr("default"));
                        listenerThread = new Thread(thread_listener);
                        listenerThread.start();
                        connected = true;
                    }

                    clientAddress = new IclStruct("addr", address);
                    queryConnect.shutdown();
                    return true;
                } catch (ConnectException eConnect) {
                    TcpPrintError("Could not connect");
                } catch (IOException e) {
                    logger.error("Exception in Communication.java connect:" + e);
                }
            } else {
                TcpPrintError("Couldn't resolve address in comConnect: " + address);
                logger.error(host + ".isVar()? " + host.isVar());
                logger.error(port + ".isVar()? " + port.isVar());
            }
        }
        return false;
    }

    /**
     * Sets the connection data (package-private).
     */
    void setClientConnectionData(Socket clientSocket, IclTerm clientAddress, RealConnection thread_listener) {
        this.clientAddress = clientAddress;
        this.thread_listener = thread_listener;
        this.clientSocket = clientSocket;
        if (thread_listener != null) {
            thread_listener.connection = this;
        }
        connected = true;
    }

    public boolean cnx_sendMsg(IclTerm t) {
        try {
            if (clientSocketOutputWriter != null) {
                clientSocketOutputWriter.close();
            }
            if (clientSocketObjectOutputStream == null) {
                clientSocketObjectOutputStream = new ObjectOutputStream(clientSocket.getOutputStream());
            }
            clientSocketObjectOutputStream.writeObject(t);
            clientSocketObjectOutputStream.flush();
            //System.out.println("Sent " + t.toString());
            return true;
        } catch (Exception e) { // output message
            logger.error("Exception in Communication.java send_msg:" + e);
            return false;
        }
    }

    public boolean cnx_sendMsgFormatWriter(IclTerm t) {
        thread_listener.sendTerm(t);
        return true;
    }

    /**