camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nferr...@apache.org
Subject [6/6] camel git commit: CAMEL-10612: Reactive streams
Date Fri, 27 Jan 2017 14:35:05 GMT
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.ferraro@gmail.com>
Authored: Wed Dec 21 18:59:01 2016 +0100
Committer: Nicola Ferraro <ni.ferraro@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();
+    }
+}


Mime
View raw message