camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject git commit: CAMEL-6403: DefaultConsumer allows to handle the lifecycle of Exchange UoW, which is needed for HTTP/TCP consumers which can send back very large streaming based replies. As the UoW work should be done after the reply has been sent.
Date Mon, 12 Aug 2013 17:48:33 GMT
Updated Branches:
  refs/heads/master 81d0a174b -> 24e01f171


CAMEL-6403: DefaultConsumer allows to handle the lifecycle of Exchange UoW, which is needed
for HTTP/TCP consumers which can send back very large streaming based replies. As the UoW
work should be done after the reply has been sent.


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/24e01f17
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/24e01f17
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/24e01f17

Branch: refs/heads/master
Commit: 24e01f17149ea3690d1fe4d2fabe4c78a5d88225
Parents: 81d0a17
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Mon Aug 12 19:45:11 2013 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Mon Aug 12 19:45:11 2013 +0200

----------------------------------------------------------------------
 .../apache/camel/component/cxf/CxfConsumer.java | 43 ++++++++++++++------
 .../camel/component/cxf/jaxrs/CxfRsInvoker.java | 24 ++++++++---
 .../cxf/AbstractCXFGreeterRouterTest.java       |  1 +
 3 files changed, 49 insertions(+), 19 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/24e01f17/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
b/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
index a8c29f8..cc8145e 100644
--- a/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
+++ b/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
@@ -82,7 +82,7 @@ public class CxfConsumer extends DefaultConsumer {
             private Object asyncInvoke(Exchange cxfExchange, final Continuation continuation)
{
                 synchronized (continuation) {
                     if (continuation.isNew()) {
-                        final org.apache.camel.Exchange camelExchange = perpareCamelExchange(cxfExchange);
+                        final org.apache.camel.Exchange camelExchange = prepareCamelExchange(cxfExchange);
                         
                         // Now we don't set up the timeout value
                         LOG.trace("Suspending continuation of exchangeId: {}", camelExchange.getExchangeId());
@@ -104,9 +104,12 @@ public class CxfConsumer extends DefaultConsumer {
                         });
                         
                     } else if (continuation.isResumed()) {
-                        org.apache.camel.Exchange camelExchange = (org.apache.camel.Exchange)continuation
-                            .getObject();
-                        setResponseBack(cxfExchange, camelExchange);
+                        org.apache.camel.Exchange camelExchange = (org.apache.camel.Exchange)continuation.getObject();
+                        try {
+                            setResponseBack(cxfExchange, camelExchange);
+                        } finally {
+                            CxfConsumer.this.doneUoW(camelExchange);
+                        }
 
                     }
                 }
@@ -127,27 +130,41 @@ public class CxfConsumer extends DefaultConsumer {
             }
             
             private Object syncInvoke(Exchange cxfExchange) {
-                org.apache.camel.Exchange camelExchange = perpareCamelExchange(cxfExchange);
              
-                // send Camel exchange to the target processor
-                LOG.trace("Processing +++ START +++");
+                org.apache.camel.Exchange camelExchange = prepareCamelExchange(cxfExchange);
                 try {
-                    getProcessor().process(camelExchange);
-                } catch (Exception e) {
-                    throw new Fault(e);
+                    try {
+                        LOG.trace("Processing +++ START +++");
+                        // send Camel exchange to the target processor
+                        getProcessor().process(camelExchange);
+                    } catch (Exception e) {
+                        throw new Fault(e);
+                    }
+
+                    LOG.trace("Processing +++ END +++");
+                    setResponseBack(cxfExchange, camelExchange);
+                } finally {
+                    doneUoW(camelExchange);
                 }
-                LOG.trace("Processing +++ END +++");
-                setResponseBack(cxfExchange, camelExchange);
+
                 // response should have been set in outMessage's content
                 return null;
             }
             
-            private org.apache.camel.Exchange perpareCamelExchange(Exchange cxfExchange)
{
+            private org.apache.camel.Exchange prepareCamelExchange(Exchange cxfExchange)
{
                 // get CXF binding
                 CxfEndpoint endpoint = (CxfEndpoint)getEndpoint();
                 CxfBinding binding = endpoint.getCxfBinding();
 
                 // create a Camel exchange, the default MEP is InOut
                 org.apache.camel.Exchange camelExchange = endpoint.createExchange();
+                // we want to handle the UoW
+                try {
+                    CxfConsumer.this.createUoW(camelExchange);
+                } catch (Exception e) {
+                    log.error("Error processing request", e);
+                    throw new Fault(e);
+                }
+
                 DataFormat dataFormat = endpoint.getDataFormat();
 
                 BindingOperationInfo boi = cxfExchange.getBindingOperationInfo();

http://git-wip-us.apache.org/repos/asf/camel/blob/24e01f17/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java
----------------------------------------------------------------------
diff --git a/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java
b/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java
index f9a8d1c..38bf420 100644
--- a/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java
+++ b/components/camel-cxf/src/main/java/org/apache/camel/component/cxf/jaxrs/CxfRsInvoker.java
@@ -76,6 +76,9 @@ public class CxfRsInvoker extends JAXRSInvoker {
                     ep = ExchangePattern.InOnly;
                 } 
                 final org.apache.camel.Exchange camelExchange = endpoint.createExchange(ep);
+                // we want to handle the UoW
+                cxfRsConsumer.createUoW(camelExchange);
+
                 CxfRsBinding binding = endpoint.getBinding();
                 binding.populateExchangeFromCxfRsRequest(cxfExchange, camelExchange, method,
paramArray);
                 // Now we don't set up the timeout value
@@ -99,9 +102,12 @@ public class CxfRsInvoker extends JAXRSInvoker {
             }
             if (continuation.isResumed()) {
                 cxfExchange.put(SUSPENED, Boolean.FALSE);
-                org.apache.camel.Exchange camelExchange = (org.apache.camel.Exchange)continuation
-                    .getObject();
-                return returnResponse(cxfExchange, camelExchange);
+                org.apache.camel.Exchange camelExchange = (org.apache.camel.Exchange)continuation.getObject();
+                try {
+                    return returnResponse(cxfExchange, camelExchange);
+                } finally {
+                    cxfRsConsumer.doneUoW(camelExchange);
+                }
             }
         }
         return null;
@@ -115,16 +121,22 @@ public class CxfRsInvoker extends JAXRSInvoker {
             ep = ExchangePattern.InOnly;
         } 
         org.apache.camel.Exchange camelExchange = endpoint.createExchange(ep);
+        // we want to handle the UoW
+        cxfRsConsumer.createUoW(camelExchange);
         CxfRsBinding binding = endpoint.getBinding();
         binding.populateExchangeFromCxfRsRequest(cxfExchange, camelExchange, method, paramArray);
-        
+
         try {
             cxfRsConsumer.getProcessor().process(camelExchange);
         } catch (Exception exception) {
             camelExchange.setException(exception);
         }
-        return returnResponse(cxfExchange, camelExchange);
-        
+
+        try {
+            return returnResponse(cxfExchange, camelExchange);
+        } finally {
+            cxfRsConsumer.doneUoW(camelExchange);
+        }
     }
     
     private Object returnResponse(Exchange cxfExchange, org.apache.camel.Exchange camelExchange)
throws Exception {

http://git-wip-us.apache.org/repos/asf/camel/blob/24e01f17/components/camel-cxf/src/test/java/org/apache/camel/component/cxf/AbstractCXFGreeterRouterTest.java
----------------------------------------------------------------------
diff --git a/components/camel-cxf/src/test/java/org/apache/camel/component/cxf/AbstractCXFGreeterRouterTest.java
b/components/camel-cxf/src/test/java/org/apache/camel/component/cxf/AbstractCXFGreeterRouterTest.java
index 7b3b92c..223ef21 100644
--- a/components/camel-cxf/src/test/java/org/apache/camel/component/cxf/AbstractCXFGreeterRouterTest.java
+++ b/components/camel-cxf/src/test/java/org/apache/camel/component/cxf/AbstractCXFGreeterRouterTest.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.cxf;
 
 import javax.xml.namespace.QName;
 import javax.xml.ws.Service;
+import javax.xml.ws.soap.SOAPFaultException;
 
 import org.apache.camel.CamelContext;
 import org.apache.camel.RuntimeCamelException;


Mime
View raw message