SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove internal dependency on BinderAwareChannelResolver

Browse Source 
Removed dependency in FunctioinConfiguration on BinderAwareChannelResolver
Improved StreamBridge to only register single pass-thru function to facilitate output type conversion

polish
This commit is contained in:
Oleg Zhurakousky
2020-05-12 20:21:02 +02:00
parent 8008ccaab9
commit 4914f8d9a9
3 changed files with 34 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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");
}

42
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
View File

@@ -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<String, SubscribableChannel> 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<Function<Object, Object>> fr = new FunctionRegistration<>(v -> v, bindingName);
this.functionRegistry.register(fr.type(FunctionType.from(Object.class).to(Object.class).message()));
messageChannel = new DirectWithAttributesChannel();
Binding<SubscribableChannel> binding = this.bindingService.bindProducer(messageChannel, bindingName, false);
this.channelCache.put(bindingName, messageChannel);
}
SubscribableChannel messageChannel = this.resolveDestination(bindingName, producerProperties);
Function<Object, Object> functionToInvoke = this.functionCatalog.lookup(bindingName, outputContentType.toString());
Function<Object, Object> 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<String, DirectWithAttributesChannel> channels = applicationContext.getBeansOfType(DirectWithAttributesChannel.class);
if (!CollectionUtils.isEmpty(channels)) { // single for all channel pass-through function to facilitate output conversion to byte[]
FunctionRegistration<Function<Object, Object>> fr = new FunctionRegistration<>(v -> v, STREAM_BRIDGE_FUNC_NAME);
this.functionRegistry.register(fr.type(FunctionType.from(Object.class).to(Object.class).message()));
}
for (Entry<String, DirectWithAttributesChannel> 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<Function<Object, Object>> 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;
}
}

2
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
View File

@@ -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");
}
}