Initial commit for reactive binding support

*  New binding target factory for FluxMessageChannel
 *  Introduce a new property to activate reactive binding -- spring.cloud.stream.reactive
 *  Refactoring BindableFunctionProxyFactory for FluxMessageChannel target binding
 *  Similar refactoring in FunctionConfiguration
 *  Test to verify FluxMessageChannel binding

Addressing PR review comments

Addressing PR review comments

Addressing PR review comments.

Allow reactor and MC based binders co-exist in an application
in order to support multiple functions use different binders
(with different target types).

Addressing PR review comments.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2022 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.
@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder;
 import java.io.IOException;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
 
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -48,10 +49,12 @@ import org.springframework.expression.Expression;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.AbstractSubscribableChannel;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.handler.BridgeHandler;
 import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
@@ -224,8 +227,6 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	public final Binding<MessageChannel> doBindProducer(final String destination,
 			MessageChannel outputChannel, final P producerProperties)
 			throws BinderException {
-		Assert.isInstanceOf(SubscribableChannel.class, outputChannel,
-				"Binding is supported only for SubscribableChannel instances");
 		final MessageHandler producerMessageHandler;
 		final ProducerDestination producerDestination;
 		try {
@@ -259,12 +260,23 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 		this.postProcessOutputChannel(outputChannel, producerProperties);
 
-		((SubscribableChannel) outputChannel)
-				.subscribe(new SendingHandler(producerMessageHandler,
-						HeaderMode.embeddedHeaders
-								.equals(producerProperties.getHeaderMode()),
-						this.headersToEmbed, useNativeEncoding(producerProperties)));
+		AtomicReference<ReactiveStreamsConsumer> reactiveStreamsConsumerRef = new AtomicReference<>();
 
+		if (outputChannel instanceof SubscribableChannel) {
+			((SubscribableChannel) outputChannel)
+				.subscribe(new SendingHandler(producerMessageHandler,
+					HeaderMode.embeddedHeaders
+						.equals(producerProperties.getHeaderMode()),
+					this.headersToEmbed, useNativeEncoding(producerProperties)));
+		}
+		else if (outputChannel instanceof FluxMessageChannel) {
+			final ReactiveStreamsConsumer reactiveStreamsConsumer = new ReactiveStreamsConsumer(outputChannel, producerMessageHandler);
+			reactiveStreamsConsumerRef.set(reactiveStreamsConsumer);
+			reactiveStreamsConsumer.start();
+		}
+		else {
+			throw new IllegalStateException("No capable binding targets found.");
+		}
 
 		Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(destination,
 				outputChannel, producerMessageHandler instanceof Lifecycle
@@ -284,6 +296,10 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			public void afterUnbind() {
 				try {
 					destroyErrorInfrastructure(producerDestination);
+					final ReactiveStreamsConsumer rsc = reactiveStreamsConsumerRef.get();
+					if (rsc != null && rsc.isRunning()) {
+						rsc.destroy();
+					}
 					if (producerMessageHandler instanceof DisposableBean) {
 						((DisposableBean) producerMessageHandler).destroy();
 					}
@@ -1128,5 +1144,4 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			gen.writeString(value.getExpressionString());
 		}
 	}
-
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2021 the original author or authors.
+ * Copyright 2015-2022 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.
@@ -54,6 +54,7 @@ import org.springframework.core.convert.support.GenericConversionService;
 import org.springframework.core.env.ConfigurableEnvironment;
 import org.springframework.core.env.MapPropertySource;
 import org.springframework.core.env.StandardEnvironment;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.util.Assert;
@@ -192,7 +193,7 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
 						Class<?> binderType = GenericsUtils.getParameterType(
 								binderInstance.getClass(), Binder.class, 0);
 						if (binderType.isAssignableFrom(bindingTargetType)) {
-							candidatesForBindableType.add(defaultCandidateConfiguration);
+							populateCandidatesForBindableType(bindingTargetType, candidatesForBindableType, defaultCandidateConfiguration);
 						}
 					}
 					if (candidatesForBindableType.size() == 1) {
@@ -227,6 +228,17 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
 		return binderInstance;
 	}
 
+	private <T> void populateCandidatesForBindableType(Class<? extends T> bindingTargetType, List<String> candidatesForBindableType,
+													String defaultCandidateConfiguration) {
+		// Going by the convention of proper reactor based binders start with the key literal - reactor
+		if (FluxMessageChannel.class.isAssignableFrom(bindingTargetType) && defaultCandidateConfiguration.startsWith("reactor")) {
+			candidatesForBindableType.add(defaultCandidateConfiguration);
+		}
+		else if (!defaultCandidateConfiguration.startsWith("reactor")) {
+			candidatesForBindableType.add(defaultCandidateConfiguration);
+		}
+	}
+
 	/**
 	 * Return true if the binder is a {@link PollableConsumerBinder} and the target type
 	 * is a {@link PollableSource} and their generic types match (e.g. MessageHandler), OR

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/FluxMessageChannelBindingTargetFactory.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2022-2022 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.stream.binding;
+
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.integration.channel.FluxMessageChannel;
+
+/**
+ * @author Soby Chacko
+ * @since 4.0.0
+ */
+public class FluxMessageChannelBindingTargetFactory extends AbstractBindingTargetFactory<FluxMessageChannel> {
+
+	private final MessageChannelConfigurer messageChannelConfigurer;
+
+	private final GenericApplicationContext context;
+
+	public FluxMessageChannelBindingTargetFactory(MessageChannelConfigurer messageChannelConfigurer,
+												GenericApplicationContext context) {
+		super(FluxMessageChannel.class);
+		this.messageChannelConfigurer = messageChannelConfigurer;
+		this.context = context;
+	}
+
+	@Override
+	public FluxMessageChannel createInput(String name) {
+		FluxMessageChannel fluxMessageChannel = fluxMessageChannel(name);
+		this.messageChannelConfigurer.configureInputChannel(fluxMessageChannel, name);
+		return fluxMessageChannel;
+	}
+
+	@Override
+	public FluxMessageChannel createOutput(String name) {
+		FluxMessageChannel fluxMessageChannel = fluxMessageChannel(name);
+		this.messageChannelConfigurer.configureOutputChannel(fluxMessageChannel, name);
+		return fluxMessageChannel;
+	}
+
+	public FluxMessageChannel fluxMessageChannel(String name) {
+		FluxMessageChannel fluxMessageChannel = null;
+		if (context != null && context.containsBean(name)) {
+			try {
+				fluxMessageChannel = context.getBean(name, FluxMessageChannel.class);
+			}
+			catch (BeanCreationException e) {
+				// ignore
+			}
+		}
+		if (fluxMessageChannel == null) {
+			FluxMessageChannel channel = new FluxMessageChannel();
+			channel.setComponentName(name);
+			if (context != null && !context.containsBean(name)) {
+				context.registerBean(name, FluxMessageChannel.class, () -> channel);
+			}
+			fluxMessageChannel = channel;
+		}
+		return fluxMessageChannel;
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SupportedBindableFeatures.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2022-2022 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.stream.binding;
+
+/**
+ * Internal abstraction for the supported bindable features.
+ *
+ * @author Soby Chacko
+ * @since 4.0.0
+ */
+public class SupportedBindableFeatures {
+
+	private boolean pollable;
+
+	private boolean reactive;
+
+	public void setPollable(boolean pollable) {
+		this.pollable = pollable;
+	}
+
+	public void setReactive(boolean reactive) {
+		this.reactive = reactive;
+	}
+
+	public boolean isPollable() {
+		return pollable;
+	}
+
+	public boolean isReactive() {
+		return reactive;
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2021 the original author or authors.
+ * Copyright 2015-2022 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.
@@ -40,6 +40,7 @@ import org.springframework.cloud.stream.binder.BinderType;
 import org.springframework.cloud.stream.binder.BinderTypeRegistry;
 import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry;
 import org.springframework.cloud.stream.binding.CompositeMessageChannelConfigurer;
+import org.springframework.cloud.stream.binding.FluxMessageChannelBindingTargetFactory;
 import org.springframework.cloud.stream.binding.MessageChannelConfigurer;
 import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
 import org.springframework.cloud.stream.binding.MessageSourceBindingTargetFactory;
@@ -50,6 +51,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.context.annotation.Role;
+import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.UrlResource;
 import org.springframework.core.io.support.PropertiesLoaderUtils;
@@ -221,6 +223,13 @@ public class BinderFactoryAutoConfiguration {
 				compositeMessageChannelConfigurer);
 	}
 
+	@Bean
+	public FluxMessageChannelBindingTargetFactory fluxMessageChannelBindingTargetFactory(
+		CompositeMessageChannelConfigurer compositeMessageChannelConfigurer, GenericApplicationContext context) {
+		return new FluxMessageChannelBindingTargetFactory(
+			compositeMessageChannelConfigurer, context);
+	}
+
 	@Bean
 	public MessageSourceBindingTargetFactory messageSourceFactory(
 			@Qualifier(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME) CompositeMessageConverter compositeMessageConverter,

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

@@ -21,10 +21,11 @@ import org.springframework.beans.factory.FactoryBean;
 import org.springframework.cloud.stream.binder.PollableMessageSource;
 import org.springframework.cloud.stream.binding.BindableProxyFactory;
 import org.springframework.cloud.stream.binding.BoundTargetHolder;
+import org.springframework.cloud.stream.binding.SupportedBindableFeatures;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.support.GenericApplicationContext;
-import org.springframework.messaging.MessageChannel;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
@@ -55,22 +56,22 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory implement
 
 	private final StreamFunctionProperties functionProperties;
 
-	private final boolean pollable;
+	private final SupportedBindableFeatures supportedBindableFeatures;
 
 	private GenericApplicationContext context;
 
 	BindableFunctionProxyFactory(String functionDefinition, int inputCount, int outputCount, StreamFunctionProperties functionProperties) {
-		this(functionDefinition, inputCount, outputCount, functionProperties, false);
+		this(functionDefinition, inputCount, outputCount, functionProperties, new SupportedBindableFeatures());
 	}
 
 	BindableFunctionProxyFactory(String functionDefinition, int inputCount, int outputCount, StreamFunctionProperties functionProperties,
-			boolean pollable) {
+			SupportedBindableFeatures supportedBindableFeatures) {
 		super(null);
 		this.inputCount = inputCount;
 		this.outputCount = outputCount;
 		this.functionDefinition = functionDefinition;
 		this.functionProperties = functionProperties;
-		this.pollable = pollable;
+		this.supportedBindableFeatures = supportedBindableFeatures;
 	}
 
 	@Override
@@ -148,13 +149,18 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory implement
 		if (this.functionProperties.getBindings().containsKey(name)) {
 			name = this.functionProperties.getBindings().get(name);
 		}
-		if (this.pollable) {
+		if (this.supportedBindableFeatures.isPollable()) {
 			PollableMessageSource pollableSource = (PollableMessageSource) getBindingTargetFactory(PollableMessageSource.class).createInput(name);
 			if (context != null && !context.containsBean(name)) {
 				context.registerBean(name, PollableMessageSource.class, () -> pollableSource);
 			}
 			this.inputHolders.put(name, new BoundTargetHolder(pollableSource, true));
 		}
+		else if (this.supportedBindableFeatures.isReactive()) {
+			this.inputHolders.put(name,
+				new BoundTargetHolder(getBindingTargetFactory(FluxMessageChannel.class)
+					.createInput(name), true));
+		}
 		else {
 			this.inputHolders.put(name,
 					new BoundTargetHolder(getBindingTargetFactory(SubscribableChannel.class)
@@ -166,9 +172,16 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory implement
 		if (this.functionProperties.getBindings().containsKey(name)) {
 			name = this.functionProperties.getBindings().get(name);
 		}
-		this.outputHolders.put(name,
-				new BoundTargetHolder(getBindingTargetFactory(MessageChannel.class)
-						.createOutput(name), true));
+		if (this.supportedBindableFeatures.isReactive()) {
+			this.outputHolders.put(name,
+				new BoundTargetHolder(getBindingTargetFactory(FluxMessageChannel.class)
+					.createOutput(name), true));
+		}
+		else {
+			this.outputHolders.put(name,
+				new BoundTargetHolder(getBindingTargetFactory(SubscribableChannel.class)
+					.createOutput(name), true));
+		}
 	}
 
 	@Override

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 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.
@@ -41,7 +41,6 @@ import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoSink;
 import reactor.util.function.Tuples;
 
-
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.config.BeanDefinition;
@@ -70,6 +69,7 @@ import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.binding.BindableProxyFactory;
 import org.springframework.cloud.stream.binding.NewDestinationBindingCallback;
+import org.springframework.cloud.stream.binding.SupportedBindableFeatures;
 import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.BindingServiceConfiguration;
@@ -84,10 +84,12 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.core.env.ConfigurableEnvironment;
 import org.springframework.core.env.Environment;
 import org.springframework.core.type.MethodMetadata;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.AbstractSubscribableChannel;
+import org.springframework.integration.channel.FluxMessageChannel;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
@@ -418,14 +420,14 @@ public class FunctionConfiguration {
 				if (StringUtils.hasText(functionDefinition) && !shouldNotProcess) {
 					FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition);
 					if (function != null && !function.isSupplier()) {
-						this.bindFunctionToDestinations(bindableProxyFactory, functionDefinition);
+						this.bindFunctionToDestinations(bindableProxyFactory, functionDefinition, applicationContext.getEnvironment());
 					}
 				}
 			}
 		}
 
 		@SuppressWarnings({ "rawtypes", "unchecked" })
-		private void bindFunctionToDestinations(BindableProxyFactory bindableProxyFactory, String functionDefinition) {
+		private void bindFunctionToDestinations(BindableProxyFactory bindableProxyFactory, String functionDefinition, ConfigurableEnvironment environment) {
 			this.assertBindingIsPossible(bindableProxyFactory);
 
 
@@ -460,7 +462,13 @@ public class FunctionConfiguration {
 					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);
+					final Boolean reactive = functionProperties.getReactive().get(functionDefinition);
+					final boolean reactiveFn = reactive != null && reactive;
+					Class<?> bindableType = MessageChannel.class;
+					if (reactiveFn) {
+						bindableType = FluxMessageChannel.class;
+					}
+					Object binder = binderFactory.getBinder(binderConfigurationName, bindableType);
 					String targetProtocol = binder.getClass().getSimpleName().startsWith("Rabbit") ? "amqp" : "kafka";
 					Field headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");
 					headersField.setAccessible(true);
@@ -484,7 +492,7 @@ public class FunctionConfiguration {
 								+ "consumer, given that project reactor maintains its own concurrency mechanism. Was '..."
 								+ inputBindingName + ".consumer.concurrency=" + consumerProperties.getConcurrency() + "'");
 					}
-					SubscribableChannel inputChannel = this.applicationContext.getBean(inputBindingName, SubscribableChannel.class);
+					MessageChannel inputChannel = this.applicationContext.getBean(inputBindingName, MessageChannel.class);
 					return IntegrationReactiveUtils.messageChannelToFlux(inputChannel).map(m -> {
 						if (m instanceof Message) {
 							m = sanitize(m);
@@ -778,7 +786,10 @@ public class FunctionConfiguration {
 					functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(1);
 					functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(0);
 					functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(new StreamFunctionProperties());
-					functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(true);
+					final SupportedBindableFeatures supportedBindableFeatures = new SupportedBindableFeatures();
+					supportedBindableFeatures.setPollable(true);
+					supportedBindableFeatures.setReactive(false);
+					functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(supportedBindableFeatures);
 					((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(sourceName + "_binding", functionBindableProxyDefinition);
 				}
 			}
@@ -842,6 +853,15 @@ public class FunctionConfiguration {
 						functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.inputCount);
 						functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.outputCount);
 						functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.streamFunctionProperties);
+
+						final Map<String, Boolean> reactiveFunctions = streamFunctionProperties.getReactive();
+						final boolean reactiveFn = reactiveFunctions.get(functionDefinition) != null;
+						if (reactiveFn) {
+							final SupportedBindableFeatures supportedBindableFeatures = new SupportedBindableFeatures();
+							supportedBindableFeatures.setPollable(false);
+							supportedBindableFeatures.setReactive(true);
+							functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(supportedBindableFeatures);
+						}
 						registry.registerBeanDefinition(functionDefinition + "_binding", functionBindableProxyDefinition);
 					}
 					else {

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

@@ -50,6 +50,8 @@ public class StreamFunctionProperties {
 
 	private boolean composeFrom;
 
+	private Map<String, Boolean> reactive = new HashMap<>();
+
 	public boolean isComposeTo() {
 		return composeTo;
 	}
@@ -111,4 +113,12 @@ public class StreamFunctionProperties {
 				.collect(Collectors.toList());
 		return list;
 	}
+
+	public Map<String, Boolean> getReactive() {
+		return this.reactive;
+	}
+
+	public void setReactive(Map<String, Boolean> reactive) {
+		this.reactive = reactive;
+	}
 }

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/FluxMessageChannelBindingTests.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2022-2022 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.stream.binding;
+
+import java.util.function.Function;
+
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.channel.FluxMessageChannel;
+import org.springframework.messaging.MessageChannel;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Soby Chacko
+ */
+public class FluxMessageChannelBindingTests {
+
+	@Test
+	public void testFluxMessageChannelBindingWhenReactiveOptIn() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+			TestChannelBinderConfiguration.getCompleteConfiguration(ReactiveFunctionConfiguration.class))
+			.web(WebApplicationType.NONE)
+			.run("--spring.jmx.enabled=false",
+				"--spring.cloud.stream.function.reactive.uppercase=true")) {
+			assertThat(context.getBean("uppercase-in-0", MessageChannel.class)).isInstanceOf(FluxMessageChannel.class);
+			assertThat(context.getBean("uppercase-out-0", MessageChannel.class)).isInstanceOf(FluxMessageChannel.class);
+		}
+	}
+
+	@EnableAutoConfiguration
+	@Configuration
+	public static class ReactiveFunctionConfiguration {
+
+		@Bean
+		public Function<Flux<String>, Flux<String>> uppercase() {
+			return s -> s.map(String::toUpperCase);
+		}
+	}
+}