This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/camel.git
commit cecbad0bab7fbd0438922ed6f2d805a69b91b44c Author: Claus Ibsen <claus.ib...@gmail.com> AuthorDate: Wed Feb 14 09:00:14 2018 -0500 CAMEL-12244: Fixed intercept send to endpoint with ServicePoolAware producers not being released back in pool. --- .../apache/camel/impl/InterceptSendToEndpoint.java | 8 +- .../impl/InterceptSendToEndpointProcessor.java | 8 +- ...nterceptSendToEndpointServicePoolProcessor.java | 31 ++++++ ...rceptSendToEndpointServicePoolProducerTest.java | 119 +++++++++++++++++++++ 4 files changed, 163 insertions(+), 3 deletions(-) diff --git a/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java b/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java index 629803c..b4d1d90 100644 --- a/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java +++ b/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpoint.java @@ -27,6 +27,7 @@ import org.apache.camel.ExchangePattern; import org.apache.camel.PollingConsumer; import org.apache.camel.Processor; import org.apache.camel.Producer; +import org.apache.camel.ServicePoolAware; import org.apache.camel.ShutdownableService; import org.apache.camel.util.ServiceHelper; import org.slf4j.Logger; @@ -98,7 +99,12 @@ public class InterceptSendToEndpoint implements Endpoint, ShutdownableService { } public Producer createProducer() throws Exception { - return new InterceptSendToEndpointProcessor(this, delegate, skip); + Producer producer = delegate.createProducer(); + if (producer instanceof ServicePoolAware) { + return new InterceptSendToEndpointServicePoolProcessor(this, delegate, producer, skip); + } else { + return new InterceptSendToEndpointProcessor(this, delegate, producer, skip); + } } public Consumer createConsumer(Processor processor) throws Exception { diff --git a/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpointProcessor.java b/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpointProcessor.java index 10c1052..1808525 100644 --- a/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpointProcessor.java +++ b/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpointProcessor.java @@ -28,6 +28,10 @@ import org.slf4j.LoggerFactory; import static org.apache.camel.processor.PipelineHelper.continueProcessing; +/** + * {@link org.apache.camel.Processor} used to interceptor and detour the routing + * when using the {@link InterceptSendToEndpoint} functionality. + */ public class InterceptSendToEndpointProcessor extends DefaultAsyncProducer { private static final Logger LOG = LoggerFactory.getLogger(InterceptSendToEndpointProcessor.class); @@ -36,11 +40,11 @@ public class InterceptSendToEndpointProcessor extends DefaultAsyncProducer { private final Producer producer; private final boolean skip; - public InterceptSendToEndpointProcessor(InterceptSendToEndpoint endpoint, Endpoint delegate, boolean skip) throws Exception { + public InterceptSendToEndpointProcessor(InterceptSendToEndpoint endpoint, Endpoint delegate, Producer producer, boolean skip) throws Exception { super(delegate); this.endpoint = endpoint; this.delegate = delegate; - this.producer = delegate.createProducer(); + this.producer = producer; this.skip = skip; } diff --git a/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpointServicePoolProcessor.java b/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpointServicePoolProcessor.java new file mode 100644 index 0000000..9de74ef --- /dev/null +++ b/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpointServicePoolProcessor.java @@ -0,0 +1,31 @@ +/** + * 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.impl; + +import org.apache.camel.Endpoint; +import org.apache.camel.Producer; +import org.apache.camel.ServicePoolAware; + +/** + * {@link InterceptSendToEndpointProcessor} used for producers that are {@link ServicePoolAware}. + */ +public class InterceptSendToEndpointServicePoolProcessor extends InterceptSendToEndpointProcessor implements ServicePoolAware { + + public InterceptSendToEndpointServicePoolProcessor(InterceptSendToEndpoint endpoint, Endpoint delegate, Producer producer, boolean skip) throws Exception { + super(endpoint, delegate, producer, skip); + } +} diff --git a/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToEndpointServicePoolProducerTest.java b/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToEndpointServicePoolProducerTest.java new file mode 100644 index 0000000..85e6e8a --- /dev/null +++ b/camel-core/src/test/java/org/apache/camel/processor/intercept/InterceptSendToEndpointServicePoolProducerTest.java @@ -0,0 +1,119 @@ +/** + * 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.util.Map; + +import org.apache.camel.Component; +import org.apache.camel.Consumer; +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Endpoint; +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.Producer; +import org.apache.camel.ServicePoolAware; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.impl.DefaultComponent; +import org.apache.camel.impl.DefaultEndpoint; +import org.apache.camel.impl.DefaultProducer; + +public class InterceptSendToEndpointServicePoolProducerTest extends ContextTestSupport { + + private MyProducer myProducer; + + public void testInterceptSendToEndpointServicePoolProducer() throws Exception { + assertNotNull(myProducer); + + getMockEndpoint("mock:result").expectedMessageCount(2); + getMockEndpoint("mock:hello").expectedMessageCount(2); + + template.sendBody("direct:start", "Hello World"); + template.sendBody("direct:start", "Bye World"); + + assertMockEndpointsSatisfied(); + } + + @Override + protected RouteBuilder createRouteBuilder() throws Exception { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + context.addComponent("mycomp", new MyComponent()); + + interceptSendToEndpoint("mycomp:*").to("mock:hello"); + + from("direct:start") + .to("mycomp:hello") + .to("mock:result"); + } + }; + } + + public class MyComponent extends DefaultComponent { + + @Override + protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception { + return new MyEndpoint(uri, this); + } + } + + public class MyEndpoint extends DefaultEndpoint { + + public MyEndpoint(String endpointUri, Component component) { + super(endpointUri, component); + } + + @Override + public Producer createProducer() throws Exception { + if (myProducer != null) { + throw new IllegalStateException("The producer should only be created once and pooled"); + } + myProducer = new MyProducer(this); + return myProducer; + } + + @Override + public Consumer createConsumer(Processor processor) throws Exception { + return null; + } + + @Override + public boolean isSingleton() { + return true; + } + } + + public class MyProducer extends DefaultProducer implements ServicePoolAware { + + public MyProducer(Endpoint endpoint) { + super(endpoint); + } + + @Override + public void process(Exchange exchange) throws Exception { + if (!isStarted()) { + throw new IllegalArgumentException("Should be started"); + } + } + + @Override + public boolean isSingleton() { + return false; + } + } + +} -- To stop receiving notification emails like this one, please contact davscl...@apache.org.