tcptransport.java

jxta_src_2.41b jxta 2.41b 最新版源码 from www.jxta.org

/*
 *
 * $Id: TcpTransport.java,v 1.158 2005/11/28 17:30:52 hamada 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.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.List;

import java.util.Timer;
import java.util.TimerTask;

import java.io.UnsupportedEncodingException;
import java.net.UnknownHostException;
import java.util.NoSuchElementException;

import org.apache.log4j.Level;
import org.apache.log4j.Logger;

import net.jxta.document.Advertisement;
import net.jxta.document.AdvertisementFactory;
import net.jxta.document.Attributable;
import net.jxta.document.Attribute;
import net.jxta.document.Element;
import net.jxta.document.MimeMediaType;
import net.jxta.document.TextElement;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.EndpointService;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.MessageReceiver;
import net.jxta.endpoint.MessageSender;
import net.jxta.endpoint.Messenger;
import net.jxta.endpoint.MessengerEvent;
import net.jxta.endpoint.MessengerEventListener;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.endpoint.WireFormatMessage;
import net.jxta.endpoint.WireFormatMessageFactory;
import net.jxta.id.ID;
import net.jxta.meter.MonitorResources;
import net.jxta.peer.PeerID;
import net.jxta.peergroup.PeerGroup;
import net.jxta.platform.Module;
import net.jxta.protocol.ConfigParams;
import net.jxta.protocol.ModuleImplAdvertisement;
import net.jxta.protocol.TransportAdvertisement;
import net.jxta.util.CountingOutputStream;
import net.jxta.util.DevNullOutputStream;
import net.jxta.util.LimitInputStream;
import net.jxta.util.WatchedStream;

import net.jxta.exception.PeerGroupException;

import net.jxta.impl.endpoint.EndpointServiceImpl;
import net.jxta.impl.endpoint.IPUtils;
import net.jxta.impl.endpoint.LoopbackMessenger;
import net.jxta.impl.endpoint.msgframing.MessagePackageHeader;
import net.jxta.impl.endpoint.msgframing.MessagePackageHeader.Header;
import net.jxta.impl.endpoint.transportMeter.TransportBindingMeter;
import net.jxta.impl.endpoint.transportMeter.TransportMeter;
import net.jxta.impl.endpoint.transportMeter.TransportMeterBuildSettings;
import net.jxta.impl.endpoint.transportMeter.TransportServiceMonitor;
import net.jxta.impl.meter.MonitorManager;
import net.jxta.impl.protocol.TCPAdv;
import net.jxta.impl.util.TimeUtils;
import net.jxta.impl.util.TimerThreadNamer;


/**
 * This class implements the TCP Transport Protocol
 *
 *  @see net.jxta.endpoint.MessageTransport
 *  @see net.jxta.endpoint.MessageSender
 *  @see net.jxta.endpoint.MessageReceiver
 *  @see net.jxta.endpoint.EndpointService
 *  @see <a href="http://spec.jxta.org/v1.0/docbook/JXTAProtocols.html#trans-tcpipt">JXTA Protocols Specification : Standard JXTA Transport Bindings</a>
 */
public class TcpTransport implements Runnable, Module, MessageSender, MessageReceiver {

    /**
     *  Log4j category
     **/
    private static final Logger LOG = Logger.getLogger(TcpTransport.class.getName());
    
    /**
     * The size of the buffer that we use to store message data being sent.
     * Make it the maximum size of a message. (smaller is permitted, though).
     **/
    static final int            SendBufferSize = 64 * 1024; // 64 KBytes
    
    /**
     * The number of bytes that we write to/read from a socket in a single write/read call.
     * Keep this reasonably small since it defines the precision at which we monitor
     * progress. With 8K, any write/read will complete in at most 1 second on a 64Kbit/s connection.
     * Watched Stream start worrying after 10 seconds, so this will still work for connections
     * as slow as 6Kbit/s. Note that we monitor progress of read only when we know data is expected.
     **/
    static final int            ChunkSize = 8 * 1024; // 8 KBytes
    
    /**
     * The buffer size that we instruct TCP to use for incoming data.
     * One full message.
     **/
    static final int            RecvBufferSize = 64 * 1024; // 64 KBytes
    
    // Note: We do not rely on Socket timeout. Java's implementation is crap.
    // These time outs are used to control the behaviour of WatchedInputStream
    // and WatchedOutputStream. These are Filters that monitor their progress.
    
    /**
     * Amount of time our input stream will wait for any kind of progress
     * on input before declaring that input has stalled. This one is for
     * when we're just waiting for a message to come in. So we'll close
     * the connection if it seems to be unused for quite a long time.
     * WatchedOutputStream breaks writes in small enough chunks that progress
     * can be reliably measured with a resolution of a few seconds.
     */
    static final int            LongTimeout = 30 * (int) TimeUtils.AMINUTE;
    
    /**
     * Amount of time our input stream will wait for any kind of progress
     * on input before declaring that input has stalled. This one is for
     * when we know data should be comming.
     * Amount of time our output stream will wait for any kind of progress
     * during a write() before declaring that the output has stalled.
     */
    static final int            ShortTimeout = 10 * (int) TimeUtils.ASECOND;
    
    /**
     *  The amount of time the socket "lingers" after we close it locally.
     *  Linger enables the remote socket to finish receiving any pending data
     *  at its own rate.
     **/
    // xxx: consider referencing a common resource
    //static final int              LingerDelay = 2 * (int) TimeUtils.AMINUTE;
    static final int              LingerDelay = 2 * 60;
    
    /**
     * The amount of time we wait for a connection to be established (effective with jdk 1.4 only).
     * We go get the same system property than URLconnection and default to 20 seconds.
     */
    static int                    connectionTimeOut = 20 * (int) TimeUtils.ASECOND;
    
    static final int              MaxAcceptCnxBacklog = 50; // Java's default is 50
    
    // Connections that are watched often - io in progress
    List                          ShortCycle = Collections.synchronizedList(new ArrayList());
    
    // Connections that are watched rarely - idle or waiting for input
    List                          LongCycle = Collections.synchronizedList(new ArrayList());
    
    private String                serverName = null;
    private List                  publicAddresses = new ArrayList();
    private EndpointAddress       publicAddress = null;
    private MessageElement        msgSrcAddrElement = null;
    
    private String                interfaceAddressStr;
    InetAddress                   usingInterface;
    private int                   serverSocketPort;
    private int                   restrictionPort = -1;
    private IncomingUnicastServer unicastServer = null;
    
    private boolean               isClosed = false;
    
    private boolean               allowMulticast = true;
    private String                multicastAddress = "224.0.1.85";
    private int                   multicastPortNb = 1234;
    private int                   multicastPacketSize = 16384;
    private EndpointAddress       mAddress = null;
    private InetAddress           propagateInetAddress;
    private int                   propagatePort;
    private int                   propagateSize;
    private Thread                multicastThread = null;
    private MulticastSocket       multicastSocket = null;
    
    PeerGroup                     group = null;
    EndpointService               endpoint = null;
    
    private String                protocolName = "tcp";
    private TransportMeter        unicastTransportMeter;
    private TransportMeter        multicastTransportMeter;
    private TransportBindingMeter multicastTransportBindingMeter;
    
    private boolean               publicAddressOnly = false;
    
    private MessengerEventListener messengerEventListener = null;
    
    /**
     *  This is the thread group into which we will place all of the threads
     *  we create. THIS HAS NO EFFECT ON SCHEDULING. Java thread groups are
     *  only for organization and naming.
     **/
    ThreadGroup                   myThreadGroup = null;
    
    final Timer                   connectionWatchTimer;
     
    /**
     * TimerTask used to watch over connections.
     */
    static class Watcher extends TimerTask {
        private Collection watchList;
        public Watcher(Collection watchList) {
            this.watchList = watchList;
        }

        public void run() {
            try {
                WatchedStream[] allStreams = (WatchedStream[]) watchList.toArray(new WatchedStream[0]);
                int len = allStreams.length;

                for (int i = 0; i < len; i++) {
                    allStreams[i].watch();
                }
            } catch (Throwable all) {
                if (LOG.isEnabledFor(Level.FATAL)) {
                    LOG.fatal("Uncaught Throwable in thread :" + Thread.currentThread().getName(), all);
                }
            }
        }
    }
   
    /**
     * Construct a new TCPTransport instance
     **/
    public TcpTransport() {
        try {
            String connectTOStr = System.getProperty("sun.net.client.defaultConnectTimeout");

            if (connectTOStr != null) {
                int i = Integer.parseInt(connectTOStr);

                connectionTimeOut = i;
            }
        } catch (Exception e) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Could not parse system property: sun.net.client.defaultConnectTimeout");
            }
            
            // Keep the default
        }
        
        connectionWatchTimer = new Timer(true);
        connectionWatchTimer.schedule(new TimerThreadNamer("TCP Transport Connection Timer"), 0);
        
        // Setup the timer for the two connection watch lists.
        connectionWatchTimer.schedule(new Watcher(LongCycle), LongTimeout, LongTimeout);
        connectionWatchTimer.schedule(new Watcher(ShortCycle), ShortTimeout, ShortTimeout);
    }
    
    /**
     * {@inheritDoc}
     **/
    public boolean equals(Object target) {
        if (this == target) {
            return true;
        }
        
        if (null == target) {
            return false;
        }