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

GH-1553 Fixed inconsistencies in function bootstrap

Browse Source 
Resolves #1553
This commit is contained in:
Oleg Zhurakousky
2018-12-11 19:45:44 +01:00
parent 1a7548fcf0
commit fd989cd712
4 changed files with 95 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -51,6 +51,7 @@ 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.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.handler.AbstractMessageHandler;
@@ -119,6 +120,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
@Autowired(required = false)
private StreamFunctionProperties streamFunctionProperties;
@Autowired(required = false)
private IntegrationFlow integrationFlow;
public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider) {
this(headersToEmbed, provisioningProvider, null);
}
@@ -193,7 +197,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
}
this.postProcessOutputChannel(outputChannel, producerProperties);
if (this.streamFunctionProperties != null && StringUtils.hasText(this.streamFunctionProperties.getDefinition())) {
if (this.streamFunctionProperties != null && StringUtils.hasText(this.streamFunctionProperties.getDefinition()) && isInheritedIntegrationFlow()) {
outputChannel = this.postProcessOutboundChannelForFunction(outputChannel, producerProperties);
}
@@ -342,7 +346,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
try {
ConsumerDestination destination = this.provisioningProvider.provisionConsumerDestination(name, group, properties);
// the function support for the inbound channel is only for Sink
if (this.streamFunctionProperties != null && StringUtils.hasText(this.streamFunctionProperties.getDefinition())) {
if (this.streamFunctionProperties != null && StringUtils.hasText(this.streamFunctionProperties.getDefinition()) && isInheritedIntegrationFlow()) {
inputChannel = this.postProcessInboundChannelForFunction(inputChannel, (ConsumerProperties) properties);
}
if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) {
@@ -806,6 +810,19 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
}
}
/*
* FUNCTION-TO-EXISTING-APP section
*
* To support composing functions into the existing apps.
* These methods do/should not participate in any way with general function
* bootstrap (e.g. brand new function based app). For that please see
* FunctionConfiguration.integrationFlowCreator
*/
private boolean isInheritedIntegrationFlow() {
return !this.getApplicationContext().containsBean("integrationFlowCreator") || integrationFlow == null;
}
private SubscribableChannel postProcessOutboundChannelForFunction(MessageChannel outputChannel, ProducerProperties producerProperties) {
if (this.integrationFlowFunctionSupport != null) {
Publisher<?> publisher = MessageChannelReactiveUtils.toPublisher(outputChannel);
@@ -846,22 +863,25 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
}
// we're doing it reflectively so we don't expose this as a property to the user
private void propagateConsumerPropertiesToFunction(ConsumerProperties consumerProperties) {
private void propagateProducerPropertiesToFunction(ProducerProperties producerProperties) {
try {
Method setConsumerProperties = ReflectionUtils.findMethod(StreamFunctionProperties.class, "setConsumerProperties", ConsumerProperties.class);
setConsumerProperties.setAccessible(true);
setConsumerProperties.invoke(this.streamFunctionProperties, consumerProperties);
Method setProducerProperties = ReflectionUtils.findMethod(StreamFunctionProperties.class,
"setProducerProperties", ProducerProperties.class);
setProducerProperties.setAccessible(true);
setProducerProperties.invoke(this.streamFunctionProperties, producerProperties);
}
catch (Exception e) {
throw new IllegalStateException(e);
}
}
// we're doing it reflectively so we don't expose this as a property to the user
private void propagateProducerPropertiesToFunction(ProducerProperties producerProperties) {
private void propagateConsumerPropertiesToFunction(ConsumerProperties consumerProperties) {
try {
Method setProducerProperties = ReflectionUtils.findMethod(StreamFunctionProperties.class, "setProducerProperties", ProducerProperties.class);
setProducerProperties.setAccessible(true);
setProducerProperties.invoke(this.streamFunctionProperties, producerProperties);
Method setConsumerProperties = ReflectionUtils.findMethod(StreamFunctionProperties.class,
"setConsumerProperties", ConsumerProperties.class);
setConsumerProperties.setAccessible(true);
setConsumerProperties.invoke(this.streamFunctionProperties, consumerProperties);
}
catch (Exception e) {
throw new IllegalStateException(e);
@@ -876,6 +896,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
}
}
// END FUNCTION-TO-EXISTING-APP section
private final class SendingHandler extends AbstractMessageHandler implements Lifecycle {
private final boolean embedHeaders;

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

@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.function;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -25,12 +26,17 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.catalog.FunctionInspector;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
/**
* @author Oleg Zhurakousky
@@ -49,12 +55,15 @@ public class FunctionConfiguration {
@Autowired(required = false)
private Processor processor;
@Autowired(required = false)
private Sink sink;
@Bean
public IntegrationFlowFunctionSupport functionSupport(FunctionCatalogWrapper functionCatalog,
FunctionInspector functionInspector, CompositeMessageConverterFactory messageConverterFactory,
StreamFunctionProperties functionProperties) {
StreamFunctionProperties functionProperties, BindingServiceProperties bindingServiceProperties) {
return new IntegrationFlowFunctionSupport(functionCatalog, functionInspector, messageConverterFactory,
functionProperties);
functionProperties, bindingServiceProperties);
}
@Bean
@@ -63,19 +72,42 @@ public class FunctionConfiguration {
}
/**
* This configuration creates an instance of {@link IntegrationFlow} appropriate for binding declared using EnableBinding.
* This configuration creates an instance of the {@link IntegrationFlow} from standard
* Spring Cloud Stream bindings such as {@link Source}, {@link Processor} and {@link Sink}
* ONLY if there are no existing instances of the {@link IntegrationFlow} already available
* in the context. This means that it only plays a role in green-field Spring Cloud Stream apps.
*
* For logic to compose functions into the existing apps please see "FUNCTION-TO-EXISTING-APP"
* section of AbstractMessageChannelBinder.
*
* The @ConditionalOnMissingBean ensures it does not collide with the the instance of the IntegrationFlow
* that may have been already defined by the existing (extended) app.
*/
@ConditionalOnMissingBean
@Bean
public IntegrationFlow integrationFlowCreator(IntegrationFlowFunctionSupport functionSupport) {
if (this.processor != null) {
return functionSupport.containsFunction(Function.class) ?
functionSupport.integrationFlowForFunction(this.processor.input(), this.processor.output()).get() : null;
if (functionSupport.containsFunction(Consumer.class) && consumerBindingPresent()) {
return functionSupport.integrationFlowForFunction(getInputChannel(), getOutputChannel()).get();
}
else if (this.source != null && this.processor == null) {
return functionSupport.containsFunction(Supplier.class) ?
functionSupport.integrationFlowFromNamedSupplier().channel(this.source.output()).get() : null;
else if (functionSupport.containsFunction(Function.class) && consumerBindingPresent()) {
return functionSupport.integrationFlowForFunction(getInputChannel(), getOutputChannel()).get();
}
else if (functionSupport.containsFunction(Supplier.class)) {
return functionSupport.integrationFlowFromNamedSupplier().channel(getOutputChannel()).get();
}
return null;
}
private boolean consumerBindingPresent() {
return this.processor != null || this.sink != null;
}
private SubscribableChannel getInputChannel() {
return this.processor != null ? this.processor.input() : this.sink.input();
}
private MessageChannel getOutputChannel() {
return this.processor != null ? this.processor.output()
: (this.source != null ? this.source.output() : new NullChannel());
}
}

23
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java
View File

@@ -29,7 +29,9 @@ import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.FunctionType;
import org.springframework.cloud.function.context.catalog.FunctionInspector;
import org.springframework.cloud.function.core.FluxSupplier;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.messaging.Message;
@@ -56,6 +58,8 @@ public class IntegrationFlowFunctionSupport {
private final StreamFunctionProperties functionProperties;
private final BindingServiceProperties bindingServiceProperties;
@Autowired
private MessageChannel errorChannel;
@@ -66,7 +70,8 @@ public class IntegrationFlowFunctionSupport {
* @param functionProperties
*/
IntegrationFlowFunctionSupport(FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
CompositeMessageConverterFactory messageConverterFactory, StreamFunctionProperties functionProperties) {
CompositeMessageConverterFactory messageConverterFactory, StreamFunctionProperties functionProperties,
BindingServiceProperties bindingServiceProperties) {
Assert.notNull(functionCatalog, "'functionCatalog' must not be null");
Assert.notNull(functionInspector, "'functionInspector' must not be null");
@@ -76,6 +81,7 @@ public class IntegrationFlowFunctionSupport {
this.functionInspector = functionInspector;
this.messageConverterFactory = messageConverterFactory;
this.functionProperties = functionProperties;
this.bindingServiceProperties = bindingServiceProperties;
}
/**
@@ -148,6 +154,21 @@ public class IntegrationFlowFunctionSupport {
public <O> IntegrationFlowBuilder integrationFlowForFunction(SubscribableChannel inputChannel,
MessageChannel outputChannel) {
if (inputChannel instanceof IntegrationObjectSupport) {
String inputBindingName = ((IntegrationObjectSupport)inputChannel).getComponentName();
if (StringUtils.hasText(inputBindingName)) {
this.functionProperties.setConsumerProperties(this.bindingServiceProperties.getConsumerProperties(inputBindingName));
}
}
if (outputChannel instanceof IntegrationObjectSupport) {
String outputBindingName = ((IntegrationObjectSupport)outputChannel).getComponentName();
if (StringUtils.hasText(outputBindingName)) {
this.functionProperties.setProducerProperties(this.bindingServiceProperties.getProducerProperties(outputBindingName));
}
}
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(inputChannel).bridge();
if (!this.andThenFunction(flowBuilder, outputChannel, this.functionProperties)) {

8
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
View File

@@ -26,7 +26,6 @@ import org.junit.Test;
import org.junit.rules.ExpectedException;
import reactor.core.publisher.Flux;
import org.springframework.beans.BeanInstantiationException;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.WebApplicationType;
@@ -49,10 +48,6 @@ import org.springframework.util.Assert;
import org.springframework.util.MimeTypeUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.hamcrest.CoreMatchers.allOf;
import static org.hamcrest.CoreMatchers.endsWith;
import static org.hamcrest.CoreMatchers.isA;
import static org.hamcrest.Matchers.hasProperty;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
@@ -169,9 +164,6 @@ public class SourceToFunctionsSupportTests {
public void testFunctionDoesNotExist() {
expectedException.expect(BeanCreationException.class);
expectedException.expectCause(
allOf(isA(BeanInstantiationException.class), hasProperty("cause", isA(IllegalArgumentException.class)),
hasProperty("message", endsWith("'doesNotExist' cannot be located."))));
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)).web(