httpmessageservlet.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.util.Enumeration;

import java.io.IOException;
import java.util.Enumeration;
import java.util.NoSuchElementException;

import java.util.logging.Logger;
import java.util.logging.Level;
import net.jxta.logging.Logging;

import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
import javax.servlet.http.Cookie;
// import javax.servlet.http.Cookie;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import net.jxta.document.MimeMediaType;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.EndpointService;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.Messenger;
import net.jxta.endpoint.WireFormatMessage;
import net.jxta.endpoint.WireFormatMessageFactory;

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;

/**
 *  This is a simple servlet that accepts JXTA Messages from clients using HTTP
 *  via {@code POST}. In addition to receiving messages via {@code POST}
 *  responses clients can also poll for messages using {@code GET}.
 *
 *  <p/>It also supports a ping operation. When the URI is <tt>/</tt> the
 *  response consists of the unique value portion of the local peer id.
 */
public class HttpMessageServlet extends HttpServlet {

    /**
     * Logger
     */
    private final static transient Logger LOG = Logger.getLogger(HttpMessageServlet.class.getName());

    /**
     *  The maximum duration in milliseconds we will keep a connection alive
     *  while generating a response.
     */
    private final static long MAXIMUM_RESPONSE_DURATION = 2 * TimeUtils.AMINUTE;

    /**
     *  Owner of this servlet.
     */
    private HttpMessageReceiver owner = null;

    /**
     *  The endpoint that the owner message transport is registered with.
     */
    private EndpointService endpoint = null;

    /**
     *  Our address.
     */
    private EndpointAddress localAddress = null;
    private byte[] pingResponseBytes;

    private ServletHttpTransport servletHttpTransport = null;

    /**
     *  If {@code true} then this servlet has been (or is being) destroyed.
     */
    private volatile boolean destroyed = false;

    /**
     *  Recovers the Message Transport which owns this servlet from the context
     *  information.
     */
    @Override
    public void init(ServletConfig config) throws ServletException {
        super.init(config);

        try {
            owner = (HttpMessageReceiver) getServletContext().getAttribute("HttpMessageReceiver");
            if (owner == null) {
                throw new ServletException("Servlet Context did not contain 'HttpMessageReceiver'");
            }
        } catch (ClassCastException e) {
            throw new ServletException("'HttpMessageReceiver' attribute was not of the proper type in the Servlet Context");
        }

        servletHttpTransport = owner.servletHttpTransport;
        endpoint = owner.getEndpointService();

        String peerId = endpoint.getGroup().getPeerID().getUniqueValue().toString();

        localAddress = new EndpointAddress("jxta", peerId, null, null);

        try {
            pingResponseBytes = peerId.getBytes("UTF-8");
        } catch (java.io.UnsupportedEncodingException never) {
            // UTF-8 is always available.
        }
    }

    /**
     *  {@inheritDoc}
     */
    @Override
    public void doGet(HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException {

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("GET " + req.getRequestURI() + " thread = " + Thread.currentThread());
        }

        processRequest(req, res);

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("GET done for thread = " + Thread.currentThread());
        }
    }

    /**
     *  {@inheritDoc}
     */
    @Override
    public void doPost(HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException {

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("POST " + req.getRequestURI() + " thread = " + Thread.currentThread());
        }

        processRequest(req, res);

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("POST done for thread = " + Thread.currentThread());
        }
    }

    /**
     *  {@inheritDoc}
     */
    @Override
    public synchronized void destroy() {
        // All we need to do is wakeup the threads that are waiting. (In truth
        // we'll miss those that are waiting on a messenger, but that'll do for
        // now, because we do that only when shutting down the group and then
        // the relay will be shutdown as well, which will take care of the
        // messengers.
        destroyed = true;
        notifyAll();
    }

    /**
     *  Handle a request and optionally generate a response.
     *
     *  @param req The request we are being asked to process.
     *  @param res The response we should use for any response.
     *  @throws IOException For failures in IO processing.
     */
    private void processRequest(HttpServletRequest req, HttpServletResponse res) throws IOException {

        long lastReadWriteTime;
        int requestSize = 0;
        TransportBindingMeter transportBindingMeter = null;

        if (Logging.SHOW_FINEST && LOG.isLoggable(Level.FINEST)) {
            printRequest(req);
        }

        if (TransportMeterBuildSettings.TRANSPORT_METERING) {
            int contentLength = req.getContentLength();

            requestSize += (contentLength != -1) ? contentLength : 0;
        }

        JxtaRequest currentRequest = new JxtaRequest(req);

        /*
         *  We accept three request formats:
         *
         *  o PING : no Message and no Requestor defined.
         *  o POLL : Requestor defined, positive response timeout and destination address.
         *  o SEND : Requestor defined, positive response timeout or no destination address.
         */

        // check if this is a ping request, no requestor peerId or incoming message
        if (null == currentRequest.requestorAddr && !currentRequest.messageContent) {
            // this is only a ping request
            pingResponse(res);

            if (TransportMeterBuildSettings.TRANSPORT_METERING) {
                long connectionTime = TimeUtils.toRelativeTimeMillis(TimeUtils.timeNow(), currentRequest.requestStartTime);
                EndpointAddress sourceAddress = new EndpointAddress("http", req.getRemoteHost(), null, null); //

                transportBindingMeter = servletHttpTransport.getTransportBindingMeter(null, sourceAddress);
                if (transportBindingMeter != null) {
                    transportBindingMeter.connectionEstablished(false, currentRequest.requestStartTime);
                    transportBindingMeter.dataReceived(false, requestSize);
                    transportBindingMeter.dataSent(false, 0);
                    transportBindingMeter.pingReceived();
                    transportBindingMeter.connectionClosed(false, connectionTime);
                }
            }

            return;
        }

        if (TransportMeterBuildSettings.TRANSPORT_METERING) {
            lastReadWriteTime = TimeUtils.timeNow();
            long connectTime = TimeUtils.toRelativeTimeMillis(TimeUtils.timeNow(), currentRequest.requestStartTime);
            EndpointAddress sourceAddress = new EndpointAddress("http", req.getRemoteHost(), null, null); //

            if (null != currentRequest.requestorAddr) {
                transportBindingMeter = servletHttpTransport.getTransportBindingMeter(currentRequest.requestorAddr.toString(),
                        sourceAddress);
            } else {
                transportBindingMeter = servletHttpTransport.getTransportBindingMeter(req.getRemoteHost(), sourceAddress);
            }
            if (transportBindingMeter != null) {
                transportBindingMeter.connectionEstablished(false, connectTime);
                transportBindingMeter.dataReceived(false, requestSize);
            }
        }

        // check if the request included polling (valid requestor peerId and timeout not -1)
        HttpServletMessenger messenger = null;

        if ((null != currentRequest.requestorAddr) && (currentRequest.responseTimeout >= 0) && (null != currentRequest.destAddr)) {
            // create the back channel messenger
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine( "Creating back channel messenger for " + currentRequest.requestorAddr + " (" + currentRequest.destAddr + ")");
            }

            long messengerAliveFor;

            if (0 == currentRequest.responseTimeout) {
                messengerAliveFor = 0;
            } else {
                messengerAliveFor = Math.max(currentRequest.responseTimeout, currentRequest.extraResponsesTimeout);