tcpmessenger.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 net.jxta.document.MimeMediaType;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.endpoint.WireFormatMessage;
import net.jxta.endpoint.WireFormatMessageFactory;
import net.jxta.id.ID;
import net.jxta.impl.endpoint.BlockingMessenger;
import net.jxta.impl.endpoint.EndpointServiceImpl;
import net.jxta.impl.endpoint.msgframing.MessagePackageHeader;
import net.jxta.impl.endpoint.msgframing.WelcomeMessage;
import net.jxta.impl.endpoint.transportMeter.TransportBindingMeter;
import net.jxta.impl.endpoint.transportMeter.TransportMeterBuildSettings;
import net.jxta.impl.util.TimeUtils;
import net.jxta.logging.Logging;
import net.jxta.peer.PeerID;

import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Implements a messenger which sends messages via raw TCP sockets.
 */
public class TcpMessenger extends BlockingMessenger implements Runnable {

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

    /**
     * The number of times we allow our write selector to be selected with no
     * progress before we give up.
     */
    private static final int MAX_WRITE_ATTEMPTS = 3;

    /**
     * Description of our current location within the stream.
     */
    private enum readState {
        /**
         * Reading the initial welcome
         */
        WELCOME,
        /**
         * Reading a message header
         */
        HEADER,
        /**
         * Reading a message
         */
        BODY
    }

    /**
     * The source address of messages sent on this messenger.
     */
    private final EndpointAddress srcAddress;

    private final MessageElement srcAddressElement;

    /**
     * Cache of the logical destination of this messenger. (It helps if it works even after close)
     */
    private EndpointAddress logicalDestAddress;

    /**
     * The message tcpTransport we are working for.
     */
    private final TcpTransport tcpTransport;

    private EndpointAddress dstAddress = null;
    private EndpointAddress origAddress = null;
    private EndpointAddress fullDstAddress = null;
    private InetAddress inetAddress = null;
    private int port = 0;

    private volatile boolean closed = false;
    private boolean closingDueToFailure = false;

    private WelcomeMessage itsWelcome = null;

    private final long createdAt = TimeUtils.timeNow();
    private long lastUsed = TimeUtils.timeNow();

    private SocketChannel socketChannel = null;

    private TransportBindingMeter transportBindingMeter;

    /**
     * If this is an incoming connection we must not close it when the messenger
     * disappears. It has many reasons to disappear while the connection must
     * keep receiving messages. This is causing some problems for incoming
     * messengers that are managed by some entity, such as the router or the
     * relay. These two do call close explicitly when they discard a messenger,
     * and their intent is truly to close the underlying connection. So
     * basically we need to distinguish between incoming messengers that are
     * abandoned without closing (for these we must protect the input side
     * because that's the only reason for the connection being there) and
     * incoming messengers that are explicitly closed (in which case we must let
     * the entire connection be closed).
     */
    private boolean initiator;

    private AtomicReference<readState> state = new AtomicReference<readState>(readState.WELCOME);

    private final static int MAX_LEN = 4096;
    private ByteBuffer buffer = ByteBuffer.allocate(MAX_LEN);

    /**
     * Header from the current incoming message (if any).
     */
    private MessagePackageHeader header = null;

    /**
     * Time at which we began receiving the current incoming message.
     */
    long receiveBeginTime = 0;

    /**
     * Enforces single writer on message write in case the messenger is being
     * used on multiple threads.
     */
    private final ReentrantLock writeLock = new ReentrantLock();

    /**
     * Create a new TcpMessenger for the specified address.
     *
     * @param socketChannel the SocketChannel for the messenger
     * @param transport     the tcp MessageSender we are working for.
     * @throws java.io.IOException if an io error occurs
     */
    TcpMessenger(SocketChannel socketChannel, TcpTransport transport) throws IOException {
        super(transport.group.getPeerGroupID(),
                new EndpointAddress(transport.getProtocolName(),
                        socketChannel.socket().getInetAddress().getHostAddress() + ":" + socketChannel.socket().getPort(), null, null), true);

        initiator = false;
        this.socketChannel = socketChannel;
        this.tcpTransport = transport;
        this.srcAddress = transport.getPublicAddress();
        this.srcAddressElement = new StringMessageElement(EndpointServiceImpl.MESSAGE_SOURCE_NAME, srcAddress.toString(), null);

        try {
            if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
                LOG.info("Connection from " + socketChannel.socket().getInetAddress().getHostAddress() + ":" + socketChannel.socket().getPort());
            }

            // Set the socket options.
            Socket socket = socketChannel.socket();

            int useBufferSize = Math.max(TcpTransport.SendBufferSize, socket.getSendBufferSize());

            socket.setSendBufferSize(useBufferSize);

            inetAddress = socketChannel.socket().getInetAddress();
            port = socketChannel.socket().getPort();

            socket.setKeepAlive(true);
            socket.setSoTimeout(TcpTransport.connectionTimeOut);
            socket.setSoLinger(true, TcpTransport.LingerDelay);

            // Disable Nagle's algorithm (We do this to reduce latency)
            socket.setTcpNoDelay(true);

            // Temporarily, our address for inclusion in the welcome message response.
            dstAddress = new EndpointAddress(this.tcpTransport.getProtocolName(), inetAddress.getHostAddress() + ":" + port, null, null);
            fullDstAddress = dstAddress;

            startMessenger();
        } catch (IOException io) {
            if (TransportMeterBuildSettings.TRANSPORT_METERING) {
                transportBindingMeter = this.tcpTransport.getUnicastTransportBindingMeter(null, dstAddress);
                if (transportBindingMeter != null) {
                    transportBindingMeter.connectionFailed(initiator, TimeUtils.timeNow() - createdAt);
                }
            }
            // If we failed for any reason, make sure the socket is closed.
            // We're the only one to know about it.
            if (socketChannel != null) {
                socketChannel.close();
            }
            throw io;
        }

        if (TransportMeterBuildSettings.TRANSPORT_METERING) {
            transportBindingMeter = this.tcpTransport.getUnicastTransportBindingMeter((PeerID) getDestinationPeerID(), dstAddress);
            if (transportBindingMeter != null) {
                transportBindingMeter.connectionEstablished(initiator, TimeUtils.timeNow() - createdAt);
            }
        }
        if (!isConnected()) {
            throw new IOException("Failed to establish connection to " + dstAddress);
        }
    }


    /**
     * Create a new TcpMessenger for the specified address.
     *
     * @param destaddr     the destination of the messenger
     * @param tcpTransport the tcp MessageSender we are working for.
     * @throws java.io.IOException if an io error occurs
     */
    TcpMessenger(EndpointAddress destaddr, TcpTransport tcpTransport) throws IOException {
        this(destaddr, tcpTransport, true);
    }

    /**
     * Create a new TcpMessenger for the specified address.
     *
     * @param destaddr     the destination of the messenger
     * @param tcpTransport the tcp MessageSender we are working for.
     * @param selfDestruct indicates whether the messenger created will self destruct when idle
     * @throws java.io.IOException if an io error occurs
     */
    TcpMessenger(EndpointAddress destaddr, TcpTransport tcpTransport, boolean selfDestruct) throws IOException {
        // We need self destruction: tcp messengers are expensive to make and they refer to
        // a connection that must eventually be closed.
        super(tcpTransport.group.getPeerGroupID(), destaddr, selfDestruct);
        this.origAddress = destaddr;

        initiator = true;
        this.tcpTransport = tcpTransport;

        this.fullDstAddress = destaddr;
        this.dstAddress = new EndpointAddress(destaddr, null, null);

        this.srcAddress = tcpTransport.getPublicAddress();

        srcAddressElement = new StringMessageElement(EndpointServiceImpl.MESSAGE_SOURCE_NAME, srcAddress.toString(), null);

        String protoAddr = destaddr.getProtocolAddress();
        int portIndex = protoAddr.lastIndexOf(":");

        if (portIndex == -1) {
            throw new IllegalArgumentException("Invalid Protocol Address (port # missing) ");
        }

        String portString = protoAddr.substring(portIndex + 1);
        try {
            port = Integer.valueOf(portString);
        } catch (NumberFormatException caught) {
            throw new IllegalArgumentException("Invalid Protocol Address (port # invalid): " + portString);
        }

        // Check for bad port number.
        if ((port <= 0) || (port > 65535)) {
            throw new IllegalArgumentException("Invalid port number in Protocol Address : " + port);
        }

        String hostString = protoAddr.substring(0, portIndex);

        inetAddress = InetAddress.getByName(hostString);

        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            LOG.info("Creating new TCP Connection to : " + dstAddress + " / " + inetAddress.getHostAddress() + ":" + port);
        }

        // See if we're attempting to use the loopback address.
        // And if so, is the peer configured for the loopback network only?
        // (otherwise the connection is not permitted). Btw, the otherway around
        // is just as wrong, so we check both at once and pretend it cannot work,
        // even if it might have.
        // FIXME 20041130 This is not an appropriate check if the other peer is
        // running on the same machine and the InetAddress.getByName returns the
        // loopback address.
        // if (inetAddress.isLoopbackAddress() != tcpTransport.usingInterface.isLoopbackAddress()) {
        // throw new IOException("Network unreachable--connect to loopback attempted.");
        // }

        try {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Connecting to " + inetAddress.getHostAddress() + ":" + port + " via "
                        + this.tcpTransport.usingInterface.getHostAddress() + ":0");
            }

            socketChannel = SocketChannel.open();
            Socket socket = socketChannel.socket();

            // Bind it to our outbound interface.
            SocketAddress bindAddress = new InetSocketAddress(this.tcpTransport.usingInterface, 0);

            socket.bind(bindAddress);
            // Set Socket options.
            int useBufferSize = Math.max(TcpTransport.SendBufferSize, socket.getSendBufferSize());
            socket.setSendBufferSize(useBufferSize);

            useBufferSize = Math.max(TcpTransport.RecvBufferSize, socket.getReceiveBufferSize());
            socket.setReceiveBufferSize(useBufferSize);

            socket.setKeepAlive(true);
            socket.setSoTimeout(TcpTransport.connectionTimeOut);
            socket.setSoLinger(true, TcpTransport.LingerDelay);
            // Disable Nagle's algorithm (We do this to reduce latency)
            socket.setTcpNoDelay(true);

            SocketAddress connectAddress = new InetSocketAddress(inetAddress, port);
            socketChannel.connect(connectAddress);