Skip to content

Add initial support for RSockets #2902

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Merged
merged 3 commits into from
Apr 25, 2019
Merged

Add initial support for RSockets #2902

merged 3 commits into from
Apr 25, 2019

Conversation

artembilan
Copy link
Member

  • Add spring-integration-rsocket module and respective dependencies
  • Implement RSocketOutboundGateway based on the Spring Messaging
    RSocketRequester.
    This component supports dynamic RSocket properties via expressions
    against request message.
    to handle Publisher for requests, it must be present in the request
    message payload instead of FluxMessageChannel upstream, since the
    last one just flattens events to be handled in the MessageHandler one
    by one.
    The result Mono is subscribed downstream in the FluxMessageChannel
    or directly by the AbstractReplyProducingMessageHandler.
    If result is a Flux it is just wrapped into the Mono to be processed
    downstream by end-user code.
    The point is that these request/replies are volatile and live in the
    particular context meanwhile a FluxMessageChannel is long living
    publisher in the application context boundaries.
  • The RSocketOutboundGatewayIntegrationTests is an adapted copy of
    RSocketClientToServerIntegrationTests from Spring Messaging
  • Add doOnError() into the Flux created in the
    AbstractMessageProducingHandler for Publisher replies

@artembilan
Add initial support for RSockets
6f763c2 
* Add `spring-integration-rsocket` module and respective dependencies
* Implement `RSocketOutboundGateway` based on the Spring Messaging
`RSocketRequester`.
This component supports dynamic RSocket properties via expressions
against request message.
to handle `Publisher` for requests, it must be present in the request
message `payload` instead of `FluxMessageChannel` upstream, since the
last one just flattens events to be handled in the `MessageHandler` one
by one.
The result `Mono` is subscribed downstream in the `FluxMessageChannel`
or directly by the `AbstractReplyProducingMessageHandler`.
If result is a `Flux` it is just wrapped into the `Mono` to be processed
downstream by end-user code.
The point is that these request/replies are volatile and live in the
particular context meanwhile a `FluxMessageChannel` is long living
publisher in the application context boundaries.
* The `RSocketOutboundGatewayIntegrationTests` is an adapted copy of
`RSocketClientToServerIntegrationTests` from Spring Messaging
* Add `doOnError()` into the `Flux` created in the
`AbstractMessageProducingHandler` for `Publisher` replies
@artembilan
Copy link
Member Author

@garyrussell ,

You can merge if it is OK: other features and Docs can be added in the upcoming PRs.
Or let me know and I'll continue in this branch with the rest of stuff

Thanks

garyrussell
garyrussell suggested changes Apr 25, 2019
View reviewed changes
Copy link
Contributor

@garyrussell garyrussell left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good work! Just a few issues.

build.gradle Outdated
@@ -596,6 +598,18 @@ project('spring-integration-rmi') {
}
}

project('spring-integration-rsocket') {
description = 'Spring Integration RSockets Support'
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It appears this should be singular RSocket.

...t/src/main/java/org/springframework/integration/rsocket/outbound/RSocketOutboundGateway.java Outdated
protected void doInit() {
super.doInit();
RSocketRequester.Builder builder = RSocketRequester.builder();
if (this.factoryConfigurer != null) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's probably cleaner to default to a no-op configurers and assert not null in setters.

...org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java Outdated
@DirtiesContext
public class RSocketOutboundGatewayIntegrationTests {

private static final int PORT = SocketUtils.findAvailableTcpPort();
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there no support for port 0 so the OS allocates the port?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I just can't swallow code complexity in Netty, so it isn't clear if it is there.
@rstoyanchev just uses a static 7000 port on the matter 😢

/CC @smaldini

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's just that we went through a process of converting all our TCP module tests to use port 0 so the OS chooses the port. These "findAvailablePort" schemes, while they work most of the time, we occasionally saw failures (BindException) on CI builds. A hard-wired port # doesn't sound good either.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK. I have figured out how to use 0 for port.

See latest commit.

Thanks

@artembilan
* Use singular for the RSocket term
fc8e83a 
* Use no-op `Consumer` for the `strategiesConfigurer` and
`factoryConfigurer` in the `RSocketOutboundGateway` and also
`Assert.notNull()` in the appropriate setters to avoid null check during
`RSocketRequester.builder()` initialization
* Use `TcpServer.create().port(0)` in the
`RSocketOutboundGatewayIntegrationTests` to allow to select free OS port
and bind into it.
The selected port is used later for client configuration in the
`RSocketOutboundGateway` bean definition
garyrussell
garyrussell reviewed Apr 25, 2019
View reviewed changes
...org/springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java Outdated
@@ -74,7 +74,7 @@
@DirtiesContext
public class RSocketOutboundGatewayIntegrationTests {

private static final int PORT = SocketUtils.findAvailableTcpPort();
private static int PORT;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This should no longer be static.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

??? It is initialized from the static @BeforeAll and it won't be available in the ClientConfig below which has to be a static as well.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry; I meant it should no longer be upper case.

@garyrussell garyrussell merged commit 89e11f2 into spring-projects:master Apr 25, 2019
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@garyrussell garyrussell garyrussell requested changes

Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
2 participants
@artembilan @garyrussell