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

GH-1581 Fixed multi-IntegrationFlow conflicts

Browse Source 
Removed dependency on IntegrationFlow from AbstractMessageChannelBinder as it was unnecessary
Fixed and re-enabled ignored tests

Resolves #1581
This commit is contained in:
Oleg Zhurakousky
2019-01-18 15:05:00 +01:00
parent 664b850996
commit 794c75f536
3 changed files with 24 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2016-2019 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.
@@ -51,7 +51,6 @@ 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;
@@ -120,8 +119,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
@Autowired(required = false)
private StreamFunctionProperties streamFunctionProperties;
@Autowired(required = false)
private IntegrationFlow integrationFlow;
private boolean producerBindingExist;
public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider) {
this(headersToEmbed, provisioningProvider, null);
@@ -197,7 +195,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
}
this.postProcessOutputChannel(outputChannel, producerProperties);
if (this.streamFunctionProperties != null && StringUtils.hasText(this.streamFunctionProperties.getDefinition()) && isInheritedIntegrationFlow()) {
if (shouldWireDunctionToChannel(true)) {
outputChannel = this.postProcessOutboundChannelForFunction(outputChannel, producerProperties);
}
@@ -232,6 +230,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
};
doPublishEvent(new BindingCreatedEvent(binding));
this.producerBindingExist = true;
return binding;
}
@@ -346,7 +345,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()) && isInheritedIntegrationFlow()) {
if (shouldWireDunctionToChannel(false)) {
inputChannel = this.postProcessInboundChannelForFunction(inputChannel, (ConsumerProperties) properties);
}
if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) {
@@ -818,9 +817,18 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
* 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 boolean shouldWireDunctionToChannel(boolean producer) {
if (!producer && this.producerBindingExist) {
return false;
}
else {
return this.streamFunctionProperties != null
&& StringUtils.hasText(this.streamFunctionProperties.getDefinition())
&& (
!this.getApplicationContext().containsBean("integrationFlowCreator")
|| this.getApplicationContext().getBean("integrationFlowCreator").equals(null)
);
}
}
private SubscribableChannel postProcessOutboundChannelForFunction(MessageChannel outputChannel, ProducerProperties producerProperties) {

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java
View File

@@ -116,7 +116,9 @@ public class GreenfieldFunctionEnableBindingTests {
restTemplate.postForLocation("http://localhost:8080", "hello");
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(10000).getPayload()).isEqualTo("HELLO".getBytes(StandardCharsets.UTF_8));
String result = new String(target.receive(10000).getPayload());
System.out.println(result);
assertThat(result).isEqualTo("HELLO");
}
}

9
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018 the original author or authors.
* Copyright 2019 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.
@@ -21,7 +21,6 @@ import java.util.function.Consumer;
import java.util.function.Function;
import org.junit.After;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
@@ -70,7 +69,6 @@ public class ProcessorToFunctionsSupportTests {
}
@Test
@Ignore
public void testSingleFunction() {
context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)).web(
@@ -84,7 +82,6 @@ public class ProcessorToFunctionsSupportTests {
}
@Test
@Ignore
public void testComposedFunction() {
context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)).web(
@@ -95,7 +92,9 @@ public class ProcessorToFunctionsSupportTests {
InputDestination source = context.getBean(InputDestination.class);
OutputDestination target = context.getBean(OutputDestination.class);
source.send(new GenericMessage<byte[]>("hello".getBytes(StandardCharsets.UTF_8)));
assertThat(target.receive(1000).getPayload()).isEqualTo("HELLO:HELLO".getBytes(StandardCharsets.UTF_8));
String result = new String(target.receive(1000).getPayload());
System.out.println(result);
assertThat(result).isEqualTo("HELLO:HELLO");
}
@Test