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:04 UTC
[5/6] camel git commit: CAMEL-10612: Reactive streams
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/resources/META-INF/LICENSE.txt
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/resources/META-INF/LICENSE.txt b/components/camel-reactive-streams/src/main/resources/META-INF/LICENSE.txt
new file mode 100644
index 0000000..6b0b127
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/resources/META-INF/LICENSE.txt
@@ -0,0 +1,203 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed 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.
+
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/resources/META-INF/NOTICE.txt
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/resources/META-INF/NOTICE.txt b/components/camel-reactive-streams/src/main/resources/META-INF/NOTICE.txt
new file mode 100644
index 0000000..2e215bf
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/resources/META-INF/NOTICE.txt
@@ -0,0 +1,11 @@
+ =========================================================================
+ == NOTICE file corresponding to the section 4 d of ==
+ == the Apache License, Version 2.0, ==
+ == in this case for the Apache Camel distribution. ==
+ =========================================================================
+
+ This product includes software developed by
+ The Apache Software Foundation (http://www.apache.org/).
+
+ Please read the different LICENSE files present in the licenses directory of
+ this distribution.
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/main/resources/META-INF/services/org/apache/camel/component/reactive-streams
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/main/resources/META-INF/services/org/apache/camel/component/reactive-streams b/components/camel-reactive-streams/src/main/resources/META-INF/services/org/apache/camel/component/reactive-streams
new file mode 100644
index 0000000..3ea3f1c
--- /dev/null
+++ b/components/camel-reactive-streams/src/main/resources/META-INF/services/org/apache/camel/component/reactive-streams
@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+class=org.apache.camel.component.reactive.streams.ReactiveStreamsComponent
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressurePublisherRoutePolicyTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressurePublisherRoutePolicyTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressurePublisherRoutePolicyTest.java
new file mode 100644
index 0000000..81d03bc
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressurePublisherRoutePolicyTest.java
@@ -0,0 +1,165 @@
+/**
+ * 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.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import io.reactivex.Observable;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.StatefulService;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.component.reactive.streams.support.TestSubscriber;
+import org.apache.camel.impl.ThrottlingInflightRoutePolicy;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+import org.reactivestreams.Publisher;
+
+public class BackpressurePublisherRoutePolicyTest extends CamelTestSupport {
+
+ @Test
+ public void testThatBackpressureCausesTemporaryRouteStop() throws Exception {
+
+ CountDownLatch generationLatch = new CountDownLatch(25);
+
+ new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ ThrottlingInflightRoutePolicy policy = new ThrottlingInflightRoutePolicy();
+ policy.setMaxInflightExchanges(10);
+ policy.setScope(ThrottlingInflightRoutePolicy.ThrottlingScope.Route);
+ policy.setResumePercentOfMax(70);
+ policy.setLoggingLevel(LoggingLevel.INFO);
+
+ from("timer:tick?period=50&repeatCount=35")
+ .id("policy-route")
+ .routePolicy(policy)
+ .process(x -> generationLatch.countDown())
+ .to("reactive-streams:pub");
+ }
+ }.addRoutesToCamelContext(context);
+
+ CountDownLatch receptionLatch = new CountDownLatch(35);
+
+ Publisher<Exchange> pub = CamelReactiveStreams.get(context()).getPublisher("pub", Exchange.class);
+ TestSubscriber<Exchange> subscriber = new TestSubscriber<Exchange>() {
+ @Override
+ public void onNext(Exchange o) {
+ super.onNext(o);
+ receptionLatch.countDown();
+ }
+ };
+ subscriber.setInitiallyRequested(10);
+ pub.subscribe(subscriber);
+
+ // Add another (fast) subscription that should not affect the backpressure on the route
+ Observable.fromPublisher(pub)
+ .subscribe();
+
+ context.start();
+
+ generationLatch.await(5, TimeUnit.SECONDS); // after 25 messages are generated
+ // The number of exchanges should be 10 (requested by the subscriber), so 35-10=25
+ assertEquals(25, receptionLatch.getCount());
+
+ // fire a delayed request from the subscriber (required by camel core)
+ subscriber.request(1);
+ Thread.sleep(250);
+
+ StatefulService service = (StatefulService) context().getRoute("policy-route").getConsumer();
+ // ensure the route is stopped or suspended
+ assertTrue(service.isStopped() || service.isSuspended());
+
+ // request all the remaining exchanges
+ subscriber.request(24);
+ assertTrue(receptionLatch.await(5, TimeUnit.SECONDS));
+ // The reception latch has gone to 0
+ }
+
+ @Test
+ public void testThatRouteRestartsOnUnsubscription() throws Exception {
+
+ CountDownLatch generationLatch = new CountDownLatch(25);
+
+ new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ ThrottlingInflightRoutePolicy policy = new ThrottlingInflightRoutePolicy();
+ policy.setMaxInflightExchanges(10);
+ policy.setScope(ThrottlingInflightRoutePolicy.ThrottlingScope.Route);
+ policy.setResumePercentOfMax(70);
+ policy.setLoggingLevel(LoggingLevel.INFO);
+
+ from("timer:tick?period=50") // unbounded
+ .id("policy-route")
+ .routePolicy(policy)
+ .process(x -> generationLatch.countDown())
+ .to("reactive-streams:pub");
+ }
+ }.addRoutesToCamelContext(context);
+
+ CountDownLatch receptionLatch = new CountDownLatch(35);
+
+ Publisher<Exchange> pub = CamelReactiveStreams.get(context()).getPublisher("pub", Exchange.class);
+ TestSubscriber<Exchange> subscriber = new TestSubscriber<Exchange>() {
+ @Override
+ public void onNext(Exchange o) {
+ super.onNext(o);
+ receptionLatch.countDown();
+ }
+ };
+ subscriber.setInitiallyRequested(10);
+ pub.subscribe(subscriber);
+
+ // Add another (fast) subscription that should not affect the backpressure on the route
+ Observable.fromPublisher(pub)
+ .subscribe();
+
+ context.start();
+
+ generationLatch.await(5, TimeUnit.SECONDS); // after 25 messages are generated
+ // The number of exchanges should be 10 (requested by the subscriber), so 35-10=25
+ assertEquals(25, receptionLatch.getCount());
+
+ // fire a delayed request from the subscriber (required by camel core)
+ subscriber.request(1);
+ Thread.sleep(250);
+
+ StatefulService service = (StatefulService) context().getRoute("policy-route").getConsumer();
+ // ensure the route is stopped or suspended
+ assertTrue(service.isStopped() || service.isSuspended());
+ subscriber.cancel();
+
+ // request other exchanges to ensure that the route works
+ CountDownLatch latch = new CountDownLatch(20);
+ Observable.fromPublisher(pub)
+ .subscribe(n -> {
+ latch.countDown();
+ });
+
+ assertTrue(latch.await(5, TimeUnit.SECONDS));
+ }
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureStrategyTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureStrategyTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureStrategyTest.java
new file mode 100644
index 0000000..8846cd1
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureStrategyTest.java
@@ -0,0 +1,164 @@
+/**
+ * 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.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import io.reactivex.Flowable;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.component.reactive.streams.support.TestSubscriber;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+public class BackpressureStrategyTest extends CamelTestSupport {
+
+ @Test
+ public void testBackpressureBufferStrategy() throws Exception {
+
+ new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("timer:gen?period=20&repeatCount=20")
+ .setBody().header(Exchange.TIMER_COUNTER)
+ .to("reactive-streams:integers");
+ }
+ }.addRoutesToCamelContext(context);
+
+ Flowable<Integer> integers = Flowable.fromPublisher(CamelReactiveStreams.get(context).getPublisher("integers", Integer.class));
+
+ ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
+ CountDownLatch latch = new CountDownLatch(1);
+
+ Flowable.interval(0, 50, TimeUnit.MILLISECONDS)
+ .zipWith(integers, (l, i) -> i)
+ .timeout(2000, TimeUnit.MILLISECONDS, Flowable.empty())
+ .doOnComplete(latch::countDown)
+ .subscribe(queue::add);
+
+ context().start();
+
+ assertTrue(latch.await(5, TimeUnit.SECONDS));
+
+ assertEquals(20, queue.size());
+ int num = 1;
+ for (int i : queue) {
+ assertEquals(num++, i);
+ }
+ }
+
+ @Test
+ public void testBackpressureDropStrategy() throws Exception {
+
+ ReactiveStreamsComponent comp = (ReactiveStreamsComponent) context().getComponent("reactive-streams");
+ comp.setBackpressureStrategy(ReactiveStreamsBackpressureStrategy.DROP);
+
+ new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("timer:gen?period=20&repeatCount=20")
+ .setBody().header(Exchange.TIMER_COUNTER)
+ .to("reactive-streams:integers");
+ }
+ }.addRoutesToCamelContext(context);
+
+
+ ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
+ final CountDownLatch latch = new CountDownLatch(1);
+ final CountDownLatch latch2 = new CountDownLatch(2);
+
+ TestSubscriber<Integer> subscriber = new TestSubscriber<Integer>() {
+ @Override
+ public void onNext(Integer o) {
+ queue.add(o);
+ latch.countDown();
+ latch2.countDown();
+ }
+ };
+ subscriber.setInitiallyRequested(1);
+ CamelReactiveStreams.get(context).getPublisher("integers", Integer.class).subscribe(subscriber);
+
+ context().start();
+
+ assertTrue(latch.await(5, TimeUnit.SECONDS));
+ Thread.sleep(1000); // wait for all numbers to be generated
+
+ subscriber.request(19);
+ assertTrue(latch2.await(1, TimeUnit.SECONDS));
+ Thread.sleep(200); // add other time to ensure no other items arrive
+ assertEquals(2, queue.size());
+ int sum = queue.stream().reduce((i, j) -> i + j).get();
+ assertEquals(3, sum); // 1 + 2 = 3
+
+ subscriber.cancel();
+ }
+
+ @Test
+ public void testBackpressureLatestStrategy() throws Exception {
+
+ ReactiveStreamsComponent comp = (ReactiveStreamsComponent) context().getComponent("reactive-streams");
+ comp.setBackpressureStrategy(ReactiveStreamsBackpressureStrategy.LATEST);
+
+ new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("timer:gen?period=20&repeatCount=20")
+ .setBody().header(Exchange.TIMER_COUNTER)
+ .to("reactive-streams:integers");
+ }
+ }.addRoutesToCamelContext(context);
+
+
+ ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
+ final CountDownLatch latch = new CountDownLatch(1);
+ final CountDownLatch latch2 = new CountDownLatch(2);
+
+ TestSubscriber<Integer> subscriber = new TestSubscriber<Integer>() {
+ @Override
+ public void onNext(Integer o) {
+ queue.add(o);
+ latch.countDown();
+ latch2.countDown();
+ }
+ };
+ subscriber.setInitiallyRequested(1);
+ CamelReactiveStreams.get(context).getPublisher("integers", Integer.class).subscribe(subscriber);
+
+ context().start();
+
+ assertTrue(latch.await(5, TimeUnit.SECONDS));
+ Thread.sleep(1000); // wait for all numbers to be generated
+
+ subscriber.request(19);
+ assertTrue(latch2.await(1, TimeUnit.SECONDS));
+ Thread.sleep(200); // add other time to ensure no other items arrive
+ assertEquals(2, queue.size());
+ int sum = queue.stream().reduce((i, j) -> i + j).get();
+ assertEquals(21, sum); // 1 + 20 = 21
+
+ subscriber.cancel();
+ }
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureSubscriberTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureSubscriberTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureSubscriberTest.java
new file mode 100644
index 0000000..edcad30
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BackpressureSubscriberTest.java
@@ -0,0 +1,107 @@
+/**
+ * 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 io.reactivex.BackpressureStrategy;
+import io.reactivex.Flowable;
+import io.reactivex.Observable;
+
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+/**
+ * Test backpressure from the consumer side.
+ */
+public class BackpressureSubscriberTest extends CamelTestSupport {
+
+ @Test
+ public void testBackpressure() throws Exception {
+
+ long start = System.currentTimeMillis();
+ Observable.range(0, 10)
+ .toFlowable(BackpressureStrategy.BUFFER)
+ .subscribe(CamelReactiveStreams.get(context).getSubscriber("slowNumbers", Integer.class));
+
+ MockEndpoint endpoint = getMockEndpoint("mock:endpoint");
+ endpoint.expectedMessageCount(10);
+ endpoint.assertIsSatisfied();
+ long end = System.currentTimeMillis();
+
+ // Maximum one inflight exchange, even if multiple consumer threads are present
+ // Must take at least 50 * 10 = 500ms
+ assertTrue("Exchange completed too early", end - start >= 500);
+ }
+
+ @Test
+ public void testSlowerBackpressure() throws Exception {
+
+ long start = System.currentTimeMillis();
+ Observable.range(0, 2)
+ .toFlowable(BackpressureStrategy.BUFFER)
+ .subscribe(CamelReactiveStreams.get(context).getSubscriber("slowerNumbers", Integer.class));
+
+ MockEndpoint endpoint = getMockEndpoint("mock:endpoint");
+ endpoint.expectedMessageCount(2);
+ endpoint.assertIsSatisfied();
+ long end = System.currentTimeMillis();
+
+ // Maximum one inflight exchange, even if multiple consumer threads are present
+ // Must take at least 300 * 2 = 600ms
+ assertTrue("Exchange completed too early", end - start >= 600);
+ }
+
+ @Test
+ public void testParallelSlowBackpressure() throws Exception {
+
+ long start = System.currentTimeMillis();
+ Flowable.range(0, 40)
+ .subscribe(CamelReactiveStreams.get(context).getSubscriber("parallelSlowNumbers", Integer.class));
+
+ MockEndpoint endpoint = getMockEndpoint("mock:endpoint");
+ endpoint.expectedMessageCount(40);
+ endpoint.assertIsSatisfied();
+ long end = System.currentTimeMillis();
+
+ // Maximum 5 inflight exchanges
+ // Must take at least 100 * (40 / 5) = 800ms
+ assertTrue("Exchange completed too early", end - start >= 800);
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("reactive-streams:slowNumbers?concurrentConsumers=10&maxInflightExchanges=1")
+ .process(x -> Thread.sleep(50))
+ .to("mock:endpoint");
+
+ from("reactive-streams:slowerNumbers?concurrentConsumers=10&maxInflightExchanges=1")
+ .process(x -> Thread.sleep(300))
+ .to("mock:endpoint");
+
+ from("reactive-streams:parallelSlowNumbers?concurrentConsumers=10&maxInflightExchanges=5")
+ .process(x -> Thread.sleep(100))
+ .to("mock:endpoint");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BasicPublisherTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BasicPublisherTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BasicPublisherTest.java
new file mode 100644
index 0000000..84b15de
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BasicPublisherTest.java
@@ -0,0 +1,112 @@
+/**
+ * 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.LinkedList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import io.reactivex.Observable;
+import io.reactivex.disposables.Disposable;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+public class BasicPublisherTest extends CamelTestSupport {
+
+ @Test
+ public void testWorking() throws Exception {
+
+ new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("timer:tick?period=5&repeatCount=30")
+ .setBody().header(Exchange.TIMER_COUNTER)
+ .to("reactive-streams:pub");
+ }
+ }.addRoutesToCamelContext(context);
+
+ int num = 30;
+ CountDownLatch latch = new CountDownLatch(num);
+ List<Integer> recv = new LinkedList<>();
+
+ Observable.fromPublisher(CamelReactiveStreams.get(context).getPublisher("pub", Integer.class))
+ .doOnNext(recv::add)
+ .doOnNext(n -> latch.countDown())
+ .subscribe();
+
+ context.start();
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertEquals(num, recv.size());
+ for (int i = 1; i <= num; i++) {
+ assertEquals(i, recv.get(i - 1).intValue());
+ }
+ }
+
+ @Test
+ public void testMultipleSubscriptions() throws Exception {
+
+ new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("timer:tick?period=50")
+ .setBody().header(Exchange.TIMER_COUNTER)
+ .to("reactive-streams:unbounded");
+ }
+ }.addRoutesToCamelContext(context);
+
+ CountDownLatch latch1 = new CountDownLatch(5);
+ Disposable disp1 = Observable.fromPublisher(CamelReactiveStreams.get(context).getPublisher("unbounded", Integer.class))
+ .subscribe(n -> latch1.countDown());
+
+ context.start();
+
+ // Add another subscription
+ CountDownLatch latch2 = new CountDownLatch(5);
+ Disposable disp2 = Observable.fromPublisher(CamelReactiveStreams.get(context).getPublisher("unbounded", Integer.class))
+ .subscribe(n -> latch2.countDown());
+
+ assertTrue(latch1.await(5, TimeUnit.SECONDS));
+ assertTrue(latch2.await(5, TimeUnit.SECONDS));
+
+ // Unsubscribe both
+ disp1.dispose();
+ disp2.dispose();
+
+ // No active subscriptions, warnings expected
+ Thread.sleep(60);
+
+ // Add another subscription
+ CountDownLatch latch3 = new CountDownLatch(5);
+ Disposable disp3 = Observable.fromPublisher(CamelReactiveStreams.get(context).getPublisher("unbounded", Integer.class))
+ .subscribe(n -> latch3.countDown());
+
+ assertTrue(latch3.await(5, TimeUnit.SECONDS));
+ disp3.dispose();
+ }
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BasicSubscriberTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BasicSubscriberTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BasicSubscriberTest.java
new file mode 100644
index 0000000..ba7bea8
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/BasicSubscriberTest.java
@@ -0,0 +1,85 @@
+/**
+ * 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.Exchange;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+
+public class BasicSubscriberTest extends CamelTestSupport {
+
+ @Test
+ public void testWorking() throws Exception {
+ int count = 2;
+ MockEndpoint e1 = getMockEndpoint("mock:sub1");
+ e1.expectedMinimumMessageCount(count);
+ e1.assertIsSatisfied();
+
+ MockEndpoint e2 = getMockEndpoint("mock:sub2");
+ e2.expectedMinimumMessageCount(count);
+ e2.assertIsSatisfied();
+
+ MockEndpoint e3 = getMockEndpoint("mock:sub3");
+ e3.expectedMinimumMessageCount(count);
+ e3.assertIsSatisfied();
+
+ for (int i = 0; i < count; i++) {
+ Exchange ex1 = e1.getExchanges().get(i);
+ Exchange ex2 = e2.getExchanges().get(i);
+ Exchange ex3 = e3.getExchanges().get(i);
+ assertEquals(ex1.getIn().getBody(), ex2.getIn().getBody());
+ assertEquals(ex1.getIn().getBody(), ex3.getIn().getBody());
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ protected void doPostSetup() throws Exception {
+
+ Subscriber<Integer> sub = CamelReactiveStreams.get(context()).getSubscriber("sub", Integer.class);
+ Subscriber<Integer> sub2 = CamelReactiveStreams.get(context()).getSubscriber("sub2", Integer.class);
+ Publisher<Integer> pub = CamelReactiveStreams.get(context()).getPublisher("pub", Integer.class);
+
+ pub.subscribe(sub);
+ pub.subscribe(sub2);
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("reactive-streams:sub")
+ .to("mock:sub1");
+
+ from("reactive-streams:sub2")
+ .to("mock:sub2");
+
+ from("timer:tick?period=50")
+ .setBody().simple("random(500)")
+ .wireTap("mock:sub3")
+ .to("reactive-streams:pub");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelPublisherConversionVerificationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelPublisherConversionVerificationTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelPublisherConversionVerificationTest.java
new file mode 100644
index 0000000..a3b1c6b
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelPublisherConversionVerificationTest.java
@@ -0,0 +1,68 @@
+/**
+ * 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.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.tck.PublisherVerification;
+import org.reactivestreams.tck.TestEnvironment;
+
+public class CamelPublisherConversionVerificationTest extends PublisherVerification<Long> {
+
+ public CamelPublisherConversionVerificationTest() {
+ super(new TestEnvironment(2000L));
+ }
+
+ @Override
+ public Publisher<Long> createPublisher(long l) {
+
+ CamelContext context = new DefaultCamelContext();
+ RouteBuilder builder = new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("timer:tick?delay=500&period=50&repeatCount=" + l)
+ .setBody().simple("random(1000)")
+ .to("reactive-streams:prod");
+ }
+ };
+
+ Publisher<Long> pub = CamelReactiveStreams.get(context).getPublisher("prod", Long.class);
+
+ try {
+ builder.addRoutesToCamelContext(context);
+ context.start();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ return pub;
+ }
+
+ @Override
+ public long maxElementsFromPublisher() {
+ // It's an active publisher
+ return publisherUnableToSignalOnComplete(); // == Long.MAX_VALUE == unbounded
+ }
+
+ @Override
+ public Publisher<Long> createFailedPublisher() {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelPublisherVerificationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelPublisherVerificationTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelPublisherVerificationTest.java
new file mode 100644
index 0000000..a29e05e
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelPublisherVerificationTest.java
@@ -0,0 +1,68 @@
+/**
+ * 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.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.tck.PublisherVerification;
+import org.reactivestreams.tck.TestEnvironment;
+
+public class CamelPublisherVerificationTest extends PublisherVerification<Exchange> {
+
+ public CamelPublisherVerificationTest() {
+ super(new TestEnvironment(2000L));
+ }
+
+ @Override
+ public Publisher<Exchange> createPublisher(long l) {
+
+ CamelContext context = new DefaultCamelContext();
+ RouteBuilder builder = new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("timer:tick?delay=500&period=50&repeatCount=" + l)
+ .to("reactive-streams:prod");
+ }
+ };
+
+ Publisher<Exchange> pub = CamelReactiveStreams.get(context).getPublisher("prod");
+
+ try {
+ builder.addRoutesToCamelContext(context);
+ context.start();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ return pub;
+ }
+
+ @Override
+ public long maxElementsFromPublisher() {
+ // It's an active publisher
+ return publisherUnableToSignalOnComplete(); // == Long.MAX_VALUE == unbounded
+ }
+
+ @Override
+ public Publisher<Exchange> createFailedPublisher() {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelSubscriberConversionVerificationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelSubscriberConversionVerificationTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelSubscriberConversionVerificationTest.java
new file mode 100644
index 0000000..8f1d157
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelSubscriberConversionVerificationTest.java
@@ -0,0 +1,62 @@
+/**
+ * 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.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.tck.SubscriberBlackboxVerification;
+import org.reactivestreams.tck.TestEnvironment;
+
+public class CamelSubscriberConversionVerificationTest extends SubscriberBlackboxVerification<Integer> {
+
+ private CamelContext context;
+
+ public CamelSubscriberConversionVerificationTest() {
+ super(new TestEnvironment(2000L));
+ }
+
+ @Override
+ public Subscriber<Integer> createSubscriber() {
+ this.context = new DefaultCamelContext();
+ RouteBuilder builder = new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("reactive-streams:sub?maxInflightExchanges=20")
+ .to("log:INFO");
+ }
+ };
+
+ Subscriber<Integer> sub = CamelReactiveStreams.get(context).getSubscriber("sub", Integer.class);
+
+ try {
+ builder.addRoutesToCamelContext(context);
+ context.start();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ return sub;
+ }
+
+ @Override
+ public Integer createElement(int element) {
+ return element;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelSubscriberVerificationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelSubscriberVerificationTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelSubscriberVerificationTest.java
new file mode 100644
index 0000000..921f468
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CamelSubscriberVerificationTest.java
@@ -0,0 +1,66 @@
+/**
+ * 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.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.impl.DefaultExchange;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.tck.SubscriberBlackboxVerification;
+import org.reactivestreams.tck.TestEnvironment;
+
+public class CamelSubscriberVerificationTest extends SubscriberBlackboxVerification<Exchange> {
+
+ private CamelContext context;
+
+ public CamelSubscriberVerificationTest() {
+ super(new TestEnvironment(2000L));
+ }
+
+ @Override
+ public Subscriber<Exchange> createSubscriber() {
+ this.context = new DefaultCamelContext();
+ RouteBuilder builder = new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("reactive-streams:sub?maxInflightExchanges=20")
+ .to("log:INFO");
+ }
+ };
+
+ Subscriber<Exchange> sub = CamelReactiveStreams.get(context).getSubscriber("sub");
+
+ try {
+ builder.addRoutesToCamelContext(context);
+ context.start();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ return sub;
+ }
+
+ @Override
+ public Exchange createElement(int element) {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody(element);
+ return exchange;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConcurrentConsumersTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConcurrentConsumersTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConcurrentConsumersTest.java
new file mode 100644
index 0000000..95f6474
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConcurrentConsumersTest.java
@@ -0,0 +1,103 @@
+/**
+ * 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.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import io.reactivex.BackpressureStrategy;
+import io.reactivex.Observable;
+
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+/**
+ * Test the behaviour of the consumer side when using a different number of consumer threads.
+ */
+public class ConcurrentConsumersTest extends CamelTestSupport {
+
+ @Test
+ public void testSingleConsumer() throws Exception {
+
+ Observable.intervalRange(0, 1000, 0, 300, TimeUnit.MICROSECONDS)
+ .toFlowable(BackpressureStrategy.BUFFER)
+ .subscribe(CamelReactiveStreams.get(context()).getSubscriber("singleConsumer", Long.class));
+
+ MockEndpoint endpoint = getMockEndpoint("mock:singleBucket");
+ endpoint.expectedMessageCount(1000);
+ endpoint.assertIsSatisfied();
+
+ Set<String> threads = endpoint.getExchanges().stream()
+ .map(x -> x.getIn().getHeader("thread", String.class))
+ .collect(Collectors.toSet());
+ assertEquals(1, threads.size());
+
+ // Ensure order is preserved when using a single consumer
+ List<Long> nums = endpoint.getExchanges().stream()
+ .map(x -> x.getIn().getBody(Long.class))
+ .collect(Collectors.toList());
+
+ long prev = -1;
+ for (long n : nums) {
+ assertEquals(prev + 1, n);
+ prev = n;
+ }
+ }
+
+ @Test
+ public void testMultipleConsumers() throws Exception {
+
+ Observable.intervalRange(0, 1000, 0, 300, TimeUnit.MICROSECONDS)
+ .toFlowable(BackpressureStrategy.BUFFER)
+ .subscribe(CamelReactiveStreams.get(context()).getSubscriber("multipleConsumers", Long.class));
+
+ MockEndpoint endpoint = getMockEndpoint("mock:multipleBucket");
+ endpoint.expectedMessageCount(1000);
+ endpoint.assertIsSatisfied();
+
+ Set<String> threads = endpoint.getExchanges().stream()
+ .map(x -> x.getIn().getHeader("thread", String.class))
+ .collect(Collectors.toSet());
+
+ assertEquals(3, threads.size());
+ // Order cannot be preserved when using multiple consumers
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+
+ from("reactive-streams:singleConsumer")
+ .process(x -> x.getIn().setHeader("thread", Thread.currentThread().getId()))
+ .to("mock:singleBucket");
+
+ from("reactive-streams:multipleConsumers?concurrentConsumers=3")
+ .process(x -> x.getIn().setHeader("thread", Thread.currentThread().getId()))
+ .to("mock:multipleBucket");
+
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/PublisherTypeConversionTest.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/PublisherTypeConversionTest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/PublisherTypeConversionTest.java
new file mode 100644
index 0000000..f117878
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/PublisherTypeConversionTest.java
@@ -0,0 +1,88 @@
+/**
+ * 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.LinkedList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import io.reactivex.Observable;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.junit.Test;
+
+public class PublisherTypeConversionTest extends CamelTestSupport {
+
+
+ @Test
+ public void testConversion() throws Exception {
+
+ CountDownLatch latch = new CountDownLatch(3);
+ List<Integer> integers = new LinkedList<>();
+
+ Observable.fromPublisher(CamelReactiveStreams.get(context).getPublisher("pub", Exchange.class))
+ .map(x -> x.getIn().getBody(Integer.class))
+ .subscribe(n -> {
+ integers.add(n);
+ latch.countDown();
+ });
+
+ Observable.fromPublisher(CamelReactiveStreams.get(context).getPublisher("pub"))
+ .map(x -> x.getIn().getBody(Integer.class))
+ .subscribe(n -> {
+ integers.add(n);
+ latch.countDown();
+ });
+
+ Observable.fromPublisher(CamelReactiveStreams.get(context).getPublisher("pub", Integer.class))
+ .subscribe(n -> {
+ integers.add(n);
+ latch.countDown();
+ });
+
+ context.start();
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertEquals(3, integers.size());
+
+ for (int i : integers) {
+ assertEquals(123, i);
+ }
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("timer:tick?period=50&repeatCount=1")
+ .setBody().constant(123)
+ .to("reactive-streams:pub");
+ }
+ };
+ }
+
+ @Override
+ public boolean isUseAdviceWith() {
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/TestPublisher.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/TestPublisher.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/TestPublisher.java
new file mode 100644
index 0000000..0165bed
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/TestPublisher.java
@@ -0,0 +1,110 @@
+/**
+ * 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.support;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+/**
+ * A publisher for tests.
+ */
+public class TestPublisher<T> implements Publisher<T> {
+
+ private Iterable<T> data;
+
+ private long delay;
+
+ public TestPublisher(Iterable<T> data) {
+ this(data, 0L);
+ }
+
+ public TestPublisher(Iterable<T> data, long delay) {
+ this.data = data;
+ this.delay = delay;
+ }
+
+ @Override
+ public void subscribe(Subscriber<? super T> subscriber) {
+ subscriber.onSubscribe(new Subscription() {
+
+ private Iterator<T> it = data.iterator();
+
+ private AtomicLong requested = new AtomicLong(0);
+
+ private Object monitor = new Object();
+
+ @Override
+ public void request(long l) {
+ this.requested.addAndGet(l);
+
+ new Thread() {
+ @Override
+ public void run() {
+ synchronized (monitor) {
+ boolean wasNonEmpty = it.hasNext();
+ while (requested.longValue() > 0 && it.hasNext()) {
+ T d = it.next();
+ requested.decrementAndGet();
+
+ if (delay > 0) {
+ try {
+ Thread.sleep(delay);
+ } catch (InterruptedException ex) {
+ }
+ }
+ subscriber.onNext(d);
+ }
+
+ if (wasNonEmpty && !it.hasNext()) {
+ // data cannot be added to this test publisher
+ subscriber.onComplete();
+ }
+ }
+ }
+ }.start();
+ }
+
+ @Override
+ public void cancel() {
+ synchronized (monitor) {
+ this.requested.set(0);
+ this.it = new Iterator<T>() {
+ @Override
+ public boolean hasNext() {
+ return false;
+ }
+
+ @Override
+ public T next() {
+ throw new NoSuchElementException();
+ }
+ };
+ new Thread() {
+ public void run() {
+ subscriber.onComplete();
+ }
+ }.start();
+ }
+ }
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/TestSubscriber.java
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/TestSubscriber.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/TestSubscriber.java
new file mode 100644
index 0000000..f49d5c9
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/support/TestSubscriber.java
@@ -0,0 +1,73 @@
+/**
+ * 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.support;
+
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+/**
+ * A subscriber for tests.
+ */
+public class TestSubscriber<T> implements Subscriber<T> {
+
+ protected Subscription subscription;
+
+ private long initiallyRequested;
+
+ public TestSubscriber() {
+ }
+
+ public long getInitiallyRequested() {
+ return initiallyRequested;
+ }
+
+ public void setInitiallyRequested(long initiallyRequested) {
+ this.initiallyRequested = initiallyRequested;
+ }
+
+ public void request(long exchanges) {
+ this.subscription.request(exchanges);
+ }
+
+ public void cancel() {
+ this.subscription.cancel();
+ }
+
+ @Override
+ public void onSubscribe(Subscription subscription) {
+ this.subscription = subscription;
+
+ if (initiallyRequested > 0) {
+ subscription.request(initiallyRequested);
+ }
+ }
+
+ @Override
+ public void onNext(T t) {
+
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+
+ }
+
+ @Override
+ public void onComplete() {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/camel-reactive-streams/src/test/resources/log4j2.properties
----------------------------------------------------------------------
diff --git a/components/camel-reactive-streams/src/test/resources/log4j2.properties b/components/camel-reactive-streams/src/test/resources/log4j2.properties
new file mode 100644
index 0000000..fbab18c
--- /dev/null
+++ b/components/camel-reactive-streams/src/test/resources/log4j2.properties
@@ -0,0 +1,33 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+
+appender.file.type = File
+appender.file.name = file
+appender.file.fileName = target/camel-reactive-streams-test.log
+appender.file.layout.type = PatternLayout
+appender.file.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+appender.stdout.type = Console
+appender.stdout.name = stdout
+appender.stdout.layout.type = PatternLayout
+appender.stdout.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+rootLogger.level = INFO
+rootLogger.appenderRef.file.ref = file
+
+#rootLogger.appenderRef.stdout.ref = stdout
+#logger.debug.name = org.apache.camel.component.reactive.streams
+#logger.debug.level = DEBUG
+#logger.debug.additivity = true
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/pom.xml
----------------------------------------------------------------------
diff --git a/components/pom.xml b/components/pom.xml
index def06f5..4276240 100644
--- a/components/pom.xml
+++ b/components/pom.xml
@@ -216,6 +216,7 @@
<module>camel-quartz2</module>
<module>camel-quickfix</module>
<module>camel-rabbitmq</module>
+ <module>camel-reactive-streams</module>
<module>camel-restlet</module>
<module>camel-rmi</module>
<module>camel-routebox</module>
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/components/readme.adoc
----------------------------------------------------------------------
diff --git a/components/readme.adoc b/components/readme.adoc
index 9813764..b8fb246 100644
--- a/components/readme.adoc
+++ b/components/readme.adoc
@@ -438,6 +438,9 @@ Components
| link:camel-rabbitmq/src/main/docs/rabbitmq-component.adoc[RabbitMQ] (camel-rabbitmq) +
`rabbitmq:hostname:portNumber/exchangeName` | The rabbitmq component allows you produce and consume messages from RabbitMQ instances.
+| link:camel-reactive-streams/src/main/docs/reactive-streams-component.adoc[Reactive Streams] (camel-reactive-streams) +
+`reactive-streams:/stream` | The Camel reactive-streams endpoint.
+
| link:camel-restlet/src/main/docs/restlet-component.adoc[Restlet] (camel-restlet) +
`restlet:protocol:host:port/uriPattern` | Component for consuming and producing Restful resources using Restlet.
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/docs/user-manual/en/SUMMARY.md
----------------------------------------------------------------------
diff --git a/docs/user-manual/en/SUMMARY.md b/docs/user-manual/en/SUMMARY.md
index 54c6702..b7dd18c 100644
--- a/docs/user-manual/en/SUMMARY.md
+++ b/docs/user-manual/en/SUMMARY.md
@@ -268,6 +268,7 @@
* [Quartz2](quartz2-component.adoc)
* [QuickFix](quickfix-component.adoc)
* [RabbitMQ](rabbitmq-component.adoc)
+ * [Reactive Streams](reactive-streams-component.adoc)
* [Restlet](restlet-component.adoc)
* [RMI](rmi-component.adoc)
* [RouteBox](routebox-component.adoc)
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/parent/pom.xml
----------------------------------------------------------------------
diff --git a/parent/pom.xml b/parent/pom.xml
index ac58a34..41edd0b 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -520,6 +520,7 @@
<rome-bundle-version>1.0_3</rome-bundle-version>
<rome-version>1.0</rome-version>
<rxjava-version>1.2.5</rxjava-version>
+ <rxjava2-version>2.0.4</rxjava2-version>
<saaj-impl-version>1.3.2_2</saaj-impl-version>
<saxon-bundle-version>9.7.0-14_1</saxon-bundle-version>
<saxon-version>9.7.0-14</saxon-version>
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/platforms/spring-boot/components-starter/camel-reactive-streams-starter/pom.xml
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-reactive-streams-starter/pom.xml b/platforms/spring-boot/components-starter/camel-reactive-streams-starter/pom.xml
new file mode 100644
index 0000000..5325d29
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-reactive-streams-starter/pom.xml
@@ -0,0 +1,51 @@
+<?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-starter</artifactId>
+ <version>2.19.0-SNAPSHOT</version>
+ </parent>
+ <artifactId>camel-reactive-streams-starter</artifactId>
+ <packaging>jar</packaging>
+ <name>Spring-Boot Starter :: Camel :: Reactive Streams</name>
+ <description>Spring-Boot Starter for Camel Reactive Streams Component</description>
+ <dependencies>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter</artifactId>
+ <version>${spring-boot-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-reactive-streams</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <!--START OF GENERATED CODE-->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core-starter</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-spring-boot-starter</artifactId>
+ </dependency>
+ <!--END OF GENERATED CODE-->
+ </dependencies>
+</project>
http://git-wip-us.apache.org/repos/asf/camel/blob/e66207a9/platforms/spring-boot/components-starter/camel-reactive-streams-starter/src/main/java/org/apache/camel/component/reactive/streams/springboot/ReactiveStreamsComponentAutoConfiguration.java
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-reactive-streams-starter/src/main/java/org/apache/camel/component/reactive/streams/springboot/ReactiveStreamsComponentAutoConfiguration.java b/platforms/spring-boot/components-starter/camel-reactive-streams-starter/src/main/java/org/apache/camel/component/reactive/streams/springboot/ReactiveStreamsComponentAutoConfiguration.java
new file mode 100644
index 0000000..10f6dad
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-reactive-streams-starter/src/main/java/org/apache/camel/component/reactive/streams/springboot/ReactiveStreamsComponentAutoConfiguration.java
@@ -0,0 +1,112 @@
+/**
+ * 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.springboot;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.reactive.streams.ReactiveStreamsComponent;
+import org.apache.camel.util.IntrospectionSupport;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionMessage;
+import org.springframework.boot.autoconfigure.condition.ConditionOutcome;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.SpringBootCondition;
+import org.springframework.boot.bind.RelaxedPropertyResolver;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+
+/**
+ * Generated by camel-package-maven-plugin - do not edit this file!
+ */
+@Configuration
+@ConditionalOnBean(type = "org.apache.camel.spring.boot.CamelAutoConfiguration")
+@Conditional(ReactiveStreamsComponentAutoConfiguration.Condition.class)
+@AutoConfigureAfter(name = "org.apache.camel.spring.boot.CamelAutoConfiguration")
+@EnableConfigurationProperties(ReactiveStreamsComponentConfiguration.class)
+public class ReactiveStreamsComponentAutoConfiguration {
+
+ @Lazy
+ @Bean(name = "reactive-streams-component")
+ @ConditionalOnClass(CamelContext.class)
+ @ConditionalOnMissingBean(ReactiveStreamsComponent.class)
+ public ReactiveStreamsComponent configureReactiveStreamsComponent(
+ CamelContext camelContext,
+ ReactiveStreamsComponentConfiguration configuration)
+ throws Exception {
+ ReactiveStreamsComponent component = new ReactiveStreamsComponent();
+ component.setCamelContext(camelContext);
+ Map<String, Object> parameters = new HashMap<>();
+ IntrospectionSupport.getProperties(configuration, parameters, null,
+ false);
+ for (Map.Entry<String, Object> entry : parameters.entrySet()) {
+ Object value = entry.getValue();
+ Class<?> paramClass = value.getClass();
+ if (paramClass.getName().endsWith("NestedConfiguration")) {
+ Class nestedClass = null;
+ try {
+ nestedClass = (Class) paramClass.getDeclaredField(
+ "CAMEL_NESTED_CLASS").get(null);
+ HashMap<String, Object> nestedParameters = new HashMap<>();
+ IntrospectionSupport.getProperties(value, nestedParameters,
+ null, false);
+ Object nestedProperty = nestedClass.newInstance();
+ IntrospectionSupport.setProperties(camelContext,
+ camelContext.getTypeConverter(), nestedProperty,
+ nestedParameters);
+ entry.setValue(nestedProperty);
+ } catch (NoSuchFieldException e) {
+ }
+ }
+ }
+ IntrospectionSupport.setProperties(camelContext,
+ camelContext.getTypeConverter(), component, parameters);
+ return component;
+ }
+
+ public static class Condition extends SpringBootCondition {
+ @Override
+ public ConditionOutcome getMatchOutcome(
+ ConditionContext conditionContext,
+ AnnotatedTypeMetadata annotatedTypeMetadata) {
+ boolean groupEnabled = isEnabled(conditionContext,
+ "camel.component.", true);
+ ConditionMessage.Builder message = ConditionMessage
+ .forCondition("camel.component.reactive-streams");
+ if (isEnabled(conditionContext,
+ "camel.component.reactive-streams.", groupEnabled)) {
+ return ConditionOutcome.match(message.because("enabled"));
+ }
+ return ConditionOutcome.noMatch(message.because("not enabled"));
+ }
+
+ private boolean isEnabled(
+ org.springframework.context.annotation.ConditionContext context,
+ java.lang.String prefix, boolean defaultValue) {
+ RelaxedPropertyResolver resolver = new RelaxedPropertyResolver(
+ context.getEnvironment(), prefix);
+ return resolver.getProperty("enabled", Boolean.class, defaultValue);
+ }
+ }
+}
\ No newline at end of file