httpclientmessenger.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.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 java.util.logging.Level;
import net.jxta.logging.Logging;
import java.util.logging.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 {
    
    /**
     *  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. This is the amount
     *  of time between our finishing sending a message or beginning a poll and 
     *  the beginning of receipt of a response.
     */
    private final static int RESPONSE_TIMEOUT = (int) (2 * TimeUtils.AMINUTE);
    
    /**
     *  Amount of time we are willing to accept for additional responses. This 
     *  is the total amount of time we are willing to wait after receiving an
     *  initial response message whether additional responses are sent or not.
     *  This setting governs the latency with which we switch back and forth 
     *  between sending and receiving messages. 
     */
    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 = 2;
    
    /**
     *  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);
        
        this.servletHttpTransport = servletHttpTransport;

        EndpointAddress srcAddress = srcAddr;
        this.srcAddressElement = new StringMessageElement(EndpointServiceImpl.MESSAGE_SOURCE_NAME, srcAddr.toString(), null);
        
        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();
        
        // Start receiving messages from the other peer
        poller = new MessagePoller(srcAddr.getProtocolAddress(), destAddr);
                
        if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
            LOG.info("New messenger : " + this );
        }
    }
    
    /*
     * 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}
     *  <p/>
     *  A simple implementation for debugging. <b>Do not parse the String
     *  returned. All of the information is available in other (simpler) ways.</b>
     */
    public String toString() {
        StringBuilder result = new StringBuilder(super.toString());
        result.append(" {");
        result.append(getDestinationAddress());
        result.append(" / ");
        result.append(getLogicalDestinationAddress());
        result.append("}");

        return result.toString();
    }
    
    /**
     *  {@inheritDoc}
     */
    void doShutdown() {
        super.shutdown();
    }
    
    /**
     * {@inheritDoc}
     */
    @Override
    public synchronized void closeImpl() {
        
        if (isClosed()) {
            return;
        }
        
        super.close();
        
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Close messenger to " + senderURL);
        }
        
        MessagePoller stopPoller = poller;
        
        poller = null;
        
        if (null != stopPoller) {
            stopPoller.stop();
        }
    }
    
    /**
     * {@inheritDoc}
     */
    @Override
    public void 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 (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Messenger was closed, it cannot be used to send messages.", failure);
            }
            
            throw failure;
        }
        
        // clone the message before modifying it.
        message = message.clone();
        
        // 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,