You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by lb...@apache.org on 2017/05/05 11:10:28 UTC
[7/9] camel git commit: CAMEL-10807: upgraded to reactor 3.0.7 to fix
a bug with emitter processor
CAMEL-10807: upgraded to reactor 3.0.7 to fix a bug with emitter processor
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/2956fca2
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/2956fca2
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/2956fca2
Branch: refs/heads/master
Commit: 2956fca2f1fd1c8609d682fa58e72e3ccc54826a
Parents: 4266cf6
Author: Nicola Ferraro <ni...@gmail.com>
Authored: Fri Apr 28 02:13:43 2017 +0200
Committer: lburgazzoli <lb...@gmail.com>
Committed: Fri May 5 13:10:06 2017 +0200
----------------------------------------------------------------------
.../reactor/engine/ReactorCamelProcessor.java | 13 ++++++-------
parent/pom.xml | 2 +-
2 files changed, 7 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/2956fca2/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorCamelProcessor.java
----------------------------------------------------------------------
diff --git a/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorCamelProcessor.java b/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorCamelProcessor.java
index 548cace..d3c270f 100644
--- a/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorCamelProcessor.java
+++ b/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorCamelProcessor.java
@@ -49,12 +49,11 @@ final class ReactorCamelProcessor implements Closeable {
this.camelProducer = null;
this.camelSink = new AtomicReference<>();
-
- // TODO: A emitter processor with buffer-size 0 would be perfect
- // The effect of having a prefetch of 1 element is that the chain buffers at least 2 elements instead of only one
- // (one in the FluxSink and one in the EmitterProcessor) when using the "latest" or "oldest" strategy.
- // This affects slightly the behavior of the backpressure strategy "latest" (but it doesn't change the semantics).
- this.publisher = EmitterProcessor.create(1);
+ // TODO: The perfect emitter processor would have no buffer (0 sized)
+ // The chain caches one more item than expected.
+ // This implementation has (almost) full control over backpressure, but it's too chatty.
+ // There's a ticket to improve chattiness of the reactive-streams internal impl.
+ this.publisher = EmitterProcessor.create(1, false);
}
@Override
@@ -76,7 +75,7 @@ final class ReactorCamelProcessor implements Closeable {
detach();
ReactiveStreamsBackpressureStrategy strategy = producer.getEndpoint().getBackpressureStrategy();
- Flux<Exchange> flux = Flux.create(camelSink::set);
+ Flux<Exchange> flux = Flux.create(camelSink::set, FluxSink.OverflowStrategy.IGNORE);
if (ObjectHelper.equal(strategy, ReactiveStreamsBackpressureStrategy.OLDEST)) {
// signal item emitted for non-dropped items only
http://git-wip-us.apache.org/repos/asf/camel/blob/2956fca2/parent/pom.xml
----------------------------------------------------------------------
diff --git a/parent/pom.xml b/parent/pom.xml
index 51e550c..30fbfa7 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -529,7 +529,7 @@
<quickfixj-version>1.6.3</quickfixj-version>
<rabbitmq-amqp-client-version>4.1.0</rabbitmq-amqp-client-version>
<reactive-streams-version>1.0.0</reactive-streams-version>
- <reactor-version>3.0.5.RELEASE</reactor-version>
+ <reactor-version>3.0.7.RELEASE</reactor-version>
<reflections-bundle-version>0.9.10_3</reflections-bundle-version>
<regexp-bundle-version>1.4_1</regexp-bundle-version>
<rest-assured-version>3.0.2</rest-assured-version>