Add a basic example to Spring Integration with Reactive Streams #3718

migroskub · 2022-02-04T16:28:53Z

I've seen that there is no clear example of how to quickly get started with Spring Integration and Reactive Streams. This PR tries to help with this.

I'm trying to help mainly for questions like this.

I've seen that there is no clear example of how to quickly get started with Spring Integration and Reactive Streams. This PR tries to help with this.

pivotal-cla · 2022-02-04T16:28:56Z

@migroskub Please sign the Contributor License Agreement!

Click here to manually synchronize the status of this Pull Request.

See the FAQ for frequently asked questions.

pivotal-cla · 2022-02-04T16:31:21Z

@migroskub Thank you for signing the Contributor License Agreement!

artembilan

Thank you for a sample!

Please, consider to move it before A reactive outbound channel adapter paragraph.

Any samples for the outbound part?

The R2dbcMessageHandler should be a good one took into.

artembilan · 2022-02-09T20:18:58Z

@migroskub ,

Gentle ping.
Otherwise we will take it from here ourselves and will merge it as we think it should be.

Thank you!

migroskub · 2022-02-11T16:04:08Z

I've committed the changes as you've suggested. I have 3 notes after reading some pages from the documentation.

There are "hidden" pages. Pages that are accessible only from links, but not from the main page. For example, take a look at the functions support page, which can't be accessed from the main page. I've seen more like this.
There are lots of grammar mistakes. It makes it very hard to read. I suggest starting documenting with Grammarly as a helper tool, and I think it would be a pretty good idea to let this tool walk through the existing pages and suggest fixes.
The most important - It's pretty hard to understand what are the correct interfaces that we should use in each case of writing channel adapters with the Java DSL. I think there should be a better guide for writing custom channel adapters (since there are not so many channel adapters, writing custom ones is a pretty common task, if not the most common task when using Spring Integration). I'm not talking here only about Reactive Channel Adapters, but also of synchronous ones. The reference doesn't mention any of the JavaDSL channel adapter interfaces, so that's super hard to understand the purpose of each interface and what fits my needs. The only solution what diving into the source code, which has inconsistency between channel adapters (some channel adapters name is XInboundChannelAdapter, and some channel adapters name is XMessageProducer. I know that there may be some reasons and conventions to it, but this whole part is just not documented). I'm saying this because I think it will be hard for newbies to contribute channel adapters this way. I've PR'd because I'm trying to help a little with this to the next developers who'll come by this challenge.

artembilan · 2022-02-11T18:00:55Z

Thank you for feedback, @migroskub !

The Function Support is part of Messaging Endpoints chapter. So, it is indeed cannot be visible from the main page. You need to go to this one to see it in the content on the left: https://docs.spring.io/spring-integration/reference/html/messaging-endpoints.html#messaging-endpoints-chapter. Same applies for may others paragraphs which are part for bigger chapters. Perhaps if you go to a single page, it would become clearer for you: https://docs.spring.io/spring-integration/docs/current/reference/html/index-single.html
I will take a look into that tool.
Please, raise a separate GH issue for that channel adapters development concern.

artembilan

Thanks

artembilan · 2022-02-11T18:07:57Z

src/reference/asciidoc/reactive-streams.adoc

+}
+```
+
+And now we will be able to use it:


I think this flow sample belongs to the producer section above.
The CustomReactiveMessageHandler deserves its own independent usage sample.
Otherwise it is confusing a little bit.

Can be resolved

migroskub · 2022-02-11T19:05:39Z

src/reference/asciidoc/reactive-streams.adoc

+
+    @Override
+    protected Mono<Void> handleMessageInternal(Message<?> message) {
+        return Mono.fromSupplier(() -> this.queryTypeExpression.getValue(this.evaluationContext, message, Type.class))


@artembilan I'm struggling with the CustomReactiveMessageHandler implementation - can you simplify this to a more basic supplier usage that will be practical and won't require evaluationContext?

Just do -> message.getHeaders().get("queryType", Type.class).
And no queryTypeExpression in the code at all!

…dapter

artembilan · 2022-02-11T19:21:25Z

src/reference/asciidoc/reactive-streams.adoc

-A returned reactive type can be subscribed immediately if we are in one-way, fire-and-forget scenario, or it is propagated downstream (request-reply scenarios) for further integration flow or an explicit subscription in the target business logic, but still downstream preserving reactive streams semantics.
+An example for a reactive inbound channel adapter:
+```java
+public class CustomReactiveMessageSource extends MessageProducerSupport {


It's not correct to call it MessageSource. Just because it does not implement a MessageSource. And definitely a reactive publisher could not be wrapped to the MessageSource anyway.
Better to call it CustomReactiveMessageProducer since you really do it correctly extending MessageProducerSupport and using its subscribeToPublisher() from the doStart() impl.

artembilan · 2022-02-11T19:21:38Z

src/reference/asciidoc/reactive-streams.adoc

+
+    private final CustomReactiveSource customReactiveSource;
+
+    public CustomReactiveSourceMessageProducer(CustomReactiveSource customReactiveSource) {


I see that constructor for the class you called properly. So, please, revise the name of the class 😄

artembilan · 2022-02-11T19:23:59Z

src/reference/asciidoc/reactive-streams.adoc

+
+    @Override
+    protected Mono<Void> handleMessageInternal(Message<?> message) {
+        return Mono.fromSupplier(() -> this.queryTypeExpression.getValue(this.evaluationContext, message, Type.class))


Just do -> message.getHeaders().get("queryType", Type.class).
And no queryTypeExpression in the code at all!

…e fix and queryTypeExpression removement

artembilan · 2022-02-11T20:48:13Z

src/reference/asciidoc/reactive-streams.adoc

+                    MessageBuilder
+                    .withPayload(event.getBody())
+                    .setHeader(MyReactiveHeaders.SOURCE_NAME, event.getSourceName())
+                    .build());)


The ; is wrong here.
I would prefer to have this Flux extracted as a variable. It's OK to have it in this bean method definition.

Do you mean like this?

public class MainFlow { @Bean public IntegrationFlow buildFlow() { Flux<Message<?>> myFlux = this.customReactiveSource .map(event - > MessageBuilder .withPayload(event.getBody()) .setHeader(MyReactiveHeaders.SOURCE_NAME, event.getSourceName()) .build()); return IntegrationFlows.from(myFlux) .handle(outputChannel) .get(); } }

Yep! This is good. It does give us a clue what is that customReactiveSource about and which from() factory is going to be for building an IntegrationFlow.

migroskub · 2022-02-11T21:22:42Z

I think the PR is ready

artembilan · 2022-02-11T21:37:48Z

src/reference/asciidoc/reactive-streams.adoc

+
+    @Override
+    protected void doStart() {
+        Flux<Message<<?>> messageFlux =


Typo for diamond definition. Has lead to this error:

Execution failed for task ':checkAsciidocLinks'. > [*** Anchor reference errors found:, Anchor '?' not found in 'reactive-streams.adoc', if in another file, it needs to be qualified with './fileName.adoc#']

But originally I definitely have missed it 😄

I haven't exactly understood this. What should be fixed?

This code is wrong: Flux<Message<<?>>.
Must be Flux<Message<?>>. You have to have exact number of closing bracket as many as you have opening.
The error is misleading, but it really helped me to spot the typo in your code.

…r example

artembilan · 2022-02-13T23:44:53Z

src/reference/asciidoc/reactive-streams.adoc

+
+    @Override
+    protected void doStart() {
+        Flux<Message<<?>> messageFlux =


This code is wrong: Flux<Message<<?>>.
Must be Flux<Message<?>>. You have to have exact number of closing bracket as many as you have opening.
The error is misleading, but it really helped me to spot the typo in your code.

artembilan · 2022-02-14T22:10:15Z

Thank you, @migroskub; looking forward for more!

Add a basic example to Spring Integration with Reactive Streams

509d7ff

I've seen that there is no clear example of how to quickly get started with Spring Integration and Reactive Streams. This PR tries to help with this.

artembilan requested changes Feb 4, 2022

View reviewed changes

artembilan added this to the 6.0 M2 milestone Feb 8, 2022

artembilan added the type: documentation label Feb 8, 2022

Add outbound channel adapter example

63eb61f

artembilan requested changes Feb 11, 2022

View reviewed changes

migroskub added 3 commits February 11, 2022 20:55

Add an example for the CustomReactiveMessageHandler usage

0e259c1

Emphasize that the examples are for Reactive Streams

b86636e

Fix a missing ` letter that caused the text to get inside the code block

7d0da5f

migroskub commented Feb 11, 2022

View reviewed changes

Emphasize that the first example is an event driven inbound channel a…

6da758e

…dapter

artembilan requested changes Feb 11, 2022

View reviewed changes

Addition of a no channel adapter example, inbound channel adapter nam…

e7663d0

…e fix and queryTypeExpression removement

artembilan requested changes Feb 11, 2022

View reviewed changes

Fix the Java DSL example

3cec200

migroskub requested a review from artembilan February 11, 2022 21:23

artembilan requested changes Feb 11, 2022

View reviewed changes

Remove the queryTypeExpression usage from the outbound channel adapte…

8bfe182

…r example

artembilan requested changes Feb 13, 2022

View reviewed changes

Fix a redundant question mark from the example code

d7daabd

artembilan approved these changes Feb 14, 2022

View reviewed changes

artembilan merged commit 285c380 into spring-projects:main Feb 14, 2022


		private final CustomReactiveSource customReactiveSource;

		public CustomReactiveSourceMessageProducer(CustomReactiveSource customReactiveSource) {

Add a basic example to Spring Integration with Reactive Streams #3718

Add a basic example to Spring Integration with Reactive Streams #3718

Uh oh!

Conversation

migroskub commented Feb 4, 2022

Uh oh!

pivotal-cla commented Feb 4, 2022

Uh oh!

pivotal-cla commented Feb 4, 2022

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

artembilan commented Feb 9, 2022

Uh oh!

migroskub commented Feb 11, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

artembilan commented Feb 11, 2022

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

migroskub Feb 11, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

migroskub commented Feb 11, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

artembilan commented Feb 14, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

migroskub commented Feb 11, 2022 •

edited

Loading

migroskub Feb 11, 2022 •

edited

Loading