Repository: camel
Updated Branches:
  refs/heads/camel-2.16.x 63dc9d9c3 -> 54a03578d


Fixed CS


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

Branch: refs/heads/camel-2.16.x
Commit: 54a03578ddbf81cbf0fd126700400f7d65b53aab
Parents: 63dc9d9
Author: Claus Ibsen <davscl...@apache.org>
Authored: Thu Apr 28 19:26:39 2016 +0200
Committer: Gregor Zurowski <gre...@zurowski.org>
Committed: Sun Sep 18 12:57:48 2016 +0200

----------------------------------------------------------------------
 .../apache/camel/component/cxf/CxfConsumer.java | 422 ++++++++++---------
 ...nsumerDuplicateNamespaceStreamCacheTest.java |   2 +-
 2 files changed, 218 insertions(+), 206 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/54a03578/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 4c29f8a..11ec2f2 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
@@ -64,210 +64,7 @@ public class CxfConsumer extends DefaultConsumer {
         cxfEndpoint = endpoint;
         // create server
         ServerFactoryBean svrBean = endpoint.createServerFactoryBean();
-        svrBean.setInvoker(new Invoker() {
-            // we receive a CXF request when this method is called
-            public Object invoke(Exchange cxfExchange, Object o) {
-                LOG.trace("Received CXF Request: {}", cxfExchange);            
    
-                Continuation continuation;
-                if (!endpoint.isSynchronous() && 
isAsyncInvocationSupported(cxfExchange)
-                    && (continuation = getContinuation(cxfExchange)) != null) {
-                    LOG.trace("Calling the Camel async processors.");
-                    return asyncInvoke(cxfExchange, continuation);
-                } else {
-                    LOG.trace("Calling the Camel sync processors.");
-                    return syncInvoke(cxfExchange);
-                }
-            }
-            // NOTE this code cannot work with CXF 2.2.x and JMSContinuation
-            // as it doesn't break out the interceptor chain when we call it 
-            private Object asyncInvoke(Exchange cxfExchange, final 
Continuation continuation) {
-                synchronized (continuation) {
-                    if (continuation.isNew()) {
-                        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());
-                        
-                        // The continuation could be called before the suspend 
is called
-                        
continuation.suspend(cxfEndpoint.getContinuationTimeout());
-
-                        // use the asynchronous API to process the exchange
-                        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) {
-                                    LOG.trace("Resuming continuation of 
exchangeId: {}", camelExchange.getExchangeId());
-                                    // resume processing after both, sync and 
async callbacks
-                                    continuation.setObject(camelExchange);
-                                    continuation.resume();
-                                }
-                            }
-                        });
-                        
-                    } else if (continuation.isResumed()) {
-                        org.apache.camel.Exchange camelExchange = 
(org.apache.camel.Exchange)continuation.getObject();
-                        try {
-                            setResponseBack(cxfExchange, camelExchange);
-                        } finally {
-                            CxfConsumer.this.doneUoW(camelExchange);
-                        }
-
-                    }
-                }
-                return null;
-            }
-            private Continuation getContinuation(Exchange cxfExchange) {
-                ContinuationProvider provider = 
-                    
(ContinuationProvider)cxfExchange.getInMessage().get(ContinuationProvider.class.getName());
-                Continuation continuation = provider == null ? null : 
provider.getContinuation();
-                // Make sure we don't return the JMSContinuation, as it 
doesn't support the Continuation we wants
-                // Don't want to introduce the dependency of 
cxf-rt-transprot-jms here
-                if (continuation != null && 
continuation.getClass().getName().equals("org.apache.cxf.transport.jms.continuations.JMSContinuation"))
 {
-                    return null;
-                } else {
-                    return continuation;
-                }
-            }
-            private Object syncInvoke(Exchange cxfExchange) {
-                org.apache.camel.Exchange camelExchange = 
prepareCamelExchange(cxfExchange);
-                try {
-                    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);
-                }
-                // response should have been set in outMessage's content
-                return null;
-            }
-            
-            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();
-
-                DataFormat dataFormat = endpoint.getDataFormat();
-
-                BindingOperationInfo boi = 
cxfExchange.getBindingOperationInfo();
-                // make sure the "boi" is remained as wrapped in PAYLOAD mode
-                if (boi != null && dataFormat == DataFormat.PAYLOAD && 
boi.isUnwrapped()) {
-                    boi = boi.getWrappedOperation();
-                    cxfExchange.put(BindingOperationInfo.class, boi);
-                }
-                
-                if (boi != null) {
-                    
camelExchange.setProperty(BindingOperationInfo.class.getName(), boi);
-                    LOG.trace("Set exchange property: BindingOperationInfo: 
{}", boi);
-                    // set the message exchange patter with the boi
-                    if (boi.getOperationInfo().isOneWay()) {
-                        camelExchange.setPattern(ExchangePattern.InOnly);
-                    }
-                } else {
-                    if 
(cxfEndpoint.getExchangePattern().equals(ExchangePattern.InOnly)) {
-                        camelExchange.setPattern(ExchangePattern.InOnly);
-                    }
-                }
-
-                
-                // set data format mode in Camel exchange
-                camelExchange.setProperty(CxfConstants.DATA_FORMAT_PROPERTY, 
dataFormat);   
-                LOG.trace("Set Exchange property: {}={}", 
DataFormat.class.getName(), dataFormat);
-                
-                camelExchange.setProperty(Message.MTOM_ENABLED, 
String.valueOf(endpoint.isMtomEnabled()));
-                
-                if (endpoint.getMergeProtocolHeaders()) {
-                    
camelExchange.setProperty(CxfConstants.CAMEL_CXF_PROTOCOL_HEADERS_MERGED, 
Boolean.TRUE);
-                }
-                // bind the CXF request into a Camel exchange
-                binding.populateExchangeFromCxfRequest(cxfExchange, 
camelExchange);
-                // extract the javax.xml.ws header
-                Map<String, Object> context = new HashMap<String, Object>();
-                binding.extractJaxWsContext(cxfExchange, context);
-                // put the context into camelExchange
-                camelExchange.setProperty(CxfConstants.JAXWS_CONTEXT, context);
-
-                // we want to handle the UoW
-                try {
-                    CxfConsumer.this.createUoW(camelExchange);
-                } catch (Exception e) {
-                    log.error("Error processing request", e);
-                    throw new Fault(e);
-                }
-                return camelExchange;
-            }
-            
-            @SuppressWarnings("unchecked")
-            private void setResponseBack(Exchange cxfExchange, 
org.apache.camel.Exchange camelExchange) {
-                CxfEndpoint endpoint = (CxfEndpoint)getEndpoint();
-                CxfBinding binding = endpoint.getCxfBinding();                
-                
-                checkFailure(camelExchange, cxfExchange);
-                
-                binding.populateCxfResponseFromExchange(camelExchange, 
cxfExchange);
-                
-                // check failure again as fault could be discovered by 
converter
-                checkFailure(camelExchange, cxfExchange);
-
-                // copy the headers javax.xml.ws header back
-                binding.copyJaxWsContext(cxfExchange, (Map<String, 
Object>)camelExchange.getProperty(CxfConstants.JAXWS_CONTEXT));
-            }
-
-            private void checkFailure(org.apache.camel.Exchange camelExchange, 
Exchange cxfExchange) throws Fault {
-                final Throwable t;
-                if (camelExchange.isFailed()) {
-                    org.apache.camel.Message camelMsg = camelExchange.hasOut() 
? camelExchange.getOut() : camelExchange.getIn();
-                    if (camelMsg.isFault()) {
-                        t = camelMsg.getBody(Throwable.class);
-                    } else {
-                        t = camelExchange.getException();
-                    }
-                    cxfExchange.getInMessage().put(FaultMode.class, 
FaultMode.UNCHECKED_APPLICATION_FAULT);
-                    if (t instanceof Fault) {
-                        cxfExchange.getInMessage().put(FaultMode.class, 
FaultMode.CHECKED_APPLICATION_FAULT);
-                        throw (Fault)t;
-                    } else if (t != null) {                        
-                        // This is not a CXF Fault. Build the CXF Fault 
manually.
-                        Fault fault = new Fault(t);
-                        if (fault.getMessage() == null) {
-                            // The Fault has no Message. This is the case if 
it has
-                            // no message, for example was a 
NullPointerException.
-                            fault.setMessage(t.getClass().getSimpleName());
-                        }
-                        WebFault faultAnnotation = 
t.getClass().getAnnotation(WebFault.class);
-                        Object faultInfo = null;
-                        try {
-                            Method method = 
t.getClass().getMethod("getFaultInfo");
-                            faultInfo = method.invoke(t, new Object[0]);
-                        } catch (Exception e) {
-                            // do nothing here                            
-                        }
-                        if (faultAnnotation != null && faultInfo == null) {
-                            // t has a JAX-WS WebFault annotation, which 
describes
-                            // in detail the Web Service Fault that should be 
thrown. Add the
-                            // detail.
-                            Element detail = fault.getOrCreateDetail();
-                            Element faultDetails = detail.getOwnerDocument()
-                                
.createElementNS(faultAnnotation.targetNamespace(), faultAnnotation.name());
-                            detail.appendChild(faultDetails);
-                        }
-
-                        throw fault;
-                    }
-
-                }
-            }
-
-        });
+        svrBean.setInvoker(new CxfConsumerInvoker(endpoint));
         server = svrBean.create();
         // Apply the server configurer if it is possible 
         if (cxfEndpoint.getCxfEndpointConfigurer() != null) {
@@ -289,6 +86,7 @@ public class CxfConsumer extends DefaultConsumer {
         server.stop();
         super.doStop();
     }
+
     private EndpointReferenceType getReplyTo(Object o) {
         try {
             return 
(EndpointReferenceType)o.getClass().getMethod("getReplyTo").invoke(o);
@@ -296,6 +94,7 @@ public class CxfConsumer extends DefaultConsumer {
             throw new Fault(t);
         }
     }
+
     protected boolean isAsyncInvocationSupported(Exchange cxfExchange) {
         Message cxfMessage = cxfExchange.getInMessage();
         Object addressingProperties = 
cxfMessage.get(CxfConstants.WSA_HEADERS_INBOUND);
@@ -316,5 +115,218 @@ public class CxfConsumer extends DefaultConsumer {
     public Server getServer() {
         return server;
     }
-    
+
+    private final class CxfConsumerInvoker implements Invoker {
+        private final CxfEndpoint endpoint;
+
+        public CxfConsumerInvoker(CxfEndpoint endpoint) {
+            this.endpoint = endpoint;
+        }
+
+        // we receive a CXF request when this method is called
+        public Object invoke(Exchange cxfExchange, Object o) {
+            LOG.trace("Received CXF Request: {}", cxfExchange);
+            Continuation continuation;
+            if (!endpoint.isSynchronous() && 
isAsyncInvocationSupported(cxfExchange)
+                && (continuation = getContinuation(cxfExchange)) != null) {
+                LOG.trace("Calling the Camel async processors.");
+                return asyncInvoke(cxfExchange, continuation);
+            } else {
+                LOG.trace("Calling the Camel sync processors.");
+                return syncInvoke(cxfExchange);
+            }
+        }
+
+        // NOTE this code cannot work with CXF 2.2.x and JMSContinuation
+        // as it doesn't break out the interceptor chain when we call it
+        private Object asyncInvoke(Exchange cxfExchange, final Continuation 
continuation) {
+            synchronized (continuation) {
+                if (continuation.isNew()) {
+                    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());
+
+                    // The continuation could be called before the suspend is 
called
+                    continuation.suspend(cxfEndpoint.getContinuationTimeout());
+
+                    // use the asynchronous API to process the exchange
+                    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) {
+                                LOG.trace("Resuming continuation of 
exchangeId: {}", camelExchange.getExchangeId());
+                                // resume processing after both, sync and 
async callbacks
+                                continuation.setObject(camelExchange);
+                                continuation.resume();
+                            }
+                        }
+                    });
+
+                } else if (continuation.isResumed()) {
+                    org.apache.camel.Exchange camelExchange = 
(org.apache.camel.Exchange)continuation.getObject();
+                    try {
+                        setResponseBack(cxfExchange, camelExchange);
+                    } finally {
+                        CxfConsumer.this.doneUoW(camelExchange);
+                    }
+
+                }
+            }
+            return null;
+        }
+
+        private Continuation getContinuation(Exchange cxfExchange) {
+            ContinuationProvider provider =
+                
(ContinuationProvider)cxfExchange.getInMessage().get(ContinuationProvider.class.getName());
+            Continuation continuation = provider == null ? null : 
provider.getContinuation();
+            // Make sure we don't return the JMSContinuation, as it doesn't 
support the Continuation we wants
+            // Don't want to introduce the dependency of cxf-rt-transprot-jms 
here
+            if (continuation != null && 
continuation.getClass().getName().equals("org.apache.cxf.transport.jms.continuations.JMSContinuation"))
 {
+                return null;
+            } else {
+                return continuation;
+            }
+        }
+
+        private Object syncInvoke(Exchange cxfExchange) {
+            org.apache.camel.Exchange camelExchange = 
prepareCamelExchange(cxfExchange);
+            try {
+                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);
+            }
+            // response should have been set in outMessage's content
+            return null;
+        }
+
+        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();
+
+            DataFormat dataFormat = endpoint.getDataFormat();
+
+            BindingOperationInfo boi = cxfExchange.getBindingOperationInfo();
+            // make sure the "boi" is remained as wrapped in PAYLOAD mode
+            if (boi != null && dataFormat == DataFormat.PAYLOAD && 
boi.isUnwrapped()) {
+                boi = boi.getWrappedOperation();
+                cxfExchange.put(BindingOperationInfo.class, boi);
+            }
+
+            if (boi != null) {
+                
camelExchange.setProperty(BindingOperationInfo.class.getName(), boi);
+                LOG.trace("Set exchange property: BindingOperationInfo: {}", 
boi);
+                // set the message exchange patter with the boi
+                if (boi.getOperationInfo().isOneWay()) {
+                    camelExchange.setPattern(ExchangePattern.InOnly);
+                }
+            } else {
+                if 
(cxfEndpoint.getExchangePattern().equals(ExchangePattern.InOnly)) {
+                    camelExchange.setPattern(ExchangePattern.InOnly);
+                }
+            }
+
+
+            // set data format mode in Camel exchange
+            camelExchange.setProperty(CxfConstants.DATA_FORMAT_PROPERTY, 
dataFormat);
+            LOG.trace("Set Exchange property: {}={}", 
DataFormat.class.getName(), dataFormat);
+
+            camelExchange.setProperty(Message.MTOM_ENABLED, 
String.valueOf(endpoint.isMtomEnabled()));
+
+            if (endpoint.getMergeProtocolHeaders()) {
+                
camelExchange.setProperty(CxfConstants.CAMEL_CXF_PROTOCOL_HEADERS_MERGED, 
Boolean.TRUE);
+            }
+            // bind the CXF request into a Camel exchange
+            binding.populateExchangeFromCxfRequest(cxfExchange, camelExchange);
+            // extract the javax.xml.ws header
+            Map<String, Object> context = new HashMap<String, Object>();
+            binding.extractJaxWsContext(cxfExchange, context);
+            // put the context into camelExchange
+            camelExchange.setProperty(CxfConstants.JAXWS_CONTEXT, context);
+
+            // we want to handle the UoW
+            try {
+                CxfConsumer.this.createUoW(camelExchange);
+            } catch (Exception e) {
+                log.error("Error processing request", e);
+                throw new Fault(e);
+            }
+            return camelExchange;
+        }
+
+        @SuppressWarnings("unchecked")
+        private void setResponseBack(Exchange cxfExchange, 
org.apache.camel.Exchange camelExchange) {
+            CxfEndpoint endpoint = (CxfEndpoint)getEndpoint();
+            CxfBinding binding = endpoint.getCxfBinding();
+
+            checkFailure(camelExchange, cxfExchange);
+
+            binding.populateCxfResponseFromExchange(camelExchange, 
cxfExchange);
+
+            // check failure again as fault could be discovered by converter
+            checkFailure(camelExchange, cxfExchange);
+
+            // copy the headers javax.xml.ws header back
+            binding.copyJaxWsContext(cxfExchange, (Map<String, 
Object>)camelExchange.getProperty(CxfConstants.JAXWS_CONTEXT));
+        }
+
+        private void checkFailure(org.apache.camel.Exchange camelExchange, 
Exchange cxfExchange) throws Fault {
+            final Throwable t;
+            if (camelExchange.isFailed()) {
+                org.apache.camel.Message camelMsg = camelExchange.hasOut() ? 
camelExchange.getOut() : camelExchange.getIn();
+                if (camelMsg.isFault()) {
+                    t = camelMsg.getBody(Throwable.class);
+                } else {
+                    t = camelExchange.getException();
+                }
+                cxfExchange.getInMessage().put(FaultMode.class, 
FaultMode.UNCHECKED_APPLICATION_FAULT);
+                if (t instanceof Fault) {
+                    cxfExchange.getInMessage().put(FaultMode.class, 
FaultMode.CHECKED_APPLICATION_FAULT);
+                    throw (Fault)t;
+                } else if (t != null) {
+                    // This is not a CXF Fault. Build the CXF Fault manually.
+                    Fault fault = new Fault(t);
+                    if (fault.getMessage() == null) {
+                        // The Fault has no Message. This is the case if it has
+                        // no message, for example was a NullPointerException.
+                        fault.setMessage(t.getClass().getSimpleName());
+                    }
+                    WebFault faultAnnotation = 
t.getClass().getAnnotation(WebFault.class);
+                    Object faultInfo = null;
+                    try {
+                        Method method = t.getClass().getMethod("getFaultInfo");
+                        faultInfo = method.invoke(t, new Object[0]);
+                    } catch (Exception e) {
+                        // do nothing here
+                    }
+                    if (faultAnnotation != null && faultInfo == null) {
+                        // t has a JAX-WS WebFault annotation, which describes
+                        // in detail the Web Service Fault that should be 
thrown. Add the
+                        // detail.
+                        Element detail = fault.getOrCreateDetail();
+                        Element faultDetails = detail.getOwnerDocument()
+                            
.createElementNS(faultAnnotation.targetNamespace(), faultAnnotation.name());
+                        detail.appendChild(faultDetails);
+                    }
+
+                    throw fault;
+                }
+
+            }
+        }
+
+    }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/54a03578/components/camel-cxf/src/test/java/org/apache/camel/component/cxf/CxfPayloadConsumerDuplicateNamespaceStreamCacheTest.java
----------------------------------------------------------------------
diff --git 
a/components/camel-cxf/src/test/java/org/apache/camel/component/cxf/CxfPayloadConsumerDuplicateNamespaceStreamCacheTest.java
 
b/components/camel-cxf/src/test/java/org/apache/camel/component/cxf/CxfPayloadConsumerDuplicateNamespaceStreamCacheTest.java
index ace11a7..3e258c8 100644
--- 
a/components/camel-cxf/src/test/java/org/apache/camel/component/cxf/CxfPayloadConsumerDuplicateNamespaceStreamCacheTest.java
+++ 
b/components/camel-cxf/src/test/java/org/apache/camel/component/cxf/CxfPayloadConsumerDuplicateNamespaceStreamCacheTest.java
@@ -25,7 +25,7 @@ public class 
CxfPayloadConsumerDuplicateNamespaceStreamCacheTest extends CxfPayl
      * be an invalid XML.
      */
     protected static final String REQUEST_MESSAGE = "<soap:Envelope 
xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\"; 
xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\"; 
xmlns:xs=\"http://www.w3.org/2001/XMLSchema\";>"
-                                                    + 
"<soap:Body><ns2:getToken 
xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\"; 
xmlns:ns2=\"http://camel.apache.org/cxf/namespace\";><arg0 
xsi:type=\"xs:string\">Send</arg0></ns2:getToken></soap:Body></soap:Envelope>";
+        + "<soap:Body><ns2:getToken 
xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\"; 
xmlns:ns2=\"http://camel.apache.org/cxf/namespace\";><arg0 
xsi:type=\"xs:string\">Send</arg0></ns2:getToken></soap:Body></soap:Envelope>";
 
     @Test
     public void testInvokeRouter() {

Reply via email to