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/01/27 14:35:05 UTC
[6/6] camel git commit: CAMEL-10612: Reactive streams
CAMEL-10612: Reactive streams
CAMEL-10612: First complete implementation
CAMEL-10612: Implemented subscriber verification and rx implementation
CAMEL-10612: Support switching to a different implementation
CAMEL-10612: Added internal engine configuration
CAMEL-10612: Make sure backpressure can be combined with throttling route policy
CAMEL-10612: more consistent structure
CAMEL-10612: fixes and test updates
CAMEL-10612: enhanced publisher backpressure with mixed slow and fast subscribers
CAMEL-10612: fixed issue with route restart
CAMEL-10612: fixed stream completion
CAMEL-10650: added backpressure strategy
CAMEL-10650: added javadoc
CAMEL-10650: rebase with master
CAMEL-10650: source check
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/e66207a9
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/e66207a9
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/e66207a9
Branch: refs/heads/master
Commit: e66207a9185df54ac7e67de027dc68f85f1ffdce
Parents: 353eeef
Author: Nicola Ferraro <ni...@gmail.com>
Authored: Wed Dec 21 18:59:01 2016 +0100
Committer: Nicola Ferraro <ni...@gmail.com>
Committed: Fri Jan 27 15:31:31 2017 +0100
----------------------------------------------------------------------
components/camel-reactive-streams/pom.xml | 96 +++++++
.../ReactiveStreamsBackpressureStrategy.java | 40 +++
.../streams/ReactiveStreamsComponent.java | 69 +++++
.../streams/ReactiveStreamsConsumer.java | 92 +++++++
.../streams/ReactiveStreamsEndpoint.java | 97 +++++++
.../streams/ReactiveStreamsProducer.java | 59 +++++
.../streams/api/CamelReactiveStreams.java | 166 ++++++++++++
.../api/CamelReactiveStreamsService.java | 114 +++++++++
.../reactive/streams/api/DispatchCallback.java | 34 +++
.../reactive/streams/engine/CamelPublisher.java | 105 ++++++++
.../engine/CamelReactiveStreamsServiceImpl.java | 133 ++++++++++
.../streams/engine/CamelSubscriber.java | 177 +++++++++++++
.../streams/engine/CamelSubscription.java | 252 +++++++++++++++++++
.../ReactiveStreamsEngineConfiguration.java | 65 +++++
.../reactive/streams/engine/StreamPayload.java | 52 ++++
.../streams/engine/UnwrappingPublisher.java | 87 +++++++
.../streams/util/ConvertingPublisher.java | 113 +++++++++
.../streams/util/ConvertingSubscriber.java | 75 ++++++
.../src/main/resources/META-INF/LICENSE.txt | 203 +++++++++++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 +
.../org/apache/camel/component/reactive-streams | 18 ++
.../BackpressurePublisherRoutePolicyTest.java | 165 ++++++++++++
.../streams/BackpressureStrategyTest.java | 164 ++++++++++++
.../streams/BackpressureSubscriberTest.java | 107 ++++++++
.../reactive/streams/BasicPublisherTest.java | 112 +++++++++
.../reactive/streams/BasicSubscriberTest.java | 85 +++++++
...amelPublisherConversionVerificationTest.java | 68 +++++
.../streams/CamelPublisherVerificationTest.java | 68 +++++
...melSubscriberConversionVerificationTest.java | 62 +++++
.../CamelSubscriberVerificationTest.java | 66 +++++
.../streams/ConcurrentConsumersTest.java | 103 ++++++++
.../streams/PublisherTypeConversionTest.java | 88 +++++++
.../reactive/streams/support/TestPublisher.java | 110 ++++++++
.../streams/support/TestSubscriber.java | 73 ++++++
.../src/test/resources/log4j2.properties | 33 +++
components/pom.xml | 1 +
components/readme.adoc | 3 +
docs/user-manual/en/SUMMARY.md | 1 +
parent/pom.xml | 1 +
.../camel-reactive-streams-starter/pom.xml | 51 ++++
...activeStreamsComponentAutoConfiguration.java | 112 +++++++++
.../ReactiveStreamsComponentConfiguration.java | 100 ++++++++
.../src/main/resources/META-INF/LICENSE.txt | 203 +++++++++++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 +
...dditional-spring-configuration-metadata.json | 10 +
.../main/resources/META-INF/spring.factories | 19 ++
.../src/main/resources/META-INF/spring.provides | 18 ++
.../spring-boot/components-starter/pom.xml | 1 +
48 files changed, 3893 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/pom.xml b/components/camel-reactive-streams/pom.xml
new file mode 100644
index 0000000..2357e70
--- /dev/null
+++ b/components/camel-reactive-streams/pom.xml
@@ -0,0 +1,96 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components</artifactId>
+ <version>2.19.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-reactive-streams</artifactId>
+ <packaging>jar</packaging>
+ <name>Camel :: Reactive Streams</name>
+ <description>Camel Reactive Streams Component</description>
+
+ <properties>
+ <camel.osgi.export.pkg>org.apache.camel.component.reactive.streams.*</camel.osgi.export.pkg>
+ </properties>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.reactivestreams</groupId>
+ <artifactId>reactive-streams</artifactId>
+ <version>1.0.0</version>
+ </dependency>
+
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>org.reactivestreams</groupId>
+ <artifactId>reactive-streams-tck</artifactId>
+ <version>1.0.0</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.reactivex.rxjava2</groupId>
+ <artifactId>rxjava</artifactId>
+ <version>${rxjava2-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <version>${mockito-version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsBackpressureStrategy.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsBackpressureStrategy.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsBackpressureStrategy.java
new file mode 100644
index 0000000..823a7b8
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsBackpressureStrategy.java
@@ -0,0 +1,40 @@
+/**
+ * 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;
+
+/**
+ * A list of possible backpressure strategy to use when the emission of upstream items cannot respect backpressure.
+ */
+public enum ReactiveStreamsBackpressureStrategy {
+
+ /**
+ * Buffers <em>all</em> onNext values until the downstream consumes it.
+ */
+ BUFFER,
+
+ /**
+ * Drops the most recent onNext value if the downstream can't keep up.
+ */
+ DROP,
+
+ /**
+ * Keeps only the latest onNext value, overwriting any previous value if the
+ * downstream can't keep up.
+ */
+ LATEST
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsComponent.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsComponent.java
new file mode 100644
index 0000000..5801877
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsComponent.java
@@ -0,0 +1,69 @@
+/**
+ * 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.util.Map;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.reactive.streams.engine.ReactiveStreamsEngineConfiguration;
+import org.apache.camel.impl.UriEndpointComponent;
+
+/**
+ * The Camel reactive-streams component.
+ */
+public class ReactiveStreamsComponent extends UriEndpointComponent {
+
+ private ReactiveStreamsEngineConfiguration internalEngineConfiguration = new ReactiveStreamsEngineConfiguration();
+
+ private ReactiveStreamsBackpressureStrategy backpressureStrategy = ReactiveStreamsBackpressureStrategy.BUFFER;
+
+ public ReactiveStreamsComponent() {
+ super(ReactiveStreamsEndpoint.class);
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ ReactiveStreamsEndpoint endpoint = new ReactiveStreamsEndpoint(uri, this);
+ endpoint.setStream(remaining);
+ setProperties(endpoint, parameters);
+
+ return endpoint;
+ }
+
+ public ReactiveStreamsEngineConfiguration getInternalEngineConfiguration() {
+ return internalEngineConfiguration;
+ }
+
+ /**
+ * Configures the internal engine for Reactive Streams.
+ */
+ public void setInternalEngineConfiguration(ReactiveStreamsEngineConfiguration internalEngineConfiguration) {
+ this.internalEngineConfiguration = internalEngineConfiguration;
+ }
+
+ public ReactiveStreamsBackpressureStrategy getBackpressureStrategy() {
+ return backpressureStrategy;
+ }
+
+ /**
+ * The backpressure strategy to use when pushing events to a slow subscriber.
+ */
+ public void setBackpressureStrategy(ReactiveStreamsBackpressureStrategy backpressureStrategy) {
+ this.backpressureStrategy = backpressureStrategy;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java
new file mode 100644
index 0000000..c345a2b
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java
@@ -0,0 +1,92 @@
+/**
+ * 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.util.concurrent.ExecutorService;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.impl.DefaultConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The Camel reactive-streams consumer.
+ */
+public class ReactiveStreamsConsumer extends DefaultConsumer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ReactiveStreamsConsumer.class);
+
+ private ReactiveStreamsEndpoint endpoint;
+
+ private ExecutorService executor;
+
+ public ReactiveStreamsConsumer(ReactiveStreamsEndpoint endpoint, Processor processor) {
+ super(endpoint, processor);
+ this.endpoint = endpoint;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ int poolSize = endpoint.getConcurrentConsumers();
+ if (executor == null) {
+ executor = getEndpoint().getCamelContext().getExecutorServiceManager().newFixedThreadPool(this, getEndpoint().getEndpointUri(), poolSize);
+ }
+
+ CamelReactiveStreams.get(endpoint.getCamelContext()).getService().attachConsumer(endpoint.getStream(), this);
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+ CamelReactiveStreams.get(endpoint.getCamelContext()).getService().detachConsumer(endpoint.getStream());
+
+ if (executor != null) {
+ endpoint.getCamelContext().getExecutorServiceManager().shutdownNow(executor);
+ executor = null;
+ }
+ }
+
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ ExecutorService executorService = this.executor;
+ if (executorService != null && this.isRunAllowed()) {
+
+ executorService.execute(() -> this.getAsyncProcessor().process(exchange, doneSync -> {
+ if (exchange.getException() != null) {
+ LOG.warn("Error processing the exchange " + exchange + " from reactive-streams", exchange.getException());
+ }
+
+ callback.done(doneSync);
+ }));
+ return false;
+
+ } else {
+ LOG.warn("Consumer not ready to process exchanges. The exchange {} will be discarded", exchange);
+ callback.done(true);
+ return true;
+ }
+ }
+
+ @Override
+ public ReactiveStreamsEndpoint getEndpoint() {
+ return endpoint;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsEndpoint.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsEndpoint.java
new file mode 100644
index 0000000..7d294f1
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsEndpoint.java
@@ -0,0 +1,97 @@
+/**
+ * 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 org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriPath;
+
+/**
+ * The Camel reactive-streams endpoint.
+ */
+@UriEndpoint(scheme = "reactive-streams", title = "Reactive Streams", syntax = "reactive-streams:/stream",
+ consumerClass = ReactiveStreamsConsumer.class, label = "streams")
+public class ReactiveStreamsEndpoint extends DefaultEndpoint {
+
+ @UriPath
+ private String stream;
+
+ @UriParam(label = "consumer", defaultValue = "128")
+ private Integer maxInflightExchanges = 128;
+
+ @UriParam(label = "consumer", defaultValue = "1")
+ private int concurrentConsumers = 1;
+
+ public ReactiveStreamsEndpoint(String endpointUri, ReactiveStreamsComponent component) {
+ super(endpointUri, component);
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return true;
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new ReactiveStreamsProducer(this, stream);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ return new ReactiveStreamsConsumer(this, processor);
+ }
+
+ public String getStream() {
+ return stream;
+ }
+
+ /**
+ * Name of the stream channel used by the endpoint to exchange messages.
+ */
+ public void setStream(String stream) {
+ this.stream = stream;
+ }
+
+ public Integer getMaxInflightExchanges() {
+ return maxInflightExchanges;
+ }
+
+ /**
+ * Maximum number of exchanges concurrently being processed by Camel.
+ * This parameter controls backpressure on the stream.
+ * Setting a non-positive value will disable backpressure.
+ */
+ public void setMaxInflightExchanges(Integer maxInflightExchanges) {
+ this.maxInflightExchanges = maxInflightExchanges;
+ }
+
+ public int getConcurrentConsumers() {
+ return concurrentConsumers;
+ }
+
+ /**
+ * Number of threads used to process exchanges in the Camel route.
+ */
+ public void setConcurrentConsumers(int concurrentConsumers) {
+ this.concurrentConsumers = concurrentConsumers;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsProducer.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsProducer.java
new file mode 100644
index 0000000..a8ec559
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsProducer.java
@@ -0,0 +1,59 @@
+/**
+ * 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 org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreamsService;
+import org.apache.camel.impl.DefaultAsyncProducer;
+
+/**
+ * The Camel reactive-streams producer.
+ */
+public class ReactiveStreamsProducer<T> extends DefaultAsyncProducer {
+
+ private String name;
+
+ private CamelReactiveStreamsService service;
+
+ public ReactiveStreamsProducer(ReactiveStreamsEndpoint endpoint, String name) {
+ super(endpoint);
+ this.name = name;
+ }
+
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ service.process(name, exchange, (data, error) -> {
+ if (error != null) {
+ data.setException(error);
+ }
+
+ if (callback != null) {
+ callback.done(false);
+ }
+ });
+ return false;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ this.service = CamelReactiveStreams.get(getEndpoint().getCamelContext()).getService();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/CamelReactiveStreams.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/CamelReactiveStreams.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/CamelReactiveStreams.java
new file mode 100644
index 0000000..dfa8185
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/CamelReactiveStreams.java
@@ -0,0 +1,166 @@
+/**
+ * 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.api;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.reactive.streams.engine.CamelReactiveStreamsServiceImpl;
+import org.apache.camel.spi.FactoryFinder;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This is the main entry-point for getting Camel streams associate to reactive-streams endpoints.
+ *
+ * It delegates main methods to an instance of {@link CamelReactiveStreamsService}. This component provides
+ * a default implementation that can be overridden in a 'META-INF/services/reactive-streams/reactiveStreamsService' file.
+ */
+public final class CamelReactiveStreams {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CamelReactiveStreams.class);
+
+ private static Map<CamelContext, CamelReactiveStreams> instances = new HashMap<>();
+
+ private CamelReactiveStreamsService service;
+
+ private CamelReactiveStreams(CamelReactiveStreamsService service) {
+ this.service = service;
+ }
+
+ public static synchronized CamelReactiveStreams get(CamelContext context) {
+ if (!instances.containsKey(context)) {
+ CamelReactiveStreamsService service = resolveReactiveStreamsService(context);
+ try {
+ context.addService(service, true, true);
+ } catch (Exception ex) {
+ throw new IllegalStateException("Cannot add the CamelReactiveStreamsService to the Camel context", ex);
+ }
+
+ instances.put(context, new CamelReactiveStreams(service));
+ }
+
+ return instances.get(context);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static CamelReactiveStreamsService resolveReactiveStreamsService(CamelContext context) {
+ Class<? extends CamelReactiveStreamsService> serviceClass = null;
+ try {
+ FactoryFinder finder = context.getFactoryFinder("META-INF/services/reactive-streams/");
+ LOG.trace("Using FactoryFinder: {}", finder);
+ serviceClass = (Class<? extends CamelReactiveStreamsService>) finder.findClass("reactiveStreamsService");
+ } catch (ClassNotFoundException e) {
+ LOG.trace("'reactive.streams.service.class' not found", e);
+ } catch (IOException e) {
+ LOG.trace("No reactive stream service defined in 'META-INF/services/org/apache/camel/component/'", e);
+ }
+
+ CamelReactiveStreamsService service = null;
+ if (serviceClass != null) {
+ try {
+ service = serviceClass.newInstance();
+ LOG.info("Created reactive stream service from class: " + serviceClass.getName());
+ } catch (InstantiationException | IllegalAccessException e) {
+ LOG.debug("Unable to create a reactive stream service of class " + serviceClass.getName(), e);
+ }
+ }
+
+ if (service == null) {
+ LOG.info("Using default reactive stream service");
+ service = new CamelReactiveStreamsServiceImpl();
+ }
+
+ return service;
+ }
+
+ /**
+ * Allows retrieving the service responsible for binding camel routes to streams.
+ *
+ * @return the stream service
+ */
+ public CamelReactiveStreamsService getService() {
+ return service;
+ }
+
+ /**
+ * Returns the publisher associated to the given stream name.
+ * A publisher can be used to push Camel exchanges to reactive-streams subscribers.
+ *
+ * @param name the stream name
+ * @return the stream publisher
+ */
+ public Publisher<Exchange> getPublisher(String name) {
+ Objects.requireNonNull(name, "name cannot be null");
+
+ return service.getPublisher(name);
+ }
+
+ /**
+ * Returns the publisher associated to the given stream name.
+ * A publisher can be used to push Camel exchange to external reactive-streams subscribers.
+ *
+ * The publisher converts automatically exchanges to the given type.
+ *
+ * @param name the stream name
+ * @param type the type of the emitted items
+ * @param <T> the type of items emitted by the publisher
+ * @return the publisher associated to the stream
+ */
+ public <T> Publisher<T> getPublisher(String name, Class<T> type) {
+ Objects.requireNonNull(name, "name cannot be null");
+
+ return service.getPublisher(name, type);
+ }
+
+ /**
+ * Returns the subscriber associated to the given stream name.
+ * A subscriber can be used to push items coming from external reactive-streams publishers to Camel routes.
+ *
+ * @param name the stream name
+ * @return the subscriber associated with the stream
+ */
+ public Subscriber<Exchange> getSubscriber(String name) {
+ Objects.requireNonNull(name, "name cannot be null");
+
+ return service.getSubscriber(name);
+ }
+
+ /**
+ * Returns the subscriber associated to the given stream name.
+ * A subscriber can be used to push items coming from external reactive-streams publishers to Camel routes.
+ *
+ * The subscriber converts automatically items of the given type to exchanges before pushing them.
+ *
+ * @param name the stream name
+ * @param type the publisher converts automatically exchanges to the given type.
+ * @param <T> the type of items accepted by the subscriber
+ * @return the subscriber associated with the stream
+ */
+ public <T> Subscriber<T> getSubscriber(String name, Class<T> type) {
+ Objects.requireNonNull(name, "name cannot be null");
+
+ return service.getSubscriber(name, type);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/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
new file mode 100644
index 0000000..147ece7
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/CamelReactiveStreamsService.java
@@ -0,0 +1,114 @@
+/**
+ * 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.api;
+
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.Exchange;
+import org.apache.camel.Service;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsConsumer;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+
+/**
+ * The interface to which any implementation of the reactive-streams engine should comply.
+ */
+public interface CamelReactiveStreamsService extends CamelContextAware, Service {
+
+ /*
+ * Main API methods.
+ */
+
+ /**
+ * Returns the publisher associated to the given stream name.
+ * A publisher can be used to push Camel exchanges to reactive-streams subscribers.
+ *
+ * @param name the stream name
+ * @return the stream publisher
+ */
+ Publisher<Exchange> getPublisher(String name);
+
+ /**
+ * Returns the publisher associated to the given stream name.
+ * A publisher can be used to push Camel exchange to external reactive-streams subscribers.
+ *
+ * The publisher converts automatically exchanges to the given type.
+ *
+ * @param name the stream name
+ * @param type the type of the emitted items
+ * @param <T> the type of items emitted by the publisher
+ * @return the publisher associated to the stream
+ */
+ <T> Publisher<T> getPublisher(String name, Class<T> type);
+
+ /**
+ * Returns the subscriber associated to the given stream name.
+ * A subscriber can be used to push items coming from external reactive-streams publishers to Camel routes.
+ *
+ * @param name the stream name
+ * @return the subscriber associated with the stream
+ */
+ Subscriber<Exchange> getSubscriber(String name);
+
+ /**
+ * Returns the subscriber associated to the given stream name.
+ * A subscriber can be used to push items coming from external reactive-streams publishers to Camel routes.
+ *
+ * The subscriber converts automatically items of the given type to exchanges before pushing them.
+ *
+ * @param name the stream name
+ * @param type the publisher converts automatically exchanges to the given type.
+ * @param <T> the type of items accepted by the subscriber
+ * @return the subscriber associated with the stream
+ */
+ <T> Subscriber<T> getSubscriber(String name, Class<T> type);
+
+ /*
+ * Methods for producers.
+ */
+
+ /**
+ * Sends the exchange to all active subscriptions on the given stream.
+ * The callback is used to signal that the exchange has been delivered to the subscribers.
+ *
+ * @param name the stream name
+ * @param exchange the exchange to be forwarded to the external subscribers
+ * @param callback the callback that signals the delivery of the exchange
+ */
+ void process(String name, Exchange exchange, DispatchCallback<Exchange> callback);
+
+ /*
+ * Methods for consumers.
+ */
+
+ /**
+ * Associate the subscriber of the stream with the given name to a specific Camel consumer.
+ * This method is used to bind a Camel route to a reactive stream.
+ *
+ * @param name the stream name
+ * @param consumer the consumer of the route
+ * @throws IllegalStateException if another consumer is already associated with the given stream name
+ */
+ void attachConsumer(String name, ReactiveStreamsConsumer consumer);
+
+ /**
+ * Deassociate the existing consumer from the given stream.
+ *
+ * @param name the stream name
+ */
+ void detachConsumer(String name);
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/DispatchCallback.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/DispatchCallback.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/DispatchCallback.java
new file mode 100644
index 0000000..c07f3d3
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/api/DispatchCallback.java
@@ -0,0 +1,34 @@
+/**
+ * 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.api;
+
+/**
+ * A callback used to signal when a item coming from a Camel route has been delivered to the external stream processor.
+ */
+public interface DispatchCallback<T> {
+
+ /**
+ * Signals the delivery of the item.
+ * If the item cannot be delivered (no subscribers registered, conversion error)
+ * the related {@link Throwable} is specified as parameter.
+ *
+ * @param data the item
+ * @param error the error occurred, if any
+ */
+ void processed(T data, Throwable error);
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelPublisher.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelPublisher.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelPublisher.java
new file mode 100644
index 0000000..bf675fb
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelPublisher.java
@@ -0,0 +1,105 @@
+/**
+ * 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.util.LinkedList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsBackpressureStrategy;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsComponent;
+import org.apache.camel.component.reactive.streams.api.DispatchCallback;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The Camel publisher. It forwards Camel exchanges to external reactive-streams subscribers.
+ */
+public class CamelPublisher implements Publisher<StreamPayload<Exchange>>, AutoCloseable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CamelPublisher.class);
+
+ private ExecutorService workerPool;
+
+ private String name;
+
+ private ReactiveStreamsBackpressureStrategy backpressureStrategy;
+
+ private List<CamelSubscription> subscriptions = new CopyOnWriteArrayList<>();
+
+ public CamelPublisher(ExecutorService workerPool, CamelContext context, String name) {
+ this.workerPool = workerPool;
+ this.backpressureStrategy = ((ReactiveStreamsComponent) context.getComponent("reactive-streams")).getBackpressureStrategy();
+ this.name = name;
+ }
+
+ @Override
+ public void subscribe(Subscriber<? super StreamPayload<Exchange>> subscriber) {
+ Objects.requireNonNull(subscriber, "subscriber must not be null");
+ CamelSubscription sub = new CamelSubscription(workerPool, this, this.backpressureStrategy, subscriber);
+ this.subscriptions.add(sub);
+ subscriber.onSubscribe(sub);
+ }
+
+ public void unsubscribe(CamelSubscription subscription) {
+ subscriptions.remove(subscription);
+ }
+
+ public void publish(StreamPayload<Exchange> data) {
+ // freeze the subscriptions
+ List<CamelSubscription> subs = new LinkedList<>(subscriptions);
+ DispatchCallback<Exchange> originalCallback = data.getCallback();
+ if (originalCallback != null && subs.size() > 0) {
+ // Notify processing once if multiple subscribers are present
+ AtomicInteger counter = new AtomicInteger(0);
+ AtomicReference<Throwable> thrown = new AtomicReference<>(null);
+ data = new StreamPayload<>(data.getItem(), (ex, error) -> {
+ int status = counter.incrementAndGet();
+ thrown.compareAndSet(null, error);
+ if (status == subs.size()) {
+ originalCallback.processed(ex, thrown.get());
+ }
+ });
+ }
+
+ if (subs.size() > 0) {
+ LOG.debug("Exchange published to {} subscriptions for the stream {}: {}", subs.size(), name, data.getItem());
+ // at least one subscriber
+ for (CamelSubscription sub : subs) {
+ sub.publish(data);
+ }
+ } else {
+ data.getCallback().processed(data.getItem(), new IllegalStateException("The stream has no active subscriptions"));
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ for (CamelSubscription sub : subscriptions) {
+ sub.signalCompletion();
+ }
+ subscriptions.clear();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelReactiveStreamsServiceImpl.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelReactiveStreamsServiceImpl.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelReactiveStreamsServiceImpl.java
new file mode 100644
index 0000000..fb6b693
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelReactiveStreamsServiceImpl.java
@@ -0,0 +1,133 @@
+/**
+ * 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.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsComponent;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsConsumer;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreamsService;
+import org.apache.camel.component.reactive.streams.api.DispatchCallback;
+import org.apache.camel.component.reactive.streams.util.ConvertingPublisher;
+import org.apache.camel.component.reactive.streams.util.ConvertingSubscriber;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+
+
+public class CamelReactiveStreamsServiceImpl implements CamelReactiveStreamsService {
+
+ private CamelContext context;
+
+ private ExecutorService workerPool;
+
+ private Map<String, CamelPublisher> publishers = new HashMap<>();
+
+ private final Map<String, CamelSubscriber> subscribers = new HashMap<>();
+
+ public CamelReactiveStreamsServiceImpl() {
+ }
+
+ @Override
+ public void start() throws Exception {
+ ReactiveStreamsComponent component = context.getComponent("reactive-streams", ReactiveStreamsComponent.class);
+ ReactiveStreamsEngineConfiguration config = component.getInternalEngineConfiguration();
+ this.workerPool = context.getExecutorServiceManager().newThreadPool(this, config.getThreadPoolName(), config.getThreadPoolMinSize(), config.getThreadPoolMaxSize());
+ }
+
+ @Override
+ public void stop() throws Exception {
+ if (this.workerPool != null) {
+ context.getExecutorServiceManager().shutdownNow(this.workerPool);
+ }
+ }
+
+ @Override
+ public Publisher<Exchange> getPublisher(String name) {
+ return new UnwrappingPublisher<>(getPayloadPublisher(name));
+ }
+
+ @SuppressWarnings("unchecked")
+ public <T> Publisher<T> getPublisher(String name, Class<T> cls) {
+ if (Exchange.class.equals(cls)) {
+ return (Publisher<T>) getPublisher(name);
+ }
+
+ return new ConvertingPublisher<T>(getPublisher(name), cls);
+ }
+
+ @Override
+ public CamelSubscriber getSubscriber(String name) {
+ synchronized (this) {
+ if (!subscribers.containsKey(name)) {
+ CamelSubscriber sub = new CamelSubscriber(name);
+ subscribers.put(name, sub);
+ }
+ return subscribers.get(name);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public <T> Subscriber<T> getSubscriber(String name, Class<T> type) {
+ if (Exchange.class.equals(type)) {
+ return (Subscriber<T>) getSubscriber(name);
+ }
+
+ return new ConvertingSubscriber<T>(getSubscriber(name), getCamelContext());
+ }
+
+ @Override
+ public void process(String name, Exchange exchange, DispatchCallback<Exchange> callback) {
+ StreamPayload<Exchange> payload = new StreamPayload<>(exchange, callback);
+ getPayloadPublisher(name).publish(payload);
+ }
+
+
+ private CamelPublisher getPayloadPublisher(String name) {
+ synchronized (this) {
+ if (!publishers.containsKey(name)) {
+ CamelPublisher publisher = new CamelPublisher(this.workerPool, this.context, name);
+ publishers.put(name, publisher);
+ }
+
+ return publishers.get(name);
+ }
+ }
+
+ @Override
+ public void attachConsumer(String name, ReactiveStreamsConsumer consumer) {
+ getSubscriber(name).attachConsumer(consumer);
+ }
+
+ @Override
+ public void detachConsumer(String name) {
+ getSubscriber(name).detachConsumer();
+ }
+
+ @Override
+ public void setCamelContext(CamelContext camelContext) {
+ this.context = camelContext;
+ }
+
+ @Override
+ public CamelContext getCamelContext() {
+ return this.context;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/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
new file mode 100644
index 0000000..3b7e5d5
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscriber.java
@@ -0,0 +1,177 @@
+/**
+ * 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.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsConsumer;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreamsService;
+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);
+
+ /**
+ * Enough to be considered unbounded. Requests are refilled once completed.
+ */
+ private static final long MAX_INFLIGHT_UNBOUNDED = Long.MAX_VALUE / 2;
+
+ 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 on stream '" + name + "'");
+ }
+ this.consumer = consumer;
+ }
+ refill();
+ }
+
+ 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) {
+ subscription.cancel();
+ } else {
+ refill();
+ }
+ }
+
+ @Override
+ public void onNext(Exchange exchange) {
+ if (exchange == null) {
+ throw new NullPointerException("exchange is null");
+ }
+
+ ReactiveStreamsConsumer target;
+ synchronized (this) {
+ 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() : MAX_INFLIGHT_UNBOUNDED;
+ long newRequest = max - requested - inflightCount;
+ if (newRequest > 0) {
+ 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);
+ synchronized (this) {
+ this.subscription = null;
+ }
+ }
+
+ @Override
+ public void onComplete() {
+ LOG.info("Reactive stream '{}' completed", name);
+ synchronized (this) {
+ this.subscription = null;
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ Subscription subscription;
+ synchronized (this) {
+ subscription = this.subscription;
+ }
+
+ if (subscription != null) {
+ subscription.cancel();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscription.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscription.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscription.java
new file mode 100644
index 0000000..e3489ef
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/CamelSubscription.java
@@ -0,0 +1,252 @@
+/**
+ * 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.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsBackpressureStrategy;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Represents a contract between a Camel published and an external subscriber.
+ * It manages backpressure in order to deal with slow subscribers.
+ */
+public class CamelSubscription implements Subscription {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CamelSubscription.class);
+
+ private ExecutorService workerPool;
+
+ private CamelPublisher publisher;
+
+ private ReactiveStreamsBackpressureStrategy backpressureStrategy;
+
+ private Subscriber<? super StreamPayload<Exchange>> subscriber;
+
+ /**
+ * The lock is used just for the time necessary to read/write shared variables.
+ */
+ private Lock mutex = new ReentrantLock(true);
+
+ private LinkedList<StreamPayload<Exchange>> buffer = new LinkedList<>();
+
+ /**
+ * The current number of exchanges requested by the subscriber.
+ */
+ private long requested;
+
+ /**
+ * Indicates that a cancel operation is to be performed.
+ */
+ private boolean terminating;
+
+ /**
+ * Indicates that the subscription is end.
+ */
+ private boolean terminated;
+
+ /**
+ * Indicates that a thread is currently sending items downstream.
+ * Items must be sent downstream by a single thread for each subscription.
+ */
+ private boolean sending;
+
+
+ public CamelSubscription(ExecutorService workerPool, CamelPublisher publisher, ReactiveStreamsBackpressureStrategy backpressureStrategy, Subscriber<? super StreamPayload<Exchange>> subscriber) {
+ this.workerPool = workerPool;
+ this.publisher = publisher;
+ this.backpressureStrategy = backpressureStrategy;
+ this.subscriber = subscriber;
+ }
+
+ @Override
+ public void request(long l) {
+ LOG.debug("Requested {} events from subscriber", l);
+ if (l <= 0) {
+ // wrong argument
+ mutex.lock();
+ terminated = true;
+ mutex.unlock();
+
+ publisher.unsubscribe(this);
+ subscriber.onError(new IllegalArgumentException("3.9"));
+ } else {
+ mutex.lock();
+ requested += l;
+ mutex.unlock();
+
+ checkAndFlush();
+ }
+ }
+
+ protected void checkAndFlush() {
+ mutex.lock();
+ boolean shouldFlush = !terminated && !sending && requested > 0 && buffer.size() > 0;
+ if (shouldFlush) {
+ sending = true;
+ }
+ mutex.unlock();
+
+ if (shouldFlush) {
+ workerPool.execute(() -> {
+
+ this.flush();
+
+ mutex.lock();
+ sending = false;
+ mutex.unlock();
+
+ // try again to flush
+ checkAndFlush();
+ });
+ } else {
+ mutex.lock();
+ boolean shouldComplete = terminating && !terminated;
+ if (shouldComplete) {
+ terminated = true;
+ }
+ mutex.unlock();
+
+ if (shouldComplete) {
+ this.publisher.unsubscribe(this);
+ this.subscriber.onComplete();
+ discardBuffer(this.buffer);
+ }
+ }
+ }
+
+ protected void flush() {
+ LinkedList<StreamPayload<Exchange>> sendingQueue = null;
+ try {
+ mutex.lock();
+
+ if (this.terminated) {
+ return;
+ }
+
+ int amount = (int) Math.min(requested, (long) buffer.size());
+ if (amount > 0) {
+ this.requested -= amount;
+ sendingQueue = new LinkedList<>();
+ while (amount > 0) {
+ sendingQueue.add(buffer.removeFirst());
+ amount--;
+ }
+ }
+
+ } finally {
+ mutex.unlock();
+ }
+
+ if (sendingQueue != null) {
+ LOG.debug("Sending {} events to the subscriber", sendingQueue.size());
+ for (StreamPayload<Exchange> data : sendingQueue) {
+ // TODO what if the subscriber throws an exception?
+ this.subscriber.onNext(data);
+
+ mutex.lock();
+ boolean shouldStop = this.terminated;
+ mutex.unlock();
+
+ if (shouldStop) {
+ break;
+ }
+ }
+ }
+ }
+
+ public void signalCompletion() throws Exception {
+ mutex.lock();
+ terminating = true;
+ mutex.unlock();
+
+ checkAndFlush();
+ }
+
+ @Override
+ public void cancel() {
+ publisher.unsubscribe(this);
+
+ mutex.lock();
+ this.terminated = true;
+ List<StreamPayload<Exchange>> bufferCopy = new LinkedList<>(buffer);
+ this.buffer.clear();
+ mutex.unlock();
+
+ discardBuffer(bufferCopy);
+ }
+
+ protected void discardBuffer(List<StreamPayload<Exchange>> remaining) {
+ for (StreamPayload<Exchange> data : remaining) {
+ data.getCallback().processed(data.getItem(), new IllegalStateException("Cannot process the exchange " + data.getItem() + ": subscription cancelled"));
+ }
+ }
+
+ public void publish(StreamPayload<Exchange> message) {
+ StreamPayload<Exchange> discardedMessage = null;
+ String discardReason = null;
+ try {
+ mutex.lock();
+ if (!this.terminating && !this.terminated) {
+ if (this.backpressureStrategy == ReactiveStreamsBackpressureStrategy.BUFFER) {
+ buffer.addLast(message);
+ } else if (this.backpressureStrategy == ReactiveStreamsBackpressureStrategy.DROP) {
+ if (buffer.size() > 0) {
+ LOG.warn("Exchange " + message.getItem() + " dropped according to the backpressure strategy " + ReactiveStreamsBackpressureStrategy.DROP);
+ discardedMessage = message;
+ discardReason = "the backpressure strategy (DROP) does not allow buffering";
+ } else {
+ buffer.addLast(message);
+ }
+ } else if (this.backpressureStrategy == ReactiveStreamsBackpressureStrategy.LATEST) {
+ if (buffer.size() > 0) {
+ StreamPayload<Exchange> older = buffer.removeFirst();
+ LOG.warn("Exchange " + message.getItem() + " dropped according to the backpressure strategy " + ReactiveStreamsBackpressureStrategy.LATEST);
+ discardedMessage = older;
+ discardReason = "the backpressure strategy (LATEST) does not allow buffering";
+ }
+ buffer.addLast(message);
+ } else {
+ throw new IllegalStateException("Unsupported backpressure strategy: " + this.backpressureStrategy);
+ }
+
+ } else {
+ discardedMessage = message;
+ discardReason = "subscription closed";
+ }
+ } finally {
+ mutex.unlock();
+ }
+
+ if (discardedMessage != null) {
+ // acknowledge
+ discardedMessage.getCallback().processed(message.getItem(), new IllegalStateException("Exchange discarded: " + discardReason));
+ }
+
+ checkAndFlush();
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/ReactiveStreamsEngineConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/ReactiveStreamsEngineConfiguration.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/ReactiveStreamsEngineConfiguration.java
new file mode 100644
index 0000000..af0fbec
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/ReactiveStreamsEngineConfiguration.java
@@ -0,0 +1,65 @@
+/**
+ * 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;
+
+/**
+ * Configuration parameters for the Camel internal reactive-streams engine.
+ */
+public class ReactiveStreamsEngineConfiguration implements Cloneable {
+
+ private String threadPoolName = "CamelReactiveStreamsWorker";
+
+ private int threadPoolMinSize;
+
+ private int threadPoolMaxSize = 10;
+
+ public ReactiveStreamsEngineConfiguration() {
+ }
+
+ public String getThreadPoolName() {
+ return threadPoolName;
+ }
+
+ /**
+ * The name of the thread pool used by the reactive streams internal engine.
+ */
+ public void setThreadPoolName(String threadPoolName) {
+ this.threadPoolName = threadPoolName;
+ }
+
+ public int getThreadPoolMinSize() {
+ return threadPoolMinSize;
+ }
+
+ /**
+ * The minimum number of threads used by the reactive streams internal engine.
+ */
+ public void setThreadPoolMinSize(int threadPoolMinSize) {
+ this.threadPoolMinSize = threadPoolMinSize;
+ }
+
+ public int getThreadPoolMaxSize() {
+ return threadPoolMaxSize;
+ }
+
+ /**
+ * The maximum number of threads used by the reactive streams internal engine.
+ */
+ public void setThreadPoolMaxSize(int threadPoolMaxSize) {
+ this.threadPoolMaxSize = threadPoolMaxSize;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/StreamPayload.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/StreamPayload.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/StreamPayload.java
new file mode 100644
index 0000000..b592c40
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/StreamPayload.java
@@ -0,0 +1,52 @@
+/**
+ * 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 org.apache.camel.component.reactive.streams.api.DispatchCallback;
+
+/**
+ * A helper object that wraps the emitted item and the corresponding dispatch callback.
+ */
+public class StreamPayload<D> {
+
+ private D item;
+
+ private DispatchCallback<D> callback;
+
+ public StreamPayload(D item, DispatchCallback<D> callback) {
+ this.item = item;
+ this.callback = callback;
+ }
+
+ public D getItem() {
+ return item;
+ }
+
+ public DispatchCallback<D> getCallback() {
+ return callback;
+ }
+
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder("StreamPayload{");
+ sb.append("item=").append(item);
+ sb.append(", callback=").append(callback);
+ sb.append('}');
+ return sb.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/UnwrappingPublisher.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/UnwrappingPublisher.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/UnwrappingPublisher.java
new file mode 100644
index 0000000..1cf73cc
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/engine/UnwrappingPublisher.java
@@ -0,0 +1,87 @@
+/**
+ * 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.util.Objects;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A publisher that extracts the item from the payload as soon as it is delivered to the subscriber.
+ * It calls the dispatch callback if defined.
+ */
+public class UnwrappingPublisher<R> implements Publisher<R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UnwrappingPublisher.class);
+
+ private Publisher<StreamPayload<R>> delegate;
+
+ public UnwrappingPublisher(Publisher<StreamPayload<R>> delegate) {
+ Objects.requireNonNull(delegate, "delegate publisher cannot be null");
+ this.delegate = delegate;
+ }
+
+ @Override
+ public void subscribe(Subscriber<? super R> subscriber) {
+ delegate.subscribe(new Subscriber<StreamPayload<R>>() {
+
+ private Subscription subscription;
+
+ @Override
+ public void onSubscribe(Subscription subscription) {
+ if (subscription == null) {
+ throw new NullPointerException("subscription is null");
+ }
+
+ if (this.subscription != null) {
+ subscription.cancel();
+ } else {
+ this.subscription = subscription;
+ subscriber.onSubscribe(subscription);
+ }
+ }
+
+ @Override
+ public void onNext(StreamPayload<R> payload) {
+ Throwable error = null;
+ try {
+ subscriber.onNext(payload.getItem());
+ } catch (Throwable t) {
+ error = t;
+ }
+
+ if (payload.getCallback() != null) {
+ payload.getCallback().processed(payload.getItem(), error);
+ }
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ subscriber.onError(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+ subscriber.onComplete();
+ }
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/util/ConvertingPublisher.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/util/ConvertingPublisher.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/util/ConvertingPublisher.java
new file mode 100644
index 0000000..e8c10dd
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/util/ConvertingPublisher.java
@@ -0,0 +1,113 @@
+/**
+ * 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.util;
+
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.TypeConversionException;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A publisher that converts Camel {@code Exchange}s into the given type.
+ */
+public class ConvertingPublisher<R> implements Publisher<R> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ConvertingPublisher.class);
+
+ private Publisher<Exchange> delegate;
+
+ private Class<R> type;
+
+ public ConvertingPublisher(Publisher<Exchange> delegate, Class<R> type) {
+ Objects.requireNonNull(delegate, "delegate publisher cannot be null");
+ this.delegate = delegate;
+ this.type = type;
+ }
+
+ @Override
+ public void subscribe(Subscriber<? super R> subscriber) {
+ delegate.subscribe(new Subscriber<Exchange>() {
+
+ private AtomicBoolean active = new AtomicBoolean(true);
+
+ private Subscription subscription;
+
+ @Override
+ public void onSubscribe(Subscription subscription) {
+ if (subscription == null) {
+ throw new NullPointerException("subscription is null");
+ }
+ if (this.subscription != null) {
+ subscription.cancel();
+ } else {
+ this.subscription = subscription;
+ subscriber.onSubscribe(subscription);
+ }
+ }
+
+ @Override
+ public void onNext(Exchange ex) {
+ if (!active.get()) {
+ return;
+ }
+
+ R r;
+ try {
+ r = ex.getIn().getBody(type);
+ } catch (TypeConversionException e) {
+ LOG.warn("Unable to convert body to the specified type: " + type.getName(), e);
+ r = null;
+ }
+
+ if (r == null && ex.getIn().getBody() != null) {
+ this.onError(new ClassCastException("Unable to convert body to the specified type: " + type.getName()));
+
+ active.set(false);
+ subscription.cancel();
+ } else {
+ subscriber.onNext(r);
+ }
+
+
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ if (!active.get()) {
+ return;
+ }
+
+ subscriber.onError(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+ if (!active.get()) {
+ return;
+ }
+
+ subscriber.onComplete();
+ }
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/util/ConvertingSubscriber.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/util/ConvertingSubscriber.java b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/util/ConvertingSubscriber.java
new file mode 100644
index 0000000..cf4d4f0
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/util/ConvertingSubscriber.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.reactive.streams.util;
+
+import java.util.Objects;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultExchange;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+/**
+ * A subscriber that converts items of the given type into Camel {@code Exchange}s.
+ */
+public class ConvertingSubscriber<R> implements Subscriber<R> {
+
+ private Subscriber<Exchange> delegate;
+
+ private CamelContext context;
+
+ public ConvertingSubscriber(Subscriber<Exchange> delegate, CamelContext context) {
+ Objects.requireNonNull(delegate, "delegate subscriber cannot be null");
+ this.delegate = delegate;
+ this.context = context;
+ }
+
+ @Override
+ public void onSubscribe(Subscription subscription) {
+ if (subscription == null) {
+ throw new NullPointerException("subscription is null");
+ }
+
+ delegate.onSubscribe(subscription);
+ }
+
+ @Override
+ public void onNext(R r) {
+ if (r == null) {
+ throw new NullPointerException("element is null");
+ }
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody(r);
+ delegate.onNext(exchange);
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ if (throwable == null) {
+ throw new NullPointerException("throwable is null");
+ }
+
+ delegate.onError(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+ delegate.onComplete();
+ }
+}