httpmessageservlet.java

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

/*
 *
 * $Id: HttpMessageServlet.java,v 1.35 2006/06/09 22:11:28 hamada 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.util.Enumeration;

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

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

import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
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.meter.*;

import net.jxta.impl.endpoint.EndpointServiceImpl;
import net.jxta.impl.endpoint.transportMeter.*;
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 {

    /**
     * Log4J 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.
     */
    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}
     */
    public void doGet(HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException {

        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("GET " + req.getRequestURI() + " thread = " + Thread.currentThread());
        }

        processRequest(req, res);

        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("GET done for thread = " + Thread.currentThread());
        }
    }

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

        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("POST " + req.getRequestURI() + " thread = " + Thread.currentThread());
        }

        processRequest(req, res);

        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("POST done for thread = " + Thread.currentThread());
        }
    }

    /**
     *  {@inheritDoc}
     */
    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 = 0;
        int requestSize = 0;
        TransportBindingMeter transportBindingMeter = null;

        //        if (LOG.isEnabledFor(Level.TRACE)) {
        //            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 (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("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);
            }