httpclientmessenger.java

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

/*
 *
 * $Id: HttpClientMessenger.java,v 1.60 2006/03/06 20:43:31 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 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.
 *
 * ====================================================================
 *
 * 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.servlethttp;


import java.io.InputStream;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;

import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.io.UnsupportedEncodingException;
import java.net.MalformedURLException;
import java.net.SocketTimeoutException;

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

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.impl.endpoint.BlockingMessenger;
import net.jxta.impl.endpoint.EndpointServiceImpl;
import net.jxta.impl.endpoint.transportMeter.TransportBindingMeter;
import net.jxta.impl.endpoint.transportMeter.TransportMeterBuildSettings;
import net.jxta.impl.util.TimeUtils;


/**
 *  Simple messenger that simply posts a message to a URL.
 *
 *  <p/>URL/HttpURLConnection is used, so (depending on your JDK) you will get
 *  reasonably good persistent connection management.
 */
final class HttpClientMessenger extends BlockingMessenger {
    
    /**
     *    Log4J Logger
     **/
    private final static transient Logger LOG = Logger.getLogger(HttpClientMessenger.class.getName());
    
    /**
     *  Minimum amount of time between poll
     **/
    private final static int MIMIMUM_POLL_INTERVAL = (int) (5 * TimeUtils.ASECOND);
    
    /**
     *  Amount of time to wait for connections to open.
     **/
    private final static int CONNECT_TIMEOUT = (int) (15 * TimeUtils.ASECOND);
    
    /**
     *  Amount of time we are willing to wait for responses.
     **/
    private final static int RESPONSE_TIMEOUT = (int) (2 * TimeUtils.AMINUTE);
    
    /**
     *  Amount of time we are willing to wait for responses.
     **/
    private final static int EXTRA_RESPONSE_TIMEOUT = (int) (2 * TimeUtils.AMINUTE);
    
    /**
     *  Messenger idle timeout.
     **/
    private final static long MESSENGER_IDLE_TIMEOUT = 15 * TimeUtils.AMINUTE;
    
    
    /**
     *  Number of attempts we will attempt to make connections.
     **/
    private final static int CONNECT_RETRIES = 3;
    
    /**
     *  Warn only once about obsolete proxies.
     **/
    private static boolean neverWarned = true;
    
    /**
     *  The URL we send messages to.
     **/
    private final URL senderURL;
    
    /**
     * The ServletHttpTransport that created this object.
     **/
    private final ServletHttpTransport servletHttpTransport;
    
    /**
     *  The Return Address element we will add to all messages we send.
     **/
    private final MessageElement srcAddressElement;
    
    /**
     *  The logical destination address of this messenger.
     **/
    private final EndpointAddress logicalDest;
    
    private TransportBindingMeter transportBindingMeter;
    
    /**
     *  The last time at which we successfully received or sent a message.
     **/
    private transient long lastUsed = TimeUtils.timeNow();
    
    /**
     *  Poller that we use to get our messages.
     **/
    private MessagePoller poller = null;
    
    /**
     *  Constructs the messenger.
     *
     *  @param servletHttpTransport The transport this messenger will work for.
     *  @param srcAddr The source address.
     *  @param destAddr The destination address.
     **/
    HttpClientMessenger(ServletHttpTransport servletHttpTransport,
            EndpointAddress srcAddr,
            EndpointAddress destAddr) throws IOException {
        
        // We do use self destruction.
        super(servletHttpTransport.getEndpointService().getGroup().getPeerGroupID(), destAddr, true);
        
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("Creating messenger for " + destAddr );
        }
        
        this.servletHttpTransport = servletHttpTransport;
        
        String protoAddr = destAddr.getProtocolAddress();
        
        String host;
        int port;
        int lastColon = protoAddr.lastIndexOf(':');
        
        if ((-1 == lastColon) || (lastColon < protoAddr.lastIndexOf(']')) || ((lastColon + 1) == protoAddr.length())) {
            // There's no port or it's an IPv6 addr with no port or the colon is the last character.
            host = protoAddr;
            port = 80;
        } else {
            host = protoAddr.substring(0, lastColon);
            port = Integer.parseInt(protoAddr.substring(lastColon + 1));
        }
        
        senderURL = new URL("http", host, port, "/");
        
        logicalDest = retreiveLogicalDestinationAddress();
        
        this.srcAddressElement = new StringMessageElement(EndpointServiceImpl.MESSAGE_SOURCE_NAME, srcAddr.toString(), null);
        
        // Start receiving messages from the other peer
        poller = new MessagePoller( srcAddr.getProtocolAddress(), destAddr );
    }
    
    /*
     * The cost of just having a finalize routine is high. The finalizer is
     * a bottleneck and can delay garbage collection all the way to heap
     * exhaustion. Leave this comment as a reminder to future maintainers.
     * Below is the reason why finalize is not needed here.
     *
     * These messengers (normally) never go to the application layer. Endpoint
     * code does call close when necessary.
     
     protected void finalize() {
     }
     
     */
    
    /**
     *  {@inheritDoc}
     **/
    void doShutdown() {
        super.shutdown();
    }
    
    /**
     * {@inheritDoc}
     **/
    public synchronized void closeImpl() {
        
        if (isClosed()) {
            return;
        }
        
        super.close();
        
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("Close messenger to " + senderURL);
        }
        
        MessagePoller stopPoller = poller;
        
        poller = null;
        
        if (null != stopPoller) {
            stopPoller.stop();
        }
    }
    
    /**
     * {@inheritDoc}
     **/
    public boolean sendMessageBImpl(Message message, String service, String serviceParam) throws IOException {
        
        if (isClosed()) {
            IOException failure = new IOException("Messenger was closed, it cannot be used to send messages.");
            
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn( "Messenger was closed, it cannot be used to send messages.", failure);
            }
            
            throw failure;
        }
        
        // Set the message with the appropriate src and dest address
        message.replaceMessageElement(EndpointServiceImpl.MESSAGE_SOURCE_NS, srcAddressElement);
        
        EndpointAddress destAddressToUse = getDestAddressToUse(service, serviceParam);
        
        MessageElement dstAddressElement = new StringMessageElement(EndpointServiceImpl.MESSAGE_DESTINATION_NAME, destAddressToUse.toString(),
                (MessageElement) null);
        
        message.replaceMessageElement(EndpointServiceImpl.MESSAGE_DESTINATION_NS, dstAddressElement);
        
        try {
            return doSend(message);
        } catch (IOException e) {
            // close this messenger
            close();
            // rethrow the exception
            throw e;
        }
    }
    
    /**
     * {@inheritDoc}