You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by nf...@apache.org on 2017/05/15 10:49:28 UTC
camel git commit: CAMEL-11140: pulling ReactiveStreamsCamelSubscriber
out of internal engine
Repository: camel
Updated Branches:
refs/heads/master 4120c1fd1 -> 000e09a80
CAMEL-11140: pulling ReactiveStreamsCamelSubscriber out of internal engine
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/000e09a8
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/000e09a8
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/000e09a8
Branch: refs/heads/master
Commit: 000e09a80874cc6b3ee748504611d4bb45be3483
Parents: 4120c1f
Author: Nicola Ferraro <ni...@gmail.com>
Authored: Mon May 15 12:46:56 2017 +0200
Committer: Nicola Ferraro <ni...@gmail.com>
Committed: Mon May 15 12:46:56 2017 +0200
----------------------------------------------------------------------
.../streams/ReactiveStreamsCamelSubscriber.java | 209 ++++++++++++++++++
.../api/CamelReactiveStreamsService.java | 4 +-
.../streams/engine/CamelSubscriber.java | 210 -------------------
.../DefaultCamelReactiveStreamsService.java | 11 +-
.../support/ReactiveStreamsTestService.java | 4 +-
.../reactor/engine/ReactorStreamsService.java | 22 +-
.../engine/ReactorStreamsServiceFactory.java | 2 +-
.../ReactiveStreamsServiceTestSupport.java | 4 +-
8 files changed, 232 insertions(+), 234 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/000e09a8/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsCamelSubscriber.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsCamelSubscriber.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsCamelSubscriber.java
new file mode 100644
index 0000000..601bb5e
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsCamelSubscriber.java
@@ -0,0 +1,209 @@
+/**
+ * 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.reactive.streams;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.camel.Exchange;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The Camel subscriber. It bridges messages from reactive streams to Camel routes.
+ */
+public class ReactiveStreamsCamelSubscriber implements Subscriber<Exchange>, Closeable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ReactiveStreamsCamelSubscriber.class);
+
+ /**
+ * Unbounded as per rule #17. No need to refill.
+ */
+ private static final long UNBOUNDED_REQUESTS = Long.MAX_VALUE;
+
+ private ReactiveStreamsConsumer consumer;
+
+ private Subscription subscription;
+
+ private String name;
+
+ private long requested;
+
+ private long inflightCount;
+
+ public ReactiveStreamsCamelSubscriber(String name) {
+ this.name = name;
+ }
+
+ public void attachConsumer(ReactiveStreamsConsumer consumer) {
+ synchronized (this) {
+ if (this.consumer != null) {
+ throw new IllegalStateException("A consumer is already attached to the stream '" + name + "'");
+ }
+ this.consumer = consumer;
+ }
+ refill();
+ }
+
+ public synchronized ReactiveStreamsConsumer getConsumer() {
+ return consumer;
+ }
+
+ public void detachConsumer() {
+ synchronized (this) {
+ this.consumer = null;
+ }
+ }
+
+ @Override
+ public void onSubscribe(Subscription subscription) {
+ if (subscription == null) {
+ throw new NullPointerException("subscription is null for stream '" + name + "'");
+ }
+
+ boolean allowed = true;
+ synchronized (this) {
+ if (this.subscription != null) {
+ allowed = false;
+ } else {
+ this.subscription = subscription;
+ }
+ }
+
+ if (!allowed) {
+ LOG.warn("There is another active subscription: cancelled");
+ subscription.cancel();
+ } else {
+ refill();
+ }
+ }
+
+ @Override
+ public void onNext(Exchange exchange) {
+ if (exchange == null) {
+ throw new NullPointerException("exchange is null");
+ }
+
+ ReactiveStreamsConsumer target;
+ synchronized (this) {
+ if (requested < UNBOUNDED_REQUESTS) {
+ // When there are UNBOUNDED_REQUESTS, they remain constant
+ requested--;
+ }
+ target = this.consumer;
+ if (target != null) {
+ inflightCount++;
+ }
+ }
+
+ if (target != null) {
+ target.process(exchange, doneSync -> {
+ synchronized (this) {
+ inflightCount--;
+ }
+
+ refill();
+ });
+ } else {
+ // This may happen when the consumer is stopped
+ LOG.warn("Message received in stream '{}', but no consumers were attached. Discarding {}.", name, exchange);
+ }
+ }
+
+ protected void refill() {
+ Long toBeRequested = null;
+ Subscription subs = null;
+ synchronized (this) {
+ if (consumer != null && this.subscription != null) {
+ Integer consMax = consumer.getEndpoint().getMaxInflightExchanges();
+ long max = (consMax != null && consMax > 0) ? consMax.longValue() : UNBOUNDED_REQUESTS;
+ if (requested < UNBOUNDED_REQUESTS) {
+ long lowWatermark = Math.max(0, Math.round(consumer.getEndpoint().getExchangesRefillLowWatermark() * max));
+ long minRequests = Math.min(max, max - lowWatermark);
+ long newRequest = max - requested - inflightCount;
+ if (newRequest > 0 && newRequest >= minRequests) {
+ toBeRequested = newRequest;
+ requested += toBeRequested;
+ subs = this.subscription;
+ }
+ }
+ }
+ }
+
+ if (toBeRequested != null) {
+ subs.request(toBeRequested);
+ }
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ if (throwable == null) {
+ throw new NullPointerException("throwable is null");
+ }
+
+ LOG.error("Error in reactive stream '" + name + "'", throwable);
+
+ ReactiveStreamsConsumer consumer;
+ synchronized (this) {
+ consumer = this.consumer;
+ this.subscription = null;
+ }
+
+ if (consumer != null) {
+ consumer.onError(throwable);
+ }
+
+ }
+
+ @Override
+ public void onComplete() {
+ LOG.info("Reactive stream '{}' completed", name);
+
+ ReactiveStreamsConsumer consumer;
+ synchronized (this) {
+ consumer = this.consumer;
+ this.subscription = null;
+ }
+
+ if (consumer != null) {
+ consumer.onComplete();
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ Subscription subscription;
+ synchronized (this) {
+ subscription = this.subscription;
+ }
+
+ if (subscription != null) {
+ subscription.cancel();
+ }
+ }
+
+ public long getRequested() {
+ return requested;
+ }
+
+ public long getInflightCount() {
+ return inflightCount;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/000e09a8/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/CamelReactiveStreamsService.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/CamelReactiveStreamsService.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/CamelReactiveStreamsService.java
index 86879ac..199f96d 100644
--- a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/CamelReactiveStreamsService.java
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/CamelReactiveStreamsService.java
@@ -22,7 +22,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.Service;
import org.apache.camel.component.reactive.streams.ReactiveStreamsConsumer;
import org.apache.camel.component.reactive.streams.ReactiveStreamsProducer;
-import org.apache.camel.component.reactive.streams.engine.CamelSubscriber;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsCamelSubscriber;
import org.apache.camel.spi.HasId;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
@@ -298,7 +298,7 @@ public interface CamelReactiveStreamsService extends Service, HasId {
* @return the associated subscriber
* @throws IllegalStateException if another consumer is already associated with the given stream name
*/
- CamelSubscriber attachCamelConsumer(String name, ReactiveStreamsConsumer consumer);
+ ReactiveStreamsCamelSubscriber attachCamelConsumer(String name, ReactiveStreamsConsumer consumer);
/**
* Used by Camel to detach the existing consumer from the given stream.
http://git-wip-us.apache.org/repos/asf/camel/blob/000e09a8/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscriber.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscriber.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscriber.java
deleted file mode 100644
index 94a9c46..0000000
--- a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscriber.java
+++ /dev/null
@@ -1,210 +0,0 @@
-/**
- * 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.reactive.streams.engine;
-
-import java.io.Closeable;
-import java.io.IOException;
-
-import org.apache.camel.Exchange;
-import org.apache.camel.component.reactive.streams.ReactiveStreamsConsumer;
-import org.reactivestreams.Subscriber;
-import org.reactivestreams.Subscription;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * The Camel subscriber. It bridges messages from reactive streams to Camel routes.
- */
-public class CamelSubscriber implements Subscriber<Exchange>, Closeable {
-
- private static final Logger LOG = LoggerFactory.getLogger(CamelSubscriber.class);
-
- /**
- * Unbounded as per rule #17. No need to refill.
- */
- private static final long UNBOUNDED_REQUESTS = Long.MAX_VALUE;
-
- private ReactiveStreamsConsumer consumer;
-
- private Subscription subscription;
-
- private String name;
-
- private long requested;
-
- private long inflightCount;
-
- public CamelSubscriber(String name) {
- this.name = name;
- }
-
- public void attachConsumer(ReactiveStreamsConsumer consumer) {
- synchronized (this) {
- if (this.consumer != null) {
- throw new IllegalStateException("A consumer is already attached to the stream '" + name + "'");
- }
- this.consumer = consumer;
- }
- refill();
- }
-
- public synchronized ReactiveStreamsConsumer getConsumer() {
- return consumer;
- }
-
- public void detachConsumer() {
- synchronized (this) {
- this.consumer = null;
- }
- }
-
- @Override
- public void onSubscribe(Subscription subscription) {
- if (subscription == null) {
- throw new NullPointerException("subscription is null for stream '" + name + "'");
- }
-
- boolean allowed = true;
- synchronized (this) {
- if (this.subscription != null) {
- allowed = false;
- } else {
- this.subscription = subscription;
- }
- }
-
- if (!allowed) {
- LOG.warn("There is another active subscription: cancelled");
- subscription.cancel();
- } else {
- refill();
- }
- }
-
- @Override
- public void onNext(Exchange exchange) {
- if (exchange == null) {
- throw new NullPointerException("exchange is null");
- }
-
- ReactiveStreamsConsumer target;
- synchronized (this) {
- if (requested < UNBOUNDED_REQUESTS) {
- // When there are UNBOUNDED_REQUESTS, they remain constant
- requested--;
- }
- target = this.consumer;
- if (target != null) {
- inflightCount++;
- }
- }
-
- if (target != null) {
- target.process(exchange, doneSync -> {
- synchronized (this) {
- inflightCount--;
- }
-
- refill();
- });
- } else {
- // This may happen when the consumer is stopped
- LOG.warn("Message received in stream '{}', but no consumers were attached. Discarding {}.", name, exchange);
- }
- }
-
- protected void refill() {
- Long toBeRequested = null;
- Subscription subs = null;
- synchronized (this) {
- if (consumer != null && this.subscription != null) {
- Integer consMax = consumer.getEndpoint().getMaxInflightExchanges();
- long max = (consMax != null && consMax > 0) ? consMax.longValue() : UNBOUNDED_REQUESTS;
- if (requested < UNBOUNDED_REQUESTS) {
- long lowWatermark = Math.max(0, Math.round(consumer.getEndpoint().getExchangesRefillLowWatermark() * max));
- long minRequests = Math.min(max, max - lowWatermark);
- long newRequest = max - requested - inflightCount;
- if (newRequest > 0 && newRequest >= minRequests) {
- toBeRequested = newRequest;
- requested += toBeRequested;
- subs = this.subscription;
- }
- }
- }
- }
-
- if (toBeRequested != null) {
- subs.request(toBeRequested);
- }
- }
-
- @Override
- public void onError(Throwable throwable) {
- if (throwable == null) {
- throw new NullPointerException("throwable is null");
- }
-
- LOG.error("Error in reactive stream '" + name + "'", throwable);
-
- ReactiveStreamsConsumer consumer;
- synchronized (this) {
- consumer = this.consumer;
- this.subscription = null;
- }
-
- if (consumer != null) {
- consumer.onError(throwable);
- }
-
- }
-
- @Override
- public void onComplete() {
- LOG.info("Reactive stream '{}' completed", name);
-
- ReactiveStreamsConsumer consumer;
- synchronized (this) {
- consumer = this.consumer;
- this.subscription = null;
- }
-
- if (consumer != null) {
- consumer.onComplete();
- }
- }
-
- @Override
- public void close() throws IOException {
- Subscription subscription;
- synchronized (this) {
- subscription = this.subscription;
- }
-
- if (subscription != null) {
- subscription.cancel();
- }
- }
-
- public long getRequested() {
- return requested;
- }
-
- public long getInflightCount() {
- return inflightCount;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/000e09a8/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/DefaultCamelReactiveStreamsService.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/DefaultCamelReactiveStreamsService.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/DefaultCamelReactiveStreamsService.java
index 06650d8..e8a599b 100644
--- a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/DefaultCamelReactiveStreamsService.java
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/DefaultCamelReactiveStreamsService.java
@@ -37,6 +37,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.api.management.ManagedOperation;
import org.apache.camel.api.management.ManagedResource;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsCamelSubscriber;
import org.apache.camel.component.reactive.streams.ReactiveStreamsConstants;
import org.apache.camel.component.reactive.streams.ReactiveStreamsConsumer;
import org.apache.camel.component.reactive.streams.ReactiveStreamsHelper;
@@ -66,7 +67,7 @@ public class DefaultCamelReactiveStreamsService extends ServiceSupport implement
private ExecutorService workerPool;
private final ConcurrentMap<String, CamelPublisher> publishers = new ConcurrentHashMap<>();
- private final ConcurrentMap<String, CamelSubscriber> subscribers = new ConcurrentHashMap<>();
+ private final ConcurrentMap<String, ReactiveStreamsCamelSubscriber> subscribers = new ConcurrentHashMap<>();
private final ConcurrentMap<String, String> publishedUriToStream = new ConcurrentHashMap<>();
private final ConcurrentMap<String, String> requestedUriToStream = new ConcurrentHashMap<>();
@@ -114,8 +115,8 @@ public class DefaultCamelReactiveStreamsService extends ServiceSupport implement
}
@Override
- public CamelSubscriber streamSubscriber(String name) {
- return subscribers.computeIfAbsent(name, n -> new CamelSubscriber(name));
+ public ReactiveStreamsCamelSubscriber streamSubscriber(String name) {
+ return subscribers.computeIfAbsent(name, n -> new ReactiveStreamsCamelSubscriber(name));
}
@SuppressWarnings("unchecked")
@@ -299,8 +300,8 @@ public class DefaultCamelReactiveStreamsService extends ServiceSupport implement
}
@Override
- public CamelSubscriber attachCamelConsumer(String name, ReactiveStreamsConsumer consumer) {
- CamelSubscriber subscriber = streamSubscriber(name);
+ public ReactiveStreamsCamelSubscriber attachCamelConsumer(String name, ReactiveStreamsConsumer consumer) {
+ ReactiveStreamsCamelSubscriber subscriber = streamSubscriber(name);
subscriber.attachConsumer(consumer);
return subscriber;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/000e09a8/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/ReactiveStreamsTestService.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/ReactiveStreamsTestService.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/ReactiveStreamsTestService.java
index f0e7bf1..6065795 100644
--- a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/ReactiveStreamsTestService.java
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/ReactiveStreamsTestService.java
@@ -22,7 +22,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.component.reactive.streams.ReactiveStreamsConsumer;
import org.apache.camel.component.reactive.streams.ReactiveStreamsProducer;
import org.apache.camel.component.reactive.streams.api.CamelReactiveStreamsService;
-import org.apache.camel.component.reactive.streams.engine.CamelSubscriber;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsCamelSubscriber;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
@@ -73,7 +73,7 @@ public class ReactiveStreamsTestService implements CamelReactiveStreamsService {
}
@Override
- public CamelSubscriber attachCamelConsumer(String name, ReactiveStreamsConsumer consumer) {
+ public ReactiveStreamsCamelSubscriber attachCamelConsumer(String name, ReactiveStreamsConsumer consumer) {
return null;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/000e09a8/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorStreamsService.java
----------------------------------------------------------------------
diff --git a/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorStreamsService.java b/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorStreamsService.java
index 9279dab..2c9dfe3 100644
--- a/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorStreamsService.java
+++ b/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorStreamsService.java
@@ -24,12 +24,11 @@ import java.util.function.Supplier;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsCamelSubscriber;
import org.apache.camel.component.reactive.streams.ReactiveStreamsConsumer;
import org.apache.camel.component.reactive.streams.ReactiveStreamsHelper;
import org.apache.camel.component.reactive.streams.ReactiveStreamsProducer;
import org.apache.camel.component.reactive.streams.api.CamelReactiveStreamsService;
-import org.apache.camel.component.reactive.streams.engine.CamelSubscriber;
-import org.apache.camel.component.reactive.streams.engine.ReactiveStreamsEngineConfiguration;
import org.apache.camel.component.reactive.streams.util.BodyConverter;
import org.apache.camel.component.reactive.streams.util.ConvertingPublisher;
import org.apache.camel.component.reactive.streams.util.ConvertingSubscriber;
@@ -39,21 +38,20 @@ import org.apache.camel.support.ServiceSupport;
import org.apache.camel.util.function.Suppliers;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
+
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
final class ReactorStreamsService extends ServiceSupport implements CamelReactiveStreamsService {
private final CamelContext context;
- private final ReactiveStreamsEngineConfiguration configuration;
private final Supplier<UnwrapStreamProcessor> unwrapStreamProcessorSupplier;
private final ConcurrentMap<String, ReactorCamelProcessor> publishers;
- private final ConcurrentMap<String, CamelSubscriber> subscribers;
+ private final ConcurrentMap<String, ReactiveStreamsCamelSubscriber> subscribers;
private final ConcurrentMap<String, String> publishedUriToStream;
private final ConcurrentMap<String, String> requestedUriToStream;
- ReactorStreamsService(CamelContext context, ReactiveStreamsEngineConfiguration configuration) {
+ ReactorStreamsService(CamelContext context) {
this.context = context;
- this.configuration = configuration;
this.publishers = new ConcurrentHashMap<>();
this.subscribers = new ConcurrentHashMap<>();
this.publishedUriToStream = new ConcurrentHashMap<>();
@@ -79,7 +77,7 @@ final class ReactorStreamsService extends ServiceSupport implements CamelReactiv
for (ReactorCamelProcessor processor : publishers.values()) {
processor.close();
}
- for (CamelSubscriber subscriber : subscribers.values()) {
+ for (ReactiveStreamsCamelSubscriber subscriber : subscribers.values()) {
subscriber.close();
}
}
@@ -105,8 +103,8 @@ final class ReactorStreamsService extends ServiceSupport implements CamelReactiv
}
@Override
- public CamelSubscriber streamSubscriber(String name) {
- return subscribers.computeIfAbsent(name, n -> new CamelSubscriber(name));
+ public ReactiveStreamsCamelSubscriber streamSubscriber(String name) {
+ return subscribers.computeIfAbsent(name, n -> new ReactiveStreamsCamelSubscriber(name));
}
@SuppressWarnings("unchecked")
@@ -296,8 +294,8 @@ final class ReactorStreamsService extends ServiceSupport implements CamelReactiv
// ******************************************
@Override
- public CamelSubscriber attachCamelConsumer(String name, ReactiveStreamsConsumer consumer) {
- CamelSubscriber subscriber = streamSubscriber(name);
+ public ReactiveStreamsCamelSubscriber attachCamelConsumer(String name, ReactiveStreamsConsumer consumer) {
+ ReactiveStreamsCamelSubscriber subscriber = streamSubscriber(name);
subscriber.attachConsumer(consumer);
return subscriber;
@@ -305,7 +303,7 @@ final class ReactorStreamsService extends ServiceSupport implements CamelReactiv
@Override
public void detachCamelConsumer(String name) {
- CamelSubscriber subscriber = streamSubscriber(name);
+ ReactiveStreamsCamelSubscriber subscriber = streamSubscriber(name);
subscriber.detachConsumer();
}
http://git-wip-us.apache.org/repos/asf/camel/blob/000e09a8/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorStreamsServiceFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorStreamsServiceFactory.java b/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorStreamsServiceFactory.java
index ae963d9..3c6123a 100644
--- a/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorStreamsServiceFactory.java
+++ b/components/camel-reactor/src/main/java/org/apache/camel/component/reactor/engine/ReactorStreamsServiceFactory.java
@@ -24,6 +24,6 @@ import org.apache.camel.component.reactive.streams.engine.ReactiveStreamsEngineC
public class ReactorStreamsServiceFactory implements CamelReactiveStreamsServiceFactory {
@Override
public CamelReactiveStreamsService newInstance(CamelContext context, ReactiveStreamsEngineConfiguration configuration) {
- return new ReactorStreamsService(context, configuration);
+ return new ReactorStreamsService(context);
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/000e09a8/platforms/spring-boot/components-starter/camel-reactive-streams-starter/src/test/java/org/apache/camel/component/reactive/streams/springboot/test/support/ReactiveStreamsServiceTestSupport.java
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-reactive-streams-starter/src/test/java/org/apache/camel/component/reactive/streams/springboot/test/support/ReactiveStreamsServiceTestSupport.java b/platforms/spring-boot/components-starter/camel-reactive-streams-starter/src/test/java/org/apache/camel/component/reactive/streams/springboot/test/support/ReactiveStreamsServiceTestSupport.java
index fe679b5..8290919 100644
--- a/platforms/spring-boot/components-starter/camel-reactive-streams-starter/src/test/java/org/apache/camel/component/reactive/streams/springboot/test/support/ReactiveStreamsServiceTestSupport.java
+++ b/platforms/spring-boot/components-starter/camel-reactive-streams-starter/src/test/java/org/apache/camel/component/reactive/streams/springboot/test/support/ReactiveStreamsServiceTestSupport.java
@@ -22,7 +22,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.component.reactive.streams.ReactiveStreamsConsumer;
import org.apache.camel.component.reactive.streams.ReactiveStreamsProducer;
import org.apache.camel.component.reactive.streams.api.CamelReactiveStreamsService;
-import org.apache.camel.component.reactive.streams.engine.CamelSubscriber;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsCamelSubscriber;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
@@ -140,7 +140,7 @@ public class ReactiveStreamsServiceTestSupport implements CamelReactiveStreamsSe
}
@Override
- public CamelSubscriber attachCamelConsumer(String s, ReactiveStreamsConsumer consumer) {
+ public ReactiveStreamsCamelSubscriber attachCamelConsumer(String s, ReactiveStreamsConsumer consumer) {
return null;
}