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

Cleanup 'requestChannel' logic

Browse Source 
Add RSocketConnectionUtils to deal with connectivity logic
This commit is contained in:
Oleg Zhurakousky
2020-07-07 19:41:40 +02:00
parent f725b5aeb8
commit bc7376f2a6
7 changed files with 466 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files

185
spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketAutoConfiguration.java
View File

@@ -16,33 +16,26 @@
package org.springframework.cloud.function.rsocket;
import java.lang.reflect.Type;
import java.nio.ByteBuffer;
import java.util.function.Function;
import java.net.InetSocketAddress;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketServer;
import io.rsocket.transport.netty.server.TcpServerTransport;
import io.rsocket.util.DefaultPayload;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.FunctionProperties;
import org.springframework.cloud.function.context.FunctionRegistration;
import org.springframework.cloud.function.context.FunctionRegistry;
import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -58,113 +51,17 @@ public class RSocketAutoConfiguration {
private static Log logger = LogFactory.getLog(RSocketAutoConfiguration.class);
@Bean
public FunctionToDestinationBinder functionToDestinationBinder(FunctionCatalog functionCatalog,
FunctionProperties functionProperties, RSocketFunctionProperties rSocketFunctionProperties) {
return new FunctionToDestinationBinder(functionCatalog, functionProperties, rSocketFunctionProperties);
}
@SuppressWarnings("rawtypes")
private static RSocket buildRSocket(String definition, Type functionType, Function function) {
RSocket clientRSocket = null;
// if (isFireAndForget(functionType)) { // fire-and-forget
// RSocket rsocket = new RSocket() { // imperative function or Function<?, Mono> = requestResponse
// @Override
// public Mono<Void> fireAndForget(Payload p) {
// System.out.println("Invoking fireAndForget");
// invokeFunction(p, function);
// return Mono.empty();
// }
// };
// return rsocket;
// }
if (isRequestRepply(functionType)) {
if (logger.isDebugEnabled()) {
logger.debug("Mapping function '" + definition + "' as RSocket `requestResponse`.");
}
clientRSocket = new RSocket() { // imperative function or Function<?, Mono> = requestResponse
@SuppressWarnings("unchecked")
@Override
public Mono<Payload> requestResponse(Payload payload) {
Object result = invokeFunction(payload, function);
Mono<Payload> invocationResult = ((Mono<Message<byte[]>>) result).map(message -> DefaultPayload.create(message.getPayload()));
return invocationResult;
}
};
}
else if (isRequestChannel(functionType)) {
if (logger.isDebugEnabled()) {
logger.debug("Mapping function '" + definition + "' as RSocket `requestChannel`.");
}
clientRSocket = new RSocket() { // imperative function or Function<?, Mono> = requestResponse
@SuppressWarnings("unchecked")
@Override
public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
return Flux.from(payloads).transform(flux -> {
return flux.map(payload -> {
ByteBuffer buffer = payload.getData();
byte[] rawData = new byte[buffer.remaining()];
buffer.get(rawData);
if (payload.hasMetadata()) {
String metadata = payload.getMetadataUtf8(); // TODO see what to do with it
}
Message<byte[]> inputMessage = MessageBuilder.withPayload(rawData).build();
return inputMessage;
});
})
.transform(function)
.transform(resultFlux -> {
return ((Flux<Message<byte[]>>) resultFlux).map(message -> {
Payload p = DefaultPayload.create(message.getPayload());
return p;
});
});
}
};
}
else {
throw new UnsupportedOperationException("Only RSocket 'requestResponse' is currently supported");
}
Assert.notNull(clientRSocket, "Failed to create RSocket for function '" + definition + "'");
return clientRSocket;
}
@SuppressWarnings("unchecked")
private static Mono<Message<byte[]>> invokeFunction(Payload p, Function<Message<byte[]>, Mono<Message<byte[]>>> function) {
ByteBuffer buffer = p.getData();
byte[] rawData = new byte[buffer.remaining()];
buffer.get(rawData);
if (p.hasMetadata()) {
String metadata = p.getMetadataUtf8(); // TODO see what to do with it
}
Message<byte[]> inputMessage = MessageBuilder.withPayload(rawData).build();
Object result = function.apply(inputMessage);
return result instanceof Mono ? (Mono<Message<byte[]>>) result : Mono.just((Message<byte[]>) result);
}
private static boolean isFireAndForget(Type functionType) {
Type inputType = FunctionTypeUtils.getInputType(functionType, 0);
return FunctionTypeUtils.isConsumer(functionType) && !FunctionTypeUtils.isPublisher(inputType);
}
private static boolean isRequestRepply(Type functionType) {
Type inputType = FunctionTypeUtils.getInputType(functionType, 0);
Type outputType = FunctionTypeUtils.getOutputType(functionType, 0);
return !FunctionTypeUtils.isPublisher(inputType) && (!FunctionTypeUtils.isPublisher(outputType) || FunctionTypeUtils.isMono(outputType));
}
private static boolean isRequestChannel(Type functionType) {
Type inputType = FunctionTypeUtils.getInputType(functionType, 0);
Type outputType = FunctionTypeUtils.getOutputType(functionType, 0);
return FunctionTypeUtils.isPublisher(inputType) && FunctionTypeUtils.isFlux(outputType);
}
/**
*
*/
private static class FunctionToDestinationBinder implements InitializingBean {
private static class FunctionToDestinationBinder implements InitializingBean, DisposableBean, ApplicationContextAware {
private final FunctionCatalog functionCatalog;
@@ -172,33 +69,75 @@ public class RSocketAutoConfiguration {
private final RSocketFunctionProperties rSocketFunctionProperties;
private RSocketFunction invocableFunction;
private GenericApplicationContext context;
FunctionToDestinationBinder(FunctionCatalog functionCatalog, FunctionProperties functionProperties,
RSocketFunctionProperties rSocketFunctionProperties) {
this.functionCatalog = functionCatalog;
this.functionProperties = functionProperties;
this.rSocketFunctionProperties = rSocketFunctionProperties;
}
@SuppressWarnings("rawtypes")
@Override
public void afterPropertiesSet() throws Exception {
String definition = this.functionProperties.getDefinition();
this.registerRsocketProxiesIfNecessary(definition);
//TODO externalize content-type
FunctionInvocationWrapper function = functionCatalog.lookup(definition, "application/json");
if (function.isSupplier()) {
throw new UnsupportedOperationException("Supplier is not currently supported for RSocket interaction");
}
Function invocableFunction = StringUtils.hasText(this.rSocketFunctionProperties.getTargetAddress())
? new RSocketFunction(this.rSocketFunctionProperties.getTargetAddress(), this.rSocketFunctionProperties.getTargetPort(), function)
: function;
InetSocketAddress bindAddress = InetSocketAddress
.createUnresolved(this.rSocketFunctionProperties.getBindAddress(), this.rSocketFunctionProperties.getBindPort());
Type functionType = function.getFunctionType();
RSocket rsocket = buildRSocket(definition, functionType, invocableFunction);
RSocketServer.create(SocketAcceptor.with(rsocket))
.bind(TcpServerTransport.create(this.rSocketFunctionProperties.getBindAddress(), this.rSocketFunctionProperties.getBindPort()))
.subscribe(); // TODO do we need to close/dispose etc?
if (this.invocableFunction == null) {
this.invocableFunction = new RSocketFunction(function, bindAddress, null);
this.invocableFunction.start();
}
}
@SuppressWarnings({ "rawtypes", "unchecked" })
private void registerRsocketProxiesIfNecessary(String definition) {
String[] names = StringUtils.delimitedListToStringArray(definition.replaceAll(",", "|").trim(), "|");
InetSocketAddress listenAddress = InetSocketAddress
.createUnresolved(this.rSocketFunctionProperties.getBindAddress(), this.rSocketFunctionProperties.getBindPort());
for (String name : names) {
if (!this.context.containsBean(name)) { // this means RSocket
String[] functionToRSocketDefinition = StringUtils.delimitedListToStringArray(name, ">");
Assert.isTrue(functionToRSocketDefinition.length == 2, "Must only contain one output redirect");
FunctionInvocationWrapper function = functionCatalog.lookup(functionToRSocketDefinition[0], "application/json");
String[] hostPort = StringUtils.delimitedListToStringArray(functionToRSocketDefinition[1], ":");
InetSocketAddress outputAddress = InetSocketAddress
.createUnresolved(hostPort[0], Integer.valueOf(hostPort[1]));
RSocketFunction rsocketFunction = new RSocketFunction(function, listenAddress, outputAddress);
FunctionRegistration functionRegistration = new FunctionRegistration(rsocketFunction, name);
functionRegistration.type(FunctionTypeUtils.discoverFunctionTypeFromClass(RSocketFunction.class));
((FunctionRegistry) this.functionCatalog).register(functionRegistration);
this.invocableFunction = rsocketFunction;
this.invocableFunction.start();
}
}
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.context = (GenericApplicationContext) applicationContext;
}
@Override
public void destroy() throws Exception {
if (this.invocableFunction != null) {
this.invocableFunction.stop();
}
}
}
}

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

@@ -0,0 +1,55 @@
/*
* 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;
}
}

193
spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketFunction.java
View File

@@ -16,6 +16,8 @@
package org.springframework.cloud.function.rsocket;
import java.lang.reflect.Type;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.function.Function;
@@ -25,10 +27,17 @@ import io.rsocket.RSocket;
import io.rsocket.core.RSocketConnector;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;
import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
@@ -40,47 +49,183 @@ import org.springframework.messaging.support.MessageBuilder;
* @author Oleg Zhurakousky
* @since 3.1
*/
class RSocketFunction implements Function<Message<byte[]>, Mono<Message<byte[]>>> {
class RSocketFunction implements Function<Message<byte[]>, Publisher<Message<byte[]>>> {
private final String bindAddress;
private static String splash = " ____ _ _______ __ ____ __ _ ___ ____ __ __ \n" +
" / __/__ ____(_)__ ___ _ / ___/ /__ __ _____/ / / __/_ _____ ____/ /_(_)__ ___ / _ \\/ __/__ ____/ /_____ / /_\n" +
" _\\ \\/ _ \\/ __/ / _ \\/ _ `/ / /__/ / _ \\/ // / _ / / _// // / _ \\/ __/ __/ / _ \\/ _ \\ / , _/\\ \\/ _ \\/ __/ '_/ -_) __/\n" +
"/___/ .__/_/ /_/_//_/\\_, / \\___/_/\\___/\\_,_/\\_,_/ /_/ \\_,_/_//_/\\__/\\__/_/\\___/_//_/ /_/|_/___/\\___/\\__/_/\\_\\\\__/\\__/ \n" +
" /_/ /___/ \n" +
"";
private final int port;
private static Log logger = LogFactory.getLog(RSocketFunction.class);
private final FunctionInvocationWrapper function;
private final InetSocketAddress listenAddress;
private final InetSocketAddress outputAddress;
private final FunctionInvocationWrapper targetFunction;
private final RSocket rSocket;
RSocketFunction(String bindAddress, int port, FunctionInvocationWrapper function) {
this.bindAddress = bindAddress;
this.port = port;
this.function = function;
this.rSocket = RSocketConnector.connectWith(TcpClientTransport.create(this.bindAddress, this.port))
.log()
.retryWhen(Retry.backoff(5, Duration.ofSeconds(1)))
.block();
private Disposable rsocketConnection;
RSocketFunction(FunctionInvocationWrapper targetFunction, InetSocketAddress listenAddress, @Nullable InetSocketAddress outputAddress) {
this.listenAddress = listenAddress;
this.outputAddress = outputAddress;
this.targetFunction = targetFunction;
this.rSocket = outputAddress == null ? null
: RSocketConnector.connectWith(TcpClientTransport.create(this.outputAddress))
.log()
.retryWhen(Retry.backoff(5, Duration.ofSeconds(1)))
.block();
}
@SuppressWarnings("unchecked")
@Override
public Mono<Message<byte[]>> apply(Message<byte[]> input) {
Message<byte[]> result = (Message<byte[]>) function.apply(input);
Mono<Message<byte[]>> resultMessage = null;
if (result != null) {
resultMessage = this.rSocket
.requestResponse(DefaultPayload.create(result.getPayload()))
.map(this::buildResultMessage);
public Publisher<Message<byte[]>> apply(Message<byte[]> input) {
if (logger.isDebugEnabled()) {
logger.debug("Executiing: " + this.targetFunction + " on " + this.listenAddress);
}
return resultMessage;
Object rawResult = this.targetFunction.apply(input);
if (rawResult instanceof Message) {
Publisher<Message<byte[]>> resultMessage = null;
if (this.outputAddress != null) {
resultMessage = this.rSocket
.requestStream(DefaultPayload.create(((Message<byte[]>) rawResult).getPayload()))
.map(this::buildResultMessage);
}
resultMessage = rawResult instanceof Publisher ? (Publisher<Message<byte[]>>) rawResult : Mono.just((Message<byte[]>) rawResult);
return resultMessage;
}
else {
return (Publisher<Message<byte[]>>) rawResult;
}
}
void start() {
Type functionType = this.targetFunction.getFunctionType();
RSocket rsocket = buildRSocket(this.targetFunction.getFunctionDefinition(), functionType, this);
if (this.listenAddress != null) {
this.rsocketConnection = RSocketConnectionUtils.createServerSocket(rsocket, this.listenAddress);
this.printSplashScreen(this.targetFunction.getFunctionDefinition(), functionType);
}
}
void stop() {
if (this.rsocketConnection != null) {
this.rsocketConnection.dispose();
}
}
private RSocket buildRSocket(String definition, Type functionType, Function<Message<byte[]>, Publisher<Message<byte[]>>> function) {
RSocket clientRSocket = new RSocket() { // imperative function or Function<?, Mono> = requestResponse
@Override
public Mono<Payload> requestResponse(Payload payload) {
if (logger.isDebugEnabled()) {
logger.debug("Invoking function '" + definition + "' as RSocket `requestResponse`.");
}
if (isFunctionReactive(functionType)) {
Flux<Payload> result = this.requestChannel(Flux.just(payload));
return Mono.from(result);
}
else {
Message<byte[]> inputMessage = deserealizePayload(payload);
Mono<Message<byte[]>> result = Mono.from(function.apply(inputMessage));
if (rSocket != null) {
return result.flatMap(message -> {
Mono<Payload> requestResponse = rSocket.requestResponse(DefaultPayload.create(message.getPayload()));
return requestResponse;
});
}
else {
return result.map(message -> DefaultPayload.create(message.getPayload()));
}
}
}
@Override
public Flux<Payload> requestStream(Payload payload) {
if (logger.isDebugEnabled()) {
logger.debug("Invoking function '" + definition + "' as RSocket `requestStream`.");
}
if (isFunctionReactive(functionType)) {
return this.requestChannel(Flux.just(payload));
}
else {
Message<byte[]> inputMessage = deserealizePayload(payload);
Flux<Message<byte[]>> result = Flux.from(function.apply(inputMessage));
return result.map(message -> DefaultPayload.create(message.getPayload()));
}
}
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
if (logger.isDebugEnabled()) {
logger.debug("Invoking function '" + definition + "' as RSocket `requestChannel`.");
}
if (isFunctionReactive(functionType)) {
return Flux.from(payloads)
.transform(inputFlux -> inputFlux.map(payload -> deserealizePayload(payload)))
.transform((Function) targetFunction)
.transform(outputFlux -> ((Flux<Message<byte[]>>) outputFlux).map(message -> DefaultPayload.create(message.getPayload())));
}
else {
return Flux.from(payloads)
.transform(flux -> {
return flux.flatMap(payload -> {
Message<byte[]> inputMessage = deserealizePayload(payload);
Flux<Message<byte[]>> result = Flux.from(function.apply(inputMessage));
return result;
});
})
.doOnNext(System.out::println)
.transform(outputFlux -> outputFlux.map(message -> DefaultPayload.create(message.getPayload())));
}
}
};
return clientRSocket;
}
private static boolean isFunctionReactive(Type functionType) {
Type inputType = FunctionTypeUtils.getInputType(functionType, 0);
Type outputType = FunctionTypeUtils.getOutputType(functionType, 0);
return FunctionTypeUtils.isPublisher(inputType) && FunctionTypeUtils.isFlux(outputType);
}
@SuppressWarnings("rawtypes")
private static Message<byte[]> deserealizePayload(Payload payload) {
ByteBuffer buffer = payload.getData();
byte[] rawData = new byte[buffer.remaining()];
buffer.get(rawData);
if (payload.hasMetadata()) {
String metadata = payload.getMetadataUtf8(); // TODO see what to do with it
}
MessageBuilder builder = MessageBuilder.withPayload(rawData);
Message<byte[]> inputMessage = builder.build();
return inputMessage;
}
private Message<byte[]> buildResultMessage(Payload payload) {
ByteBuffer payloadBuffer = payload.getData();
byte[] payloadData = new byte[payloadBuffer.remaining()];
payloadBuffer.get(payloadData);
// ByteBuffer headersBuffer = responsePayload.getMetadata();
// byte[] rawData = new byte[payloadBuffer.remaining()];
// payloadBuffer.get(rawData);
return MessageBuilder.withPayload(payloadData).build();
}
private void printSplashScreen(String definition, Type type) {
System.out.println(splash);
System.out.println("Function Definition: " + definition + ":[" + type + "]");
System.out.println("RSocket Listen Address: " + this.listenAddress);
System.out.println("RSocket Target Address: " + this.outputAddress);
System.out.println("======================================================\n");
}
}

38
spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketFunctionProperties.java
View File

@@ -29,13 +29,13 @@ import org.springframework.cloud.function.context.FunctionProperties;
@ConfigurationProperties(prefix = FunctionProperties.PREFIX + ".rsocket")
public class RSocketFunctionProperties {
private String bindAddress;
private String bindAddress = "localhost";
private int bindPort;
private int bindPort = 55555;
private String targetAddress;
private int targetPort;
// private String targetAddress;
//
// private int targetPort;
public String getBindAddress() {
return bindAddress;
@@ -53,19 +53,19 @@ public class RSocketFunctionProperties {
this.bindPort = bindPort;
}
public String getTargetAddress() {
return targetAddress;
}
// public String getTargetAddress() {
// return targetAddress;
// }
//
// public void setTargetAddress(String targetAddress) {
// this.targetAddress = targetAddress;
// }
public void setTargetAddress(String targetAddress) {
this.targetAddress = targetAddress;
}
public int getTargetPort() {
return targetPort;
}
public void setTargetPort(int targetPort) {
this.targetPort = targetPort;
}
// public int getTargetPort() {
// return targetPort;
// }
//
// public void setTargetPort(int targetPort) {
// this.targetPort = targetPort;
// }
}

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

@@ -16,26 +16,18 @@
package org.springframework.cloud.function.rsocket;
import java.time.Duration;
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.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;
@@ -50,9 +42,8 @@ import org.springframework.util.SocketUtils;
* @since 3.1
*/
public class RSocketAutoConfigurationTests {
@Test
public void testRequestReplyFunction() throws Exception {
public void testImperativeFunctionAsRequestReply() throws Exception {
int port = SocketUtils.findAvailableTcpPort();
new SpringApplicationBuilder(SampleFunctionConfiguration.class).web(WebApplicationType.NONE).run(
"--logging.level.org.springframework.cloud.function=DEBUG",
@@ -60,15 +51,125 @@ public class RSocketAutoConfigurationTests {
"--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();
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Mono<String> result = socket.requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
StepVerifier
.create(result)
.expectNext("\"HELLO\"")
.expectComplete()
.verify();
.create(result)
.expectNext("\"HELLO\"")
.expectComplete()
.verify();
}
@Test
public void testImperativeFunctionAsRequestStream() 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=" + port);
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Flux<String> result = socket.requestStream(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
StepVerifier
.create(result)
.expectNext("\"HELLO\"")
.expectComplete()
.verify();
}
@Test
public void testImperativeFunctionAsRequestChannel() 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=" + port);
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Flux<String> 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 testReactiveFunctionAsRequestReply() 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 = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Mono<String> result = socket.requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
StepVerifier
.create(result)
.expectNext("\"HELLO\"")
.expectComplete()
.verify();
}
@Test
public void testReactiveFunctionAsRequestStream() 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 = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Flux<String> result = socket.requestStream(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
StepVerifier
.create(result)
.expectNext("\"HELLO\"")
.expectComplete()
.verify();
}
@Test
public void testReactiveFunctionAsRequestChannel() 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 = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Flux<String> 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
@@ -77,25 +178,23 @@ public class RSocketAutoConfigurationTests {
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.definition=uppercase|concat",
"--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=" + portB,
"--spring.cloud.function.rsocket.target-address=localhost",
"--spring.cloud.function.rsocket.target-port=" + portA);
"--spring.cloud.function.definition=reverse>localhost:" + portA,
"--spring.cloud.function.rsocket.bind-address=localhost",
"--spring.cloud.function.rsocket.bind-port=" + portB);
RSocket socket = RSocketConnector.connectWith(TcpClientTransport.create("localhost", portB)).log()
.retryWhen(Retry.backoff(5, Duration.ofSeconds(1))).block();
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", portB), null);
Mono<String> result = socket.requestResponse(DefaultPayload.create("\"hello\"")).map(Payload::getDataUtf8);
StepVerifier
.create(result)
.expectNext("\"OLLEH\"")
.expectComplete()
.verify();
.create(result)
.expectNext("\"OLLEHOLLEH\"")
.expectComplete()
.verify();
}
@Test
@@ -107,8 +206,8 @@ public class RSocketAutoConfigurationTests {
"--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();
RSocket socket = RSocketConnectionUtils.createClientSocket(InetSocketAddress.createUnresolved("localhost", port), null);
Flux<String> result = socket.requestChannel(Flux.just(
DefaultPayload.create("\"Ricky\""),
DefaultPayload.create("\"Julien\""),
@@ -117,12 +216,12 @@ public class RSocketAutoConfigurationTests {
.map(Payload::getDataUtf8);
StepVerifier
.create(result)
.expectNext("\"RICKY\"")
.expectNext("\"JULIEN\"")
.expectNext("\"BUBBLES\"")
.expectComplete()
.verify();
.create(result)
.expectNext("\"RICKY\"")
.expectNext("\"JULIEN\"")
.expectNext("\"BUBBLES\"")
.expectComplete()
.verify();
}
@@ -152,7 +251,21 @@ public class RSocketAutoConfigurationTests {
public static class SampleFunctionConfiguration {
@Bean
public Function<String, String> uppercase() {
return v -> v.toUpperCase();
return v -> {
return v.toUpperCase();
};
}
@Bean
public Function<String, String> concat() {
return v -> {
return v + v;
};
}
@Bean
public Function<String, String> echo() {
return v -> v;
}
@Bean
@@ -176,7 +289,16 @@ public class RSocketAutoConfigurationTests {
public static class AdditionalFunctionConfiguration {
@Bean
public Function<String, String> reverse() {
return v -> new StringBuilder(v).reverse().toString();
return v -> {
return new StringBuilder(v).reverse().toString();
};
}
@Bean
public Function<String, String> wrap() {
return v -> {
return "(" + v + ")";
};
}
}
}