GH-654 Initial support for non-SpEL based routing mechanism
This commit is contained in:
@@ -32,6 +32,7 @@ import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.cloud.function.context.FunctionCatalog;
|
||||
import org.springframework.cloud.function.context.FunctionProperties;
|
||||
import org.springframework.cloud.function.context.MessageRoutingCallback;
|
||||
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
|
||||
import org.springframework.cloud.function.context.config.RoutingFunction;
|
||||
import org.springframework.core.MethodParameter;
|
||||
@@ -59,6 +60,7 @@ import org.springframework.messaging.rsocket.annotation.support.RSocketFrameType
|
||||
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
|
||||
import org.springframework.messaging.rsocket.annotation.support.RSocketPayloadReturnValueHandler;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
import org.springframework.util.MimeTypeUtils;
|
||||
import org.springframework.util.ReflectionUtils;
|
||||
import org.springframework.util.RouteMatcher;
|
||||
@@ -179,7 +181,10 @@ class FunctionRSocketMessageHandler extends RSocketMessageHandler {
|
||||
private String discoverAndInjectDestinationHeader(Message<?> message) {
|
||||
|
||||
String destination;
|
||||
if (StringUtils.hasText(this.functionProperties.getRoutingExpression())) {
|
||||
if (!CollectionUtils.isEmpty(this.getApplicationContext().getBeansOfType(MessageRoutingCallback.class))) {
|
||||
destination = RoutingFunction.FUNCTION_NAME;
|
||||
}
|
||||
else if (StringUtils.hasText(this.functionProperties.getRoutingExpression())) {
|
||||
destination = RoutingFunction.FUNCTION_NAME;
|
||||
this.updateMessageHeaders(message, destination);
|
||||
}
|
||||
|
||||
@@ -78,7 +78,8 @@ final class FunctionRSocketUtils {
|
||||
if (functionCatalog.lookup(name) == null) { // this means RSocket
|
||||
String[] functionToRSocketDefinition = StringUtils.delimitedListToStringArray(name, ">");
|
||||
if (functionToRSocketDefinition.length == 1) {
|
||||
throw new IllegalArgumentException("Function definition '" + name + "' does not exist in Function Catalog");
|
||||
//throw new IllegalArgumentException("Function definition '" + name + "' does not exist in Function Catalog");
|
||||
return;
|
||||
}
|
||||
if (LOGGER.isDebugEnabled()) {
|
||||
LOGGER.debug("Registering RSocket forwarder for '" + name + "' function.");
|
||||
|
||||
@@ -29,6 +29,8 @@ import reactor.test.StepVerifier;
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.function.context.FunctionProperties;
|
||||
import org.springframework.cloud.function.context.MessageRoutingCallback;
|
||||
import org.springframework.cloud.function.context.config.RoutingFunction;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
@@ -173,6 +175,59 @@ public class RSocketAutoConfigurationRoutingTests {
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRoutingWithRoutingCallback() {
|
||||
int port = SocketUtils.findAvailableTcpPort();
|
||||
try (
|
||||
ConfigurableApplicationContext applicationContext =
|
||||
new SpringApplicationBuilder(RoutingCallbackFunctionConfiguration.class)
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--logging.level.org.springframework.cloud.function=DEBUG",
|
||||
"--spring.cloud.function.expected-content-type=text/plain",
|
||||
"--spring.rsocket.server.port=" + port);
|
||||
) {
|
||||
RSocketRequester.Builder rsocketRequesterBuilder =
|
||||
applicationContext.getBean(RSocketRequester.Builder.class);
|
||||
|
||||
rsocketRequesterBuilder.tcp("localhost", port)
|
||||
.route("foo")
|
||||
.metadata("{\"func_name\":\"uppercase\"}", MimeTypeUtils.APPLICATION_JSON)
|
||||
.data("hello")
|
||||
.retrieveMono(String.class)
|
||||
.as(StepVerifier::create)
|
||||
.expectNext("HELLO")
|
||||
.expectComplete()
|
||||
.verify();
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@Configuration
|
||||
public static class RoutingCallbackFunctionConfiguration {
|
||||
@Bean
|
||||
public MessageRoutingCallback customRouter() {
|
||||
return new MessageRoutingCallback() {
|
||||
@Override
|
||||
public String route(Message<?> message, FunctionProperties functionProperties) {
|
||||
return (String) message.getHeaders().get("func_name");
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<String, String> uppercase() {
|
||||
return v -> v.toUpperCase();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<String, String> concat() {
|
||||
return v -> v + v;
|
||||
}
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@Configuration
|
||||
public static class SampleFunctionConfiguration {
|
||||
@@ -223,6 +278,7 @@ public class RSocketAutoConfigurationRoutingTests {
|
||||
return () -> "test data";
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user