clienthandler.java

来自「开源的axis2框架的源码。用于开发WEBSERVER」· Java 代码 · 共 420 行 · 第 1/2 页

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.axis2.transport.nhttp;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;

import org.apache.axiom.soap.SOAP11Constants;
import org.apache.axiom.soap.SOAP12Constants;
import org.apache.axiom.soap.SOAPFactory;
import org.apache.axis2.addressing.AddressingConstants;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.MessageContext;
import org.apache.axis2.description.WSDL2Constants;
import org.apache.axis2.engine.MessageReceiver;
import org.apache.axis2.transport.nhttp.util.PipeImpl;
import org.apache.axis2.transport.nhttp.util.WorkerPool;
import org.apache.axis2.transport.nhttp.util.WorkerPoolFactory;
import org.apache.axis2.wsdl.WSDLConstants;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.http.ConnectionReuseStrategy;
import org.apache.http.Header;
import org.apache.http.HttpConnection;
import org.apache.http.HttpException;
import org.apache.http.HttpRequest;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.HttpVersion;
import org.apache.http.entity.BasicHttpEntity;
import org.apache.http.impl.DefaultConnectionReuseStrategy;
import org.apache.http.nio.ContentDecoder;
import org.apache.http.nio.ContentEncoder;
import org.apache.http.nio.NHttpClientConnection;
import org.apache.http.nio.NHttpClientHandler;
import org.apache.http.params.HttpParams;
import org.apache.http.params.HttpParamsLinker;
import org.apache.http.protocol.BasicHttpProcessor;
import org.apache.http.protocol.HttpContext;
import org.apache.http.protocol.HttpExecutionContext;
import org.apache.http.protocol.HttpProcessor;
import org.apache.http.protocol.RequestConnControl;
import org.apache.http.protocol.RequestContent;
import org.apache.http.protocol.RequestExpectContinue;
import org.apache.http.protocol.RequestTargetHost;
import org.apache.http.protocol.RequestUserAgent;

/**
 * The client connection handler. An instance of this class is used by each IOReactor, to
 * process every connection. Hence this class should not store any data related to a single
 * connection - as this is being shared.
 */
public class ClientHandler implements NHttpClientHandler {

    private static final Log log = LogFactory.getLog(ClientHandler.class);

    /** the HTTP protocol parameters to adhere to for outgoing messages */
    private final HttpParams params;
    /** the HttpProcessor for response messages received */
    private final HttpProcessor httpProcessor;
    /** the connection re-use strategy */
    private final ConnectionReuseStrategy connStrategy;

    /** the Axis2 configuration context */
    ConfigurationContext cfgCtx = null;
    /** the nhttp configuration */
    private NHttpConfiguration cfg = null;

    private WorkerPool workerPool = null;

    private static final String REQUEST_BUFFER = "request-buffer";
    private static final String RESPONSE_BUFFER = "response-buffer";
    private static final String OUTGOING_MESSAGE_CONTEXT = "axis2_message_context";
    private static final String REQUEST_SOURCE_CHANNEL = "request-source-channel";
    private static final String RESPONSE_SINK_CHANNEL = "request-sink-channel";

    private static final String CONTENT_TYPE = "Content-Type";

    /**
     * Create an instance of this client connection handler using the Axis2 configuration
     * context and Http protocol parameters given
     * @param cfgCtx the Axis2 configuration context
     * @param params the Http protocol parameters to adhere to
     */
    public ClientHandler(final ConfigurationContext cfgCtx, final HttpParams params) {
        super();
        this.cfgCtx = cfgCtx;
        this.params = params;
        this.httpProcessor = getHttpProcessor();
        this.connStrategy = new DefaultConnectionReuseStrategy();

        this.cfg = NHttpConfiguration.getInstance();
        workerPool = WorkerPoolFactory.getWorkerPool(
            cfg.getClientCoreThreads(),
            cfg.getClientMaxThreads(),
            cfg.getClientKeepalive(),
            cfg.getClientQueueLen(),
            "Client Worker thread group", "HttpClientWorker");
    }

    public void requestReady(final NHttpClientConnection conn) {
        // The connection is ready for submission of a new request
    }

    /**
     * Submit a new request over an already established connection, which has been 
     * 'kept alive'
     * @param conn the connection to use to send the request, which has been kept open
     * @param axis2Req the new request
     */
    public void submitRequest(final NHttpClientConnection conn, Axis2HttpRequest axis2Req) {

        try {
            HttpContext context = conn.getContext();

            context.setAttribute(HttpExecutionContext.HTTP_CONNECTION, conn);
            context.setAttribute(HttpExecutionContext.HTTP_TARGET_HOST, axis2Req.getHttpHost());

            context.setAttribute(OUTGOING_MESSAGE_CONTEXT, axis2Req.getMsgContext());
            context.setAttribute(REQUEST_SOURCE_CHANNEL, axis2Req.getSourceChannel());

            HttpRequest request = axis2Req.getRequest();
            HttpParamsLinker.link(request, this.params);
            this.httpProcessor.process(request, context);

            conn.submitRequest(request);
            context.setAttribute(HttpExecutionContext.HTTP_REQUEST, request);

        } catch (IOException e) {
            handleException("I/O Error : " + e.getMessage(), e, conn);
        } catch (HttpException e) {
            handleException("Unexpected HTTP protocol error: " + e.getMessage(), e, conn);
        }
    }

    /**
     * Invoked when the destination is connected
     * @param conn the connection being processed
     * @param attachment the attachment set previously
     */
    public void connected(final NHttpClientConnection conn, final Object attachment) {
        try {
            HttpContext context = conn.getContext();
            Axis2HttpRequest axis2Req = (Axis2HttpRequest) attachment;

            context.setAttribute(HttpExecutionContext.HTTP_CONNECTION, conn);
            context.setAttribute(HttpExecutionContext.HTTP_TARGET_HOST, axis2Req.getHttpHost());

            // allocate temporary buffers to process this request
            context.setAttribute(REQUEST_BUFFER, ByteBuffer.allocate(cfg.getBufferZise()));
            context.setAttribute(RESPONSE_BUFFER, ByteBuffer.allocate(cfg.getBufferZise()));

            context.setAttribute(OUTGOING_MESSAGE_CONTEXT, axis2Req.getMsgContext());
            context.setAttribute(REQUEST_SOURCE_CHANNEL, axis2Req.getSourceChannel());

            HttpRequest request = axis2Req.getRequest();
            HttpParamsLinker.link(request, this.params);
            this.httpProcessor.process(request, context);

            conn.submitRequest(request);
            context.setAttribute(HttpExecutionContext.HTTP_REQUEST, request);

        } catch (IOException e) {
            handleException("I/O Error : " + e.getMessage(), e, conn);
        } catch (HttpException e) {
            handleException("Unexpected HTTP protocol error: " + e.getMessage(), e, conn);
        }
    }

    public void closed(final NHttpClientConnection conn) {
        log.trace("Connection closed");
    }

    /**
     * Handle connection timeouts by shutting down the connections
     * @param conn the connection being processed
     */
    public void timeout(final NHttpClientConnection conn) {
        log.debug("Connection Timeout");
        shutdownConnection(conn);
    }

    /**
     * Handle Http protocol violations encountered while reading from underlying channels
     * @param conn the connection being processed
     * @param e the exception encountered
     */
    public void exception(final NHttpClientConnection conn, final HttpException e) {
        log.error("HTTP protocol violation : " + e.getMessage());
        shutdownConnection(conn);
    }