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()));