diff --git a/pom.xml b/pom.xml
index 25ad5c116..8d9568019 100644
--- a/pom.xml
+++ b/pom.xml
@@ -25,7 +25,7 @@
 		<java.version>1.8</java.version>
 		<reactor.version>2020.0.0-RC2</reactor.version>
 		<objenesis.version>2.1</objenesis.version>
-		<spring-cloud-function.version>3.1.0-SNAPSHOT</spring-cloud-function.version>
+		<spring-cloud-function.version>3.1.0-RC1</spring-cloud-function.version>
 		<maven-checkstyle-plugin.failsOnError>true</maven-checkstyle-plugin.failsOnError>
 		<maven-checkstyle-plugin.failsOnViolation>true</maven-checkstyle-plugin.failsOnViolation>
 		<maven-checkstyle-plugin.includeTestSourceDirectory>true</maven-checkstyle-plugin.includeTestSourceDirectory>
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
index 762b2d46c..495407913 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
@@ -63,7 +63,9 @@ import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry
 import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration;
 import org.springframework.cloud.function.context.config.FunctionContextUtils;
 import org.springframework.cloud.function.context.config.RoutingFunction;
+import org.springframework.cloud.function.context.message.MessageUtils;
 import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.binder.BinderFactory;
 import org.springframework.cloud.stream.binder.BindingCreatedEvent;
 import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.ProducerProperties;
@@ -403,6 +405,7 @@ public class FunctionConfiguration {
 		private void bindFunctionToDestinations(BindableProxyFactory bindableProxyFactory, String functionDefinition) {
 			this.assertBindingIsPossible(bindableProxyFactory);
 
+
 			Set<String> inputBindingNames = bindableProxyFactory.getInputs();
 			Set<String> outputBindingNames = bindableProxyFactory.getOutputs();
 
@@ -428,6 +431,24 @@ public class FunctionConfiguration {
 			}
 
 			if (isReactiveOrMultipleInputOutput(bindableProxyFactory, functionType)) {
+				//String targetProtocol = null;
+				AtomicReference<Function<Message<?>, Message<?>>> targetProtocolEnhancer = new AtomicReference<>();
+				if (!CollectionUtils.isEmpty(outputBindingNames)) {
+					String outputBindingName = outputBindingNames.iterator().next(); // TODO only gets the first one
+					String binderConfigurationName = this.serviceProperties.getBinder(outputBindingName);
+					BinderFactory binderFactory = applicationContext.getBean(BinderFactory.class);
+					Object binder = binderFactory.getBinder(binderConfigurationName, MessageChannel.class);
+					String targetProtocol = binder.getClass().getSimpleName().startsWith("Rabbit") ? "amqp" : "kafka";
+					Field headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");
+					headersField.setAccessible(true);
+					targetProtocolEnhancer.set(message -> {
+						Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
+								.getField(headersField, ((Message) message).getHeaders());
+						headersMap.putIfAbsent(MessageUtils.TARGET_PROTOCOL, targetProtocol);
+						return message;
+					});
+				}
+
 				Publisher[] inputPublishers = inputBindingNames.stream().map(inputBindingName -> {
 					BindingProperties bindingProperties = this.serviceProperties.getBindings().get(inputBindingName);
 					ConsumerProperties consumerProperties = bindingProperties == null ? null : bindingProperties.getConsumer();
@@ -439,7 +460,16 @@ public class FunctionConfiguration {
 					}
 					SubscribableChannel inputChannel = this.applicationContext.getBean(inputBindingName, SubscribableChannel.class);
 					return IntegrationReactiveUtils.messageChannelToFlux(inputChannel);
-				}).toArray(Publisher[]::new);
+				})
+				.map(publisher -> {
+					if (targetProtocolEnhancer.get() != null) {
+						return publisher.map(targetProtocolEnhancer.get());
+					}
+					else {
+						return publisher;
+					}
+				})
+				.toArray(Publisher[]::new);
 
 				Function functionToInvoke = function;
 				if (!CollectionUtils.isEmpty(outputBindingNames)) {
@@ -481,7 +511,8 @@ public class FunctionConfiguration {
 								}
 								outputChannel.send((Message) message);
 							}
-						}).doOnError(e -> {
+						})
+						.doOnError(e -> {
 							logger.error("Failure was detected during execution of the reactive function '" +  functionDefinition + "'");
 							((Throwable) e).printStackTrace();
 						});
@@ -513,8 +544,9 @@ public class FunctionConfiguration {
 			ProducerProperties producerProperties = StringUtils.hasText(outputChannelName)
 					? this.serviceProperties.getBindingProperties(outputChannelName).getProducer()
 							: null;
+
 			ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function, consumerProperties,
-					producerProperties, applicationContext)) {
+					producerProperties, applicationContext, this.determineTargetProtocol(outputChannelName))) {
 				@Override
 				protected void sendOutputs(Object result, Message<?> requestMessage) {
 					if (result instanceof Iterable) {
@@ -551,7 +583,16 @@ public class FunctionConfiguration {
 			return handler;
 		}
 
-
+		private String determineTargetProtocol(String outputBindingName) {
+			if (StringUtils.hasText(outputBindingName)) {
+				String binderConfigurationName = this.serviceProperties.getBinder(outputBindingName);
+				BinderFactory binderFactory = applicationContext.getBean(BinderFactory.class);
+				Object binder = binderFactory.getBinder(binderConfigurationName, MessageChannel.class);
+				String protocol = binder.getClass().getSimpleName().startsWith("Rabbit") ? "amqp" : "kafka";
+				return protocol;
+			}
+			return null;
+		}
 
 		private boolean isReactiveOrMultipleInputOutput(BindableProxyFactory bindableProxyFactory, Type functionType) {
 			boolean reactiveInputsOutputs = FunctionTypeUtils.isPublisher(FunctionTypeUtils.getInputType(functionType)) ||
@@ -624,8 +665,10 @@ public class FunctionConfiguration {
 
 		private final boolean isRoutingFunction;
 
+		private final String targetProtocol;
+
 		FunctionWrapper(Function function, ConsumerProperties consumerProperties,
-				ProducerProperties producerProperties, ConfigurableApplicationContext applicationContext) {
+				ProducerProperties producerProperties, ConfigurableApplicationContext applicationContext, String targetProtocol) {
 
 			isRoutingFunction = ((FunctionInvocationWrapper) function).getTarget() instanceof RoutingFunction;
 			this.applicationContext = applicationContext;
@@ -640,17 +683,20 @@ public class FunctionConfiguration {
 			}
 			this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");
 			this.headersField.setAccessible(true);
+			this.targetProtocol = targetProtocol;
 		}
 
 		@SuppressWarnings("unchecked")
 		@Override
 		public Object apply(Message<byte[]> message) {
+			Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
+					.getField(this.headersField, message.getHeaders());
+			if (StringUtils.hasText(targetProtocol)) {
+				headersMap.putIfAbsent(MessageUtils.TARGET_PROTOCOL, targetProtocol);
+			}
 			if (message != null && consumerProperties != null) {
-				Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
-						.getField(this.headersField, message.getHeaders());
 				headersMap.put(FunctionProperties.SKIP_CONVERSION_HEADER, consumerProperties.isUseNativeDecoding());
 			}
-
 			Object result = function.apply(message);
 			if (result instanceof Publisher && this.isRoutingFunction) {
 				throw new IllegalStateException("Routing to functions that return Publisher "
@@ -745,7 +791,6 @@ public class FunctionConfiguration {
 							functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.streamFunctionProperties);
 							try {
 								String name = functionDefinition + "_binding";
-								System.out.println(registry.containsBeanDefinition(name));
 								registry.registerBeanDefinition(name, functionBindableProxyDefinition);
 							}
 							catch (Exception e) {