jxtaserverpipe.java

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

/*
 *  $Id: JxtaServerPipe.java,v 1.19 2006/03/24 01:17:22 bondolo 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 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.concurrent.BlockingQueue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;

import java.io.IOException;
import java.net.SocketException;
import java.net.SocketTimeoutException;

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

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.InputStreamMessageElement;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.Messenger;
import net.jxta.endpoint.TextDocumentMessageElement;
import net.jxta.id.IDFactory;
import net.jxta.peergroup.PeerGroup;
import net.jxta.peergroup.PeerGroupID;
import net.jxta.pipe.InputPipe;
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;


/**
 *  JxtaServerPipe is a bi-directional Pipe, that behaves very much like
 *  ServerSocket, it creates a inputpipe and listens for pipe connection requests.
 *  JxtaServerPipe also defines it own protocol, requests arrive as a JXTA Message
 *  with the following elements :
 *
 * <p>
 *  &lt;Cred> to determine whether requestor has the proper access to be granted a connection  &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;/reqPeer>
 *  <p>
 *  &lt;reliable> Reliability setting ("true", or "false") &lt;/reliable>
 *  <p>
 *  &lt;data> data &lt;/data> 
 *  <p>
 *  JxtaServerPipe then creates a new private pipe, and listens for messages on that pipe
 *  resolves the Requestor's pipe, and sends <remPipe> private pipecreated </remotePipe>
 *  advertisement back, where the remove side resolves back.
 */

public class JxtaServerPipe implements PipeMsgListener {

    private static final Logger LOG = Logger.getLogger(JxtaServerPipe.class.getName());
    protected static final String  nameSpace = "JXTABIP";
    protected static final String    credTag = "Cred";
    protected static final String reqPipeTag = "reqPipe";
    protected static final String remPeerTag = "remPeer";
    protected static final String remPipeTag = "remPipe";
    protected static final String   closeTag = "close";
    protected static final String   reliableTag = "reliable";
    private PeerGroup group;
    private InputPipe serverPipe;
    private PipeAdvertisement pipeadv;
    private int backlog = 50;
    private long timeout = 60 * 1000L;
    private final Object closeLock = new String("closeLock");
    protected BlockingQueue queue = null;
    private boolean bound = false;
    private boolean closed = false;
    protected StructuredDocument myCredentialDoc = null;

    /**
     * Default constructor for the JxtaServerPipe
     * <p>
     * backlog default of 50
     * <p>
     * timeout defaults to 60 seconds, i.e. blocking.
     * <p>
     * @param  group                JXTA PeerGroup
     * @param  pipeadv              PipeAdvertisement on which pipe requests are accepted
     * @exception  IOException  if an I/O error occurs
     */
    public JxtaServerPipe(PeerGroup group, PipeAdvertisement pipeadv) throws IOException {
        this(group, pipeadv, 50);
    }

    /**
     *Constructor for the JxtaServerPipe
     *
     * @param  group                JXTA PeerGroup
     * @param  pipeadv              PipeAdvertisement on which pipe requests are accepted
     * @param  backlog              the maximum length of the queue.
     * @param  timeout              the specified timeout, in relative milliseconds
     * @exception  IOException  if an I/O error occurs
     */
    public JxtaServerPipe(PeerGroup group, PipeAdvertisement pipeadv, int backlog, int timeout) throws IOException {
        this(group, pipeadv, backlog);
        this.timeout = timeout;
    }

    /**
     *Constructor for the JxtaServerPipe object
     *
     * @param  group                JXTA PeerGroup
     * @param  pipeadv              PipeAdvertisement on which pipe requests are accepted
     * @param  backlog              the maximum length of the queue.
     ** @exception  IOException  if an I/O error occurs
     */
    public JxtaServerPipe(PeerGroup group, PipeAdvertisement pipeadv, int backlog) throws IOException {
        this.group = group;
        this.pipeadv = pipeadv;
        this.backlog = backlog;
        queue = new ArrayBlockingQueue(backlog);
        PipeService pipeSvc = group.getPipeService();

        serverPipe = pipeSvc.createInputPipe(pipeadv, this);
        setBound();
    }

    /**
     *  Binds the <code>JxtaServerPipe</code> to a specific pipe advertisement
     *
     * @param  group                JXTA PeerGroup
     * @param  pipeadv              PipeAdvertisement on which pipe requests are accepted
     * @exception  IOException  if an I/O error occurs
     */
    public void bind(PeerGroup group, PipeAdvertisement pipeadv) throws IOException {
        bind(group, pipeadv, backlog);
    }

    /**
     *  Binds the <code>JxtaServerPipe</code> to a specific pipe advertisement
     *
     * @param  group                JXTA PeerGroup
     * @param  pipeadv              PipeAdvertisement on which pipe requests are accepted
     * @param  backlog              the maximum length of the queue.
     * @exception  IOException  if an I/O error occurs
     */
    public void bind(PeerGroup group, PipeAdvertisement pipeadv, int backlog) throws IOException {
        this.backlog = backlog;
        this.queue = new ArrayBlockingQueue(backlog);
        this.group = group;
        this.pipeadv = pipeadv;
        PipeService pipeSvc = group.getPipeService();

        serverPipe = pipeSvc.createInputPipe(pipeadv, this);
        setBound();
    }

    /**
     * Listens for a connection to be made to this socket and accepts 
     * it. The method blocks until a connection is made.
     * @return                  JxtaBiDiPipe
     * @exception  IOException  if an I/O error occurs
     */
    public JxtaBiDiPipe accept() throws IOException {
        if (isClosed()) {
            throw new SocketException("JxtaServerPipe is closed");
        }
        if (!isBound()) {
            throw new SocketException("JxtaServerPipe is not bound yet");
        }
        try {
            while (true) {
                Message msg = (Message) queue.poll(timeout, TimeUnit.MILLISECONDS);

                if (msg == null) {
                    throw new SocketTimeoutException("Timeout reached");
                }
                JxtaBiDiPipe bidi = processMessage(msg);

                // make sure we have a socket returning
                if (bidi != null) {
                    return bidi;
                }
            }
        } catch (InterruptedException ie) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Interrupted", ie);
            }
            throw new SocketException("interrupted");
        }
    }

    /**
     *  Gets the group associated with this JxtaServerPipe
     *
     * @return    The group value
     */
    public PeerGroup getGroup() {
        return group;
    }

    /**
     *  Gets the PipeAdvertisement associated with this JxtaServerPipe
     *
     * @return    The pipeAdv value
     */
    public PipeAdvertisement getPipeAdv() {
        return pipeadv;
    }

    /**
     *  Closes this JxtaServerPipe (closes the underlying input pipe).
     *
     * @exception  IOException  if an I/O error occurs
     */
    public void close() throws IOException {
        synchronized (closeLock) {
            if (isClosed()) {
                return;
            }
            if (bound) {
                // close all the pipe
                serverPipe.close();
                queue.clear();