httpmessagereceiver.java

提供ESB 应用mule源代码 提供ESB 应用mule源代码

                boolean endpointOverride = Boolean.parseBoolean((String) endpoint.getProperty("keepAlive"));
                
                Header connectionHeader = request.getFirstHeader("Connection");
                if (connectionHeader != null)
                {
                    String value = connectionHeader.getValue();
					if ("keep-alive".equalsIgnoreCase(value) && !endpointOverride) 
                    {
                    	response.setKeepAlive(true);
                        Header header = new Header(HttpConstants.HEADER_KEEP_ALIVE, "timeout=" 
                            + ((HttpConnector) connector).getKeepAliveTimeout());
                        response.addHeader(header);   
                    }
                    else if ("close".equalsIgnoreCase(value))
                    {
                    	response.setKeepAlive(false);
                    } 
                    else if (response.getHttpVersion().greaterEquals(HttpVersion.HTTP_1_1))
                    {
                        response.setKeepAlive(true);
                    }
                    else
                    {
                        response.setKeepAlive(false);
                    }
                }
            }
            else
            {
                response = buildFailureResponse(requestLine, message);
            }
            return response;
        }

        protected HttpResponse doOtherValid(RequestLine requestLine, String method) throws MuleException
        {
            MuleMessage message = new DefaultMuleMessage(NullPayload.getInstance());
            MuleEvent event = new DefaultMuleEvent(message, endpoint, new DefaultMuleSession(message, new NullSessionHandler(), connector.getMuleContext()), true);
            OptimizedRequestContext.unsafeSetEvent(event);
            HttpResponse response = new HttpResponse();
            response.setStatusLine(requestLine.getHttpVersion(), HttpConstants.SC_METHOD_NOT_ALLOWED);
            response.setBody(HttpMessages.methodNotAllowed(method).toString() + HttpConstants.CRLF);
            return transformResponse(response);
        }

        protected HttpResponse doBad(RequestLine requestLine) throws MuleException
        {
            MuleMessage message = new DefaultMuleMessage(NullPayload.getInstance());
            MuleEvent event = new DefaultMuleEvent(message, endpoint, new DefaultMuleSession(message, new NullSessionHandler(), connector.getMuleContext()), true);
            OptimizedRequestContext.unsafeSetEvent(event);
            HttpResponse response = new HttpResponse();
            response.setStatusLine(requestLine.getHttpVersion(), HttpConstants.SC_BAD_REQUEST);
            response.setBody(HttpMessages.malformedSyntax().toString() + HttpConstants.CRLF);
            return transformResponse(response);
        }

        protected MessageAdapter buildStandardAdapter(final HttpRequest request,
                                                         final Map headers) throws MessagingException, TransformerException, IOException
        {
            final RequestLine requestLine = request.getRequestLine();

            sendExpect100(headers, requestLine);

            Object body = request.getBody();
            if (body == null)
            {
                body = requestLine.getUri();
            }

            return connector.getMessageAdapter(new Object[]{body, headers});
        }

        private void sendExpect100(Map headers, RequestLine requestLine)
            throws TransformerException, IOException
        {
            // respond with status code 100, for Expect handshake
            // according to rfc 2616 and http 1.1
            // the processing will continue and the request will be fully
            // read immediately after
            if (HttpConstants.HTTP11.equals(headers.get(HttpConnector.HTTP_VERSION_PROPERTY)))
            {
                // just in case we have something other than String in
                // the headers map
                String expectHeaderValue = ObjectUtils.toString(
                        headers.get(HttpConstants.HEADER_EXPECT)).toLowerCase();
                if (HttpConstants.HEADER_EXPECT_CONTINUE_REQUEST_VALUE.equals(expectHeaderValue))
                {
                    HttpResponse expected = new HttpResponse();
                    expected.setStatusLine(requestLine.getHttpVersion(), HttpConstants.SC_CONTINUE);
                    final DefaultMuleEvent event = new DefaultMuleEvent(new DefaultMuleMessage(expected), endpoint,
                            new DefaultMuleSession(service, connector.getMuleContext()), true);
                    RequestContext.setEvent(event);
                    conn.writeResponse(transformResponse(expected));
                }
            }
        }

        protected HttpResponse buildFailureResponse(RequestLine requestLine, MuleMessage message) throws TransformerException
        {
            EndpointURI uri = endpoint.getEndpointURI();
            String failedPath = uri.getScheme() + "://" + uri.getHost() + ":" + uri.getPort()
                    + getRequestPath(message);

            if (logger.isDebugEnabled())
            {
                logger.debug("Failed to bind to " + failedPath);
            }

            HttpResponse response = new HttpResponse();
            response.setStatusLine(requestLine.getHttpVersion(), HttpConstants.SC_NOT_FOUND);
            response.setBody(HttpMessages.cannotBindToAddress(failedPath).toString());
            RequestContext.setEvent(new DefaultMuleEvent(new DefaultMuleMessage(response), endpoint,
                    new DefaultMuleSession(service, connector.getMuleContext()), true));
            // The DefaultResponseTransformer will set the necessary headers
            return transformResponse(response);
        }

        protected Map parseHeaders(HttpRequest request) throws MalformedCookieException
        {
            RequestLine requestLine = request.getRequestLine();
            Map headers = new HashMap();

            for (Iterator rhi = request.getHeaderIterator(); rhi.hasNext();)
            {
                Header header = (Header) rhi.next();
                String headerName = header.getName();
                Object headerValue = header.getValue();

                // fix Mule headers?
                if (headerName.startsWith("X-MULE"))
                {
                    headerName = headerName.substring(2);
                }
                // Parse cookies?
                else if (headerName.equals(HttpConnector.HTTP_COOKIES_PROPERTY))
                {
                    if (enableCookies)
                    {
                        Cookie[] cookies = CookieHelper.parseCookies(header, cookieSpec);
                        if (cookies.length > 0)
                        {
                            // yum!
                            headerValue = cookies;
                        }
                        else
                        {
                            // bad cookies?!
                            continue;
                        }
                    }
                    else
                    {
                        // no cookies for you!
                        continue;
                    }
                }

                // accept header & value
                headers.put(headerName, headerValue);
            }

            headers.put(HttpConnector.HTTP_METHOD_PROPERTY, requestLine.getMethod());
            headers.put(HttpConnector.HTTP_REQUEST_PROPERTY, requestLine.getUri());
            headers.put(HttpConnector.HTTP_VERSION_PROPERTY, requestLine.getHttpVersion().toString());
            headers.put(HttpConnector.HTTP_COOKIE_SPEC_PROPERTY, cookieSpec);
            return headers;
        }

        protected void preRouteMessage(MuleMessage message)
        {
            message.setProperty(MuleProperties.MULE_REMOTE_CLIENT_ADDRESS, remoteClientAddress);
        }

        public void release()
        {
            conn.close();
            conn = null;
        }
    }

    protected String getRequestPath(MuleMessage message)
    {
        String path = (String) message.getProperty(HttpConnector.HTTP_REQUEST_PROPERTY);
        int i = path.indexOf('?');
        if (i > -1)
        {
            path = path.substring(0, i);
        }
        return path;
    }

    protected MessageReceiver getTargetReceiver(MuleMessage message, ImmutableEndpoint endpoint)
            throws ConnectException
    {
        String path = (String) message.getProperty(HttpConnector.HTTP_REQUEST_PROPERTY);
        int i = path.indexOf('?');
        if (i > -1)
        {
            path = path.substring(0, i);
        }

        StringBuffer requestUri = new StringBuffer(80);
        if (path.indexOf("://") == -1)
        {
            requestUri.append(endpoint.getProtocol()).append("://");
            requestUri.append(endpoint.getEndpointURI().getHost());
            requestUri.append(':').append(endpoint.getEndpointURI().getPort());
        }
        // first check that there is a receiver on the root address
        if (logger.isTraceEnabled())
        {
            logger.trace("Looking up receiver on connector: " + connector.getName() + " with URI key: "
                    + requestUri.toString());
        }

        MessageReceiver receiver = connector.lookupReceiver(requestUri.toString());

        // If no receiver on the root and there is a request path, look up the
        // received based on the root plus request path
        if (receiver == null && !"/".equals(path))
        {
            // remove anything after the last '/'
            int x = path.lastIndexOf('/');
            if (x > 1 && path.indexOf('.') > x)
            {
                requestUri.append(path.substring(0, x));
            }
            else
            {
                requestUri.append(path);
            }

            if (logger.isDebugEnabled())
            {
                logger.debug("Secondary lookup of receiver on connector: " + connector.getName()
                        + " with URI key: " + requestUri.toString());
            }

            // try again
            String uriStr = requestUri.toString();
            receiver = connector.lookupReceiver(uriStr);

            if (receiver == null)
            {
                receiver = findReceiverByStem(connector.getReceivers(), uriStr);
            }

            if (receiver == null && logger.isWarnEnabled())
            {
                logger.warn("No receiver found with secondary lookup on connector: " + connector.getName()
                        + " with URI key: " + requestUri.toString());
                logger.warn("Receivers on connector are: "
                        + MapUtils.toString(connector.getReceivers(), true));
            }
        }

        return receiver;
    }

    protected HttpResponse transformResponse(Object response) throws TransformerException
    {
        MuleMessage message;
        if (response instanceof MuleMessage)
        {
            message = (MuleMessage) response;
        }
        else
        {
            message = new DefaultMuleMessage(response);
        }
        //TODO RM*: Maybe we can have a generic Transformer wrapper rather that using DefaultMuleMessage (or another static utility
        //class
        message.applyTransformers(connector.getDefaultResponseTransformers(), HttpResponse.class);
        return (HttpResponse) message.getPayload();
    }

    public static MessageReceiver findReceiverByStem(Map receivers, String uriStr)
    {
        int match = 0;
        MessageReceiver receiver = null;
        for (Iterator itr = receivers.entrySet().iterator(); itr.hasNext();)
        {
            Map.Entry e = (Map.Entry) itr.next();
            String key = (String) e.getKey();
            MessageReceiver candidate = (MessageReceiver) e.getValue();
            if (uriStr.startsWith(key) && match < key.length())
            {
                match = key.length();
                receiver = candidate;
            }
        }
        return receiver;
    }

}