From f725b5aeb84a5da94370c675e7cc442f1c2f9980 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 7 Jul 2020 19:30:44 +0200 Subject: [PATCH] Add StepVerifier to RSocket tests --- spring-cloud-function-rsocket/pom.xml | 5 + .../RSocketAutoConfigurationTests.java | 109 +++++++++++------- 2 files changed, 73 insertions(+), 41 deletions(-) diff --git a/spring-cloud-function-rsocket/pom.xml b/spring-cloud-function-rsocket/pom.xml index 7b6a57766..69e88b1ae 100644 --- a/spring-cloud-function-rsocket/pom.xml +++ b/spring-cloud-function-rsocket/pom.xml @@ -43,6 +43,11 @@ spring-boot-configuration-processor true + + io.projectreactor + reactor-test + test + diff --git a/spring-cloud-function-rsocket/src/test/java/org/springframework/cloud/function/rsocket/RSocketAutoConfigurationTests.java b/spring-cloud-function-rsocket/src/test/java/org/springframework/cloud/function/rsocket/RSocketAutoConfigurationTests.java index 26c5aebde..a89847011 100644 --- a/spring-cloud-function-rsocket/src/test/java/org/springframework/cloud/function/rsocket/RSocketAutoConfigurationTests.java +++ b/spring-cloud-function-rsocket/src/test/java/org/springframework/cloud/function/rsocket/RSocketAutoConfigurationTests.java @@ -25,18 +25,24 @@ import io.rsocket.RSocket; import io.rsocket.core.RSocketConnector; import io.rsocket.transport.netty.client.TcpClientTransport; import io.rsocket.util.DefaultPayload; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; import reactor.util.retry.Retry; +import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.util.SocketUtils; -import static org.assertj.core.api.Assertions.assertThat; /** * @@ -47,59 +53,80 @@ public class RSocketAutoConfigurationTests { @Test public void testRequestReplyFunction() throws Exception { - new SpringApplicationBuilder(SampleFunctionConfiguration.class).run( - "--logging.level.org.springframework.cloud.function=DEBUG", - "--spring.cloud.function.definition=uppercase", - "--spring.cloud.function.rsocket.bind-address=localhost", - "--spring.cloud.function.rsocket.bind-port=12345"); - - RSocket socket = RSocketConnector.connectWith(TcpClientTransport.create("localhost", 12345)).log() - .retryWhen(Retry.backoff(5, Duration.ofSeconds(1))).block(); - socket.requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8).subscribe(System.out::println); - - Thread.sleep(1000); -// assertThat(result).isEqualTo("\"HELLO\""); - } - - @Test - public void testRequestChannelFunction() throws Exception { - new SpringApplicationBuilder(SampleFunctionConfiguration.class).run( - "--logging.level.org.springframework.cloud.function=DEBUG", - "--spring.cloud.function.definition=uppercaseReactive", - "--spring.cloud.function.rsocket.bind-address=localhost", - "--spring.cloud.function.rsocket.bind-port=12345"); - - RSocket socket = RSocketConnector.connectWith(TcpClientTransport.create("localhost", 12345)).log() - .retryWhen(Retry.backoff(5, Duration.ofSeconds(1))).block(); - socket.requestChannel(Flux.just(DefaultPayload.create("\"Ricky\""), DefaultPayload.create("\"Julien\""), DefaultPayload.create("\"Bubbles\""))) - .subscribe(System.out::println); - - Thread.sleep(1000); -// assertThat(result).isEqualTo("\"HELLO\""); - } - - @Test - public void testRequestReplyFunctionWithComposition() throws Exception { + int port = SocketUtils.findAvailableTcpPort(); new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run( "--logging.level.org.springframework.cloud.function=DEBUG", "--spring.cloud.function.definition=uppercase", "--spring.cloud.function.rsocket.bind-address=localhost", - "--spring.cloud.function.rsocket.bind-port=12345"); + "--spring.cloud.function.rsocket.bind-port=" + port); + + RSocket socket = RSocketConnector.connectWith(TcpClientTransport.create("localhost", port)).log() + .retryWhen(Retry.backoff(5, Duration.ofSeconds(1))).block(); + Mono result = socket.requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8); + + StepVerifier + .create(result) + .expectNext("\"HELLO\"") + .expectComplete() + .verify(); + } + + @Test + public void testRequestReplyFunctionWithComposition() throws Exception { + int portA = SocketUtils.findAvailableTcpPort(); + int portB = SocketUtils.findAvailableTcpPort(); + new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run( + "--logging.level.org.springframework.cloud.function=DEBUG", + "--spring.cloud.function.definition=uppercase", + "--spring.cloud.function.rsocket.bind-address=localhost", + "--spring.cloud.function.rsocket.bind-port=" + portA); new SpringApplicationBuilder(AdditionalFunctionConfiguration.class).web(WebApplicationType.NONE).run( "--logging.level.org.springframework.cloud.function=DEBUG", "--spring.cloud.function.definition=reverse", "--spring.cloud.function.rsocket.bind-address=localhost", - "--spring.cloud.function.rsocket.bind-port=12346", + "--spring.cloud.function.rsocket.bind-port=" + portB, "--spring.cloud.function.rsocket.target-address=localhost", - "--spring.cloud.function.rsocket.target-port=12345"); + "--spring.cloud.function.rsocket.target-port=" + portA); - RSocket socket = RSocketConnector.connectWith(TcpClientTransport.create("localhost", 12346)).log() + RSocket socket = RSocketConnector.connectWith(TcpClientTransport.create("localhost", portB)).log() .retryWhen(Retry.backoff(5, Duration.ofSeconds(1))).block(); - socket.requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8).subscribe(System.out::println); - Thread.sleep(1000); -// assertThat(result).isEqualTo("\"OLLEH\""); + Mono result = socket.requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8); + StepVerifier + .create(result) + .expectNext("\"OLLEH\"") + .expectComplete() + .verify(); } + @Test + public void testRequestChannelFunction() throws Exception { + int port = SocketUtils.findAvailableTcpPort(); + new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run( + "--logging.level.org.springframework.cloud.function=DEBUG", + "--spring.cloud.function.definition=uppercaseReactive", + "--spring.cloud.function.rsocket.bind-address=localhost", + "--spring.cloud.function.rsocket.bind-port=" + port); + + RSocket socket = RSocketConnector.connectWith(TcpClientTransport.create("localhost", port)).log() + .retryWhen(Retry.backoff(5, Duration.ofSeconds(1))).block(); + Flux result = socket.requestChannel(Flux.just( + DefaultPayload.create("\"Ricky\""), + DefaultPayload.create("\"Julien\""), + DefaultPayload.create("\"Bubbles\"")) + ) + .map(Payload::getDataUtf8); + + StepVerifier + .create(result) + .expectNext("\"RICKY\"") + .expectNext("\"JULIEN\"") + .expectNext("\"BUBBLES\"") + .expectComplete() + .verify(); + } + + + // @Test // public void testFireAndForgetConsumer() throws Exception { // new SpringApplicationBuilder(SampleFunctionConfiguration.class)