From 669b820f989f7464b1e37aac9012e41f04b6c67a Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 2 Feb 2022 19:58:13 -0500 Subject: [PATCH] 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. --- .../binder/AbstractMessageChannelBinder.java | 33 ++++++--- .../stream/binder/DefaultBinderFactory.java | 16 +++- ...luxMessageChannelBindingTargetFactory.java | 74 +++++++++++++++++++ .../binding/SupportedBindableFeatures.java | 46 ++++++++++++ .../BinderFactoryAutoConfiguration.java | 11 ++- .../BindableFunctionProxyFactory.java | 31 +++++--- .../function/FunctionConfiguration.java | 34 +++++++-- .../function/StreamFunctionProperties.java | 10 +++ .../FluxMessageChannelBindingTests.java | 62 ++++++++++++++++ 9 files changed, 289 insertions(+), 28 deletions(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/FluxMessageChannelBindingTargetFactory.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SupportedBindableFeatures.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/FluxMessageChannelBindingTests.java diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index da0cf9c39..f56dd2194 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/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 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 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 binding = new DefaultBinding(destination, outputChannel, producerMessageHandler instanceof Lifecycle @@ -284,6 +296,10 @@ public abstract class AbstractMessageChannelBinder 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 void populateCandidatesForBindableType(Class bindingTargetType, List 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 diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/FluxMessageChannelBindingTargetFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/FluxMessageChannelBindingTargetFactory.java new file mode 100644 index 000000000..a0990a3e4 --- /dev/null +++ b/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 { + + 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; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SupportedBindableFeatures.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SupportedBindableFeatures.java new file mode 100644 index 000000000..b182c9ff3 --- /dev/null +++ b/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; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java index 5f0451077..602e76dd1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java +++ b/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, diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java index 8c605f08c..2e75bf079 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java +++ b/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 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 57171e140..ffdca9a3a 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 @@ -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 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 { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java index 3ab84e86c..3d60eba87 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java +++ b/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 reactive = new HashMap<>(); + public boolean isComposeTo() { return composeTo; } @@ -111,4 +113,12 @@ public class StreamFunctionProperties { .collect(Collectors.toList()); return list; } + + public Map getReactive() { + return this.reactive; + } + + public void setReactive(Map reactive) { + this.reactive = reactive; + } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/FluxMessageChannelBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/FluxMessageChannelBindingTests.java new file mode 100644 index 000000000..252f31251 --- /dev/null +++ b/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> uppercase() { + return s -> s.map(String::toUpperCase); + } + } +}