tcptransport.java

jxme的一些相关程序,主要是手机上程序开发以及手机和计算机通信的一些程序资料,程序编译需要Ant支持

/************************************************************************
 *
 * $Id: TcpTransport.java,v 1.2 2002/03/04 21:42:57 echtcherbina Exp $
 *
 * Copyright (c) 2001 Sun Microsystems, Inc.  All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 *
 * 3. The end-user documentation included with the redistribution,
 *    if any, must include the following acknowledgment:
 *       "This product includes software developed by the
 *       Sun Microsystems, Inc. for Project JXTA."
 *    Alternately, this acknowledgment may appear in the software itself,
 *    if and wherever such third-party acknowledgments normally appear.
 *
 * 4. The names "Sun", "Sun Microsystems, Inc.", "JXTA" and "Project JXTA"
 *    must not be used to endorse or promote products derived from this
 *    software without prior written permission. For written
 *    permission, please contact Project JXTA at http://www.jxta.org.
 *
 * 5. Products derived from this software may not be called "JXTA",
 *    nor may "JXTA" appear in their name, without prior written
 *    permission of Sun.
 *
 * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED.  IN NO EVENT SHALL SUN MICROSYSTEMS OR
 * ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
 * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * ====================================================================
 *
 * This software consists of voluntary contributions made by many
 * individuals on behalf of Project JXTA.  For more
 * information on Project JXTA, please see
 * <http://www.jxta.org/>.
 *
 * This license is based on the BSD license adopted by the Apache Foundation.
 *********************************************************************************/

package net.jxta.impl.endpoint.tcp;

import java.net.*;
import java.io.*;
import java.util.*;
import org.apache.log4j.Category; import org.apache.log4j.Priority;

import net.jxta.document.MimeMediaType;
import net.jxta.endpoint.*;
import net.jxta.protocol.*;
import net.jxta.peergroup.PeerGroup;
import net.jxta.id.ID;
import net.jxta.document.Element;
import net.jxta.document.TextElement;
import net.jxta.document.Advertisement;
import net.jxta.document.AdvertisementFactory;
import net.jxta.exception.PeerGroupException;

import net.jxta.impl.endpoint.MessageImpl;
import net.jxta.impl.endpoint.MessageWireFormat;
import net.jxta.impl.endpoint.MessageWireFormatFactory;
import net.jxta.impl.protocol.*;

import net.jxta.impl.endpoint.Address;

import net.jxta.platform.Module;

/***
 * This class implements the TCP Transport Protocol
 */

public class TcpTransport implements EndpointProtocol, Runnable, Module {
    private static final Category LOG = Category.getInstance(TcpTransport.class.getName());

    public  ServerSocket          unicastSocket;
    public  InetAddress           usingInterface;
    private InetAddress           propagateInetAddress;
    private int                   propagatePort;
    private int                   propagateSize;
    public  int                   serverSocketPort;

    private Thread                multicastThread = null;
    private Thread                unicastThread = null;

    private MulticastSocket       multicastSocket = null;

    private PeerGroup             group = null;
    public  EndpointService              endpoint = null;

    private EndpointAddress       publicAddress = null;
    private EndpointAdvertisement epadv= null;
    private EndpointAddress       mAddress = null;
    private String                localSubnet = null;
    private String                protocolName = "tcp";

    private boolean               allowMulticast = true;

    private TcpConnectionManager connManager = null;

    /**
     * This is a portion of the API which might not be useful when
     * the configuration part of the JXTA platform will be completed.
     * XXX: to revisit.
     */

    public static int unicastPortNb = 9701;
    public static String multicastAddress = "224.0.1.85";
    public static int multicastPortNb = 1234;
    public static int multicastPacketSize = 16384;
    public static String serverName = null;
    public static String interfaceAddress = null;

    private IncomingUnicastThreads unicastThreads = null;
    public  static final int DefaultNbOfUnicastThreads = 1;
    public  static final int MaxNbOfUnicastThreads = 50;
    public  static final int MaxNbOfUnicastKeepAliveThreads = 40;
    public  static final int MaxKeepAliveDelay = 15 * 60 * 1000; // 15 Minute

    private IncomingMulticastThreads multicastThreads = null;
    public  static final int DefaultNbOfMulticastThreads = 0;
    public  static final int MaxNbOfMulticastThreads = 20;
    public  static final int MaxNbOfPendingSockets = 100;
    public  static final int MaxCnxBacklog = 50; // Java's default is 50

    public TcpTransport() {
    }

    public int getPort() {
        return serverSocketPort;
    }

    public InetAddress getMcastAddr() {
        return propagateInetAddress;
    }

    public int getMcastPort() {
        return propagatePort;
    }

    public int getMcastSize() {
        return propagateSize;
    }

    /**
     * Returns true if this protocol accepts to be overloaded.
     * That is let a protocol with the name protocol name in a
     * descendant group be registered.
     *
     * @return boolean true if over load is allowed.
     */
    public boolean allowOverLoad() {
        return false;
    }

    /**
     * Mark this module as up and running.
     */
    public int startApp(String[] arg) {
        return 0;
    }

    /**
     * closes this TransportProtocol.
     */
    public void stopApp() {
        endpoint.removeEndpointProtocol(this);
        if (LOG.isEnabledFor(Priority.WARN)) LOG.warn("There may be more things to be done here");
    }

    /**
     * Initialization of the TcpTransport (called by Platform)
     */

    public void init(PeerGroup g, ID assignedID, Advertisement impl)
        throws PeerGroupException
    {

	// First, create our TcpConnection Manager
	connManager = new TcpConnectionManager (this);

        try {

            ModuleImplAdvertisement implAdv = (ModuleImplAdvertisement) impl;
            PeerAdvertisement configAdv = (PeerAdvertisement)
                g.getConfigAdvertisement();

            // Get out invariable parameters from the implAdv
            Element param = implAdv.getParam();
            if (param != null) {
                Enumeration list = param.getChildren("Proto");
                if (list.hasMoreElements()) {
                    TextElement pname = (TextElement) list.nextElement();
                    protocolName = pname.getTextValue();
                }
            }

            // Get our peer-defined parameters in the configAdv
            param = configAdv.getServiceParam(assignedID);

            // FIXME 20011220 [e-mail] Temporarily accept both nodes of
            // type TCPAdv and TransportAdvertisement.
            Enumeration tcpChilds = param.getChildren(
            TransportAdvertisement.getAdvertisementType());

            // get the TransportAdv from either TransportAdv or HttpAdv
            if( tcpChilds.hasMoreElements() ) {
                param = (Element) tcpChilds.nextElement();
            } else {
                tcpChilds = param.getChildren(
                TCPAdv.getAdvertisementType());

                if( tcpChilds.hasMoreElements() ) {
                    param = (Element) tcpChilds.nextElement();
                }
            }

            TCPAdv adv = (TCPAdv)
                AdvertisementFactory.newAdvertisement((TextElement) param);

            unicastPortNb = new Integer(adv.getPort()).intValue();

            // FIXME: these three should not be peer configurable but
            // that will do for now (no worse than before).
            multicastAddress = adv.getMulticastAddr();
            multicastPortNb = new Integer(adv.getMulticastPort()).intValue();
            multicastPacketSize =
                new Integer(adv.getMulticastSize()).intValue();

            interfaceAddress = adv.getInterfaceAddress();
            serverName = adv.getServer();

            if (LOG.isEnabledFor(Priority.INFO)) LOG.info("Configuring TCP Transport");
            if (LOG.isEnabledFor(Priority.INFO)) LOG.info("UnicastPort:"+ unicastPortNb);
            if (LOG.isEnabledFor(Priority.INFO)) LOG.info("MulticastAddr:"  + multicastAddress);
            if (LOG.isEnabledFor(Priority.INFO)) LOG.info("MulticastPort:"  + multicastPortNb);
            if (LOG.isEnabledFor(Priority.INFO)) LOG.info("MulticastPacketSize:"  + multicastPacketSize);
            if (LOG.isEnabledFor(Priority.INFO)) LOG.info("Interface address:"
                     + interfaceAddress == null ? "(unspecified)"
                     : interfaceAddress);
            if (LOG.isEnabledFor(Priority.INFO)) LOG.info("Public address:"
                     + serverName == null ? "(unspecified)" : serverName);


            unicastThreads = new IncomingUnicastThreads(this, DefaultNbOfUnicastThreads);
            multicastThreads = new IncomingMulticastThreads(this, DefaultNbOfMulticastThreads);

            /* Open the incoming socket. */
            if (interfaceAddress == null) {
                interfaceAddress = InetAddress.getLocalHost().getHostAddress();
            }
            usingInterface = InetAddress.getByName( interfaceAddress );
            unicastSocket = new ServerSocket(unicastPortNb, MaxCnxBacklog,
                                             usingInterface);

            // First create a listening thread waiting for incoming message
            // on serverSocket

            serverSocketPort = unicastPortNb;

            // Build the actual publicAddress and locally bound InetAddress
            if (serverName == null) {
                serverName = interfaceAddress + ":" + unicastPortNb;
            }

            endpoint = g.getEndpointService();

            publicAddress = endpoint.newEndpointAddress(protocolName
                                                        + "://"
                                                        + serverName);
            mAddress =
                endpoint.newEndpointAddress(protocolName
                                            + "://"
                                            + multicastAddress);

            if (LOG.isEnabledFor(Priority.INFO)) LOG.info( "Binding to : " +  usingInterface.getHostAddress() );
            if (LOG.isEnabledFor(Priority.INFO)) LOG.info( "Publishing Address : " +  publicAddress.toString() );

            unicastThread = new Thread(this,
                                       "TcpTransport:unicast server thread" );
            unicastThread.start();

            // Create the multicast input socket
            propagatePort = multicastPortNb;
            propagateSize = multicastPacketSize;
            propagateInetAddress = InetAddress.getByName(multicastAddress);

            ServerSocket propagateSocket = null;

            multicastSocket = new MulticastSocket(propagatePort);
            multicastSocket.setInterface( usingInterface );
            multicastSocket.joinGroup(propagateInetAddress);
            if (LOG.isEnabledFor(Priority.INFO)) LOG.info( "Multicast : " + propagateInetAddress.getHostAddress() +
                      ":" + propagatePort +
                      " on interface " +
                      multicastSocket.getInterface().getHostAddress() );

            multicastThread = new Thread(this,
                                         "TcpTransport:multicast server thread" );
            multicastThread.start();

            // We're fully ready to function.
            endpoint.addEndpointProtocol(this);

        } catch(Exception e) {
            if (LOG.isEnabledFor(Priority.INFO)) LOG.info("Not initialized: " + e.toString());
            if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("Initialization exception", e);
            if (LOG.isEnabledFor(Priority.WARN)) LOG.warn("FIXME: there may be threads that need killing.");
            throw new PeerGroupException(e.getMessage());
        }
    }


    public void run() {

        try {
            Thread current = Thread.currentThread();

            if (current.equals(unicastThread)) {
                runUnicastServer();
            }

            if (current.equals(multicastThread)) {
                runMulticastServer();
            }
        } catch ( Throwable all ) {
            if (LOG.isEnabledFor(Priority.FATAL)) LOG.fatal( "Uncaught Throwable in thread :" + Thread.currentThread().getName(), all );
        }

    }

    public void runUnicastServer() {

        Socket socket;
        for (;;) {
            try {
                socket = unicastSocket.accept();
		// Configure the socket
		socket.setSoTimeout(MaxKeepAliveDelay);

        //PDA requirement 18.02.2002
        //method did not exist in jdk 1.1.8
		// socket.setKeepAlive (true);
        //PDA requirement 18.02.2002
            } catch (IOException e1) {
                if (LOG.isEnabledFor(Priority.DEBUG)) LOG.debug("ServerSocket.accept() on port " +
                serverSocketPort +
                " has failed with: " + e1.toString());

                continue;