Author: davsclaus
Date: Fri Aug 13 09:04:03 2010
New Revision: 985137
URL: http://svn.apache.org/viewvc?rev=985137&view=rev
Log:
CAMEL-3049: Fixed issue with InterceptStrategy not starting child service when
bridged.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/interceptor/AuditInterceptorAsyncDelegateIssueTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/interceptor/AuditInterceptorDelegateIssueTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/InterceptorToAsyncProcessorBridge.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/InterceptorToAsyncProcessorBridge.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/InterceptorToAsyncProcessorBridge.java?rev=985137&r1=985136&r2=985137&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/InterceptorToAsyncProcessorBridge.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/InterceptorToAsyncProcessorBridge.java
Fri Aug 13 09:04:03 2010
@@ -23,6 +23,7 @@ import org.apache.camel.Processor;
import org.apache.camel.impl.ServiceSupport;
import org.apache.camel.impl.converter.AsyncProcessorTypeConverter;
import org.apache.camel.util.AsyncProcessorHelper;
+import org.apache.camel.util.ServiceHelper;
/**
* A bridge to have regular interceptors implemented as {...@link
org.apache.camel.Processor}
@@ -99,12 +100,13 @@ public class InterceptorToAsyncProcessor
@Override
protected void doStart() throws Exception {
- // noop
+ ServiceHelper.startServices(target, interceptor);
}
@Override
protected void doStop() throws Exception {
callback.remove();
interceptorDone.remove();
+ ServiceHelper.stopServices(interceptor, target);
}
}
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/interceptor/AuditInterceptorAsyncDelegateIssueTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/interceptor/AuditInterceptorAsyncDelegateIssueTest.java?rev=985137&view=auto
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/interceptor/AuditInterceptorAsyncDelegateIssueTest.java
(added)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/interceptor/AuditInterceptorAsyncDelegateIssueTest.java
Fri Aug 13 09:04:03 2010
@@ -0,0 +1,105 @@
+/**
+ * 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.interceptor;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelContext;
+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.model.ProcessorDefinition;
+import org.apache.camel.processor.DelegateAsyncProcessor;
+import org.apache.camel.spi.InterceptStrategy;
+
+/**
+ * @version $Revision$
+ */
+public class AuditInterceptorAsyncDelegateIssueTest extends ContextTestSupport
{
+
+ private MyIntercepStrategy strategy;
+
+ @Override
+ protected void setUp() throws Exception {
+ strategy = new MyIntercepStrategy();
+ super.setUp();
+ }
+
+ public void testOk() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+ getMockEndpoint("mock:handled").expectedMessageCount(0);
+ getMockEndpoint("mock:error").expectedMessageCount(0);
+ getMockEndpoint("mock:dead").expectedMessageCount(0);
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals(true, strategy.isInvoked());
+ }
+
+ public void testILE() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+ getMockEndpoint("mock:handled").expectedMessageCount(1);
+ getMockEndpoint("mock:error").expectedMessageCount(0);
+ getMockEndpoint("mock:dead").expectedMessageCount(0);
+
+ template.sendBody("direct:iae", "Hello World");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals(true, strategy.isInvoked());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ getContext().addInterceptStrategy(strategy);
+
+ onException(IllegalArgumentException.class)
+ .handled(true)
+ .to("mock:handled");
+
+
errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(1));
+
+ from("direct:start").to("mock:result");
+
+ from("direct:iae").throwException(new
IllegalArgumentException("Damn"));
+ }
+ };
+ }
+
+ private final class MyIntercepStrategy implements InterceptStrategy {
+ private volatile boolean invoked;
+
+ public Processor wrapProcessorInInterceptors(CamelContext context,
ProcessorDefinition<?> definition, Processor target, Processor nextTarget)
throws Exception {
+ return new DelegateAsyncProcessor(target) {
+ protected boolean processNext(Exchange exchange, AsyncCallback
callback) {
+ invoked = true;
+ return super.processNext(exchange, callback);
+ }
+ };
+ }
+
+ public boolean isInvoked() {
+ return invoked;
+ }
+ }
+
+}
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/interceptor/AuditInterceptorDelegateIssueTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/interceptor/AuditInterceptorDelegateIssueTest.java?rev=985137&view=auto
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/interceptor/AuditInterceptorDelegateIssueTest.java
(added)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/interceptor/AuditInterceptorDelegateIssueTest.java
Fri Aug 13 09:04:03 2010
@@ -0,0 +1,104 @@
+/**
+ * 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.interceptor;
+
+import org.apache.camel.CamelContext;
+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.model.ProcessorDefinition;
+import org.apache.camel.processor.DelegateProcessor;
+import org.apache.camel.spi.InterceptStrategy;
+
+/**
+ * @version $Revision$
+ */
+public class AuditInterceptorDelegateIssueTest extends ContextTestSupport {
+
+ private MyIntercepStrategy strategy;
+
+ @Override
+ protected void setUp() throws Exception {
+ strategy = new MyIntercepStrategy();
+ super.setUp();
+ }
+
+ public void testOk() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(1);
+ getMockEndpoint("mock:handled").expectedMessageCount(0);
+ getMockEndpoint("mock:error").expectedMessageCount(0);
+ getMockEndpoint("mock:dead").expectedMessageCount(0);
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals(true, strategy.isInvoked());
+ }
+
+ public void testILE() throws Exception {
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+ getMockEndpoint("mock:handled").expectedMessageCount(1);
+ getMockEndpoint("mock:error").expectedMessageCount(0);
+ getMockEndpoint("mock:dead").expectedMessageCount(0);
+
+ template.sendBody("direct:iae", "Hello World");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals(true, strategy.isInvoked());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ getContext().addInterceptStrategy(strategy);
+
+ onException(IllegalArgumentException.class)
+ .handled(true)
+ .to("mock:handled");
+
+
errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(1));
+
+ from("direct:start").to("mock:result");
+
+ from("direct:iae").throwException(new
IllegalArgumentException("Damn"));
+ }
+ };
+ }
+
+ private final class MyIntercepStrategy implements InterceptStrategy {
+ private volatile boolean invoked;
+
+ public Processor wrapProcessorInInterceptors(CamelContext context,
ProcessorDefinition<?> definition, Processor target, Processor nextTarget)
throws Exception {
+ return new DelegateProcessor(target) {
+ protected void processNext(Exchange exchange) throws Exception
{
+ invoked = true;
+ super.processNext(exchange);
+ }
+ };
+ }
+
+ public boolean isInvoked() {
+ return invoked;
+ }
+ }
+
+}
