GH-499,498 Add support for SupplierExporter to control output content-type

- Add 'contentType' property to ExporterProperties to assist SupplierExporter with delegating it to function catalog
- Add additional logging and testing
- Change JsonMapper to abstract class providing special handling of conversion of Json Sting to byte[]

spring-cloud-function-adapters/spring-cloud-function-adapter-aws/src/main/java/org/springframework/cloud/function/adapter/aws/LambdaDestinationResolver.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,22 +18,39 @@ package org.springframework.cloud.function.adapter.aws;
 
 import java.util.function.Supplier;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.cloud.function.web.source.DestinationResolver;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 
+/**
+ * Implementation of {@link DestinationResolver}for AWS Lambda which resolves destination
+ * from `lambda-runtime-aws-request-id` message header.
+ *
+ * @author Dave Syer
+ * @author Oleg Zhurakousky
+ *
+ */
 public class LambdaDestinationResolver implements DestinationResolver {
 
+	private static Log logger = LogFactory.getLog(LambdaDestinationResolver.class);
+
 	@Override
 	public String destination(Supplier<?> supplier, String name, Object value) {
+		String destination = "unknown";
 		if (value instanceof Message) {
 			Message<?> message = (Message<?>) value;
 			MessageHeaders headers = message.getHeaders();
 			if (headers.containsKey("lambda-runtime-aws-request-id")) {
-				return (String) headers.get("lambda-runtime-aws-request-id");
+				destination = (String) headers.get("lambda-runtime-aws-request-id");
 			}
 		}
-		return "unknown";
+		if (logger.isDebugEnabled()) {
+			logger.debug("Lambda destination resolved to: " + destination);
+		}
+		return destination;
 	}
 
 }

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java

@@ -319,7 +319,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 
 				registrationsByFunction.putIfAbsent(function, registration);
 				registrationsByName.putIfAbsent(name, registration);
-				function = new FunctionInvocationWrapper(function, currentFunctionType, name, acceptedOutputTypes);
+				function = new FunctionInvocationWrapper(function, currentFunctionType, name, names.length > 1 ? new String[] {} : acceptedOutputTypes);
 
 				if (originFunctionType == null) {
 					originFunctionType = currentFunctionType;
@@ -447,6 +447,10 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 			this.headersField.setAccessible(true);
 		}
 
+		public String getFunctionDefinition() {
+			return this.functionDefinition;
+		}
+
 		@Override
 		public void accept(Object input) {
 			this.doApply(input, true, null);
@@ -505,6 +509,11 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 			return target;
 		}
 
+		@Override
+		public String toString() {
+			return "definition: " + this.functionDefinition + "; type: " + this.functionType;
+		}
+
 		@SuppressWarnings({"rawtypes", "unchecked"})
 		private Object invokeFunction(Object input) {
 			Object invocationResult = null;

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/json/GsonMapper.java

@@ -27,7 +27,7 @@ import com.google.gson.JsonElement;
  * @author Dave Syer
  * @author Oleg Zhurakousky
  */
-public class GsonMapper implements JsonMapper {
+public class GsonMapper extends JsonMapper {
 
 	private final Gson gson;
 
@@ -65,7 +65,11 @@ public class GsonMapper implements JsonMapper {
 
 	@Override
 	public byte[] toJson(Object value) {
-		return this.gson.toJson(value).getBytes(StandardCharsets.UTF_8);
+		byte[] jsonBytes = super.toJson(value);
+		if (jsonBytes == null) {
+			jsonBytes = this.gson.toJson(value).getBytes(StandardCharsets.UTF_8);
+		}
+		return jsonBytes;
 	}
 
 }

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/json/JacksonMapper.java

@@ -23,12 +23,16 @@ import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.type.TypeFactory;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
 /**
  * @author Dave Syer
  * @author Oleg Zhurakousky
  */
-public class JacksonMapper implements JsonMapper {
+public class JacksonMapper extends JsonMapper {
+
+	private static Log logger = LogFactory.getLog(JsonMapper.class);
 
 	private final ObjectMapper mapper;
 
@@ -58,20 +62,23 @@ public class JacksonMapper implements JsonMapper {
 			}
 		}
 		catch (Exception e) {
-			//ignore and let other converters have a chance
+			logger.warn("Failed to convert. Possible bug as the conversion probably shouldn't have been attampted here", e);
 		}
 		return convertedValue;
 	}
 
 	@Override
 	public byte[] toJson(Object value) {
-		try {
+		byte[] jsonBytes = super.toJson(value);
-			return this.mapper.writeValueAsBytes(value);
+		if (jsonBytes == null) {
+			try {
+				jsonBytes = this.mapper.writeValueAsBytes(value);
+			}
+			catch (Exception e) {
+				//ignore and let other converters have a chance
+			}
 		}
-		catch (Exception e) {
+		return jsonBytes;
-			//ignore and let other converters have a chance
-		}
-		return null;
 	}
 
 	@Override

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/json/JsonMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,35 +17,41 @@
 package org.springframework.cloud.function.json;
 
 import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.json.JSONObject;
+
 import org.springframework.core.ResolvableType;
 
 /**
  * @author Dave Syer
  * @author Oleg Zhurakousky
  */
-public interface JsonMapper {
+public abstract class JsonMapper {
+
+	private static Log logger = LogFactory.getLog(JsonMapper.class);
 
 	/**
-	 * @param <T> type for list arguments
+	 * @param <T>  type for list arguments
 	 * @param json JSON input
 	 * @param type type of list arguments
 	 * @return list of elements
 	 * @deprecated since v2.0 in favor of {@link #toObject(String, Type)}
 	 */
 	@Deprecated
-	default <T> List<T> toList(String json, Class<T> type) {
+	<T> List<T> toList(String json, Class<T> type) {
 		Type actualType = (json.startsWith("[") && !List.class.isAssignableFrom(type))
-				? ResolvableType.forClassWithGenerics(ArrayList.class, (Class<?>) type)
+				? ResolvableType.forClassWithGenerics(ArrayList.class, (Class<?>) type).getType()
-						.getType()
 				: type;
 		return toObject(json, actualType);
 	}
 
 	/**
-	 * @param <T> return type
+	 * @param <T>  return type
 	 * @param json JSON input
 	 * @param type type
 	 * @return object
@@ -53,24 +59,39 @@ public interface JsonMapper {
 	 * @deprecated since v3.0.4 in favor of {@link #fromJson(Object, Type)}
 	 */
 	@Deprecated
-	<T> T toObject(String json, Type type);
+	abstract <T> T toObject(String json, Type type);
 
-	<T> T fromJson(Object json, Type type);
+	public abstract <T> T fromJson(Object json, Type type);
 
-	byte[] toJson(Object value);
+	public byte[] toJson(Object value) {
+		if (value instanceof String) {
+			try {
+				new JSONObject((String) value);
+				if (logger.isDebugEnabled()) {
+					logger.debug(
+							"String already represents JSON. Skipping conversion in favor of 'getBytes(StandardCharsets.UTF_8'.");
+				}
+				return ((String) value).getBytes(StandardCharsets.UTF_8);
+			}
+			catch (Exception ex) {
+				// ignore
+			}
+		}
+		return null;
+	}
 
 	/**
-	 * @param <T> type for list arguments
+	 * @param <T>  type for list arguments
 	 * @param json JSON input
 	 * @param type type of list arguments
 	 * @return single object
 	 * @deprecated since v2.0 in favor of {@link #toObject(String, Type)}
 	 */
 	@Deprecated
-	default <T> T toSingle(String json, Class<T> type) {
+	<T> T toSingle(String json, Class<T> type) {
 		return toObject(json, type);
 	}
 
-	String toString(Object value);
+	public abstract String toString(Object value);
 
 }

spring-cloud-function-context/src/test/java/org/springframework/cloud/function/utils/JsonMapperTests.java

@@ -56,7 +56,7 @@ public class JsonMapperTests {
 	@Test
 	public void vanillaArray() {
 		String json = "[{\"value\":\"foo\"},{\"value\":\"foo\"}]";
-		List<Foo> list = this.mapper.toObject(json,
+		List<Foo> list = this.mapper.fromJson(json,
 				ResolvableType.forClassWithGenerics(List.class, Foo.class).getType());
 		assertThat(list).hasSize(2);
 		assertThat(list.get(0).getValue()).isEqualTo("foo");
@@ -65,7 +65,7 @@ public class JsonMapperTests {
 
 	@Test
 	public void intArray() {
-		List<Integer> list = this.mapper.toObject("[123,456]",
+		List<Integer> list = this.mapper.fromJson("[123,456]",
 				ResolvableType.forClassWithGenerics(List.class, Integer.class).getType());
 		assertThat(list).hasSize(2);
 		assertThat(list.get(0)).isEqualTo(123);
@@ -73,7 +73,7 @@ public class JsonMapperTests {
 
 	@Test
 	public void emptyArray() {
-		List<Foo> list = this.mapper.toObject("[]",
+		List<Foo> list = this.mapper.fromJson("[]",
 				ResolvableType.forClassWithGenerics(List.class, Foo.class).getType());
 		assertThat(list).hasSize(0);
 	}
@@ -81,20 +81,27 @@ public class JsonMapperTests {
 	@Test
 	public void vanillaObject() {
 		String json = "{\"value\":\"foo\"}";
-		Foo foo = this.mapper.toObject(json, Foo.class);
+		Foo foo = this.mapper.fromJson(json, Foo.class);
 		assertThat(foo.getValue()).isEqualTo("foo");
 		assertThat(this.mapper.toString(foo)).isEqualTo(json);
 	}
 
+	@Test
+	public void stringRepresentingJson() {
+		String json = "{\"value\":\"foo\"}";
+		byte[] bytes = this.mapper.toJson(json);
+		assertThat(new String(bytes)).isEqualTo(json);
+	}
+
 	@Test
 	public void intValue() {
-		int foo = this.mapper.toObject("123", Integer.class);
+		int foo = this.mapper.fromJson("123", Integer.class);
 		assertThat(foo).isEqualTo(123);
 	}
 
 	@Test
 	public void empty() {
-		Foo foo = this.mapper.toObject("{}", Foo.class);
+		Foo foo = this.mapper.fromJson("{}", Foo.class);
 		assertThat(foo.getValue()).isNull();
 	}
 

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RequestProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -127,8 +127,10 @@ public class RequestProcessor {
 
 	public Mono<ResponseEntity<?>> post(FunctionWrapper wrapper,
 			ServerWebExchange exchange) {
-		return Mono.from(body(wrapper.handler(), exchange))
+		Mono<ResponseEntity<?>> responseEntity = Mono.from(body(wrapper.handler(), exchange))
 				.flatMap(body -> response(wrapper, body, false));
+
+		return responseEntity;
 	}
 
 	public Mono<ResponseEntity<?>> post(FunctionWrapper wrapper, String body,
@@ -149,7 +151,7 @@ public class RequestProcessor {
 					jsonType = ResolvableType.forClassWithGenerics((Class<?>) jsonType,
 							(Class<?>) itemType).getType();
 				}
-				input = this.mapper.toObject((String) input, jsonType);
+				input = this.mapper.fromJson((String) input, jsonType);
 			}
 			else {
 				input = this.converter.convert(function, (String) input);
@@ -178,7 +180,6 @@ public class RequestProcessor {
 
 	private Mono<ResponseEntity<?>> response(FunctionWrapper request, Object handler,
 			Publisher<?> result, Boolean single, boolean getter) {
-
 		BodyBuilder builder = ResponseEntity.ok();
 		if (this.inspector.isMessage(handler)) {
 			result = Flux.from(result)
@@ -389,7 +390,11 @@ public class RequestProcessor {
 								exchange.getLogPrefix() + "0..1 [" + elementType + "]");
 					}
 					Mono<?> mono = reader.readMono(actualType, elementType, request,
-							response, readHints);
+							response, readHints).doOnNext(v -> {
+								if (logger.isDebugEnabled()) {
+									logger.debug("received: " + v);
+								}
+							});
 					mono = mono.onErrorResume(
 							ex -> Mono.error(handleReadError(bodyParam, ex)));
 					if (isBodyRequired) {

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FunctionController.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/source/ExporterProperties.java

@@ -24,6 +24,7 @@ import org.springframework.cloud.function.context.FunctionProperties;
 
 /**
  * @author Dave Syer
+ * @author Oleg Zhurakousky
  *
  */
 @ConfigurationProperties(prefix = FunctionProperties.PREFIX + ".web.export")
@@ -149,6 +150,11 @@ public class ExporterProperties {
 		 */
 		private String name;
 
+		/**
+		 * Content type to use when serializing source's output for transport (default 'application/json`).
+		 */
+		private String contentType = "application/json";
+
 		public String getName() {
 			return this.name;
 		}
@@ -169,6 +175,13 @@ public class ExporterProperties {
 			return this.headers;
 		}
 
+		public String getContentType() {
+			return contentType;
+		}
+
+		public void setContentType(String contentType) {
+			this.contentType = contentType;
+		}
 	}
 
 }

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/source/SupplierExporter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -58,6 +58,8 @@ public class SupplierExporter implements SmartLifecycle {
 
 	private final String supplier;
 
+	private final String contentType;
+
 	private volatile boolean running;
 
 	private volatile boolean ok = true;
@@ -70,14 +72,15 @@ public class SupplierExporter implements SmartLifecycle {
 
 	SupplierExporter(RequestBuilder requestBuilder,
 			DestinationResolver destinationResolver, FunctionCatalog catalog,
-			WebClient client, ExporterProperties props) {
+			WebClient client, ExporterProperties exporterProperties) {
 		this.requestBuilder = requestBuilder;
 		this.destinationResolver = destinationResolver;
 		this.catalog = catalog;
 		this.client = client;
-		this.debug = props.isDebug();
+		this.debug = exporterProperties.isDebug();
-		this.autoStartup = props.isAutoStartup();
+		this.autoStartup = exporterProperties.isAutoStartup();
-		this.supplier = props.getSink().getName();
+		this.supplier = exporterProperties.getSink().getName();
+		this.contentType = exporterProperties.getSink().getContentType();
 	}
 
 	@Override
@@ -93,7 +96,7 @@ public class SupplierExporter implements SmartLifecycle {
 
 		boolean suppliersPresent = false;
 		for (String name : names) {
-			Supplier<Publisher<Object>> supplier = this.catalog.lookup(Supplier.class, name);
+			Supplier<Publisher<Object>> supplier = this.catalog.lookup(name, this.contentType);
 			if (supplier == null) {
 				logger.warn("No such Supplier: " + name);
 				continue;
@@ -163,8 +166,7 @@ public class SupplierExporter implements SmartLifecycle {
 
 	private Flux<ClientResponse> forward(Supplier<Publisher<Object>> supplier, String name) {
 		return Flux.from(supplier.get()).flatMap(value -> {
-			String destination = this.destinationResolver.destination(supplier, name,
+			String destination = this.destinationResolver.destination(supplier, name, value);
-					value);
 			if (this.debug) {
 				logger.info("Posting to: " + destination);
 			}
@@ -178,9 +180,17 @@ public class SupplierExporter implements SmartLifecycle {
 			Message<?> message = (Message<?>) value;
 			body = message.getPayload();
 		}
+		if (this.debug) {
+			logger.debug("Sending BODY as type: " + body.getClass().getName());
+		}
 		Mono<ClientResponse> result = this.client.post().uri(uri)
 				.headers(headers -> headers(headers, destination, value)).bodyValue(body)
-				.exchange();
+				.exchange()
+				.doOnNext(response -> {
+					if (this.debug) {
+						logger.debug("Response STATUS: " + response.statusCode());
+					}
+				});
 		if (this.debug) {
 			result = result.log();
 		}

spring-cloud-function-web/src/test/java/org/springframework/cloud/function/test/FunctionalExporterTests.java

@@ -52,6 +52,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 		"spring.cloud.function.web.export.sink.url=http://localhost:${my.port}",
 		"spring.cloud.function.web.export.source.url=http://localhost:${my.port}",
 		"spring.cloud.function.web.export.sink.name=origin|uppercase",
+		"spring.cloud.function.web.export.sink.contentType=text/plain",
 		"spring.cloud.function.web.export.debug=true" })
 public class FunctionalExporterTests {
 

spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/source/FunctionAutoConfigurationIntegrationTests.java

@@ -85,8 +85,8 @@ public class FunctionAutoConfigurationIntegrationTests {
 		}
 		// It completed
 		assertThat(this.forwarder.isOk()).isTrue();
-		assertThat(this.app.inputs).contains("HELLO");
+		assertThat(this.app.inputs).contains("\"HELLO\"");
-		assertThat(this.app.inputs).contains("WORLD");
+		assertThat(this.app.inputs).contains("\"WORLD\"");
 	}
 
 	@EnableAutoConfiguration