[camel] branch camel-2.22.x updated (c4229f2 -> dae34c0)

Previous Topic Next Topic
 
classic Classic list List threaded Threaded
3 messages Options
Reply | Threaded
Open this post in threaded view
|

[camel] branch camel-2.22.x updated (c4229f2 -> dae34c0)

davsclaus-2
This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a change to branch camel-2.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git.


    from c4229f2  CAMEL-12713: Fixed CS
     new d6e3851  CAMEL-12709: Fixed UseOriginalAggregationStrategy concurrency issue with splitter. Thanks to Matthias Humbert for reporting and test case.
     new dae34c0  Update sjms-component.adoc

The 2 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:
 .../java/org/apache/camel/processor/Splitter.java  |  6 +--
 .../aggregate/UseOriginalAggregationStrategy.java  | 16 ++++++--
 ...yTest.java => SplitterUseOriginalLoopTest.java} | 46 +++++++++-------------
 .../camel-sjms/src/main/docs/sjms-component.adoc   |  4 --
 4 files changed, 33 insertions(+), 39 deletions(-)
 copy camel-core/src/test/java/org/apache/camel/processor/{RoutingSlipInOutAndInOnlyTest.java => SplitterUseOriginalLoopTest.java} (58%)

Reply | Threaded
Open this post in threaded view
|

[camel] 01/02: CAMEL-12709: Fixed UseOriginalAggregationStrategy concurrency issue with splitter. Thanks to Matthias Humbert for reporting and test case.

davsclaus-2
This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch camel-2.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git

commit d6e38512ff2540bcdadf52150f298cbb03bc9c61
Author: Claus Ibsen <[hidden email]>
AuthorDate: Tue Aug 7 15:38:47 2018 +0200

    CAMEL-12709: Fixed UseOriginalAggregationStrategy concurrency issue with splitter. Thanks to Matthias Humbert for reporting and test case.
---
 .../java/org/apache/camel/processor/Splitter.java  |  6 +--
 .../aggregate/UseOriginalAggregationStrategy.java  | 16 +++++--
 .../processor/SplitterUseOriginalLoopTest.java     | 56 ++++++++++++++++++++++
 3 files changed, 71 insertions(+), 7 deletions(-)

diff --git a/camel-core/src/main/java/org/apache/camel/processor/Splitter.java b/camel-core/src/main/java/org/apache/camel/processor/Splitter.java
index 2a50bc6..1f53748 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/Splitter.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/Splitter.java
@@ -102,10 +102,10 @@ public class Splitter extends MulticastProcessor implements AsyncProcessor, Trac
 
         // set original exchange if not already pre-configured
         if (strategy instanceof UseOriginalAggregationStrategy) {
+            // need to create a new private instance, as we can also have concurrency issue so we cannot store state
             UseOriginalAggregationStrategy original = (UseOriginalAggregationStrategy) strategy;
-            if (original.getOriginal() == null) {
-                original.setOriginal(exchange);
-            }
+            UseOriginalAggregationStrategy clone = original.newInstance(exchange);
+            setAggregationStrategyOnExchange(exchange, clone);
         }
 
         // if no custom aggregation strategy is being used then fallback to keep the original
diff --git a/camel-core/src/main/java/org/apache/camel/processor/aggregate/UseOriginalAggregationStrategy.java b/camel-core/src/main/java/org/apache/camel/processor/aggregate/UseOriginalAggregationStrategy.java
index 5b3ac53..e659026 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/aggregate/UseOriginalAggregationStrategy.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/aggregate/UseOriginalAggregationStrategy.java
@@ -24,11 +24,11 @@ import org.apache.camel.Exchange;
  * and then you may want to keep routing using the original {@link Exchange}.
  *
  * @see org.apache.camel.processor.Splitter
- * @version
+ * @version
  */
 public class UseOriginalAggregationStrategy implements AggregationStrategy {
 
-    private Exchange original;
+    private final Exchange original;
     private final boolean propagateException;
 
     public UseOriginalAggregationStrategy() {
@@ -36,7 +36,7 @@ public class UseOriginalAggregationStrategy implements AggregationStrategy {
     }
 
     public UseOriginalAggregationStrategy(boolean propagateException) {
-        this.propagateException = propagateException;
+        this(null, propagateException);
     }
 
     public UseOriginalAggregationStrategy(Exchange original, boolean propagateException) {
@@ -44,6 +44,13 @@ public class UseOriginalAggregationStrategy implements AggregationStrategy {
         this.propagateException = propagateException;
     }
 
+    /**
+     * Creates a new instance as a clone of this strategy with the new given original.
+     */
+    public UseOriginalAggregationStrategy newInstance(Exchange original) {
+        return new UseOriginalAggregationStrategy(original, propagateException);
+    }
+
     public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
         if (propagateException) {
             Exception exception = checkException(oldExchange, newExchange);
@@ -72,8 +79,9 @@ public class UseOriginalAggregationStrategy implements AggregationStrategy {
         return original;
     }
 
+    @Deprecated
     public void setOriginal(Exchange original) {
-        this.original = original;
+        throw new UnsupportedOperationException("This method is deprecated");
     }
 
     @Override
diff --git a/camel-core/src/test/java/org/apache/camel/processor/SplitterUseOriginalLoopTest.java b/camel-core/src/test/java/org/apache/camel/processor/SplitterUseOriginalLoopTest.java
new file mode 100644
index 0000000..55c31ae
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/processor/SplitterUseOriginalLoopTest.java
@@ -0,0 +1,56 @@
+/**
+ * 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;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.processor.aggregate.UseOriginalAggregationStrategy;
+
+public class SplitterUseOriginalLoopTest extends ContextTestSupport {
+
+    public void testUseOriginalLoop() throws Exception {
+        getMockEndpoint("mock:line").expectedMessageCount(6);
+        getMockEndpoint("mock:result").expectedBodiesReceived("Hello,World");
+        getMockEndpoint("mock:result").expectedHeaderReceived("looping", 2);
+        getMockEndpoint("mock:result").message(0).header("myHeader").isNull();
+
+        template.sendBody("direct:start", "Hello,World");
+
+        assertMockEndpointsSatisfied();
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                from("direct:start")
+                    .loop(3)
+                        .setHeader("looping", exchangeProperty(Exchange.LOOP_INDEX))
+                        .split(body(), new UseOriginalAggregationStrategy(null, false))
+                            .setHeader("myHeader", exchangeProperty(Exchange.LOOP_INDEX))
+                            .to("mock:line")
+                        .end()
+                    .end()
+                    .log("${headers}")
+                    .to("mock:result");
+            }
+        };
+    }
+
+}

Reply | Threaded
Open this post in threaded view
|

[camel] 02/02: Update sjms-component.adoc

davsclaus-2
In reply to this post by davsclaus-2
This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch camel-2.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git

commit dae34c001f04b11ba5dd65dbfff2ff760ddf2849
Author: Claus Ibsen <[hidden email]>
AuthorDate: Tue Aug 7 14:56:46 2018 +0200

    Update sjms-component.adoc
---
 components/camel-sjms/src/main/docs/sjms-component.adoc | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/components/camel-sjms/src/main/docs/sjms-component.adoc b/components/camel-sjms/src/main/docs/sjms-component.adoc
index 8de6cb4..f47e399 100644
--- a/components/camel-sjms/src/main/docs/sjms-component.adoc
+++ b/components/camel-sjms/src/main/docs/sjms-component.adoc
@@ -353,10 +353,6 @@ camelContext.addComponent("sjms", component);
 To see the full example of its usage please refer to the
 https://svn.apache.org/repos/asf/camel/trunk/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/it/ConnectionResourceIT.java[`ConnectionResourceIT`].
 
-#### Session, Consumer, & Producer Pooling & Caching Management
-
-Coming soon ...
-
 #### Batch Message Support
 
 The SjmsProducer supports publishing a collection of messages by