camel-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Claus Ibsen <claus.ib...@gmail.com>
Subject Re: svn commit: r966815 - in /camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty: CamelContinuationServlet.java CamelMultipartFilter.java JettyHttpComponent.java
Date Fri, 23 Jul 2010 04:25:31 GMT
Hi

Great catch.

Btw there is a typo in the constant for the attribute :)


On Thu, Jul 22, 2010 at 9:24 PM,  <krasserm@apache.org> wrote:
> Author: krasserm
> Date: Thu Jul 22 19:24:24 2010
> New Revision: 966815
>
> URL: http://svn.apache.org/viewvc?rev=966815&view=rev
> Log:
> Closes CAMEL-2986: IllegalStateException in CamelContinuationServlet under heavy load
>
> Added:
>    camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/CamelMultipartFilter.java
> Modified:
>    camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/CamelContinuationServlet.java
>    camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/JettyHttpComponent.java
>
> Modified: camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/CamelContinuationServlet.java
> URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/CamelContinuationServlet.java?rev=966815&r1=966814&r2=966815&view=diff
> ==============================================================================
> --- camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/CamelContinuationServlet.java
(original)
> +++ camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/CamelContinuationServlet.java
Thu Jul 22 19:24:24 2010
> @@ -38,6 +38,8 @@ import org.eclipse.jetty.continuation.Co
>  */
>  public class CamelContinuationServlet extends CamelServlet {
>
> +    static final String EXCHANGE_ATRRIBUTE_NAME = "CamelExchange";
> +
>     private static final long serialVersionUID = 1L;
>
>     @Override
> @@ -50,15 +52,15 @@ public class CamelContinuationServlet ex
>                 return;
>             }
>
> -            // are we suspended?
> -            if (consumer.isSuspended()) {
> +            final Continuation continuation = ContinuationSupport.getContinuation(request);
> +
> +            // are we suspended and a request is dispatched initially?
> +            if (consumer.isSuspended() && continuation.isInitial()) {
>                 response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
>                 return;
>             }
>
> -            final Continuation continuation = ContinuationSupport.getContinuation(request);
>             if (continuation.isInitial()) {
> -
>                 // a new request so create an exchange
>                 final Exchange exchange = new DefaultExchange(consumer.getEndpoint(),
ExchangePattern.InOut);
>                 if (consumer.getEndpoint().isBridgeEndpoint()) {
> @@ -69,44 +71,28 @@ public class CamelContinuationServlet ex
>                 }
>                 exchange.setIn(new HttpMessage(exchange, request, response));
>
> +                if (log.isTraceEnabled()) {
> +                    log.trace("Suspending continuation of exchangeId: " +
exchange.getExchangeId());
> +                }
> +                continuation.suspend();
> +
>                 // use the asynchronous API to process the exchange
> -                boolean sync = consumer.getAsyncProcessor().process(exchange,
new AsyncCallback() {
> +                consumer.getAsyncProcessor().process(exchange, new AsyncCallback()
{
>                     public void done(boolean doneSync) {
> -                        // we only have to handle async completion
> -                        if (doneSync) {
> -                            return;
> -                        }
> -
> -                        // we should resume the continuation now that we
are done asynchronously
>                         if (log.isTraceEnabled()) {
>                             log.trace("Resuming continuation of exchangeId:
" + exchange.getExchangeId());
>                         }
> -                        continuation.setAttribute("CamelExchange", exchange);
> +                        // resume processing after both, sync and async
callbacks
> +                        continuation.setAttribute(EXCHANGE_ATRRIBUTE_NAME,
exchange);
>                         continuation.resume();
>                     }
>                 });
> -
> -                if (!sync) {
> -                    // wait for the exchange to get processed.
> -                    // this might block until it completes or it might return
via an exception and
> -                    // then this method is re-invoked once the the exchange
has finished processing
> -                    if (log.isTraceEnabled()) {
> -                        log.trace("Suspending continuation of exchangeId:
" + exchange.getExchangeId());
> -                    }
> -                    continuation.suspend(response);
> -                    return;
> -                }
> -
> -                // now lets output to the response
> -                if (log.isTraceEnabled()) {
> -                    log.trace("Writing response of exchangeId: " + exchange.getExchangeId());
> -                }
> -                consumer.getBinding().writeResponse(exchange, response);
>                 return;
>             }
>
>             if (continuation.isResumed()) {
> -                Exchange exchange = (Exchange) continuation.getAttribute("CamelExchange");
> +                // a re-dispatched request containing the processing result
> +                Exchange exchange = (Exchange) continuation.getAttribute(EXCHANGE_ATRRIBUTE_NAME);
>                 if (log.isTraceEnabled()) {
>                     log.trace("Resuming continuation of exchangeId: " + exchange.getExchangeId());
>                 }
> @@ -116,8 +102,10 @@ public class CamelContinuationServlet ex
>                     log.trace("Writing response of exchangeId: " + exchange.getExchangeId());
>                 }
>                 consumer.getBinding().writeResponse(exchange, response);
> -                return;
>             }
> +        } catch (IOException e) {
> +            log.error("Error processing request", e);
> +            throw e;
>         } catch (Exception e) {
>             log.error("Error processing request", e);
>             throw new ServletException(e);
>
> Added: camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/CamelMultipartFilter.java
> URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/CamelMultipartFilter.java?rev=966815&view=auto
> ==============================================================================
> --- camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/CamelMultipartFilter.java
(added)
> +++ camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/CamelMultipartFilter.java
Thu Jul 22 19:24:24 2010
> @@ -0,0 +1,43 @@
> +/**
> + * 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.camel.component.jetty;
> +
> +import java.io.IOException;
> +
> +import javax.servlet.FilterChain;
> +import javax.servlet.ServletException;
> +import javax.servlet.ServletRequest;
> +import javax.servlet.ServletResponse;
> +
> +import org.eclipse.jetty.servlets.MultiPartFilter;
> +
> +/**
> + * A multipart filter that processes only initially dispatched requests.
> + * Re-dispatched requests are ignored.
> + */
> +class CamelMultipartFilter extends MultiPartFilter {
> +
> +    @Override
> +    public void doFilter(ServletRequest request, ServletResponse response, FilterChain
chain) throws IOException, ServletException {
> +        if (request.getAttribute(CamelContinuationServlet.EXCHANGE_ATRRIBUTE_NAME)
== null) {
> +            super.doFilter(request, response, chain);
> +        } else {
> +            chain.doFilter(request, response);
> +        }
> +    }
> +
> +}
>
> Modified: camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/JettyHttpComponent.java
> URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/JettyHttpComponent.java?rev=966815&r1=966814&r2=966815&view=diff
> ==============================================================================
> --- camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/JettyHttpComponent.java
(original)
> +++ camel/trunk/components/camel-jetty/src/main/java/org/apache/camel/component/jetty/JettyHttpComponent.java
Thu Jul 22 19:24:24 2010
> @@ -55,7 +55,6 @@ import org.eclipse.jetty.server.ssl.SslS
>  import org.eclipse.jetty.servlet.FilterHolder;
>  import org.eclipse.jetty.servlet.ServletContextHandler;
>  import org.eclipse.jetty.servlet.ServletHolder;
> -import org.eclipse.jetty.servlets.MultiPartFilter;
>  import org.eclipse.jetty.util.component.LifeCycle;
>  import org.eclipse.jetty.util.thread.QueuedThreadPool;
>  import org.eclipse.jetty.util.thread.ThreadPool;
> @@ -599,7 +598,7 @@ public class JettyHttpComponent extends
>             }
>             context.setAttribute("javax.servlet.context.tempdir", file);
>         }
> -        filterHolder.setFilter(new MultiPartFilter());
> +        filterHolder.setFilter(new CamelMultipartFilter());
>         //add the default MultiPartFilter filter for it
>         context.addFilter(filterHolder, "/*", 0);
>         context.addServlet(holder, "/*");
>
>
>



-- 
Claus Ibsen
Apache Camel Committer

Author of Camel in Action: http://www.manning.com/ibsen/
Open Source Integration: http://fusesource.com
Blog: http://davsclaus.blogspot.com/
Twitter: http://twitter.com/davsclaus

Mime
View raw message