camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ningji...@apache.org
Subject svn commit: r1002095 - /camel/trunk/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
Date Tue, 28 Sep 2010 10:13:58 GMT
Author: ningjiang
Date: Tue Sep 28 10:13:57 2010
New Revision: 1002095

URL: http://svn.apache.org/viewvc?rev=1002095&view=rev
Log:
CAMEL-3169 Add continuation synchronize block to avoid continuation is resumed befor it is
suspended by the other thread

Modified:
    camel/trunk/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java

Modified: camel/trunk/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java?rev=1002095&r1=1002094&r2=1002095&view=diff
==============================================================================
--- camel/trunk/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
(original)
+++ camel/trunk/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
Tue Sep 28 10:13:57 2010
@@ -71,47 +71,54 @@ public class CxfConsumer extends Default
             }            
             
             private Object asyncInvoke(Exchange cxfExchange, final Continuation continuation)
{
-                if (continuation.isNew()) {
-                    final org.apache.camel.Exchange camelExchange = perpareCamelExchange(cxfExchange);
-
-                    // use the asynchronous API to process the exchange
-                    boolean sync = getAsyncProcessor().process(camelExchange, new AsyncCallback()
{
-                        public void done(boolean doneSync) {
+                synchronized (continuation) {
+                    if (continuation.isNew()) {
+                        final org.apache.camel.Exchange camelExchange = perpareCamelExchange(cxfExchange);
+
+                        // use the asynchronous API to process the exchange
+                        boolean sync = getAsyncProcessor().process(camelExchange, new AsyncCallback()
{
+                            public void done(boolean doneSync) {
+                                // make sure the continuation resume will not be called before
the suspend method in other thread
+                                synchronized (continuation) {
+                                    if (LOG.isTraceEnabled()) {
+                                        LOG.trace("Resuming continuation of exchangeId: "
+                                                  + camelExchange.getExchangeId());
+                                    }
+                                    // resume processing after both, sync and async callbacks
+                                    continuation.setObject(camelExchange);
+                                    continuation.resume();
+                                }
+                            }
+                        });
+                        // just need to avoid the continuation.resume is called
+                        // before the continuation.suspend is called
+                        if (continuation.getObject() != camelExchange && !sync) {
+                            // Now we don't set up the timeout value
                             if (LOG.isTraceEnabled()) {
-                                LOG.trace("Resuming continuation of exchangeId: "
+                                LOG.trace("Suspending continuation of exchangeId: "
                                           + camelExchange.getExchangeId());
                             }
-                            // resume processing after both, sync and async callbacks
-                            continuation.setObject(camelExchange);
-                            continuation.resume();
-                        }
-                    });
-                    // just need to avoid the continuation.resume is called
-                    // before the continuation.suspend is called
-                    if (continuation.getObject() != camelExchange && !sync) {
-                        // Now we don't set up the timeout value
-                        if (LOG.isTraceEnabled()) {
-                            LOG.trace("Suspending continuation of exchangeId: "
-                                      + camelExchange.getExchangeId());
-                        }
-                        // The continuation could be called before the suspend
-                        // is called
-                        continuation.suspend(0);
-                    } else {
-                        // just set the response back, as the invoking thread is
-                        // not changed
-                        if (LOG.isTraceEnabled()) {
-                            LOG.trace("Processed the Exchange : " + camelExchange.getExchangeId());
+                            // The continuation could be called before the
+                            // suspend
+                            // is called
+                            continuation.suspend(0);
+                        } else {
+                            // just set the response back, as the invoking
+                            // thread is
+                            // not changed
+                            if (LOG.isTraceEnabled()) {
+                                LOG.trace("Processed the Exchange : " + camelExchange.getExchangeId());
+                            }
+                            setResponseBack(cxfExchange, camelExchange);
                         }
-                        setResponseBack(cxfExchange, camelExchange);
-                    }
 
-                }
-                if (continuation.isResumed()) {
-                    org.apache.camel.Exchange camelExchange = (org.apache.camel.Exchange)continuation
-                        .getObject();
-                    setResponseBack(cxfExchange, camelExchange);
+                    }
+                    if (continuation.isResumed()) {
+                        org.apache.camel.Exchange camelExchange = (org.apache.camel.Exchange)continuation
+                            .getObject();
+                        setResponseBack(cxfExchange, camelExchange);
 
+                    }
                 }
                 return null;
             }



Mime
View raw message