From 26c11b4463b86fa2becd29e0aa509e60146c6461 Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@vmware.com>
Date: Tue, 23 Feb 2021 16:58:37 +0100
Subject: [PATCH] GH-654 Fix how RoutingFunction is treated by
 RSocketListenerFunction

---
 .../cloud/function/rsocket/RSocketListenerFunction.java    | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java
index 04ca9027e..3787f740e 100644
--- a/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java
+++ b/spring-cloud-function-rsocket/src/main/java/org/springframework/cloud/function/rsocket/RSocketListenerFunction.java
@@ -69,7 +69,12 @@ class RSocketListenerFunction implements Function<Message<Flux<byte[]>>, Publish
 
 	@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()));