GH-654 Fix how RoutingFunction is treated by RSocketListenerFunction

Author: olegz

spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java

@@ -69,7 +69,12 @@ public Publisher<?> apply(Message<Flux<byte[]>> input) {
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	private Mono<Void> handle(Message<Flux<byte[]>> messageToProcess) {
-		if (this.targetFunction.isConsumer()) {
+		if (this.targetFunction.isRoutingFunction()) {
+			Flux<?> dataFlux = messageToProcess.getPayload()
+						.map((payload) -> MessageBuilder.createMessage(payload, messageToProcess.getHeaders()));
+			return dataFlux.doOnNext(this.targetFunction).then();
+		}
+		else if (this.targetFunction.isConsumer()) {
 			Flux<?> dataFlux =
 				messageToProcess.getPayload()
 					.map((payload) -> MessageBuilder.createMessage(payload, messageToProcess.getHeaders()));