Add support for configuring additional routers

[ferblaca]

We have been analyzing this spring-cloud/spring-cloud-stream#1790, and we see a clear backward step in functionality with FunctionRouter versus the filter conditions in @StreamListener.

These are the limitations we found with the FunctionRouter:

• Only one FunctionRouter can be defined per application.
• The FunctionRouter can only belong to one Binder: for example, and although it is rare, if your application consumes Rabbit and Kafka, the FunctionRouterhas to be configured with one of the binders, it will never be able to read two different data sources.
• The FunctionRouter does not know the origin of the message consumed and therefore, even if it consumes from several topics, you can not customize a routing-expression depending on the origin of the message.

For all this, we see that the FunctionRouter is another functionality totally different from the filtering in the @StreamListener, which is a significant loss in terms of usability in Spring Cloud Stream with the functional approach.

We believe it would be important to reconsider this functionality and add at least some of the following features to the FunctionRouter:

• Being able to define more than one FunctionRouter per application.
• To be able to define one routing-expression per message origin. For example, if the FunctionRouter has configured topic1 and topic2 destinations, it would be nice to be able to define one routing-expression for topic1 messages and another routing-expression for topic2 messages.

Thank you very much, we hope that the request will be considered.

[olegz]

Let me address your comments as there is some confusion which we may need to rectify with better documentation.

1. Only one FunctionRouter can be defined per application. - not true. The RoutingFunction is just a bean and you can define as many instances of it as you want

@Bean
RoutingFunction mySpecialRouter(FunctionCatalog functionCatalog, FunctionProperties functionProperties,
				BeanFactory beanFactory, @Nullable MessageRoutingCallback routingCallback) {
	return new RoutingFunction(functionCatalog, functionProperties, new BeanFactoryResolver(beanFactory), routingCallback);
}

and then you can configure mySpecialRouter-in/out-0 bindings just like you would configure then for any other function

2. Again, not true. Based on the answer in 1 (above), it is just a function bean with its own bindings configurations, hence can be associated with any binder.

3. What origin of the message you are referring to? i mean in comparison to the previous approach you mentioned with StreamListener? Or you are requesting a new feature to include the origin of the message in a message header for example?

Can you please address these 3 comments?

[ferblaca]

Hi @olegz ! thank you very much for your quick reply.

Yes, as you say it is possible to define more than one FunctionRouter, and therefore assign more than one binder per application. The only thing is that all the defined FunctionRouter share by default the same bean FunctionProperties and therefore the same routing expression. Would it be possible to assign by configuration (.yml or properties) a routing expression to each defined FunctionRouter?

Regarding the origin of the message: Yes, it is a request, if the topic/queue from which the message has been consumed were known it could be very useful to compose one expression or another depending on this value, always talking about the case in which the FunctionRouter has several destinations configured.

[olegz]

Makes sense, i'll change the title of an issue and mark it as feature request

[olegz]

Work in progress but here is what I am thinking. . .

Basically I am adding a second constructor to RoutingFunction where you can provide definition and routing-expression as map

Here is the test configuration:

@EnableAutoConfiguration
@Configuration
protected static class MultipleRouterConfiguration {

	@Bean
	RoutingFunction mySpecialRouter(FunctionCatalog functionCatalog, BeanFactory beanFactory, @Nullable MessageRoutingCallback routingCallback) {
		Map<String, String> propertiesMap = new HashMap<>();
		propertiesMap.put(FunctionProperties.PREFIX + ".routing-expression", "'reverse'");
		return new RoutingFunction(functionCatalog, propertiesMap, new BeanFactoryResolver(beanFactory), routingCallback);
	}

	@Bean
	public Function<String, String> reverse() {
		return v -> new StringBuilder(v).reverse().toString();
	}

	@Bean
	public Function<String, String> uppercase() {
		return String::toUpperCase;
	}
}

and here is the actual test

@Test
public void testMultipleRouters() {
	System.setProperty(FunctionProperties.PREFIX + ".routing-expression", "'uppercase'");
	FunctionCatalog functionCatalog = this.configureCatalog(MultipleRouterConfiguration.class);
	Function function = functionCatalog.lookup(RoutingFunction.FUNCTION_NAME);
	assertThat(function).isNotNull();
	Message<String> message = MessageBuilder.withPayload("hello").build();
	assertThat(function.apply(message)).isEqualTo("HELLO");

	function = functionCatalog.lookup("mySpecialRouter");
	assertThat(function).isNotNull();
	message = MessageBuilder.withPayload("hello").build();
	assertThat(function.apply(message)).isEqualTo("olleh");
}