GH-1174 Add support for configurable Message conversion helper

This commit includes only the interfaces and interaction callbacks. The rest of the implementation is provided by binders in s-c-stream

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

@@ -1439,7 +1439,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
 		@SuppressWarnings("unchecked")
 		private Object convertOutputMessageIfNecessary(Object output, String expectedOutputContetntType) {
 			String contentType;
-			if (((Message) output).getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) {
+			if (this.isOutputTypeMessage() && ((Message) output).getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) {
 				contentType = ((Message) output).getHeaders().get(MessageHeaders.CONTENT_TYPE).toString();
 			}
 			else {

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

@@ -129,7 +129,9 @@ public class ContextFunctionCatalogAutoConfiguration {
 		mcList.add(new StringMessageConverter());
 		mcList.add(new PrimitiveTypesFromStringMessageConverter(conversionService));
 
-		messageConverter = new SmartCompositeMessageConverter(mcList);
+		messageConverter = new SmartCompositeMessageConverter(mcList, () -> {
+			return context.getBeansOfType(MessageConverterHelper.class).values();
+		});
 		if (functionInvocationHelper instanceof CloudEventsFunctionInvocationHelper) {
 			((CloudEventsFunctionInvocationHelper) functionInvocationHelper).setMessageConverter(messageConverter);
 		}

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

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2015-2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.function.context.config;
+
+import org.springframework.messaging.Message;
+
+/**
+ * @author Oleg Zhurakousky
+ */
+public interface MessageConverterHelper {
+
+	/**
+	 * This method will be called by the framework in cases when a message failed to convert.
+	 * It allows you to signal to the framework if such failure should be considered fatal or not.
+	 *
+	 * @param message failed message
+	 * @return true if conversion failure must be considered fatal.
+	 */
+	default boolean shouldFailIfCantConvert(Message<?> message) {
+		return false;
+	}
+
+	/**
+	 * This method will be called by the framework in cases when a single message within batch of messages failed to convert.
+	 * It provides a place for providing post-processing logic before message converter returns.
+	 *
+	 * @param message failed message.
+	 * @param index index of failed message within the batch
+	 */
+	default void postProcessBatchMessageOnFailure(Message<?> message, int index) {
+	}
+}

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

@@ -19,8 +19,10 @@ package org.springframework.cloud.function.context.config;
 import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.function.Supplier;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -31,6 +33,7 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.AbstractMessageConverter;
 import org.springframework.messaging.converter.CompositeMessageConverter;
+import org.springframework.messaging.converter.MessageConversionException;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.support.MessageBuilder;
@@ -48,13 +51,22 @@ public class SmartCompositeMessageConverter extends CompositeMessageConverter {
 
 	private Log logger = LogFactory.getLog(this.getClass());
 
+	private final Supplier<Collection<MessageConverterHelper>> messageConverterHelpersSupplier;
+
 	public SmartCompositeMessageConverter(Collection<MessageConverter> converters) {
+		this(converters, null);
+	}
+
+	public SmartCompositeMessageConverter(Collection<MessageConverter> converters, Supplier<Collection<MessageConverterHelper>> messageConverterHelpersSupplier) {
 		super(converters);
+		this.messageConverterHelpersSupplier = messageConverterHelpersSupplier;
 	}
 
 	@Override
 	@Nullable
 	public Object fromMessage(Message<?> message, Class<?> targetClass) {
+		Collection<MessageConverterHelper> messageConverterHelpers = this.messageConverterHelpersSupplier != null
+				? this.messageConverterHelpersSupplier.get() : Collections.emptyList();
 		for (MessageConverter converter : getConverters()) {
 			if (!(message.getPayload() instanceof byte[]) && targetClass.isInstance(message.getPayload()) && !(message.getPayload() instanceof Collection<?>)) {
 				return message.getPayload();
@@ -71,12 +83,15 @@ public class SmartCompositeMessageConverter extends CompositeMessageConverter {
 				}
 			}
 		}
+		this.failConversionIfNecessary(message, messageConverterHelpers);
 		return null;
 	}
 
 	@SuppressWarnings("unchecked")
 	@Override
 	public Object fromMessage(Message<?> message, Class<?> targetClass, @Nullable Object conversionHint) {
+		Collection<MessageConverterHelper> messageConverterHelpers = this.messageConverterHelpersSupplier != null
+				? this.messageConverterHelpersSupplier.get() : Collections.emptyList();
 		if (!(message.getPayload() instanceof byte[]) && targetClass.isInstance(message.getPayload()) && !(message.getPayload() instanceof Collection<?>)) {
 			return message.getPayload();
 		}
@@ -105,8 +120,12 @@ public class SmartCompositeMessageConverter extends CompositeMessageConverter {
 						}
 					}
 				}
+				if (!isConverted) {
+					this.postProcessBatchMessage(message, messageConverterHelpers, resultList.size());
+					this.failConversionIfNecessary(message, messageConverterHelpers);
+				}
 			}
-			result = resultList;
+			return resultList;
 		}
 		else {
 			for (MessageConverter converter : getConverters()) {
@@ -120,10 +139,25 @@ public class SmartCompositeMessageConverter extends CompositeMessageConverter {
 				}
 			}
 		}
-
+		this.failConversionIfNecessary(message, messageConverterHelpers);
 		return result;
 	}
 
+	private void failConversionIfNecessary(Message<?> message, Collection<MessageConverterHelper> messageConverterHelpers) {
+		for (MessageConverterHelper messageConverterHelper : messageConverterHelpers) {
+			if (messageConverterHelper.shouldFailIfCantConvert(message)) {
+				throw new MessageConversionException("Failed to convert Message: " + message
+						+ ". None of the available Message converters were able to convert this Message");
+			}
+		}
+	}
+
+	private void postProcessBatchMessage(Message<?> message, Collection<MessageConverterHelper> messageConverterHelpers, int index) {
+		for (MessageConverterHelper messageConverterHelper : messageConverterHelpers) {
+			messageConverterHelper.postProcessBatchMessageOnFailure(message, index);
+		}
+	}
+
 	@Override
 	@Nullable
 	public Message<?> toMessage(Object payload, @Nullable MessageHeaders headers) {