GH-1536 Polishing

Resolves #1538

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java

@@ -16,7 +16,7 @@
 
 package org.springframework.cloud.stream.binder;
 
-import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.function.Consumer;
@@ -198,7 +198,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		this.postProcessOutputChannel(outputChannel, producerProperties);
 
 		if (this.streamFunctionProperties != null && StringUtils.hasText(this.streamFunctionProperties.getDefinition()) && this.processor == null) {
-			outputChannel = this.postProcessOutboundChannelForFunction(outputChannel);
+			outputChannel = this.postProcessOutboundChannelForFunction(outputChannel, producerProperties);
 		}
 
 		((SubscribableChannel) outputChannel).subscribe(
@@ -810,7 +810,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 	}
 
-	private SubscribableChannel postProcessOutboundChannelForFunction(MessageChannel outputChannel) {
+	private SubscribableChannel postProcessOutboundChannelForFunction(MessageChannel outputChannel, ProducerProperties producerProperties) {
 		if (this.integrationFlowFunctionSupport != null) {
 			Publisher<?> publisher = MessageChannelReactiveUtils.toPublisher(outputChannel);
 			// If the app has an explicit Supplier bean defined, make that as the publisher
@@ -818,6 +818,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				IntegrationFlowBuilder integrationFlowBuilder = IntegrationFlows.from(outputChannel).bridge();
 				publisher = integrationFlowBuilder.toReactivePublisher();
 			}
+			this.propagateProducerPropertiesToFunction(producerProperties);
 			if (this.integrationFlowFunctionSupport.containsFunction(Function.class,
 					this.streamFunctionProperties.getDefinition())) {
 				DirectChannel actualOutputChannel = new DirectChannel();
@@ -851,9 +852,20 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	// we're doing it reflectively so we don't expose this as a property to the user
 	private void propagateConsumerPropertiesToFunction(ConsumerProperties consumerProperties) {
 		try {
-			Field f = ReflectionUtils.findField(StreamFunctionProperties.class, "consumerProperties");
-			f.setAccessible(true);
-			f.set(this.streamFunctionProperties, consumerProperties);
+			Method setConsumerProperties = ReflectionUtils.findMethod(StreamFunctionProperties.class, "setConsumerProperties", ConsumerProperties.class);
+			setConsumerProperties.setAccessible(true);
+			setConsumerProperties.invoke(this.streamFunctionProperties, consumerProperties);
+		}
+		catch (Exception e) {
+			throw new IllegalStateException(e);
+		}
+	}
+	// we're doing it reflectively so we don't expose this as a property to the user
+	private void propagateProducerPropertiesToFunction(ProducerProperties producerProperties) {
+		try {
+			Method setProducerProperties = ReflectionUtils.findMethod(StreamFunctionProperties.class, "setProducerProperties", ProducerProperties.class);
+			setProducerProperties.setAccessible(true);
+			setProducerProperties.invoke(this.streamFunctionProperties, producerProperties);
 		}
 		catch (Exception e) {
 			throw new IllegalStateException(e);

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java

@@ -42,6 +42,7 @@ import org.springframework.util.Assert;
  *
  * @author Oleg Zhurakousky
  * @author David Turanski
+ * @author Tolga Kavukcu
  *
  * @param <I> the payload type of the input Message
  * @param <O> the payload type of the output Message
@@ -73,12 +74,16 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 		this(functionProperties, functionCatalog, functionInspector, compositeMessageConverterFactory, null);
 	}
 
+	@SuppressWarnings("unchecked")
 	FunctionInvoker(StreamFunctionProperties functionProperties, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
 			CompositeMessageConverterFactory compositeMessageConverterFactory, MessageChannel errorChannel) {
-		this.userFunction = functionCatalog.lookup(functionProperties.getDefinition());
+		Object originalUserFunction = functionCatalog.lookup(functionProperties.getDefinition());
+
+		//TODO needs handling for when not a FluxWrapper
+		this.userFunction = (Function<Flux<?>, Flux<?>>) originalUserFunction;
 		Assert.isInstanceOf(Function.class, this.userFunction);
 		this.messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
-		FunctionType functionType = functionInspector.getRegistration(this.userFunction).getType();
+		FunctionType functionType = functionInspector.getRegistration(originalUserFunction).getType();
 		this.isInputArgumentMessage = functionType.isMessage();
 		this.inputClass = functionType.getInputType();
 		this.outputClass = functionType.getOutputType();
@@ -121,31 +126,36 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 
 	@SuppressWarnings("unchecked")
 	private <T> Message<O> toMessage(T value, Message<I> originalMessage) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Converting result back to message using the original message: " + originalMessage);
+		}
+
+		Message<O> returnMessage;
 		if(producerProperties.isUseNativeEncoding()){
 			if (logger.isDebugEnabled()) {
 				logger.debug("Native encoding enabled wrapping result to message using the original message: " + originalMessage);
 			}
-			return wrapOutputToMessage(value,originalMessage);
+			returnMessage = wrapOutputToMessage(value,originalMessage);
 		}
-		if (logger.isDebugEnabled()) {
-			logger.debug("Converting result back to message using the original message: " + originalMessage);
-		}
-		Message<O> returnMessage = (Message<O>)
-				(value instanceof Message
-						? value
-						: this.messageConverter.toMessage(value, originalMessage.getHeaders(), this.outputClass));
-		if (returnMessage == null) {
-			if (value.getClass().isAssignableFrom(this.outputClass)) {
-				returnMessage = wrapOutputToMessage(value, originalMessage);
+		else {
+			returnMessage = (Message<O>)
+					(value instanceof Message
+							? value
+							: this.messageConverter.toMessage(value, originalMessage.getHeaders(), this.outputClass));
+			if (returnMessage == null) {
+				if (value.getClass().isAssignableFrom(this.outputClass)) {
+					returnMessage = wrapOutputToMessage(value, originalMessage);
+				}
 			}
+			Assert.notNull(returnMessage, "Failed to convert result value '" + value + "' to message.");
 		}
-		Assert.notNull(returnMessage, "Failed to convert result value '" + value + "' to message.");
 		return returnMessage;
 	}
 
+	@SuppressWarnings("unchecked")
 	private <T> Message<O> wrapOutputToMessage(T value, Message<I> originalMessage) {
-		Message<O> returnMessage;
-		returnMessage = (Message<O>) MessageBuilder.withPayload(value).copyHeaders(originalMessage.getHeaders()).removeHeader(MessageHeaders.CONTENT_TYPE).build();
+		Message<O> returnMessage = (Message<O>) MessageBuilder.withPayload(value).copyHeaders(originalMessage.getHeaders())
+				.removeHeader(MessageHeaders.CONTENT_TYPE).build();
 		return returnMessage;
 	}
 

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java

@@ -23,6 +23,8 @@ import org.springframework.cloud.stream.binder.ProducerProperties;
 /**
  *
  * @author Oleg Zhurakousky
+ * @author Tolga Kavukcu
+ *
  * @since 2.1
  */
 @ConfigurationProperties("spring.cloud.stream.function")
@@ -50,7 +52,7 @@ public class StreamFunctionProperties {
 		return consumerProperties;
 	}
 
-	public void setConsumerProperties(ConsumerProperties consumerProperties) {
+	void setConsumerProperties(ConsumerProperties consumerProperties) {
 		this.consumerProperties = consumerProperties;
 	}
 
@@ -58,7 +60,7 @@ public class StreamFunctionProperties {
 		return producerProperties;
 	}
 
-	public void setProducerProperties(ProducerProperties producerProperties) {
+	void setProducerProperties(ProducerProperties producerProperties) {
 		this.producerProperties = producerProperties;
 	}
 }

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FuntionPropertiesTest.java

@@ -1,62 +0,0 @@
-/*
- * Copyright 2018 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
- *
- *      http://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.stream.function;
-
-
-
-import org.junit.Test;
-import org.junit.runner.RunWith;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.PropertySource;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/**
- *
- * @author Tolga Kavukcu
- *
- */
-@RunWith(SpringJUnit4ClassRunner.class)
-@SpringBootTest(classes = { FuntionPropertiesTest.MyFunctionsConfiguration.class })
-public class FuntionPropertiesTest {
-
-	@Autowired
-	StreamFunctionProperties streamFunctionProperties;
-
-	@Test
-	public void converterCorrectlyInstalled() {
-		assertThat(streamFunctionProperties.getProducerProperties().isUseNativeEncoding()).isEqualTo(Boolean.TRUE);
-		assertThat(streamFunctionProperties.getConsumerProperties().getMaxAttempts()).isEqualTo(5);
-	}
-
-	@EnableAutoConfiguration
-	@PropertySource("classpath:/org/springframework/cloud/stream/binder/cloud-function-test.properties")
-	@Configuration
-	public static class MyFunctionsConfiguration {
-
-		@Bean
-		public StreamFunctionProperties streamFunctionProperties() {
-			return new StreamFunctionProperties();
-		}
-	}
-}