SpringCloud/spring-cloud-function
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add RSocketRequester configuration for tests

Browse Source
This commit is contained in:
Oleg Zhurakousky
2020-08-26 14:35:34 +02:00
parent c1240ebb91
commit 18b0f436d3
4 changed files with 91 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketConnectionUtils.java
View File

@@ -1,55 +0,0 @@
/*
* Copyright 2020-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.function.rsocket;
import java.net.InetSocketAddress;
import java.time.Duration;
import io.rsocket.RSocket;
import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketConnector;
import io.rsocket.core.RSocketServer;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.transport.netty.server.TcpServerTransport;
import reactor.core.Disposable;
import reactor.util.retry.Retry;
import reactor.util.retry.RetrySpec;
import org.springframework.lang.Nullable;
/**
*
* @author Oleg Zhurakousky
* @since 3.1
*/
public abstract class RSocketConnectionUtils {
public static Disposable createServerSocket(RSocket rsocket, InetSocketAddress address) {
Disposable server = RSocketServer.create(SocketAcceptor.with(rsocket))
.bind(TcpServerTransport.create(address)) //TODO transport can actually be selected based on address (local or tcp)??
.subscribe();
return server;
}
public static RSocket createClientSocket(InetSocketAddress address, @Nullable RetrySpec retrySpec) {
RSocket socket = RSocketConnector.connectWith(TcpClientTransport.create(address)).log()
.retryWhen(retrySpec == null ? Retry.backoff(5, Duration.ofSeconds(1)) : retrySpec)
.block();
return socket;
}
}

60
spring-cloud-function-rsocket/src/test/java/org/springframework/cloud/function/rsocket/RSocketAutoConfigurationTests.java
View File

@@ -16,12 +16,10 @@
package org.springframework.cloud.function.rsocket;
import java.net.InetSocketAddress;
import java.util.function.Consumer;
import java.util.function.Function;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.util.DefaultPayload;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
@@ -31,8 +29,11 @@ import reactor.test.StepVerifier;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.util.SocketUtils;
@@ -45,13 +46,13 @@ public class RSocketAutoConfigurationTests {
@Test
public void testImperativeFunctionAsRequestReply() throws Exception {
int port = SocketUtils.findAvailableTcpPort();
new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
ApplicationContext context = new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
"--logging.level.org.springframework.cloud.function=DEBUG",
"--spring.cloud.function.definition=uppercase",
"--spring.rsocket.server.port=" + port);
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Mono<String> result = socket.requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
RSocketRequester requester = context.getBean(RSocketRequester.class);
Mono<String> result = requester.rsocket().requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
StepVerifier
.create(result)
@@ -63,13 +64,13 @@ public class RSocketAutoConfigurationTests {
@Test
public void testImperativeFunctionAsRequestStream() throws Exception {
int port = SocketUtils.findAvailableTcpPort();
new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
ApplicationContext context = new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
"--logging.level.org.springframework.cloud.function=DEBUG",
"--spring.cloud.function.definition=uppercase",
"--spring.rsocket.server.port=" + port);
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Flux<String> result = socket.requestStream(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
RSocketRequester requester = context.getBean(RSocketRequester.class);
Flux<String> result = requester.rsocket().requestStream(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
StepVerifier
.create(result)
@@ -81,13 +82,13 @@ public class RSocketAutoConfigurationTests {
@Test
public void testImperativeFunctionAsRequestChannel() throws Exception {
int port = SocketUtils.findAvailableTcpPort();
new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
ApplicationContext context = new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
"--logging.level.org.springframework.cloud.function=DEBUG",
"--spring.cloud.function.definition=uppercase",
"--spring.rsocket.server.port=" + port);
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Flux<String> result = socket.requestChannel(Flux.just(
RSocketRequester requester = context.getBean(RSocketRequester.class);
Flux<String> result = requester.rsocket().requestChannel(Flux.just(
DefaultPayload.create("\"Ricky\""),
DefaultPayload.create("\"Julien\""),
DefaultPayload.create("\"Bubbles\""))
@@ -105,14 +106,13 @@ public class RSocketAutoConfigurationTests {
@Test
public void testReactiveFunctionAsRequestReply() throws Exception {
int port = SocketUtils.findAvailableTcpPort();
new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
ApplicationContext context = new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
"--logging.level.org.springframework.cloud.function=DEBUG",
"--spring.cloud.function.definition=uppercaseReactive",
"--spring.rsocket.server.port=" + port);
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Mono<String> result = socket.requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
RSocketRequester requester = context.getBean(RSocketRequester.class);
Mono<String> result = requester.rsocket().requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
StepVerifier
.create(result)
@@ -124,14 +124,13 @@ public class RSocketAutoConfigurationTests {
@Test
public void testReactiveFunctionAsRequestStream() throws Exception {
int port = SocketUtils.findAvailableTcpPort();
new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
ApplicationContext context = new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
"--logging.level.org.springframework.cloud.function=DEBUG",
"--spring.cloud.function.definition=uppercaseReactive",
"--spring.rsocket.server.port=" + port);
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Flux<String> result = socket.requestStream(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
RSocketRequester requester = context.getBean(RSocketRequester.class);
Flux<String> result = requester.rsocket().requestStream(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
StepVerifier
.create(result)
@@ -143,14 +142,13 @@ public class RSocketAutoConfigurationTests {
@Test
public void testReactiveFunctionAsRequestChannel() throws Exception {
int port = SocketUtils.findAvailableTcpPort();
new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
ApplicationContext context = new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
"--logging.level.org.springframework.cloud.function=DEBUG",
"--spring.cloud.function.definition=uppercaseReactive",
"--spring.rsocket.server.port=" + port);
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Flux<String> result = socket.requestChannel(Flux.just(
RSocketRequester requester = context.getBean(RSocketRequester.class);
Flux<String> result = requester.rsocket().requestChannel(Flux.just(
DefaultPayload.create("\"Ricky\""),
DefaultPayload.create("\"Julien\""),
DefaultPayload.create("\"Bubbles\""))
@@ -175,13 +173,14 @@ public class RSocketAutoConfigurationTests {
"--spring.cloud.function.definition=uppercase|concat",
"--spring.rsocket.server.port=" + portA);
new SpringApplicationBuilder(AdditionalFunctionConfiguration.class).web(WebApplicationType.NONE).run(
ApplicationContext bContext = new SpringApplicationBuilder(AdditionalFunctionConfiguration.class).web(WebApplicationType.NONE).run(
"--logging.level.org.springframework.cloud.function=DEBUG",
"--spring.cloud.function.definition=reverse>localhost:" + portA + "|wrap",
"--spring.rsocket.server.port=" + portB);
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", portB), null);
Mono<String> result = socket.requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
RSocketRequester requester = bContext.getBean(RSocketRequester.class);
Mono<String> result = requester.rsocket().requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
StepVerifier
.create(result)
.expectNext("\"(OLLEHOLLEH)\"")
@@ -192,14 +191,13 @@ public class RSocketAutoConfigurationTests {
@Test
public void testRequestChannelFunction() throws Exception {
int port = SocketUtils.findAvailableTcpPort();
new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
ApplicationContext context = new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
"--logging.level.org.springframework.cloud.function=DEBUG",
"--spring.cloud.function.definition=uppercaseReactive",
"--spring.rsocket.server.port=" + port);
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Flux<String> result = socket.requestChannel(Flux.just(
RSocketRequester requester = context.getBean(RSocketRequester.class);
Flux<String> result = requester.rsocket().requestChannel(Flux.just(
DefaultPayload.create("\"Ricky\""),
DefaultPayload.create("\"Julien\""),
DefaultPayload.create("\"Bubbles\""))
@@ -239,6 +237,7 @@ public class RSocketAutoConfigurationTests {
@EnableAutoConfiguration
@Configuration
@Import(RSocketTestConfiguration.class)
public static class SampleFunctionConfiguration {
@Bean
public Function<String, String> uppercase() {
@@ -277,6 +276,7 @@ public class RSocketAutoConfigurationTests {
@EnableAutoConfiguration
@Configuration
@Import(RSocketTestConfiguration.class)
public static class AdditionalFunctionConfiguration {
@Bean
public Function<String, String> reverse() {

61
spring-cloud-function-rsocket/src/test/java/org/springframework/cloud/function/rsocket/RSocketTestConfiguration.java Normal file
View File

@@ -0,0 +1,61 @@
/*
* Copyright 2020-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.function.rsocket;
import java.net.InetSocketAddress;
import java.time.Duration;
import io.rsocket.RSocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.transport.netty.client.TcpClientTransport;
import reactor.util.retry.Retry;
import reactor.util.retry.RetrySpec;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.core.env.Environment;
import org.springframework.lang.Nullable;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.util.Assert;
import org.springframework.util.MimeTypeUtils;
/**
*
* @author Oleg Zhurakousky
*
*/
@Configuration
public class RSocketTestConfiguration {
@Bean
@Scope("prototype")
RSocketRequester rSocketRequester(RSocketStrategies rSocketStrategies, Environment environment,
@Nullable RetrySpec retrySpec) {
String port = environment.getProperty("spring.rsocket.server.port");
Assert.hasText(port, "'spring.rsocket.server.port' must be specified");
String host = environment.getProperty("spring.rsocket.server.address", "localhost");
RSocket socket = RSocketConnector
.connectWith(
TcpClientTransport.create(InetSocketAddress.createUnresolved(host, Integer.parseInt(port))))
.log()
.retryWhen(retrySpec == null ? Retry.backoff(5, Duration.ofSeconds(1)) : retrySpec).block();
return RSocketRequester.wrap(socket, MimeTypeUtils.APPLICATION_JSON, MimeTypeUtils.APPLICATION_JSON,
rSocketStrategies);
}
}