tcptransport.java

JXTA&#8482 is a set of open, generalized peer-to-peer (P2P) protocols that allow any networked devi

/*
 * Copyright (c) 2001-2007 Sun Microsystems, Inc.  All rights reserved.
 *  
 *  The Sun Project JXTA(TM) Software License
 *  
 *  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 Sun Microsystems, Inc. for JXTA(TM) technology." 
 *     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.
 *  
 *  JXTA is a registered trademark of Sun Microsystems, Inc. in the United 
 *  States and other countries.
 *  
 *  Please see the license information page at :
 *  <http://www.jxta.org/project/www/license.html> for instructions on use of 
 *  the license in source files.
 *  
 *  ====================================================================
 *  
 *  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.IOException;
import java.io.InterruptedIOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.IllegalBlockingModeException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.EmptyStackException;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.Stack;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.logging.Level;
import java.util.logging.Logger;

import net.jxta.document.Advertisement;
import net.jxta.document.AdvertisementFactory;
import net.jxta.document.Attribute;
import net.jxta.document.XMLElement;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.EndpointService;
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.exception.PeerGroupException;
import net.jxta.id.ID;
import net.jxta.logging.Logging;
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.impl.endpoint.IPUtils;
import net.jxta.impl.endpoint.LoopbackMessenger;
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.peergroup.StdPeerGroup;
import net.jxta.impl.protocol.TCPAdv;
import net.jxta.impl.util.TimeUtils;


/**
 * This class implements the TCP Message Transport.
 *
 * @see net.jxta.endpoint.MessageTransport
 * @see net.jxta.endpoint.MessagePropagater
 * @see net.jxta.endpoint.MessageReceiver
 * @see net.jxta.endpoint.MessageSender
 * @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 Module, MessageSender, MessageReceiver {

    /**
     * Logger
     */
    private static final Logger LOG = Logger.getLogger(TcpTransport.class.getName());

    /**
     * The TCP send buffer size.
     * The size of the buffer used to store outgoing messages
     * This should be set to the maximum message size (smaller is allowed).
     */
    static final int SendBufferSize = 64 * 1024; // 64 KBytes

    /**
     * The TCP receive buffer size
     */
    static final int RecvBufferSize = 64 * 1024; // 64 KBytes

    /**
     * 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.
     * Note: LingerDelay time unit is seconds
     */
    static final int LingerDelay = 2 * 60;

    /**
     * Connection  timeout
     * use the same system property defined by URLconnection, otherwise default to 10 seconds.
     */
    static int connectionTimeOut = 10 * (int) TimeUtils.ASECOND;

    // Java's default is 50
    static final int MaxAcceptCnxBacklog = 50;

    private String serverName = null;
    private final List<EndpointAddress> publicAddresses = new ArrayList<EndpointAddress>();
    private EndpointAddress publicAddress = null;

    private String interfaceAddressStr;
    InetAddress usingInterface;
    private int serverSocketPort;
    private int restrictionPort = -1;
    private IncomingUnicastServer unicastServer = null;

    private boolean isClosed = false;

    private long messagesSent = 0;
    private long messagesReceived = 0;
    private long bytesSent = 0;
    private long bytesReceived = 0;
    private long connectionsAccepted = 0;

    PeerGroup group = null;
    EndpointService endpoint = null;
    Executor executor;

    private String protocolName = "tcp";
    private TransportMeter unicastTransportMeter;
    private TransportMeter multicastTransportMeter;

    private boolean publicAddressOnly = false;

    private MessengerEventListener messengerEventListener = null;

    private Thread messengerSelectorThread;
    Selector messengerSelector = null;

    private final Map<TcpMessenger, SocketChannel> regisMap = new ConcurrentHashMap<TcpMessenger, SocketChannel>();
    private final Set<SocketChannel> unregisMap = Collections.synchronizedSet(new HashSet<SocketChannel>());

    /**
     * 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;

    /**
     * The maximum number of write selectors we will maintain in our cache per
     * transport instance.
     */
    protected final static int MAX_WRITE_SELECTORS = 50;

    /**
     * A cache we maintain for selectors writing messages to the socket.
     */
    private final static Stack<Selector> writeSelectorCache = new Stack<Selector>();

    /**
     * The number of excess write selectors believed to be in the pool.
     */
    private int extraWriteSelectors = 0;

    /**
     * Construct a new TcpTransport instance
     */
    public TcpTransport() {
        // Add some selectors to the pool.
        try {
            for (int i = 0; i < MAX_WRITE_SELECTORS; i++) {
                writeSelectorCache.add(Selector.open());
            }
        } catch (IOException ex) {
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.severe("Failed adding selector to  write selector pool");
            }
        }

        try {
            String connectTOStr = System.getProperty("sun.net.client.defaultConnectTimeout");

            if (connectTOStr != null) {
                connectionTimeOut = Integer.parseInt(connectTOStr);
            }
        } catch (Exception e) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("Could not parse system property: sun.net.client.defaultConnectTimeout");
            }
        }
    }

    /**
     * Gets the number of 'connectionsAccepted'.
     *
     * @return the number of 'connectionsAccepted'.
     */
    public long getConnectionsAccepted() {
        return connectionsAccepted;
    }

    /**
     * increment the number of connectionsAccepted sent by 1
     */
    public void incrementConnectionsAccepted() {
        connectionsAccepted++;
    }

    /**
     * increment the number of messages sent by 1
     */
    public void incrementMessagesSent() {
        messagesSent++;
    }

    /**
     * increment the number of messages received by 1
     */
    public void incrementMessagesReceived() {
        messagesReceived++;
    }

    /**
     * increment the number of bytes sent
     *
     * @param bytes the number of bytes to be added
     */
    public void incrementBytesSent(long bytes) {
        bytesSent += bytes;
    }

    /**
     * increment the number of bytes received
     *
     * @param bytes the number of bytes to be added
     */
    public void incrementBytesReceived(long bytes) {
        bytesReceived += bytes;
    }

    /**
     * Gets the number of 'messagesSent'.
     *
     * @return the number of 'messagesSent'.
     */
    public long getMessagesSent() {