jxtasocket.java

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

/*
 *  $Id: JxtaSocket.java,v 1.65 2006/08/23 22:22:23 hamada Exp $
 *
 *  Copyright (c) 2006 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 THE APACHE SOFTWARE FOUNDATION 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.socket;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.SocketImplFactory;
import java.util.Collections;
import java.util.Iterator;

import net.jxta.credential.Credential;
import net.jxta.document.AdvertisementFactory;
import net.jxta.document.MimeMediaType;
import net.jxta.document.StructuredDocument;
import net.jxta.document.StructuredDocumentFactory;
import net.jxta.document.XMLDocument;
import net.jxta.endpoint.ByteArrayMessageElement;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.EndpointService;
import net.jxta.endpoint.TextDocumentMessageElement;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.Messenger;
import net.jxta.endpoint.OutgoingMessageEvent;
import net.jxta.endpoint.OutgoingMessageEventListener;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.id.ID;
import net.jxta.impl.util.UnbiasedQueue;
import net.jxta.impl.util.pipe.reliable.Defs;
import net.jxta.impl.util.pipe.reliable.FixedFlowControl;
import net.jxta.impl.util.pipe.reliable.OutgoingMsgrAdaptor;
import net.jxta.impl.util.pipe.reliable.ReliableInputStream;
import net.jxta.impl.util.pipe.reliable.ReliableOutputStream;
import net.jxta.membership.MembershipService;
import net.jxta.peer.PeerID;
import net.jxta.peergroup.PeerGroup;
import net.jxta.pipe.InputPipe;
import net.jxta.pipe.OutputPipe;
import net.jxta.pipe.OutputPipeEvent;
import net.jxta.pipe.OutputPipeListener;
import net.jxta.pipe.PipeMsgEvent;
import net.jxta.pipe.PipeMsgListener;
import net.jxta.pipe.PipeService;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.protocol.PipeAdvertisement;

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

/**
 *  JxtaSocket is a bi-directional Pipe that behaves very much like a
 *  Socket, it creates an InputPipe and listens for pipe connection request.
 *  JxtaSocket defines its own protocol. requests arrive as a JXTA Message
 *  with the following elements :
 *
 * <p>
 *  &lt;Cred> Credentials which can be used to determine trust &lt;/Cred>
 * <p>
 *  &lt;reqPipe> requestor's pipe advertisement &lt;/reqPipe>
 * <p>
 *  &lt;remPipe> Remote pipe advertisement &lt;/remPipe>
 * <p>
 *  &lt;reqPeer> Remote peer advertisement &lt;/remPeer>
 * <p>
 *  &lt;stream> determine whether the connection is reliable, or not &lt;/stream>
 * <p>
 *  &lt;close> close request &lt;/close>
 * <p>
 *  &lt;data> Data &lt;/data>
 * <p>
 *
 */
public class JxtaSocket extends Socket implements PipeMsgListener, OutputPipeListener {
    private final static Logger LOG = Logger.getLogger(JxtaSocket.class.getName());
    private final static int MAXRETRYTIMEOUT = 120000;
    protected PeerGroup group;
    protected PipeAdvertisement pipeAdv;
    protected PipeAdvertisement myPipeAdv;
    protected PipeService pipeSvc;
    protected PeerID peerid;
    protected InputPipe in;
    protected OutputPipe connectOutpipe;
    protected Messenger msgr;
    protected InputStream stream;
    /**
	 * The timeout of the read() of this socket's input stream
	 */
    private int soTimeout = 0;
    
    /**
     * timeout for connect and close
     */
    protected int timeout = 60000;
    protected int retryTimeout = 60000;
    protected int maxRetryTimeout = MAXRETRYTIMEOUT;
    protected int windowSize = 20;
    protected final String closeLock  = new String("closeLock");
    protected final String acceptLock = new String("acceptLock");
    protected final String instrLock  = new String("instrLock");
    protected final String finalLock  = new String("finalLock");
    protected boolean closed = false;
    protected boolean bound = false;
    protected final UnbiasedQueue queue = UnbiasedQueue.synchronizedQueue(new UnbiasedQueue(windowSize, false));
    protected Credential credential = null;
    protected StructuredDocument credentialDoc = null;
    protected StructuredDocument myCredentialDoc = null;
    // reliable is the default mode of operation
    protected boolean isStream = true;
    protected OutgoingMsgrAdaptor outgoing = null;
    protected ReliableInputStream ris = null;
    protected ReliableOutputStream ros = null;
    protected boolean waiting;
    private int outputBufferSize = 16384;
    private boolean osCreated = false;
    private InputStream currentMsgStream = null;

    /**
     *  Constructor for the JxtaSocket, this constructor does not establish a connection
     *  use this constructor when altering the default parameters, and options of the socket
     *  by default connections are unreliable, and the default timeout is 60 seconds
     *  to alter a connection a call to create(true) changes the connection to a reliable one.
     */
    public JxtaSocket() { }

    /**
     *  Constructor for the JxtaSocket, this constructor does not establish a connection
     *  use this constructor when altering the default parameters, and options of the socket
     *  by default connections are unreliable, and the default timeout is 60 seconds
     *  to alter a connection a call to create(true) changes the connection to a reliable one.
     *  
     *
     *@param  group            group context
     *@param  msgr             lightweight output pipe
     *@param  pipe             The ephemeral PipeAdvertisement
     *@param  credDoc          remote node's credential StructuredDocument 
     *@exception  IOException  if an io error occurs
     */
    protected JxtaSocket(PeerGroup group,
                         Messenger msgr,
                         PipeAdvertisement pipe,
                         StructuredDocument credDoc,
                         boolean isStream) throws IOException {

        if (msgr == null) {
            throw new IOException("Null Messenger");
        }
        this.group = group;
        this.pipeAdv = pipe;
        this.credentialDoc = credDoc;
        this.credentialDoc = credDoc != null ? credDoc : getCredDoc(group);
        this.pipeSvc = group.getPipeService();
        this.in = pipeSvc.createInputPipe(pipe, this);
        this.msgr = msgr;
        this.isStream = isStream;

        if (isStream) {
            // Force the creation of the inputStream now. So that we do not
            // need a queue, which only benefit would be to store messages
            // until someone calls getInputStream.
            createRis();
            // FIXME: need to delete the input pipe if createRis failed.
            // can it fail ?
        }
        setBound();
    }

    /**
     * Create a JxtaSocket connected to the give JxtaSocketAddress with
     * the default connect timeout (60 seconds).
     * 
     * @param address JxtaSocketAddress to connect to
     * @throws IOException
     */
    public JxtaSocket(SocketAddress address) throws IOException {
        connect(address, timeout);
    }

    /**
     *  Create a JxtaSocket to any peer listening on pipeAdv
     *
     *@param  group            group context
     *@param  pipeAd             PipeAdvertisement
     *@exception  IOException  if an io error occurs
     */
    public JxtaSocket(PeerGroup group,
                      PipeAdvertisement pipeAd) throws IOException {

        this.group = group;
        this.pipeAdv = pipeAd;
        connect(group, pipeAd);
    }

    /**
     * Create a JxtaSocket to the given JxtaSocketAddress, within the timeout
     * specified in milliseconds. 
     * 
     * @param address JxtaSocket address to connect to
     * @param timeout timeout for connection in milliseconds
     * @throws IOException
     */
    public JxtaSocket(SocketAddress address, int timeout) throws IOException {
        connect(address, timeout);
    }

    /**
     *  Create a JxtaSocket to any peer listening on pipeAdv
     *  this attempts establish a connection to specified
     *  pipe within the context of the specified group within
     *  timeout specified in milliseconds
     *
     *@param  group            group context
     *@param  pipeAd           PipeAdvertisement
     *@param  timeout          JxtaSocket connect timeout in milliseconds
     *@exception  IOException  if an io error occurs
     */
    public JxtaSocket(PeerGroup group,
                      PipeAdvertisement pipeAd,
                      int timeout) throws IOException {

        this.group = group;
        this.pipeAdv = pipeAd;
        this.timeout = timeout;
        connect(group, pipeAd, timeout);
    }

    /**
     *  Create a JxtaSocket to any peer listening on pipeAdv
     *  this attempts establish a connection to specified
     *  pipe within a context of group and within timeout specified in milliseconds
     *
     *@param  group            group context
     *@param  peerid           peer to connect to
     *@param  pipeAd           PipeAdvertisement
     *@param  timeout          JxtaSocket connect timeout in milliseconds
     *@exception  IOException  if an io error occurs
     */
    public JxtaSocket(PeerGroup group,
                      PeerID peerid,
                      PipeAdvertisement pipeAd,
                      int timeout) throws IOException {

        this.group = group;
        this.pipeAdv = pipeAd;
        this.timeout = timeout;
        connect(group, peerid, pipeAd, timeout);
    }

    /**
     * Create a JxtaSocket to the given JxtaSocketAddress, within the timeout
     * specified in milliseconds. The JxtaSocket can be reliable (stream) or 
     * not (datagram). If you want to use a SocketAddress in the constructor,
     * this is the preferred method. Either that, or use JxtaSocket(), followed
     * by create(boolean) to turn on reliability, followed by 
     * connect(SocketAddress, int) or connect(SocketAddress) to make the 
     * connection.
     * 
     * @param address JxtaSocket address to connect to
     * @param timeout timeout for connection in milliseconds
     * @param stream true for reliable connection, false o/w
     * @throws IOException
     */
    public JxtaSocket(SocketAddress address,
                      int timeout,
                      boolean stream) throws IOException {
        this.isStream = stream;
        connect(address, timeout);
    }

    /**
     *  Create a JxtaSocket to any peer listening on pipeAdv
     *  this attempts establish a connection to specified
     *  pipe within a context of group and within timeout specified in milliseconds
     *
     *@param  group            group context
     *@param  peerid           peer to connect to
     *@param  pipeAd             PipeAdvertisement
     *@param  timeout          JxtaSocket connect timeout in milliseconds
     *@exception  IOException  if an io error occurs
     */
    public JxtaSocket(PeerGroup group,
                      PeerID peerid,
                      PipeAdvertisement pipeAd,
                      int timeout,
                      boolean stream) throws IOException {

        this.group = group;
        this.pipeAdv = pipeAd;
        this.timeout = timeout;
        this.isStream = stream;
        connect(group, peerid, pipeAd, timeout);
    }

    /**
     *  Creates either a stream or a datagram socket. default is a datagram
     *
     *@param  stream           if <code>true</code>, create a stream socket;
     *      otherwise, create a datagram socket.
     *@exception  IOException  if an I/O error occurs while creating the
     *      socket.
     */
    public void create(boolean stream) throws IOException {
        if (isBound()) {
            throw new IOException("Socket already connected, it is not possible to change connection type");
        }
        this.isStream = stream;
    }

    /**
     * {@inheritDoc}
     *
     *  <p/>Unsupported operation, an IOException will be thrown.
     *
     *  @throws IOException Thrown in all cases as this operation is not supported.
     */
    public void bind(SocketAddress address) throws IOException {
        throw new IOException("Unsupported operation, use java.net.Socket instead");
    }

    /**
     * {@inheritDoc}
     *
     * <p/>The default connect timeout of 60 seconds is used
     * <p/>If SocketAddress is not an instance of JxtaSocketAddress, an 
     * IOException will be thrown.
     */
    public void connect(SocketAddress address) throws IOException {
        connect(address, timeout);
    }

    /**
     * {@inheritDoc}
     *
     * <p/>If SocketAddress is not an instance of JxtaSocketAddress, an 
     * IOException will be thrown.
     */
    public void connect(SocketAddress address, int timeout) throws IOException {
        if (!(address instanceof JxtaSocketAddress)) {
            throw new IOException("Subclass of SocketAddress not supported." +
                                  " Use JxtaSocketAddress instead.");
        }
        JxtaSocketAddress socketAddress = (JxtaSocketAddress) address;
        PeerGroup pg = PeerGroup.globalRegistry.lookupInstance(
                           socketAddress.getPeerGroupId());
        if (pg == null) {
            throw new IOException("Can't connect socket in PeerGroup with id " +
                                  socketAddress.getPeerGroupId().toString() +
                                  ". No running instance of the group is registered.");
        }
        connect(pg.getWeakInterface(), socketAddress.getPeerId(), socketAddress.getPipeAdv(), timeout);
        pg.unref();
    }

    /**
     *  Connects to a remote JxtaSocket on any peer within the default timeout of 60 seconds
     *
     *@param  group            group context
     *@param  pipeAd           PipeAdvertisement
     *@exception  IOException  if an io error occurs
     */
    public void connect(PeerGroup group, PipeAdvertisement pipeAd) throws IOException {