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
The following commit(s) were added to refs/heads/master by this push: new ab18d30 CAMEL-16091: Fixed Enricher EIP to handover onCompletions to UoW so they are executed at end of routing. PollEnricher already did this. Thanks to Michał Ostrowski for test case in camel-netty-http. ab18d30 is described below commit ab18d30c2e65d2706ea953d8ff956aaaf3b30d0d Author: Claus Ibsen <claus.ib...@gmail.com> AuthorDate: Sat Jan 30 16:07:27 2021 +0100 CAMEL-16091: Fixed Enricher EIP to handover onCompletions to UoW so they are executed at end of routing. PollEnricher already did this. Thanks to Michał Ostrowski for test case in camel-netty-http. --- .../netty/http/NettyEnricherLeakTest.java | 75 ++++++++++++++++++++++ .../java/org/apache/camel/processor/Enricher.java | 4 ++ .../aggregate/StringAggregationStrategy.java | 2 +- 3 files changed, 80 insertions(+), 1 deletion(-) diff --git a/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyEnricherLeakTest.java b/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyEnricherLeakTest.java new file mode 100644 index 0000000..c0316de --- /dev/null +++ b/components/camel-netty-http/src/test/java/org/apache/camel/component/netty/http/NettyEnricherLeakTest.java @@ -0,0 +1,75 @@ +/* + * 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.component.netty.http; + +import io.netty.util.ResourceLeakDetector; +import org.apache.camel.builder.AggregationStrategies; +import org.apache.camel.builder.RouteBuilder; +import org.junit.jupiter.api.Test; + +public class NettyEnricherLeakTest extends BaseNettyTest { + + @Override + public boolean isUseRouteBuilder() { + return false; + } + + @Test + public void leakNoTest() throws Exception { + context.addRoutes(new RouteBuilder() { + @Override + public void configure() throws Exception { + ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.PARANOID); + + from("netty-http:http://localhost:" + getPort() + "/test") + .transform().simple("${body}"); + + from("direct:outer") + .to("netty-http:http://localhost:" + getPort() + "/test?disconnect=true"); + } + }); + context.start(); + + ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.PARANOID); + for (int i = 0; i < 10; ++i) { + template.requestBody("direct:outer", "input", String.class); + } + } + + @Test + public void leakTest() throws Exception { + context.addRoutes(new RouteBuilder() { + @Override + public void configure() throws Exception { + ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.PARANOID); + + from("netty-http:http://localhost:" + getPort() + "/test") + .transform().simple("${body}"); + + from("direct:outer") + .enrich("netty-http:http://localhost:" + getPort() + "/test?disconnect=true", + AggregationStrategies.string(), false, false); + } + }); + context.start(); + + ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.PARANOID); + for (int i = 0; i < 10; ++i) { + template.requestBody("direct:outer", "input", String.class); + } + } +} diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java index ac2565d..4038125 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/Enricher.java @@ -237,6 +237,10 @@ public class Enricher extends AsyncProcessorSupport implements IdAware, RouteIdA if (aggregatedExchange != null) { // copy aggregation result onto original exchange (preserving pattern) copyResultsPreservePattern(exchange, aggregatedExchange); + // handover any synchronization + if (resourceExchange != null) { + resourceExchange.adapt(ExtendedExchange.class).handoverCompletions(exchange); + } } } catch (Throwable e) { // if the aggregationStrategy threw an exception, set it on the original exchange diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/StringAggregationStrategy.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/StringAggregationStrategy.java index 43422a6..b0121bc 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/StringAggregationStrategy.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/aggregate/StringAggregationStrategy.java @@ -35,7 +35,7 @@ public class StringAggregationStrategy implements AggregationStrategy { /** * Set delimiter used for joining aggregated String * - * @param delimiter The delimiter to join with. Default empty String + * @param delimiter The delimiter to join with. Default empty String */ public StringAggregationStrategy delimiter(String delimiter) { this.delimiter = delimiter;