Add special handling for JSON Strings

Added support to JsonMessageConverter to pass string as is if input type is String
Added guard condition to RSocketListenerFunction to avoid NPE if target function can not be discovered

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/JsonMessageConverter.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.function.context.config;
 
 import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
 
 import org.springframework.cloud.function.json.JsonMapper;
 import org.springframework.lang.Nullable;
@@ -76,12 +77,16 @@ public class JsonMessageConverter extends AbstractMessageConverter {
 			return message.getPayload();
 		}
 
+
+
 		Type convertToType = conversionHint == null ? targetClass : (Type) conversionHint;
 		try {
 			return this.jsonMapper.fromJson(message.getPayload(), convertToType);
 		}
 		catch (Exception e) {
-			// ignore
+			if (message.getPayload() instanceof byte[] && targetClass.isAssignableFrom(String.class)) {
+				return new String((byte[]) message.getPayload(), StandardCharsets.UTF_8);
+			}
 		}
 		return null;
 	}

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

@@ -29,6 +29,7 @@ import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry
 import org.springframework.messaging.Message;
 import org.springframework.messaging.rsocket.annotation.support.RSocketFrameTypeMessageCondition;
 import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.Assert;
 
 
 
@@ -50,6 +51,10 @@ class RSocketListenerFunction implements Function<Message<Flux<byte[]>>, Publish
 
 	@Override
 	public Publisher<?> apply(Message<Flux<byte[]>> input) {
+		Assert.isTrue(this.targetFunction != null, "Failed to discover target function. \n"
+				+ "To fix it you should either provide 'spring.cloud.function.definition' property "
+				+ "or if you are using RSocketRequester provide valid function definition via 'route' "
+				+ "operator (e.g., requester.route(\"echo\"))");
 		FrameType frameType = RSocketFrameTypeMessageCondition.getFrameType(input);
 		switch (frameType) {
 			case REQUEST_FNF: