Skip to content

Commit

Permalink
Fix RSocket module according changes in SF
Browse files Browse the repository at this point in the history 
Related to spring-projects/spring-framework#23999

* Since Spring Integration inbound endpoints are generic in their method
signature we can't rely on a new `EMPTY_CONDITION` because it turns on
configuration merge into just `FrameType.REQUEST_FNF` &
`FrameType.REQUEST_RESPONSE`.
So, use `FrameType.REQUEST_FNF`, `FrameType.REQUEST_RESPONSE`,
`FrameType.REQUEST_STREAM` & `FrameType.REQUEST_CHANNEL` explicitly to
cover all the valid request-response models for SI endpoints
* Rework `RSocketOutboundGatewayIntegrationTests` according new logic.
Plus refactor to earlier subscription into `FluxMessageChannel` to
avoid potential race conditions
  • Loading branch information
@artembilan
artembilan committed Nov 19, 2019
1 parent 0f5bd4a commit 25e9459
Show file tree
Hide file tree
Showing 2 changed files with 89 additions and 88 deletions.
8 changes: 7 additions & 1 deletion ...c/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java
View file
Expand Up @@ -32,6 +32,8 @@
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
import org.springframework.util.ReflectionUtils;

import io.rsocket.frame.FrameType;

/**
* The {@link RSocketMessageHandler} extension for Spring Integration needs.
* <p>
Expand Down Expand Up @@ -79,7 +81,11 @@ public boolean detectEndpoints() {
public void addEndpoint(IntegrationRSocketEndpoint endpoint) {
registerHandlerMethod(endpoint, HANDLE_MESSAGE_METHOD,
new CompositeMessageCondition(
RSocketFrameTypeMessageCondition.REQUEST_CONDITION,
new RSocketFrameTypeMessageCondition(
FrameType.REQUEST_FNF,
FrameType.REQUEST_RESPONSE,
FrameType.REQUEST_STREAM,
FrameType.REQUEST_CHANNEL),
new DestinationPatternsMessageCondition(endpoint.getPath(), getRouteMatcher()))); // NOSONAR
}

Expand Down
169 changes: 82 additions & 87 deletions .../springframework/integration/rsocket/outbound/RSocketOutboundGatewayIntegrationTests.java
View file
Expand Up @@ -180,20 +180,23 @@ void serverEcho() {
private void echo(MessageChannel inputChannel, FluxMessageChannel resultChannel,
RSocketRequester rsocketRequester) {

StepVerifier verifier =
StepVerifier.create(
Flux.from(resultChannel)
.map(Message::getPayload)
.cast(String.class))
.expectNext("Hello")
.thenCancel()
.verifyLater();

inputChannel.send(
MessageBuilder.withPayload("Hello")
.setHeader(ROUTE_HEADER, "echo")
.setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse)
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());

StepVerifier.create(
Flux.from(resultChannel)
.map(Message::getPayload)
.cast(String.class))
.expectNext("Hello")
.thenCancel()
.verify();
verifier.verify();
}

@Test
Expand All @@ -209,20 +212,23 @@ void serverEchoAsync() {
private void echoAsync(MessageChannel inputChannel, FluxMessageChannel resultChannel,
RSocketRequester rsocketRequester) {

StepVerifier verifier =
StepVerifier.create(
Flux.from(resultChannel)
.map(Message::getPayload)
.cast(String.class))
.expectNext("Hello async")
.thenCancel()
.verifyLater();

inputChannel.send(
MessageBuilder.withPayload("Hello")
.setHeader(ROUTE_HEADER, "echo-async")
.setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse)
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());

StepVerifier.create(
Flux.from(resultChannel)
.map(Message::getPayload)
.cast(String.class))
.expectNext("Hello async")
.thenCancel()
.verify();
verifier.verify();
}

@Test
Expand All @@ -238,29 +244,25 @@ void serverEchoStream() {
private void echoStream(MessageChannel inputChannel, FluxMessageChannel resultChannel,
RSocketRequester rsocketRequester) {

@SuppressWarnings("unchecked")
StepVerifier verifier =
StepVerifier.create(
Flux.from(resultChannel)
.next()
.map(Message::getPayload)
.flatMapMany((payload) -> (Flux<String>) payload))
.expectNext("Hello 0").expectNextCount(6).expectNext("Hello 7")
.thenCancel()
.verifyLater();

inputChannel.send(
MessageBuilder.withPayload("Hello")
.setHeader(ROUTE_HEADER, "echo-stream")
.setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestStreamOrChannel)
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());

Message<?> resultMessage =
Flux.from(resultChannel)
.blockFirst();

assertThat(resultMessage)
.isNotNull()
.extracting(Message::getPayload)
.isInstanceOf(Flux.class);

@SuppressWarnings("unchecked")
Flux<String> resultStream = (Flux<String>) resultMessage.getPayload();
StepVerifier.create(resultStream)
.expectNext("Hello 0").expectNextCount(6).expectNext("Hello 7")
.thenCancel()
.verify();

verifier.verify();
}

@Test
Expand All @@ -276,28 +278,25 @@ void serverEchoChannel() {
private void echoChannel(MessageChannel inputChannel, FluxMessageChannel resultChannel,
RSocketRequester rsocketRequester) {

@SuppressWarnings("unchecked")
StepVerifier verifier =
StepVerifier.create(
Flux.from(resultChannel)
.next()
.map(Message::getPayload)
.flatMapMany((payload) -> (Flux<String>) payload))
.expectNext("Hello 1 async").expectNextCount(8).expectNext("Hello 10 async")
.thenCancel()
.verifyLater();

inputChannel.send(
MessageBuilder.withPayload(Flux.range(1, 10).map(i -> "Hello " + i))
.setHeader(ROUTE_HEADER, "echo-channel")
.setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestStreamOrChannel)
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());

Message<?> resultMessage =
Flux.from(resultChannel)
.blockFirst();

assertThat(resultMessage)
.isNotNull()
.extracting(Message::getPayload)
.isInstanceOf(Flux.class);

@SuppressWarnings("unchecked")
Flux<String> resultStream = (Flux<String>) resultMessage.getPayload();
StepVerifier.create(resultStream)
.expectNext("Hello 1 async").expectNextCount(8).expectNext("Hello 10 async")
.thenCancel()
.verify();
verifier.verify();
}


Expand All @@ -314,26 +313,21 @@ void serverVoidReturnValue() {
private void voidReturnValue(MessageChannel inputChannel, FluxMessageChannel resultChannel,
RSocketRequester rsocketRequester) {

StepVerifier verifier =
StepVerifier.create(resultChannel)
.expectSubscription()
.expectNoEvent(Duration.ofMillis(100))
.thenCancel()
.verifyLater();

inputChannel.send(
MessageBuilder.withPayload("Hello")
.setHeader(ROUTE_HEADER, "void-return-value")
.setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestStreamOrChannel)
.setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse)
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());

Message<?> resultMessage =
Flux.from(resultChannel)
.blockFirst();

assertThat(resultMessage)
.isNotNull()
.extracting(Message::getPayload)
.isInstanceOf(Flux.class);

Flux<?> resultStream = (Flux<?>) resultMessage.getPayload();
StepVerifier.create(resultStream)
.expectComplete()
.verify();
verifier.verify();
}

@Test
Expand All @@ -349,26 +343,21 @@ void serverVoidReturnValueFromExceptionHandler() {
private void voidReturnValueFromExceptionHandler(MessageChannel inputChannel, FluxMessageChannel resultChannel,
RSocketRequester rsocketRequester) {

StepVerifier verifier =
StepVerifier.create(resultChannel)
.expectSubscription()
.expectNoEvent(Duration.ofMillis(100))
.thenCancel()
.verifyLater();

inputChannel.send(
MessageBuilder.withPayload("bad")
.setHeader(ROUTE_HEADER, "void-return-value")
.setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestStreamOrChannel)
.setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse)
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());

Message<?> resultMessage =
Flux.from(resultChannel)
.blockFirst();

assertThat(resultMessage)
.isNotNull()
.extracting(Message::getPayload)
.isInstanceOf(Flux.class);

Flux<?> resultStream = (Flux<?>) resultMessage.getPayload();
StepVerifier.create(resultStream)
.expectComplete()
.verify();
verifier.verify();
}

@Test
Expand All @@ -384,20 +373,23 @@ void serverHandleWithThrownException() {
private void handleWithThrownException(MessageChannel inputChannel, FluxMessageChannel resultChannel,
RSocketRequester rsocketRequester) {

StepVerifier verifier =
StepVerifier.create(
Flux.from(resultChannel)
.map(Message::getPayload)
.cast(String.class))
.expectNext("Invalid input error handled")
.thenCancel()
.verifyLater();

inputChannel.send(
MessageBuilder.withPayload("a")
.setHeader(ROUTE_HEADER, "thrown-exception")
.setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse)
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());

StepVerifier.create(
Flux.from(resultChannel)
.map(Message::getPayload)
.cast(String.class))
.expectNext("Invalid input error handled")
.thenCancel()
.verify();
verifier.verify();
}

@Test
Expand All @@ -413,20 +405,23 @@ void serverHandleWithErrorSignal() {
private void handleWithErrorSignal(MessageChannel inputChannel, FluxMessageChannel resultChannel,
RSocketRequester rsocketRequester) {

StepVerifier verifier =
StepVerifier.create(
Flux.from(resultChannel)
.map(Message::getPayload)
.cast(String.class))
.expectNext("Invalid input error handled")
.thenCancel()
.verifyLater();

inputChannel.send(
MessageBuilder.withPayload("a")
.setHeader(ROUTE_HEADER, "error-signal")
.setHeader(COMMAND_HEADER, RSocketOutboundGateway.Command.requestResponse)
.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, rsocketRequester)
.build());

StepVerifier.create(
Flux.from(resultChannel)
.map(Message::getPayload)
.cast(String.class))
.expectNext("Invalid input error handled")
.thenCancel()
.verify();
verifier.verify();
}

@Test
Expand Down

0 comments on commit 25e9459

Please sign in to comment.