clienthandler.java

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

    /**
     * Handle IO errors while reading or writing to underlying channels
     * @param conn the connection being processed
     * @param e the exception encountered
     */
    public void exception(final NHttpClientConnection conn, final IOException e) {
        log.error("I/O error : " + e.getMessage(), e);
        shutdownConnection(conn);
    }

    /**
     * Process ready input (i.e. response from remote server)
     * @param conn connection being processed
     * @param decoder the content decoder in use
     */
    public void inputReady(final NHttpClientConnection conn, final ContentDecoder decoder) {
        HttpContext context = conn.getContext();
        HttpResponse response = conn.getHttpResponse();
        WritableByteChannel sink = (WritableByteChannel) context.getAttribute(RESPONSE_SINK_CHANNEL);
        ByteBuffer inbuf = (ByteBuffer) context.getAttribute(REQUEST_BUFFER);

        try {
            while (decoder.read(inbuf) > 0) {
                inbuf.flip();
                sink.write(inbuf);
                inbuf.compact();
            }

            if (decoder.isCompleted()) {
                if (sink != null) sink.close();
                if (!connStrategy.keepAlive(response, context)) {
                    conn.close();
                } else {
                    ConnectionPool.release(conn);
                }
            }

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

    /**
     * Process ready output (i.e. write request to remote server)
     * @param conn the connection being processed
     * @param encoder the encoder in use
     */
    public void outputReady(final NHttpClientConnection conn, final ContentEncoder encoder) {
        HttpContext context = conn.getContext();

        ReadableByteChannel source = (ReadableByteChannel) context.getAttribute(REQUEST_SOURCE_CHANNEL);
        ByteBuffer outbuf = (ByteBuffer) context.getAttribute(RESPONSE_BUFFER);

        try {
            int bytesRead = source.read(outbuf);
            if (bytesRead == -1) {
                encoder.complete();
            } else {
                outbuf.flip();
                encoder.write(outbuf);
                outbuf.compact();
            }

            if (encoder.isCompleted()) {
                source.close();
            }

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

    /**
     * Process a response received for the request sent out
     * @param conn the connection being processed
     */
    public void responseReceived(final NHttpClientConnection conn) {
        HttpContext context = conn.getContext();
        HttpResponse response = conn.getHttpResponse();

        switch (response.getStatusLine().getStatusCode()) {
            case HttpStatus.SC_ACCEPTED : {
                log.debug("Received a 202 Accepted response");

                // create a dummy message with an empty SOAP envelope and a property
                // NhttpConstants.SC_ACCEPTED set to Boolean.TRUE to indicate this is a
                // placeholder message for the transport to send a HTTP 202 to the
                // client. Should / would be ignored by any transport other than
                // nhttp. For example, JMS would not send a reply message for one-way
                // operations.
                MessageContext outMsgCtx =
                    (MessageContext) context.getAttribute(OUTGOING_MESSAGE_CONTEXT);
                MessageReceiver mr = outMsgCtx.getAxisOperation().getMessageReceiver();

                try {
                    MessageContext responseMsgCtx = outMsgCtx.getOperationContext().
                        getMessageContext(WSDL2Constants.MESSAGE_LABEL_IN);
                    if (responseMsgCtx == null) {
                        // to support Sandesha.. however, this means that we received a 202 accepted
                        // for an out-only , for which we do not need a dummy message anyway
                        return;
                    }
                    responseMsgCtx.setServerSide(true);
                    responseMsgCtx.setDoingREST(outMsgCtx.isDoingREST());
                    responseMsgCtx.setProperty(MessageContext.TRANSPORT_IN,
                        outMsgCtx.getProperty(MessageContext.TRANSPORT_IN));
                    responseMsgCtx.setTransportIn(outMsgCtx.getTransportIn());
                    responseMsgCtx.setTransportOut(outMsgCtx.getTransportOut());

                    responseMsgCtx.setAxisMessage(outMsgCtx.getAxisOperation().
                        getMessage(WSDLConstants.MESSAGE_LABEL_IN_VALUE));
                    responseMsgCtx.setOperationContext(outMsgCtx.getOperationContext());
                    responseMsgCtx.setConfigurationContext(outMsgCtx.getConfigurationContext());
                    responseMsgCtx.setTo(null);

                    responseMsgCtx.setEnvelope(
                        ((SOAPFactory)outMsgCtx.getEnvelope().getOMFactory()).getDefaultEnvelope());
                    responseMsgCtx.setProperty(AddressingConstants.DISABLE_ADDRESSING_FOR_OUT_MESSAGES, Boolean.TRUE);
                    responseMsgCtx.setProperty(NhttpConstants.SC_ACCEPTED, Boolean.TRUE);
                    mr.receive(responseMsgCtx);

                } catch (org.apache.axis2.AxisFault af) {
                    log.error("Unable to report back 202 Accepted state to the message receiver", af);
                }

                return;
            }
            case HttpStatus.SC_INTERNAL_SERVER_ERROR : {
                Header contentType = response.getFirstHeader(CONTENT_TYPE);
                if (contentType != null &&
                    (contentType.getValue().indexOf(SOAP11Constants.SOAP_11_CONTENT_TYPE) >= 0) ||
                     contentType.getValue().indexOf(SOAP12Constants.SOAP_12_CONTENT_TYPE) >=0) {
                    log.debug("Received an internal server error with a SOAP payload");
                    processResponse(conn, context, response);
                    return;
                }
                log.error("Received an internal server error : " +
                    response.getStatusLine().getReasonPhrase());
                return;
            }
            case HttpStatus.SC_OK : {
                processResponse(conn, context, response);
            }
        }
    }

    /**
     * Perform processing of the received response though Axis2
     * @param conn
     * @param context
     * @param response
     */
    private void processResponse(final NHttpClientConnection conn, HttpContext context, HttpResponse response) {

        try {
            PipeImpl responsePipe = new PipeImpl();
            context.setAttribute(RESPONSE_SINK_CHANNEL, responsePipe.sink());

            BasicHttpEntity entity = new BasicHttpEntity();
            if (response.getStatusLine().getHttpVersion().greaterEquals(HttpVersion.HTTP_1_1)) {
                entity.setChunked(true);
            }
            response.setEntity(entity);
            context.setAttribute(HttpContext.HTTP_RESPONSE, response);

            workerPool.execute(
                new ClientWorker(cfgCtx, Channels.newInputStream(responsePipe.source()), response,
                    (MessageContext) context.getAttribute(OUTGOING_MESSAGE_CONTEXT)));

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

    // ----------- utility methods -----------

    private void handleException(String msg, Exception e, NHttpClientConnection conn) {
        log.error(msg, e);
        if (conn != null) {
            shutdownConnection(conn);
        }
    }

    /**
     * Shutdown the connection ignoring any IO errors during the process
     * @param conn the connection to be shutdown
     */
    private void shutdownConnection(final HttpConnection conn) {
        try {
            conn.shutdown();
        } catch (IOException ignore) {}
    }

    /**
     * Return the HttpProcessor for requests
     * @return the HttpProcessor that processes requests
     */
    private HttpProcessor getHttpProcessor() {
        BasicHttpProcessor httpProcessor = new BasicHttpProcessor();
        httpProcessor.addInterceptor(new RequestContent());
        httpProcessor.addInterceptor(new RequestTargetHost());
        httpProcessor.addInterceptor(new RequestConnControl());
        httpProcessor.addInterceptor(new RequestUserAgent());
        httpProcessor.addInterceptor(new RequestExpectContinue());
        return httpProcessor;
    }

}