libcomtcpconnection.java

SRI international 发布的OAA框架软件

     * Closes the Socket connection
     */
    public void cnx_Disconnect() {
        try {
            /*
          if (thread_listener!=null)
          thread_listener.stop();
          else
          System.out.println("Thread server NULL");
            */
            if (clientSocket != null) {
                clientSocket.close();
                thread_listener.disconnect();
                listenerThread.join();
                
                if(!clientSocket.isClosed()) {
                  System.out.println("cnx_Disconnect() clientSocket not closed!");
                }
                
            }

            if (serverSocket != null) {
                serverThread.disconnect();
                serverSocket.close();
            }

            if (binaryServerSocket != null) {
                binaryServerThread.disconnect();
                binaryServerSocket.close();
            }
            connected = false;
        } catch (Exception e) {
            logger.error("Exception in LibComTcp.java disconnect:" + e);
        }
    }

    private static final IclTerm resolveFalseTerm = new IclStruct(
            "resolve_vars", new IclStr("false"));

    public boolean cnx_ListenAt(IclTerm address, IclTerm params) {
        boolean ret = false;
        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
            if (!host.isVar() && !port.isVar()) {
                // Check if the host is localhost
                String hostStr = host.toIdentifyingString();

                try {
                    String localHostName = InetAddress.getLocalHost().getHostName();

                    if (!hostStr.equals(localHostName) && !hostStr.equals("localhost")) {
                        logger.warn("The specified host, " + hostStr +
                                ", (from commandline,\n  environment, or setup file), " +
                                "isn''t the machine\n  on which I am running");
                    }

                    hostStr = InetAddress.getLocalHost().getHostAddress();

                    // Creates, stores and uses the new connection
                    try {
                        // Pass in the InetAddress for the case where a multi-homed
                        // machine such as a firewall needs to specify an address
                        // to bind to.
                        serverSocket = new ServerSocket(ToInt.getInstance().from(port, -1),
                                LibCom.MAXBACKLOG, InetAddress.getByName(hostStr));
                        serverThread = new LibComTcpDefaultServerThread(this, mClient, serverSocket);
                        new Thread(serverThread).start();
                        address.replaceElement(0, new IclStr(hostStr));
                        listenAddress = address;
                        logger.info("Listening on : " + listenAddress);


                        // XXX Temporary until we resolve binary port design...
                        IclTerm binaryAddress = LibComUtils.oaaResolveVariable("oaa_listen_binary", mCmdLine);

                        if (binaryAddress == null) {
                            binaryAddress = IclTerm.fromString(true, "tcp('localhost'," +
                                    (ToInt.getInstance().from(port, -1) + 10) + ")");
                        }
                        IclTerm binaryPortTerm = binaryAddress.getTerm(1);
                        int binaryPort = ToInt.getInstance().from(binaryPortTerm, -1);
                        binaryServerSocket = new ServerSocket(binaryPort,
                                LibCom.MAXBACKLOG, InetAddress.getByName(hostStr));
                        binaryServerThread = new LibComTcpBinaryServerThread(this, mClient, binaryServerSocket);
                        new Thread(binaryServerThread).start();
                        logger.info("Listening for binary on : " + binaryAddress);
                        serverThread.setBinaryPort(binaryPort);
                        ret = true;
                    } catch (IOException e) {
                        // XXX prolog facilitator does more at this point; it tries
                        // to increase the port number, or waits and tries to
                        // reconnect. This should be implemented here also.
                        logger.error("Failed to start server socket", e);
                    }
                } catch (UnknownHostException e) {
                    logger.error(e);
                }
            } else {
                TcpPrintError("Couldn't resolve address in comConnect: " + address);
                logger.error(host + ".isVar()? " + host.isVar());
                logger.error(port + ".isVar()? " + port.isVar());
            }
        }

        connected = true;
        return ret;
    }

    public IclTerm getListenAddress() {
        return listenAddress;
    }

    public IclTerm getClientAddress() {
        return clientAddress;
    }

    public void setLibComServerListener(LibComServerListener listener) {
        serverListener = listener;
    }

    public void setLibComListener(LibComListener listener) {
        this.listener = listener;
    }

    private void TcpPrintError(String inMessage) {
        logger.error("LibComTcpConnection Error --> " + inMessage);
    }
}

/**
 * Real connection to the facilitator.  This manages sending and receiving of IclTerms
 */
class RealConnection extends SimpleFacConnection {
    LibComClient client;
    LibComTcpConnection connection;

    protected void setClient(LibComClient c) {
        client = c;
    }

    protected LibComClient getClient() {
        return client;
    }

    public RealConnection(ConnectionId id, Socket s, LibComClient c, LibComTcpConnection connection) {
        super(id, s, null);
        this.connection = connection;
        setClient(c);
        setName(new IclStr("'RealConnection:" + id.toString() + "'"));
    }

    public void shutdown(Throwable ex) {
        super.shutdown(ex);
        if (connection != null && connection.listener != null) {
            connection.listener.notifyConnectionShutDown(connection, getId(), ex);
        }
    }

    /**
     * Get the next term that has been sent to us.
     */
    public IclTerm getNextTerm() {
        try {
            return getTermReceiver().getNextTerm();
        } catch (Exception e) {
            RuntimeException re = new RuntimeException(e.toString());
            re.fillInStackTrace();
            throw re;
        }
    }

    static boolean SENDTERMNOTCONNONCE = false;

    /**
     * Send a term in the current thread.  This *will* block if the receiver does not
     * have enough buffer space.
     */
    public void sendTerm(IclTerm toSend) {
        if (logger.isDebugEnabled()) {
            logger.debug("LibComTcpConnection::RealConnection.sendTerm() sending " + toSend);
        }

        if (!isConnected()) {
            if (!SENDTERMNOTCONNONCE) {
                SENDTERMNOTCONNONCE = true;
                throw new RuntimeException("RealConnection.sendTerm(IclTerm) cannot send--not connected");
            } else {
                return;
            }
        } else {
            SENDTERMNOTCONNONCE = false;
        }

        //logger.info(getId() + ":Sending term = " + toSend);

        IclTerm wrapper = new IclStruct("term");
        wrapper.add(toSend);
        OutgoingMessageHandler handler = new OutgoingMessageHandler(wrapper, this);
        handler.setForce(true);
        synchronized (this) {
            handler.run();
        }
    }

    /**
     * Read terms and give them to the client as long as we are connected
     */
    static Logger logger = Logger.getLogger(LibComTcpConnection.class.getName());

    public void run() {
        IclTerm t;
        GETTING_TERMS:
        while (true) {
            if (!isConnected()) {
                shutdown(new Exception("Unexpected: not connected"));
                return;
            }
            t = null;
            try {
                t = getTermReceiver().getNextTerm();
            } catch (Exception e) {
                StringWriter sw = new StringWriter();
                PrintWriter pw = new PrintWriter(sw);
                e.printStackTrace(pw);
                e.printStackTrace();
                logger.error(sw.toString());
                shutdown(e);
                return;
            }
            if (!getTermReceiver().isConnected()) {
                shutdown(getTermReceiver().getCloseException());
                return;
            }
            if (t != null) {
                if (logger.isDebugEnabled()) {
                    logger.debug("LibComTcpConnection: processing term");
                    logger.debug(getId() + ": Received term = " + t);
                }
                getClient().processIncomingTerm(getId().toString(), t);
                if (logger.isDebugEnabled()) {
                    logger.debug("LibComTcpConnection: processed term");
                }
                continue GETTING_TERMS;
            }
        }
    }
}

/**
 * This is the initial connection which will be used to determine what
 * formats the facilitator accepts.
 */
class InitialConnection extends SimpleFacConnection {
    public InitialConnection(Socket s) {
        super(s, null);
    }

    public InitialConnection(ConnectionId id, Socket s) {
        super(id, s, null);
        setName(new IclStr("'LibComTcpConnection_formatRequest_shouldDisconnect'"));
    }

    /**
     * Send a term to the facilitator.  The OutgoingMessageHandler will always use a
     * GenericFormatWriter, which writes strings.
     */
    public void sendTerm(IclTerm toSend) {
        if (!isConnected()) {
            throw new RuntimeException("InitialConnection.sendTerm(IclTerm) cannot send--not connected");
        }

        IclTerm wrapper = new IclStruct("term");
        wrapper.add(toSend);
        OutgoingMessageHandler handler = new OutgoingMessageHandler(wrapper, this);
        handler.setForce(true);
        synchronized (this) {
            handler.run();
        }
    }

    /**
     * Get the next term that has been sent to us.
     */
    public IclTerm getNextTerm() {
        try {
            return getTermReceiver().getNextTerm();
        } catch (Exception e) {
            RuntimeException re = new RuntimeException(e.toString());
            re.fillInStackTrace();
            throw re;
        }
    }

    /**
     * Probably won't ever need this, but if we do, it just discards all terms.
     */
    public void run() {
        while (true) {
            if (!isConnected()) {
                shutdown(new Exception("Unexpected: not connected"));
                return;
            }
            try {
                getTermReceiver().getNextTerm();
            } catch (Exception e) {
                if (!getTermReceiver().isConnected()) {
                    shutdown(getTermReceiver().getCloseException());
                    return;
                }
            }
        }
    }
}