Author: davsclaus
Date: Thu Jul 30 08:51:20 2009
New Revision: 799205
URL: http://svn.apache.org/viewvc?rev=799205&view=rev
Log:
CAMEL-1864: fixed intercept send to endpoint not working for send processors in
onException.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToEndpointInOnException.java
(with props)
camel/trunk/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringInterceptSendToEndpointInOnExceptionTest.java
(with props)
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/interceptSendToEndpointInOnExceptionTest.xml
- copied, changed from r799169,
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringInterceptFromTest.xml
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/InterceptSendToEndpointDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/model/RoutesDefinition.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/FromFileInterceptSendToIssue.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToIssue.java
camel/trunk/components/camel-spring/src/main/java/org/apache/camel/spring/CamelContextFactoryBean.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/InterceptSendToEndpointDefinition.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/InterceptSendToEndpointDefinition.java?rev=799205&r1=799204&r2=799205&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/model/InterceptSendToEndpointDefinition.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/model/InterceptSendToEndpointDefinition.java
Thu Jul 30 08:51:20 2009
@@ -42,6 +42,11 @@
// TODO: Support lookup endpoint by ref (requires a bit more work)
+ // TODO: interceptSendToEndpoint needs to proxy the endpoints at very first
+ // so when other processors uses an endpoint its already proxied, see
workaround in SendProcessor
+ // needed when we havent proxied beforehand. This requires some work in
the route builder in Camel
+ // to implement so that should be a part of a bigger rework/improvement in
the future
+
@XmlAttribute(required = true)
private String uri;
@XmlAttribute(required = false)
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/model/RoutesDefinition.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/model/RoutesDefinition.java?rev=799205&r1=799204&r2=799205&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/model/RoutesDefinition.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/model/RoutesDefinition.java
Thu Jul 30 08:51:20 2009
@@ -201,7 +201,7 @@
// lets configure the route
route.setCamelContext(getCamelContext());
- // configure intercepts
+ // configure intercept
for (InterceptDefinition intercept : getIntercepts()) {
// add as first output so intercept is handled before the acutal
route and that gives
// us the needed head start to init and be able to intercept all
the remaining processing steps
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java?rev=799205&r1=799204&r2=799205&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
Thu Jul 30 08:51:20 2009
@@ -24,6 +24,7 @@
import org.apache.camel.ProducerCallback;
import org.apache.camel.impl.ProducerCache;
import org.apache.camel.impl.ServiceSupport;
+import org.apache.camel.impl.InterceptSendToEndpoint;
import org.apache.camel.util.ObjectHelper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -38,6 +39,7 @@
protected ProducerCache producerCache;
protected Endpoint destination;
protected ExchangePattern pattern;
+ private boolean init;
public SendProcessor(Endpoint destination) {
ObjectHelper.notNull(destination, "destination");
@@ -59,6 +61,20 @@
}
public void process(final Exchange exchange) throws Exception {
+ // the destination could since have been intercepted by a
interceptSendToEndpoint so we got to
+ // init this before we can use the destination
+ if (!init) {
+ init = true;
+ Endpoint lookup =
exchange.getContext().getEndpoint(destination.getEndpointUri());
+ if (lookup instanceof InterceptSendToEndpoint) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("SendTo is intercepted using a
interceptSendToEndpoint: " + lookup.getEndpointUri());
+ }
+ destination = lookup;
+ }
+ }
+
+ // send the exchange to the destination using a producer
getProducerCache(exchange).doInProducer(destination, exchange,
pattern, new ProducerCallback<Exchange>() {
public Exchange doInProducer(Producer producer, Exchange exchange,
ExchangePattern pattern) throws Exception {
exchange = configureExchange(exchange, pattern);
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java?rev=799205&r1=799204&r2=799205&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ExchangeHelper.java
Thu Jul 30 08:51:20 2009
@@ -77,7 +77,7 @@
* @return the binding object of the given type or null if it could not be
found or converted
*/
public static <T> T getBinding(Exchange exchange, Class<T> type) {
- return exchange != null ? (T) exchange.getProperty(Exchange.BINDING,
type) : null;
+ return exchange != null ? exchange.getProperty(Exchange.BINDING, type)
: null;
}
/**
@@ -103,8 +103,7 @@
return endpoint;
}
- public static <T> T getMandatoryProperty(Exchange exchange, String
propertyName, Class<T> type)
- throws NoSuchPropertyException {
+ public static <T> T getMandatoryProperty(Exchange exchange, String
propertyName, Class<T> type) throws NoSuchPropertyException {
T result = exchange.getProperty(propertyName, type);
if (result != null) {
return result;
@@ -112,8 +111,7 @@
throw new NoSuchPropertyException(exchange, propertyName, type);
}
- public static <T> T getMandatoryHeader(Exchange exchange, String
propertyName, Class<T> type)
- throws NoSuchHeaderException {
+ public static <T> T getMandatoryHeader(Exchange exchange, String
propertyName, Class<T> type) throws NoSuchHeaderException {
T answer = exchange.getIn().getHeader(propertyName, type);
if (answer == null) {
throw new NoSuchHeaderException(exchange, propertyName, type);
Modified:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/FromFileInterceptSendToIssue.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/FromFileInterceptSendToIssue.java?rev=799205&r1=799204&r2=799205&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/FromFileInterceptSendToIssue.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/FromFileInterceptSendToIssue.java
Thu Jul 30 08:51:20 2009
@@ -31,14 +31,14 @@
public void testInterceptSendTo() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:foo");
mock.expectedMessageCount(1);
- mock.expectedHeaderReceived(Exchange.INTERCEPTED_ENDPOINT,
"direct:foo");
+ mock.expectedHeaderReceived(Exchange.INTERCEPTED_ENDPOINT,
"direct://foo");
template.sendBodyAndHeader("file://target/intercept", "Hello World",
Exchange.FILE_NAME, "input.txt");
assertMockEndpointsSatisfied();
Exchange exchange = mock.getReceivedExchanges().get(0);
- assertEquals("direct:start",
exchange.getFromEndpoint().getEndpointUri());
+ assertEquals("file://target/intercept",
exchange.getFromEndpoint().getEndpointUri());
}
@Override
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToEndpointInOnException.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToEndpointInOnException.java?rev=799205&view=auto
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToEndpointInOnException.java
(added)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToEndpointInOnException.java
Thu Jul 30 08:51:20 2009
@@ -0,0 +1,114 @@
+/**
+ * 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.processor.intercept;
+
+import java.io.IOException;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+
+/**
+ * @version $Revision$
+ */
+public class InterceptSendToEndpointInOnException extends ContextTestSupport {
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ onException(IOException.class)
+ .handled(true)
+ .to("mock:io");
+
+ interceptSendToEndpoint("mock:io")
+ .skipSendToOriginalEndpoint()
+ .to("mock:intercepted");
+
+ from("direct:start")
+ .to("mock:foo")
+ .to("mock:result");
+ }
+ };
+ }
+
+ public void testOk() throws Exception {
+ getMockEndpoint("mock:io").expectedMessageCount(0);
+ getMockEndpoint("mock:intercepted").expectedMessageCount(0);
+ getMockEndpoint("mock:foo").expectedMessageCount(2);
+ getMockEndpoint("mock:result").expectedMessageCount(2);
+
+ template.sendBody("direct:start", "Hello World");
+ template.sendBody("direct:start", "Bye World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testAnotherError() throws Exception {
+ getMockEndpoint("mock:io").expectedMessageCount(0);
+ getMockEndpoint("mock:intercepted").expectedMessageCount(0);
+ getMockEndpoint("mock:foo").expectedMessageCount(2);
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+
+ MockEndpoint foo = getMockEndpoint("mock:foo");
+ foo.whenAnyExchangeReceived(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ throw new IllegalArgumentException("Forced");
+ }
+ });
+
+ try {
+ template.sendBody("direct:start", "Hello World");
+ fail("Should have thrown exception");
+ } catch (CamelExecutionException e) {
+ assertIsInstanceOf(IllegalArgumentException.class, e.getCause());
+ }
+
+ try {
+ template.sendBody("direct:start", "Bye World");
+ fail("Should have thrown exception");
+ } catch (CamelExecutionException e) {
+ assertIsInstanceOf(IllegalArgumentException.class, e.getCause());
+ }
+
+ assertMockEndpointsSatisfied();
+ }
+
+ public void testIOError() throws Exception {
+ getMockEndpoint("mock:io").expectedMessageCount(0);
+ getMockEndpoint("mock:intercepted").expectedMessageCount(2);
+ getMockEndpoint("mock:foo").expectedMessageCount(2);
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+
+ MockEndpoint foo = getMockEndpoint("mock:foo");
+ foo.whenAnyExchangeReceived(new Processor() {
+ public void process(Exchange exchange) throws Exception {
+ throw new IOException("Forced");
+ }
+ });
+
+ template.sendBody("direct:start", "Hello World");
+ template.sendBody("direct:start", "Bye World");
+
+ assertMockEndpointsSatisfied();
+ }
+
+}
Propchange:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToEndpointInOnException.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToEndpointInOnException.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToIssue.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToIssue.java?rev=799205&r1=799204&r2=799205&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToIssue.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToIssue.java
Thu Jul 30 08:51:20 2009
@@ -31,14 +31,14 @@
public void testInterceptSendTo() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:foo");
mock.expectedMessageCount(1);
- mock.expectedHeaderReceived(Exchange.INTERCEPTED_ENDPOINT,
"direct:foo");
+ mock.expectedHeaderReceived(Exchange.INTERCEPTED_ENDPOINT,
"direct://foo");
template.sendBody("direct:start", "Hello World");
assertMockEndpointsSatisfied();
Exchange exchange = mock.getReceivedExchanges().get(0);
- assertEquals("direct:start",
exchange.getFromEndpoint().getEndpointUri());
+ assertEquals("direct://start",
exchange.getFromEndpoint().getEndpointUri());
}
@Override
Modified:
camel/trunk/components/camel-spring/src/main/java/org/apache/camel/spring/CamelContextFactoryBean.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-spring/src/main/java/org/apache/camel/spring/CamelContextFactoryBean.java?rev=799205&r1=799204&r2=799205&view=diff
==============================================================================
---
camel/trunk/components/camel-spring/src/main/java/org/apache/camel/spring/CamelContextFactoryBean.java
(original)
+++
camel/trunk/components/camel-spring/src/main/java/org/apache/camel/spring/CamelContextFactoryBean.java
Thu Jul 30 08:51:20 2009
@@ -17,6 +17,7 @@
package org.apache.camel.spring;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import javax.xml.bind.annotation.XmlAccessType;
@@ -374,7 +375,10 @@
for (int i = 0; i < route.getOutputs().size(); i++) {
index = i;
ProcessorDefinition out = route.getOutputs().get(i);
- if (!(out instanceof InterceptDefinition)) {
+ if (out instanceof InterceptDefinition || out instanceof
InterceptSendToEndpointDefinition) {
+ continue;
+ } else {
+ // we found the spot
break;
}
}
@@ -385,9 +389,12 @@
private void initPolicies(RouteDefinition route) {
// setup the policies as JAXB yet again have not created a correct
model for us
List<ProcessorDefinition> types = route.getOutputs();
- // we need to types as transacted cannot extend policy due JAXB
limitations
+
+ // we need two types as transacted cannot extend policy due JAXB
limitations
PolicyDefinition policy = null;
TransactedDefinition transacted = null;
+
+ // add to correct type
for (ProcessorDefinition type : types) {
if (type instanceof PolicyDefinition) {
policy = (PolicyDefinition) type;
@@ -401,6 +408,7 @@
transacted.addOutput(type);
}
}
+
// did we find a policy if so replace it as the only output on the
route
if (policy != null) {
route.clearOutput();
@@ -539,9 +547,7 @@
* the default routes on startup
*/
public void setRouteBuilders(RouteBuilder[] builders) {
- for (RouteBuilder builder : builders) {
- additionalBuilders.add(builder);
- }
+ additionalBuilders.addAll(Arrays.asList(builders));
}
public ApplicationContext getApplicationContext() {
@@ -787,25 +793,22 @@
* Strategy method to try find {...@link RouteBuilder} instances on the
classpath
*/
protected void findRouteBuilders() throws Exception {
-
PackageScanClassResolver resolver =
getContext().getPackageScanClassResolver();
addPackageElementContentsToScanDefinition();
PackageScanDefinition packageScanDef = getPackageScan();
-
if (packageScanDef != null && packageScanDef.getPackages().size() > 0)
{
-
+ // use package scan filter
PatternBasedPackageScanFilter filter = new
PatternBasedPackageScanFilter();
filter.addIncludePatterns(packageScanDef.getIncludes());
filter.addExcludePatterns(packageScanDef.getExcludes());
resolver.addFilter(filter);
String[] normalized =
normalizePackages(packageScanDef.getPackages());
- RouteBuilderFinder finder = new RouteBuilderFinder(getContext(),
normalized, getContextClassLoaderOnStart(), getBeanPostProcessor(), getContext()
- .getPackageScanClassResolver());
+ RouteBuilderFinder finder = new RouteBuilderFinder(getContext(),
normalized, getContextClassLoaderOnStart(),
+ getBeanPostProcessor(),
getContext().getPackageScanClassResolver());
finder.appendBuilders(getAdditionalBuilders());
}
-
}
private void addPackageElementContentsToScanDefinition() {
Added:
camel/trunk/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringInterceptSendToEndpointInOnExceptionTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringInterceptSendToEndpointInOnExceptionTest.java?rev=799205&view=auto
==============================================================================
---
camel/trunk/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringInterceptSendToEndpointInOnExceptionTest.java
(added)
+++
camel/trunk/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringInterceptSendToEndpointInOnExceptionTest.java
Thu Jul 30 08:51:20 2009
@@ -0,0 +1,32 @@
+/**
+ * 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.spring.processor;
+
+import org.apache.camel.CamelContext;
+import
org.apache.camel.processor.intercept.InterceptSendToEndpointInOnException;
+import static
org.apache.camel.spring.processor.SpringTestHelper.createSpringCamelContext;
+
+/**
+ * @version $Revision$
+ */
+public class SpringInterceptSendToEndpointInOnExceptionTest extends
InterceptSendToEndpointInOnException {
+
+ protected CamelContext createCamelContext() throws Exception {
+ return createSpringCamelContext(this,
"org/apache/camel/spring/processor/interceptSendToEndpointInOnExceptionTest.xml");
+ }
+
+}
Propchange:
camel/trunk/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringInterceptSendToEndpointInOnExceptionTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
camel/trunk/components/camel-spring/src/test/java/org/apache/camel/spring/processor/SpringInterceptSendToEndpointInOnExceptionTest.java
------------------------------------------------------------------------------
svn:keywords = Rev Date
Copied:
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/interceptSendToEndpointInOnExceptionTest.xml
(from r799169,
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringInterceptFromTest.xml)
URL:
http://svn.apache.org/viewvc/camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/interceptSendToEndpointInOnExceptionTest.xml?p2=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/interceptSendToEndpointInOnExceptionTest.xml&p1=camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringInterceptFromTest.xml&r1=799169&r2=799205&rev=799205&view=diff
==============================================================================
---
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/SpringInterceptFromTest.xml
(original)
+++
camel/trunk/components/camel-spring/src/test/resources/org/apache/camel/spring/processor/interceptSendToEndpointInOnExceptionTest.xml
Thu Jul 30 08:51:20 2009
@@ -22,20 +22,23 @@
http://camel.apache.org/schema/spring
http://camel.apache.org/schema/spring/camel-spring.xsd
">
- <!-- START SNIPPET: example -->
<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring";>
- <!-- intercept incoming messages and route them to the mock:middle1
endpoint
- before we proceed and continue routing from the point of
interceptions, that
- is mock:end will be the next target -->
- <interceptFrom>
- <to uri="mock:middle1"/>
- </interceptFrom>
- <!-- here we have a very simple route -->
+ <onException>
+ <exception>java.io.IOException</exception>
+ <handled><constant>true</constant></handled>
+ <to uri="mock:io"/>
+ </onException>
+
+ <interceptSendToEndpoint uri="mock:io"
skipSendToOriginalEndpoint="true">
+ <to uri="mock:intercepted"/>
+ </interceptSendToEndpoint>
+
<route>
<from uri="direct:start"/>
- <to uri="mock:end"/>
+ <to uri="mock:foo"/>
+ <to uri="mock:result"/>
</route>
+
</camelContext>
- <!-- END SNIPPET: example -->
</beans>
