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 0cdc3a4b3..ff4474b60 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 @@ -64,7 +64,6 @@ import org.springframework.cloud.stream.binder.BindingCreatedEvent; 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.BinderAwareChannelResolver; import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration; import org.springframework.cloud.stream.config.BindingBeansRegistrar; import org.springframework.cloud.stream.config.BindingProperties; @@ -109,7 +108,6 @@ import org.springframework.util.StringUtils; * @author Ilayaperumal Gopinathan * @since 2.1 */ -@SuppressWarnings("deprecation") @Configuration @EnableConfigurationProperties(StreamFunctionProperties.class) @Import({ BindingBeansRegistrar.class, BinderFactoryAutoConfiguration.class }) @@ -121,7 +119,6 @@ public class FunctionConfiguration { private final static String SOURCE_PROPERY = "spring.cloud.stream.source"; @Bean -// @ConditionalOnProperty(SOURCE_PROPERY) public StreamBridge streamBridgeUtils(FunctionCatalog functionCatalog, FunctionRegistry functionRegistry, BindingServiceProperties bindingServiceProperties, ConfigurableApplicationContext applicationContext) { return new StreamBridge(functionCatalog, functionRegistry, bindingServiceProperties, applicationContext); @@ -137,14 +134,14 @@ public class FunctionConfiguration { public InitializingBean functionInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector, StreamFunctionProperties functionProperties, @Nullable BindableProxyFactory[] bindableProxyFactories, BindingServiceProperties serviceProperties, ConfigurableApplicationContext applicationContext, - FunctionBindingRegistrar bindingHolder, BinderAwareChannelResolver dynamicDestinationResolver) { + FunctionBindingRegistrar bindingHolder, StreamBridge streamBridge) { boolean shouldCreateInitializer = applicationContext.containsBean("output") || ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class)); return shouldCreateInitializer ? new FunctionToDestinationBinder(functionCatalog, functionProperties, - serviceProperties, dynamicDestinationResolver) + serviceProperties, streamBridge) : null; } @@ -154,7 +151,7 @@ public class FunctionConfiguration { @Bean InitializingBean supplierInitializer(FunctionCatalog functionCatalog, StreamFunctionProperties functionProperties, GenericApplicationContext context, BindingServiceProperties serviceProperties, - @Nullable BindableFunctionProxyFactory[] proxyFactories, BinderAwareChannelResolver dynamicDestinationResolver, + @Nullable BindableFunctionProxyFactory[] proxyFactories, StreamBridge streamBridge, TaskScheduler taskScheduler) { if (!ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class)) || proxyFactories == null) { @@ -193,7 +190,8 @@ public class FunctionConfiguration { .route(Message.class, message -> { if (message.getHeaders().get("spring.cloud.stream.sendto.destination") != null) { String destinationName = (String) message.getHeaders().get("spring.cloud.stream.sendto.destination"); - return dynamicDestinationResolver.resolveDestination(destinationName); + return streamBridge.resolveDestination(destinationName, producerProperties); + //return dynamicDestinationResolver.resolveDestination(destinationName); } return outputName; }).get(); @@ -305,14 +303,14 @@ public class FunctionConfiguration { private final BindingServiceProperties serviceProperties; - private final BinderAwareChannelResolver dynamicDestinationResolver; + private final StreamBridge streamBridge; FunctionToDestinationBinder(FunctionCatalog functionCatalog, StreamFunctionProperties functionProperties, - BindingServiceProperties serviceProperties, BinderAwareChannelResolver dynamicDestinationResolver) { + BindingServiceProperties serviceProperties, StreamBridge streamBridge) { this.functionCatalog = functionCatalog; this.functionProperties = functionProperties; this.serviceProperties = serviceProperties; - this.dynamicDestinationResolver = dynamicDestinationResolver; + this.streamBridge = streamBridge; } @Override @@ -441,7 +439,7 @@ public class FunctionConfiguration { protected void sendOutputs(Object result, Message requestMessage) { if (result instanceof Message && ((Message) result).getHeaders().get("spring.cloud.stream.sendto.destination") != null) { String destinationName = (String) ((Message) result).getHeaders().get("spring.cloud.stream.sendto.destination"); - MessageChannel outputChannel = dynamicDestinationResolver.resolveDestination(destinationName); + SubscribableChannel outputChannel = streamBridge.resolveDestination(destinationName, producerProperties); if (logger.isInfoEnabled()) { logger.info("Output message is sent to '" + destinationName + "' destination"); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index 228710be1..0bdb65333 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -31,7 +31,6 @@ import org.springframework.cloud.function.context.FunctionRegistration; import org.springframework.cloud.function.context.FunctionRegistry; import org.springframework.cloud.function.context.FunctionType; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper; -import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.binding.BindingService; import org.springframework.cloud.stream.config.BindingServiceProperties; @@ -39,6 +38,7 @@ import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.messaging.Message; import org.springframework.messaging.SubscribableChannel; +import org.springframework.util.CollectionUtils; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; @@ -60,7 +60,9 @@ import org.springframework.util.MimeTypeUtils; */ public final class StreamBridge implements SmartInitializingSingleton { - protected final Log logger = LogFactory.getLog(getClass()); + private static String STREAM_BRIDGE_FUNC_NAME = "streamBridge"; + + private final Log logger = LogFactory.getLog(getClass()); private final Map channelCache; @@ -74,6 +76,7 @@ public final class StreamBridge implements SmartInitializingSingleton { private boolean initialized; + @Autowired private BindingService bindingService; @@ -128,20 +131,12 @@ public final class StreamBridge implements SmartInitializingSingleton { * @param outputContentType content type to be used to deal with output type conversion * @return true if data was sent successfully, otherwise false or throws an exception. */ - @SuppressWarnings({ "unchecked", "unused" }) + @SuppressWarnings("unchecked") public boolean send(String bindingName, Object data, MimeType outputContentType) { - SubscribableChannel messageChannel = this.channelCache.get(bindingName); ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName); - if (messageChannel == null) { - producerProperties.setRequiredGroups(bindingName); - FunctionRegistration> fr = new FunctionRegistration<>(v -> v, bindingName); - this.functionRegistry.register(fr.type(FunctionType.from(Object.class).to(Object.class).message())); - messageChannel = new DirectWithAttributesChannel(); - Binding binding = this.bindingService.bindProducer(messageChannel, bindingName, false); - this.channelCache.put(bindingName, messageChannel); - } + SubscribableChannel messageChannel = this.resolveDestination(bindingName, producerProperties); - Function functionToInvoke = this.functionCatalog.lookup(bindingName, outputContentType.toString()); + Function functionToInvoke = this.functionCatalog.lookup(STREAM_BRIDGE_FUNC_NAME, outputContentType.toString()); if (producerProperties != null && producerProperties.isPartitioned()) { functionToInvoke = new PartitionAwareFunctionWrapper((FunctionInvocationWrapper) functionToInvoke, this.applicationContext, producerProperties); } @@ -156,15 +151,26 @@ public final class StreamBridge implements SmartInitializingSingleton { return; } Map channels = applicationContext.getBeansOfType(DirectWithAttributesChannel.class); + if (!CollectionUtils.isEmpty(channels)) { // single for all channel pass-through function to facilitate output conversion to byte[] + FunctionRegistration> fr = new FunctionRegistration<>(v -> v, STREAM_BRIDGE_FUNC_NAME); + this.functionRegistry.register(fr.type(FunctionType.from(Object.class).to(Object.class).message())); + } for (Entry channelEntry : channels.entrySet()) { if (channelEntry.getValue().getAttribute("type").equals("output")) { this.channelCache.put(channelEntry.getKey(), channelEntry.getValue()); - // we're registering a dummy pass-through function to ensure that it goes through the - // same process (type conversion, etc) as other function invocation. - FunctionRegistration> fr = new FunctionRegistration<>(v -> v, channelEntry.getKey()); - this.functionRegistry.register(fr.type(FunctionType.from(Object.class).to(Object.class).message())); - this.initialized = true; } } + this.initialized = true; + } + + SubscribableChannel resolveDestination(String destinationName, ProducerProperties producerProperties) { + SubscribableChannel messageChannel = this.channelCache.get(destinationName); + if (messageChannel == null) { + producerProperties.setRequiredGroups(destinationName); + messageChannel = new DirectWithAttributesChannel(); + this.bindingService.bindProducer(messageChannel, destinationName, false); + this.channelCache.put(destinationName, messageChannel); + } + return messageChannel; } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java index 2ae1750ec..639d98968 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java @@ -357,7 +357,7 @@ public class ImplicitFunctionBindingTests { TestChannelBinderConfiguration.getCompleteConfiguration(LegacyConfiguration.class)) .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false")) { - assertThat(context.getBean("supplierInitializer")).isEqualTo(null); + assertThat(context.getBean("supplierInitializer").getClass().getSimpleName()).isEqualTo("NullBean"); } }