diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java index 295356678..09045830a 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java @@ -541,6 +541,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect logger.debug("Applying function: " + this.functionDefinition); } + Object result; if (input instanceof Publisher) { input = this.composed ? input : diff --git a/spring-cloud-function-rsocket/pom.xml b/spring-cloud-function-rsocket/pom.xml index 69e88b1ae..beec9dbc7 100644 --- a/spring-cloud-function-rsocket/pom.xml +++ b/spring-cloud-function-rsocket/pom.xml @@ -50,20 +50,4 @@ - - - - org.springframework.boot - spring-boot-maven-plugin - - - org.springframework.boot.experimental - spring-boot-thin-layout - ${wrapper.version} - - - - - - diff --git a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketAutoConfiguration.java b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketAutoConfiguration.java index f5a3646a4..3e1d738fb 100644 --- a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketAutoConfiguration.java +++ b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketAutoConfiguration.java @@ -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 = requestResponse -// @Override -// public Mono 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 = requestResponse - @SuppressWarnings("unchecked") - @Override - public Mono requestResponse(Payload payload) { - Object result = invokeFunction(payload, function); - Mono invocationResult = ((Mono>) 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 = requestResponse - @SuppressWarnings("unchecked") - @Override - public Flux requestChannel(Publisher 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 inputMessage = MessageBuilder.withPayload(rawData).build(); - return inputMessage; - }); - }) - .transform(function) - .transform(resultFlux -> { - return ((Flux>) 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> invokeFunction(Payload p, Function, Mono>> 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 inputMessage = MessageBuilder.withPayload(rawData).build(); - Object result = function.apply(inputMessage); - return result instanceof Mono ? (Mono>) result : Mono.just((Message) 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(); + } + } } } diff --git a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketConnectionUtils.java b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketConnectionUtils.java new file mode 100644 index 000000000..a319f3267 --- /dev/null +++ b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketConnectionUtils.java @@ -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; + } +} diff --git a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketFunction.java b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketFunction.java index 417391670..5f3aacacd 100644 --- a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketFunction.java +++ b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketFunction.java @@ -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, Mono>> { +class RSocketFunction implements Function, Publisher>> { - 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> apply(Message input) { - Message result = (Message) function.apply(input); - Mono> resultMessage = null; - if (result != null) { - resultMessage = this.rSocket - .requestResponse(DefaultPayload.create(result.getPayload())) - .map(this::buildResultMessage); + public Publisher> apply(Message 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> resultMessage = null; + if (this.outputAddress != null) { + resultMessage = this.rSocket + .requestStream(DefaultPayload.create(((Message) rawResult).getPayload())) + .map(this::buildResultMessage); + } + resultMessage = rawResult instanceof Publisher ? (Publisher>) rawResult : Mono.just((Message) rawResult); + return resultMessage; + } + else { + return (Publisher>) 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, Publisher>> function) { + RSocket clientRSocket = new RSocket() { // imperative function or Function = requestResponse + @Override + public Mono requestResponse(Payload payload) { + if (logger.isDebugEnabled()) { + logger.debug("Invoking function '" + definition + "' as RSocket `requestResponse`."); + } + + if (isFunctionReactive(functionType)) { + Flux result = this.requestChannel(Flux.just(payload)); + return Mono.from(result); + } + else { + Message inputMessage = deserealizePayload(payload); + Mono> result = Mono.from(function.apply(inputMessage)); + if (rSocket != null) { + return result.flatMap(message -> { + Mono requestResponse = rSocket.requestResponse(DefaultPayload.create(message.getPayload())); + return requestResponse; + }); + } + else { + return result.map(message -> DefaultPayload.create(message.getPayload())); + } + } + } + + @Override + public Flux 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 inputMessage = deserealizePayload(payload); + Flux> result = Flux.from(function.apply(inputMessage)); + return result.map(message -> DefaultPayload.create(message.getPayload())); + } + } + + @SuppressWarnings({ "unchecked", "rawtypes" }) + @Override + public Flux requestChannel(Publisher 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>) outputFlux).map(message -> DefaultPayload.create(message.getPayload()))); + } + else { + return Flux.from(payloads) + .transform(flux -> { + return flux.flatMap(payload -> { + Message inputMessage = deserealizePayload(payload); + Flux> 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 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 inputMessage = builder.build(); + return inputMessage; + } private Message 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"); + } + } diff --git a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketFunctionProperties.java b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketFunctionProperties.java index 228c271e3..c7c5419fd 100644 --- a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketFunctionProperties.java +++ b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketFunctionProperties.java @@ -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; +// } } 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 a89847011..c0e19108e 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 @@ -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 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 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 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 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 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 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 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 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 uppercase() { - return v -> v.toUpperCase(); + return v -> { + return v.toUpperCase(); + }; + } + + @Bean + public Function concat() { + return v -> { + return v + v; + }; + } + + @Bean + public Function echo() { + return v -> v; } @Bean @@ -176,7 +289,16 @@ public class RSocketAutoConfigurationTests { public static class AdditionalFunctionConfiguration { @Bean public Function reverse() { - return v -> new StringBuilder(v).reverse().toString(); + return v -> { + return new StringBuilder(v).reverse().toString(); + }; + } + + @Bean + public Function wrap() { + return v -> { + return "(" + v + ")"; + }; } } }