You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2019/07/30 15:41:15 UTC
[camel] 02/03: CAMEL-13801 - Fixed CS for Camel-reactive-streams
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 58031a62cfec744e94a94b30aaeab963dc98f8f7
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Tue Jul 30 17:39:53 2019 +0200
CAMEL-13801 - Fixed CS for Camel-reactive-streams
---
.../reactive/streams/DirectClientAPITest.java | 84 ++++++----------------
1 file changed, 22 insertions(+), 62 deletions(-)
diff --git a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/DirectClientAPITest.java b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/DirectClientAPITest.java
index f4b7724..23d9a07 100644
--- a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/DirectClientAPITest.java
+++ b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/DirectClientAPITest.java
@@ -35,9 +35,9 @@ import org.junit.Test;
import org.reactivestreams.Publisher;
public class DirectClientAPITest extends ReactiveStreamsTestSupport {
-
- @BindToRegistry("hello")
- private SampleBean bean = new SampleBean();
+
+ @BindToRegistry("hello")
+ private SampleBean bean = new SampleBean();
@Test
public void testFromDirect() throws Exception {
@@ -46,10 +46,7 @@ public class DirectClientAPITest extends ReactiveStreamsTestSupport {
BlockingQueue<Integer> queue = new LinkedBlockingDeque<>();
- Flowable.fromPublisher(data)
- .map(i -> -i)
- .doOnNext(queue::add)
- .subscribe();
+ Flowable.fromPublisher(data).map(i -> -i).doOnNext(queue::add).subscribe();
context.start();
template.sendBody("direct:endpoint", 1);
@@ -69,10 +66,7 @@ public class DirectClientAPITest extends ReactiveStreamsTestSupport {
BlockingQueue<Integer> queue = new LinkedBlockingDeque<>();
- Flowable.fromPublisher(data)
- .map(i -> -i)
- .doOnNext(queue::add)
- .subscribe();
+ Flowable.fromPublisher(data).map(i -> -i).doOnNext(queue::add).subscribe();
template.sendBody("direct:endpoint", 1);
@@ -87,10 +81,7 @@ public class DirectClientAPITest extends ReactiveStreamsTestSupport {
BlockingQueue<String> queue = new LinkedBlockingDeque<>();
- Flowable.just(1, 2, 3)
- .flatMap(camel.to("bean:hello", String.class)::apply)
- .doOnNext(queue::add)
- .subscribe();
+ Flowable.just(1, 2, 3).flatMap(camel.to("bean:hello", String.class)::apply).doOnNext(queue::add).subscribe();
check3HelloInQueue(queue);
}
@@ -101,15 +92,13 @@ public class DirectClientAPITest extends ReactiveStreamsTestSupport {
new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:data")
- .to("mock:result");
+ from("direct:data").to("mock:result");
}
}.addRoutesToCamelContext(context);
context.start();
- Flowable.just(1, 2, 3)
- .subscribe(camel.subscriber("direct:data", Integer.class));
+ Flowable.just(1, 2, 3).subscribe(camel.subscriber("direct:data", Integer.class));
MockEndpoint mock = getMockEndpoint("mock:result");
mock.expectedMessageCount(3);
@@ -129,10 +118,7 @@ public class DirectClientAPITest extends ReactiveStreamsTestSupport {
BlockingQueue<String> queue = new LinkedBlockingDeque<>();
- Flowable.just(1, 2, 3)
- .flatMap(e -> camel.to("bean:hello", e, String.class))
- .doOnNext(queue::add)
- .subscribe();
+ Flowable.just(1, 2, 3).flatMap(e -> camel.to("bean:hello", e, String.class)).doOnNext(queue::add).subscribe();
check3HelloInQueue(queue);
}
@@ -143,12 +129,8 @@ public class DirectClientAPITest extends ReactiveStreamsTestSupport {
BlockingQueue<String> queue = new LinkedBlockingDeque<>();
- Flowable.just(1, 2, 3)
- .flatMap(camel.to("bean:hello")::apply)
- .map(ex -> ex.getOut().getBody(String.class))
- .doOnNext(queue::add)
- .subscribe();
-
+ Flowable.just(1, 2, 3).flatMap(camel.to("bean:hello")::apply).map(ex -> ex.getOut().getBody(String.class)).doOnNext(queue::add).subscribe();
+
check3HelloInQueue(queue);
}
@@ -167,11 +149,7 @@ public class DirectClientAPITest extends ReactiveStreamsTestSupport {
BlockingQueue<String> queue = new LinkedBlockingDeque<>();
- Flowable.just(1, 2, 3)
- .flatMap(e -> camel.to("bean:hello", e))
- .map(ex -> ex.getOut().getBody(String.class))
- .doOnNext(queue::add)
- .subscribe();
+ Flowable.just(1, 2, 3).flatMap(e -> camel.to("bean:hello", e)).map(ex -> ex.getOut().getBody(String.class)).doOnNext(queue::add).subscribe();
check3HelloInQueue(queue);
}
@@ -183,18 +161,13 @@ public class DirectClientAPITest extends ReactiveStreamsTestSupport {
new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:proxy")
- .to("bean:hello")
- .setBody().simple("proxy to ${body}");
+ from("direct:proxy").to("bean:hello").setBody().simple("proxy to ${body}");
}
}.addRoutesToCamelContext(context);
BlockingQueue<String> queue = new LinkedBlockingDeque<>();
- Flowable.just(1, 2, 3)
- .flatMap(camel.to("direct:proxy", String.class)::apply)
- .doOnNext(queue::add)
- .subscribe();
+ Flowable.just(1, 2, 3).flatMap(camel.to("direct:proxy", String.class)::apply).doOnNext(queue::add).subscribe();
for (int i = 1; i <= 3; i++) {
String res = queue.poll(1, TimeUnit.SECONDS);
@@ -209,23 +182,17 @@ public class DirectClientAPITest extends ReactiveStreamsTestSupport {
new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:source")
- .to("direct:stream")
- .setBody().simple("after stream: ${body}")
- .to("mock:dest");
+ from("direct:source").to("direct:stream").setBody().simple("after stream: ${body}").to("mock:dest");
}
}.addRoutesToCamelContext(context);
context.start();
- camel.process("direct:stream", p ->
- Flowable.fromPublisher(p)
- .map(exchange -> {
- int val = exchange.getIn().getBody(Integer.class);
- exchange.getOut().setBody(-val);
- return exchange;
- })
- );
+ camel.process("direct:stream", p -> Flowable.fromPublisher(p).map(exchange -> {
+ int val = exchange.getIn().getBody(Integer.class);
+ exchange.getOut().setBody(-val);
+ return exchange;
+ }));
for (int i = 1; i <= 3; i++) {
template.sendBody("direct:source", i);
@@ -242,26 +209,19 @@ public class DirectClientAPITest extends ReactiveStreamsTestSupport {
}
}
-
@Test
public void testDirectCallFromCamelWithConversion() throws Exception {
new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:source")
- .to("direct:stream")
- .setBody().simple("after stream: ${body}")
- .to("mock:dest");
+ from("direct:source").to("direct:stream").setBody().simple("after stream: ${body}").to("mock:dest");
}
}.addRoutesToCamelContext(context);
context.start();
- camel.process("direct:stream", Integer.class, p ->
- Flowable.fromPublisher(p)
- .map(i -> -i)
- );
+ camel.process("direct:stream", Integer.class, p -> Flowable.fromPublisher(p).map(i -> -i));
for (int i = 1; i <= 3; i++) {
template.sendBody("direct:source", i);