You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2018/02/14 14:34:31 UTC
[camel] branch master updated (f102d82 -> cecbad0)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from f102d82 Upgrade Chronicle Thread to version 1.9.8
new 0c3f2f2 CAMEL-12244: Intercept send to endpoint moved inner processor to external class.
new 6195679 Regen
new cecbad0 CAMEL-12244: Fixed intercept send to endpoint with ServicePoolAware producers not being released back in pool.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/camel/impl/InterceptSendToEndpoint.java | 113 ++--------------
.../impl/InterceptSendToEndpointProcessor.java | 147 +++++++++++++++++++++
...terceptSendToEndpointServicePoolProcessor.java} | 19 +--
...ceptSendToEndpointServicePoolProducerTest.java} | 72 +++++-----
.../camel-spring-boot-dependencies/pom.xml | 2 +-
5 files changed, 203 insertions(+), 150 deletions(-)
create mode 100644 camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpointProcessor.java
copy camel-core/src/main/java/org/apache/camel/impl/{DefaultProducerServicePool.java => InterceptSendToEndpointServicePoolProcessor.java} (65%)
copy camel-core/src/test/java/org/apache/camel/processor/{DefaultScheduledPollConsumerBridgeErrorHandlerTest.java => intercept/InterceptSendToEndpointServicePoolProducerTest.java} (57%)
--
To stop receiving notification emails like this one, please contact
davsclaus@apache.org.
[camel] 01/03: CAMEL-12244: Intercept send to endpoint moved inner
processor to external class.
Posted by da...@apache.org.
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 0c3f2f28ede94f8e7a47934d523b29d1eacc7720
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Feb 14 06:57:24 2018 -0500
CAMEL-12244: Intercept send to endpoint moved inner processor to external class.
---
.../apache/camel/impl/InterceptSendToEndpoint.java | 107 +--------------
.../impl/InterceptSendToEndpointProcessor.java | 143 +++++++++++++++++++++
2 files changed, 148 insertions(+), 102 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 b94b3eb..629803c 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
@@ -18,8 +18,6 @@ package org.apache.camel.impl;
import java.util.Map;
-import org.apache.camel.AsyncCallback;
-import org.apache.camel.AsyncProcessor;
import org.apache.camel.CamelContext;
import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
@@ -34,8 +32,6 @@ import org.apache.camel.util.ServiceHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.camel.processor.PipelineHelper.continueProcessing;
-
/**
* This is an endpoint when sending to it, is intercepted and is routed in a detour
*
@@ -46,7 +42,6 @@ public class InterceptSendToEndpoint implements Endpoint, ShutdownableService {
private static final Logger LOG = LoggerFactory.getLogger(InterceptSendToEndpoint.class);
private final Endpoint delegate;
- private Producer producer;
private Processor detour;
private boolean skip;
@@ -65,6 +60,10 @@ public class InterceptSendToEndpoint implements Endpoint, ShutdownableService {
this.detour = detour;
}
+ public Processor getDetour() {
+ return detour;
+ }
+
public Endpoint getDelegate() {
return delegate;
}
@@ -99,103 +98,7 @@ public class InterceptSendToEndpoint implements Endpoint, ShutdownableService {
}
public Producer createProducer() throws Exception {
- producer = delegate.createProducer();
- return new DefaultAsyncProducer(delegate) {
-
- public Endpoint getEndpoint() {
- return producer.getEndpoint();
- }
-
- public Exchange createExchange() {
- return producer.createExchange();
- }
-
- public Exchange createExchange(ExchangePattern pattern) {
- return producer.createExchange(pattern);
- }
-
- @Deprecated
- public Exchange createExchange(Exchange exchange) {
- return producer.createExchange(exchange);
- }
-
- @Override
- public boolean process(Exchange exchange, AsyncCallback callback) {
- // process the detour so we do the detour routing
- if (LOG.isDebugEnabled()) {
- LOG.debug("Sending to endpoint: {} is intercepted and detoured to: {} for exchange: {}", new Object[]{getEndpoint(), detour, exchange});
- }
- // add header with the real endpoint uri
- exchange.getIn().setHeader(Exchange.INTERCEPTED_ENDPOINT, delegate.getEndpointUri());
-
- // detour the exchange using synchronous processing
- try {
- detour.process(exchange);
- } catch (Exception e) {
- exchange.setException(e);
- }
-
- // Decide whether to continue or not; similar logic to the Pipeline
- // check for error if so we should break out
- if (!continueProcessing(exchange, "skip sending to original intended destination: " + getEndpoint(), LOG)) {
- callback.done(true);
- return true;
- }
-
- // determine if we should skip or not
- boolean shouldSkip = skip;
-
- // if then interceptor had a when predicate, then we should only skip if it matched
- Boolean whenMatches = (Boolean) exchange.removeProperty(Exchange.INTERCEPT_SEND_TO_ENDPOINT_WHEN_MATCHED);
- if (whenMatches != null) {
- shouldSkip = skip && whenMatches;
- }
-
- if (!shouldSkip) {
- if (exchange.hasOut()) {
- // replace OUT with IN as detour changed something
- exchange.setIn(exchange.getOut());
- exchange.setOut(null);
- }
-
- // route to original destination leveraging the asynchronous routing engine if possible
- if (producer instanceof AsyncProcessor) {
- AsyncProcessor async = (AsyncProcessor) producer;
- return async.process(exchange, callback);
- } else {
- try {
- producer.process(exchange);
- } catch (Exception e) {
- exchange.setException(e);
- }
- callback.done(true);
- return true;
- }
- } else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Stop() means skip sending exchange to original intended destination: {} for exchange: {}", getEndpoint(), exchange);
- }
- callback.done(true);
- return true;
- }
- }
-
- public boolean isSingleton() {
- return producer.isSingleton();
- }
-
- public void start() throws Exception {
- ServiceHelper.startService(detour);
- // here we also need to start the producer
- ServiceHelper.startService(producer);
- }
-
- public void stop() throws Exception {
- // do not stop detour as it should only be stopped when the interceptor stops
- // we should stop the producer here
- ServiceHelper.stopService(producer);
- }
- };
+ return new InterceptSendToEndpointProcessor(this, delegate, 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
new file mode 100644
index 0000000..10c1052
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/impl/InterceptSendToEndpointProcessor.java
@@ -0,0 +1,143 @@
+/**
+ * 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.AsyncCallback;
+import org.apache.camel.AsyncProcessor;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Producer;
+import org.apache.camel.util.ServiceHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.camel.processor.PipelineHelper.continueProcessing;
+
+public class InterceptSendToEndpointProcessor extends DefaultAsyncProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(InterceptSendToEndpointProcessor.class);
+ private final InterceptSendToEndpoint endpoint;
+ private final Endpoint delegate;
+ private final Producer producer;
+ private final boolean skip;
+
+ public InterceptSendToEndpointProcessor(InterceptSendToEndpoint endpoint, Endpoint delegate, boolean skip) throws Exception {
+ super(delegate);
+ this.endpoint = endpoint;
+ this.delegate = delegate;
+ this.producer = delegate.createProducer();
+ this.skip = skip;
+ }
+
+ public Endpoint getEndpoint() {
+ return producer.getEndpoint();
+ }
+
+ public Exchange createExchange() {
+ return producer.createExchange();
+ }
+
+ public Exchange createExchange(ExchangePattern pattern) {
+ return producer.createExchange(pattern);
+ }
+
+ @Deprecated
+ public Exchange createExchange(Exchange exchange) {
+ return producer.createExchange(exchange);
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ // process the detour so we do the detour routing
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Sending to endpoint: {} is intercepted and detoured to: {} for exchange: {}", new Object[]{getEndpoint(), endpoint.getDetour(), exchange});
+ }
+ // add header with the real endpoint uri
+ exchange.getIn().setHeader(Exchange.INTERCEPTED_ENDPOINT, delegate.getEndpointUri());
+
+ if (endpoint.getDetour() != null) {
+ // detour the exchange using synchronous processing
+ try {
+ endpoint.getDetour().process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
+ }
+
+ // Decide whether to continue or not; similar logic to the Pipeline
+ // check for error if so we should break out
+ if (!continueProcessing(exchange, "skip sending to original intended destination: " + getEndpoint(), LOG)) {
+ callback.done(true);
+ return true;
+ }
+
+ // determine if we should skip or not
+ boolean shouldSkip = skip;
+
+ // if then interceptor had a when predicate, then we should only skip if it matched
+ Boolean whenMatches = (Boolean) exchange.removeProperty(Exchange.INTERCEPT_SEND_TO_ENDPOINT_WHEN_MATCHED);
+ if (whenMatches != null) {
+ shouldSkip = skip && whenMatches;
+ }
+
+ if (!shouldSkip) {
+ if (exchange.hasOut()) {
+ // replace OUT with IN as detour changed something
+ exchange.setIn(exchange.getOut());
+ exchange.setOut(null);
+ }
+
+ // route to original destination leveraging the asynchronous routing engine if possible
+ if (producer instanceof AsyncProcessor) {
+ AsyncProcessor async = (AsyncProcessor) producer;
+ return async.process(exchange, callback);
+ } else {
+ try {
+ producer.process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
+ callback.done(true);
+ return true;
+ }
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Stop() means skip sending exchange to original intended destination: {} for exchange: {}", getEndpoint(), exchange);
+ }
+ callback.done(true);
+ return true;
+ }
+ }
+
+ public boolean isSingleton() {
+ return producer.isSingleton();
+ }
+
+ public void start() throws Exception {
+ ServiceHelper.startService(endpoint.getDetour());
+ // here we also need to start the producer
+ ServiceHelper.startService(producer);
+ }
+
+ public void stop() throws Exception {
+ // do not stop detour as it should only be stopped when the interceptor stops
+ // we should stop the producer here
+ ServiceHelper.stopService(producer);
+ }
+
+}
--
To stop receiving notification emails like this one, please contact
davsclaus@apache.org.
[camel] 02/03: Regen
Posted by da...@apache.org.
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 6195679a1cdadd31e9654b021c28a2f668d12dda
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Feb 14 07:47:08 2018 -0500
Regen
---
.../spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/platforms/spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml b/platforms/spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml
index 7488bc2..14c6f32 100644
--- a/platforms/spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml
+++ b/platforms/spring-boot/spring-boot-dm/camel-spring-boot-dependencies/pom.xml
@@ -3369,7 +3369,7 @@
<dependency>
<groupId>org.knowm.xchange</groupId>
<artifactId>xchange-core</artifactId>
- <version>4.3.2</version>
+ <version>4.3.3</version>
</dependency>
<dependency>
<groupId>org.mozilla</groupId>
--
To stop receiving notification emails like this one, please contact
davsclaus@apache.org.
[camel] 03/03: CAMEL-12244: Fixed intercept send to endpoint with
ServicePoolAware producers not being released back in pool.
Posted by da...@apache.org.
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 <cl...@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
davsclaus@apache.org.