From 85b591cb89f40c73975860ee623e4c61b7fd3ebc Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@pivotal.io>
Date: Mon, 31 Aug 2020 16:35:18 +0200
Subject: [PATCH] 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
---
 .../function/context/config/JsonMessageConverter.java      | 7 ++++++-
 .../cloud/function/rsocket/RSocketListenerFunction.java    | 5 +++++
 2 files changed, 11 insertions(+), 1 deletion(-)

diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/JsonMessageConverter.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/JsonMessageConverter.java
index 9bad95929..5764f4377 100644
--- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/JsonMessageConverter.java
+++ b/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;
 	}
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 6984fa623..5c6e38d95 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
@@ -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: