jxtabidipipe.java

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

/*
 *  $Id: JxtaBiDiPipe.java,v 1.33 2006/06/14 16:30:09 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.util;

import java.io.InputStream;
import java.util.Collections;
import java.util.Iterator;

import java.io.IOException;

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.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.id.ID;
import net.jxta.peer.PeerID;
import net.jxta.pipe.PipeID;
import net.jxta.membership.MembershipService;
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 net.jxta.impl.util.UnbiasedQueue;
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.impl.util.pipe.reliable.FixedFlowControl;
import net.jxta.impl.util.pipe.reliable.Defs;

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

/**
 * JxtaBiDiPipe is a bi-directional Pipe, it creates an InputPipe for incoming
 * Messages, and EndpointMessenger for outgoing messages
 * JxtaBiDiPipe defines its own protocol for negotiating connections. Connection
 * requests arrive as a JXTA Message with the following elements :
 *
 *  <p>
 *  &lt;Credential> to determine whether requestor has the proper access to be 
 *              granted a connection
 *  <p>
 *  &lt;reqPipe> Requestor's pipe advertisement &lt;/reqPipe>
 *  <p>
 *  &lt;remPipe> remote pipe advertisement &lt;/remPipe>
 *  <p>
 *  &lt;remPeer> remote peer advertisement &lt;/remPeer>
 *  <p>
 *  &lt;reliable> Reliability setting ("true", or "false") &lt;/reliable>
 *  <p>
 *  &lt;data> data &lt;data>
 *
 */
public class JxtaBiDiPipe implements PipeMsgListener,
            OutputPipeListener,
    ReliableInputStream.MsgListener {

    private final static Logger LOG = Logger.getLogger(JxtaBiDiPipe.class.getName());
    private final static int MAXRETRYTIMEOUT = 120000;
    private PipeAdvertisement remotePipeAdv;
    private PeerID peerid;
    private PeerAdvertisement remotePeerAdv;
    protected int timeout = 60000;
    protected int retryTimeout = 60000;
    protected int maxRetryTimeout = MAXRETRYTIMEOUT;
    private UnbiasedQueue queue;
    protected int windowSize = 20;
    protected PeerGroup group;
    protected PipeAdvertisement pipeAdv;
    protected PipeAdvertisement myPipeAdv;
    protected PipeService pipeSvc;
    protected InputPipe in;
    protected OutputPipe connectOutpipe;
    protected Messenger msgr;
    protected InputStream stream;
    protected final String closeLock  = new String("closeLock");
    protected final String acceptLock = new String("acceptLock");
    protected final String finalLock  = new String("finalLock");
    protected boolean closed = false;
    protected boolean bound = false;
    protected PipeMsgListener msgListener;
    protected PipeEventListener eventListener;
    protected Credential credential = null;
    protected boolean waiting;
    protected boolean isReliable = false;
    protected OutgoingMsgrAdaptor outgoing = null;
    protected ReliableInputStream ris = null;
    protected ReliableOutputStream ros = null;
    protected StructuredDocument credentialDoc = null;
    protected StructuredDocument myCredentialDoc = null;

    /**
     * Pipe close Event
     */
     public static final int PIPE_CLOSED_EVENT = 1;
     
    /**
     *  JxtaBiDiPipe A bidirectional pipe
     *
     *@param  group            group context
     *@param  msgr             lightweight output pipe
     *@param  pipe             PipeAdvertisement
     *@param  isReliable       Whether the connection is reliable or not
     *@exception  IOException  if an io error occurs
     */
    protected JxtaBiDiPipe(PeerGroup group,
                           Messenger msgr,
                           PipeAdvertisement pipe,
                           StructuredDocument credDoc,
                           boolean isReliable) 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.isReliable = isReliable;
        queue = UnbiasedQueue.synchronizedQueue(new UnbiasedQueue(windowSize, false));
        createRLib();
        setBound();
    }

    /**
     * JxtaBiDiPipe A bidirectional pipe
     * Creates a new object with a default timeout of 60,000ms, and a reliability 
     * setting of false
     *
     */
    public JxtaBiDiPipe() {}

    /**
     * attempts to create a bidirectional connection to remote peer within default
     * timeout of 60,000ms, and initiates a connection
     *
     *@param  group            group context
     *@param  pipeAd           PipeAdvertisement
     *@param msgListener          application PipeMsgListener 
     *@exception  IOException  if an io error occurs
     */
    public JxtaBiDiPipe(PeerGroup group,
                        PipeAdvertisement pipeAd,
                        PipeMsgListener msgListener) throws IOException {

        connect(group, null, pipeAd, timeout, msgListener);
    }

    /**
     * attempts to create a bidirectional connection to remote peer within default
     * timeout of 1 minutes, and initiates a connection
     *
     *@param  group            group context
     *@param  pipeAd           PipeAdvertisement
     *@param msgListener          application PipeMsgListener 
     *@exception  IOException  if an io error occurs
     */
    public JxtaBiDiPipe(PeerGroup group,
                        PipeAdvertisement pipeAd,
                        int timeout,
                        PipeMsgListener msgListener) throws IOException {

        connect(group, null, pipeAd, timeout, msgListener);
    }

    /**
     * attempts to create a bidirectional connection to remote peer within default
     * timeout of 60,000ms, and initiates a connection
     *
     *@param  group            group context
     *@param  pipeAd           PipeAdvertisement
     *@param msgListener          application PipeMsgListener 
     *@exception  IOException  if an io error occurs
     */
    public JxtaBiDiPipe(PeerGroup group,
                        PipeAdvertisement pipeAd,
                        int timeout,
                        PipeMsgListener msgListener,
                        boolean reliable) throws IOException {

        connect(group, null, pipeAd, timeout, msgListener, reliable);
    }


    /**
     *  Connect to JxtaBiDiPipe with default timeout
     *
     *@param  group            group context
     *@param  pipeAd           PipeAdvertisement
     *@exception  IOException  if an io error occurs
     */
    public void connect(PeerGroup group, PipeAdvertisement pipeAd) throws IOException {
        connect(group, pipeAd, timeout);
    }


    /**
     *  Connects to a remote JxtaBiDiPipe
     *
     *@param  group            group context
     *@param  pipeAd           PipeAdvertisement
     *@param  timeout          timeout in ms, also reset object default timeout
                               to that of timeout
     *@exception  IOException  if an io error occurs
     */
    public void connect(PeerGroup group,
                        PipeAdvertisement pipeAd,
                        int timeout) throws IOException {

        connect(group, null, pipeAd, timeout, null);
    }

    /**
     *  Connects to a remote JxtaBiDiPipe
     *
     *@param  group            group context
     *@param  peerid           peer to connect to
     *@param  pipeAd           PipeAdvertisement
     *@param  timeout          timeout in ms, also reset object default timeout to that of timeout
     *@exception  IOException  if an io error occurs
     */
    public void connect(PeerGroup group,
                        PeerID peerid,
                        PipeAdvertisement pipeAd,
                        int timeout,
                        PipeMsgListener msgListener) throws IOException {

        connect(group, peerid, pipeAd, timeout, msgListener, isReliable);
    }

    /**
     *  Connects to a remote JxtaBiDiPipe
     *
     *@param  group            group context
     *@param  peerid           peer to connect to
     *@param  pipeAd           PipeAdvertisement
     *@param  timeout          timeout in ms, also reset object default timeout to that of timeout
     *@param  reliable         Reliable connection
     *@exception  IOException  if an io error occurs
     */
    public void connect(PeerGroup group,
                        PeerID peerid,
                        PipeAdvertisement pipeAd,
                        int timeout,
                        PipeMsgListener msgListener,
                        boolean reliable) throws IOException {

        if (isBound()) {
            throw new IOException("Pipe already bound");
        }
        if (timeout <= 0) {
            throw new IllegalArgumentException("Invalid timeout :"+timeout);
        }

        this.pipeAdv = pipeAd;
        this.group = group;
        this.msgListener = msgListener;
        this.isReliable = reliable;
        pipeSvc = group.getPipeService();
        this.timeout = timeout;
        this.peerid = peerid;
        myPipeAdv = JxtaServerPipe.newInputPipe(group, pipeAd);
        this.in = pipeSvc.createInputPipe(myPipeAdv, this);
        this.credentialDoc = getCredDoc(group);
        Message openMsg = createOpenMessage(group, myPipeAdv);
        // create the output pipe and send this message
        if (peerid == null) {
            pipeSvc.createOutputPipe(pipeAd, this);
        } else {
            pipeSvc.createOutputPipe(pipeAd, Collections.singleton(peerid), this);
        }
        try {
            synchronized (acceptLock) {
                // check connectOutpipe within lock to prevent a race with modification.
                if (connectOutpipe == null) {
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("Waiting for "+timeout+" msec");
                    }
                    acceptLock.wait(timeout);
                }
            }
        } catch (InterruptedException ie) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Interrupted", ie);
            }
        }
        if (connectOutpipe == null) {
            throw new IOException("connection timeout");
        }
        // send connect message
        waiting = true;
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("Sending a backchannel message");
        }
        connectOutpipe.send(openMsg);
        //wait for the second op
        try {
            synchronized (finalLock) {
                if(waiting) {
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("Waiting for "+timeout+" msec for back channel to be established");
                    }
                    finalLock.wait(timeout);
                    //Need to check for creation
                    if (msgr == null) {
                        throw new IOException("connection timeout");
                    }
                }
            }
        } catch (InterruptedException ie) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Interrupted", ie);
            }
            throw new IOException("Interrupted");
        }
        if (msgListener == null) {
            queue = UnbiasedQueue.synchronizedQueue(new UnbiasedQueue());
        }
        setBound();
    }

    /**
     * creates all the reliability objects
     */
    private void createRLib() {
        if(isReliable) {
            if(outgoing == null) {
                outgoing = new OutgoingMsgrAdaptor(msgr, retryTimeout);
            }
            if (ros == null) {
                ros = new ReliableOutputStream(outgoing, new FixedFlowControl(windowSize));
            }
            if (ris == null) {
                ris = new ReliableInputStream(outgoing, retryTimeout, this);
            }