Add RSocketInboundGateway; refactoring
#2923
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
artembilan
changed the title
RSocketInboundGateway; refactoringRSocketInboundGateway; refactoring
|
This is not fully ready yet. At least tests are needed to cover an introduced functionality. The server inbound side and docs might be deferred into a separate PR... Right now I just want to share what I have so far to gather any possible feedback. Thanks |
garyrussell
suggested changes
May 9, 2019
...ration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java
Show resolved
Hide resolved
| int refCount = refCount(dataBuffer); | ||
| return Mono | ||
| .defer(() -> { | ||
| this.applicationEventPublisher.publishEvent( |
There was a problem hiding this comment.
null check or no-op implementation needed.
| @Override | ||
| public Mono<RSocket> accept(ConnectionSetupPayload setupPayload, RSocket sendingRSocket) { | ||
| IntegrationRSocket rsocket = createRSocket(sendingRSocket); | ||
| rsocket.setApplicationEventPublisher(this.applicationEventPublisher); |
There was a problem hiding this comment.
null check or no-op default implementation.
...ion-rsocket/src/main/java/org/springframework/integration/rsocket/RSocketConnectedEvent.java
Show resolved
Hide resolved
artembilan
force-pushed
the
RSocketInboundGateway
branch
from
3001d03 to
ecee757
Compare
|
Pushed some further implements and implementation. Thanks |
artembilan
changed the title
RSocketInboundGateway; refactoringRSocketInboundGateway; refactoring
garyrussell
suggested changes
May 10, 2019
| public String toString() { | ||
| return "RSocketConnectedEvent{" + | ||
| "destination='" + this.destination + '\'' + | ||
| ", data=" + Arrays.toString(this.data.asByteBuffer().array()) + |
There was a problem hiding this comment.
I don't think rendering a byte[] as a bunch of ints is useful, maybe omit this (or just include the length, perhaps).
|
OK. Now I think it is very close for merging. Pay attention, please, how |
garyrussell
reviewed
May 10, 2019
| @@ -67,7 +65,7 @@ public RSocketRequester getRequester() { | |||
| public String toString() { | |||
| return "RSocketConnectedEvent{" + | |||
| "destination='" + this.destination + '\'' + | |||
| ", data=" + Arrays.toString(this.data.asByteBuffer().array()) + | |||
| ", data length=" + this.data.asByteBuffer().array().length + | |||
There was a problem hiding this comment.
This still seems to be a lot of overhead in a toString().
Why do we need this DataBuffer in the event?
* Extract an `AbstractRSocketConnector` for common client and server connectors logic * Introduce an `IntegrationRSocketAcceptor` and `IntegrationRSocket` for the mapping and handling logic of RSockets and messages in between * Introduce an `IntegrationRSocketEndpoint` marker interface for Inbound Gateway mappings * Add `RSocketInboundGateway` implementation, which is called from the `IntegrationRSocketAcceptor` by the `IntegrationRSocketEndpoint` mapping * Add `RSocketConnectedEvent` to emit when the client is connected to the server. It does not make sense in Spring Integration to delegate such a logic into the `RSocketInboundGateway`
container for connected `RSocketRequester`s from clients * Extract `accept(ConnectionSetupPayload setupPayload, RSocket sendingRSocket)` server logic into an internal `ServerRSocketAcceptor` extension for the `IntegrationRSocketAcceptor` * Address PR comments: - `RSocketConnectedEvent.toString()` - `ApplicationEventPublisherAware` into the `ServerRSocketConnector` - Log RSocket connection if no `this.applicationEventPublisher` * Remove `handleConnectionSetupPayload()` from the `IntegrationRSocket` since it is not delegated to the target handler * Provide reasonable default `RSocketStrategies` for the `AbstractRSocketConnector` and `RSocketInboundGateway` * Add initial `RSocketInboundGatewayIntegrationTests`
* Always `payload.retain()` when we convert `Payload` to `DataBuffer` * Fix `IntegrationRSocketAcceptor.detectEndpoints()` stream logic to really iterate over all the `IntegrationRSocketEndpoint` beans * Fix test to use an explicit `ClientConfig` class for the `@SpringJUnitConfig`: looks like JUnit 5 is OK to scan for `package protected` classes as well * Add request/reply tests into the `RSocketInboundGatewayIntegrationTests` for both server and client sides
artembilan
force-pushed
the
RSocketInboundGateway
branch
from
149f43d to
18e1d59
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
AbstractRSocketConnectorfor common client and server connectors logicIntegrationRSocketAcceptorandIntegrationRSocketfor themapping and handling logic of RSockets and messages in between
IntegrationRSocketEndpointmarker interface for Inbound Gateway mappingsRSocketInboundGatewayimplementation, which is called from theIntegrationRSocketAcceptorby theIntegrationRSocketEndpointmappingRSocketConnectedEventto emit when the client is connected to the server.It does not make sense in Spring Integration to delegate such a logic into the
RSocketInboundGateway