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

GH-1465 Added Function support to channel binder

Browse Source 
Move IntegrationFlowFunctionSupport injection to AbstractBinder

Add function support in TestSupportBinder and fix *FunctionsSupportTests

Use BeanPostProcessor and revert TestSupportBinder

Clean up

Set errorChannel in afterPropertiesSet

Refactor IntegrationFlowFunctionSupportWiring

Close context

Revert context.close()
This commit is contained in:
David Turanski
2018-08-31 18:36:02 -04:00
committed by Oleg Zhurakousky
parent 268b0197ff
commit c3fdbb93dc
7 changed files with 159 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java
View File

@@ -49,7 +49,7 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
*/
public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends ProducerProperties>
implements ApplicationContextAware, InitializingBean, Binder<T, C, P> {
implements ApplicationContextAware, InitializingBean, Binder<T, C, P> {
/**
* The delimiter between a group and index when constructing a binder
@@ -61,10 +61,9 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
private volatile AbstractApplicationContext applicationContext;
private volatile EvaluationContext evaluationContext;
@Autowired(required=false) // this would need to be refactored into constructor in the future
@Autowired(required = false) // this would need to be refactored into constructor in the future
@StreamRetryTemplate
private RetryTemplate consumerBindingRetryTemplate;
@@ -72,7 +71,7 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
* For binder implementations that support a prefix, apply the prefix to the name.
*
* @param prefix the prefix.
* @param name the name.
* @param name the name.
*/
public static String applyPrefix(String prefix, String name) {
return prefix + name;
@@ -143,7 +142,7 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
/**
* Construct a name comprised of the name and group.
*
* @param name the name.
* @param name the name.
* @param group the group.
* @return the constructed name.
*/

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

@@ -20,14 +20,16 @@ import java.util.LinkedHashMap;
import java.util.Map;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.logging.Log;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.DefaultSingletonBeanRegistry;
import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
import org.springframework.cloud.stream.function.BinderFunctionSupport;
import org.springframework.cloud.stream.function.IntegrationFlowFunctionSupport;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.cloud.stream.provisioning.ProvisioningException;
@@ -101,6 +103,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
private ApplicationEventPublisher applicationEventPublisher;
private IntegrationFlowFunctionSupport integrationFlowFunctionSupport;
public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider) {
this(headersToEmbed, provisioningProvider, null);
}
@@ -173,12 +177,18 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
if (producerMessageHandler instanceof Lifecycle) {
((Lifecycle) producerMessageHandler).start();
}
postProcessOutputChannel(outputChannel, producerProperties);
outputChannel = BinderFunctionSupport.andThenFunctionDefinition(this.integrationFlowFunctionSupport,
outputChannel);
((SubscribableChannel) outputChannel).subscribe(
new SendingHandler(producerMessageHandler, HeaderMode.embeddedHeaders
.equals(producerProperties.getHeaderMode()), this.headersToEmbed,
useNativeEncoding(producerProperties)));
Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(destination, outputChannel,
producerMessageHandler instanceof Lifecycle ? (Lifecycle) producerMessageHandler : null) {
@@ -207,6 +217,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
return binding;
}
/**
* Whether the producer for the destination being created should be configured to use
* native encoding which may, or may not, be determined from the properties. For
@@ -728,6 +740,18 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
return destination.getName() + ".errors";
}
@Override
protected void onInit() throws Exception {
super.onInit();
try {
this.integrationFlowFunctionSupport = getApplicationContext().getBean(IntegrationFlowFunctionSupport.class);
}
catch (NoSuchBeanDefinitionException e) {
//ignore
}
}
private Map<String, Object> doGetExtendedInfo(Object destination, Object properties) {
Map<String, Object> extendedInfo = new LinkedHashMap<>();
extendedInfo.put("bindingDestination", destination.toString());

42
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BinderFunctionSupport.java Normal file
View File

@@ -0,0 +1,42 @@
/*
* Copyright 2018 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
*
* http://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.function;
import java.util.function.Function;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannelReactiveUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
/**
* @author David Turanski
* @since 2.1
**/
public abstract class BinderFunctionSupport {
public static SubscribableChannel andThenFunctionDefinition(IntegrationFlowFunctionSupport
integrationFlowFunctionSupport, MessageChannel outputChannel) {
if (integrationFlowFunctionSupport != null && integrationFlowFunctionSupport.containsFunction(Function.class)) {
DirectChannel actualOutputChannel = new DirectChannel();
integrationFlowFunctionSupport.andThenFunction(MessageChannelReactiveUtils.toPublisher(outputChannel),
actualOutputChannel);
return actualOutputChannel;
}
return (SubscribableChannel) outputChannel;
}
}

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

@@ -32,30 +32,29 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.integration.dsl.IntegrationFlow;
/**
*
* @author Oleg Zhurakousky
*
* @author David Turanski
* @since 2.1
*/
@Configuration
@ConditionalOnProperty("spring.cloud.stream.function.definition")
public class FunctionConfiguration {
@Autowired(required=false)
@Autowired(required = false)
private Source source;
@Autowired(required=false)
@Autowired(required = false)
private Processor processor;
@Autowired(required=false)
@Autowired(required = false)
private Sink sink;
@Bean
public IntegrationFlowFunctionSupport functionSupport(FunctionCatalogWrapper functionCatalog,
FunctionInspector functionInspector, CompositeMessageConverterFactory messageConverterFactory,
StreamFunctionProperties functionProperties) {
FunctionInspector functionInspector, CompositeMessageConverterFactory messageConverterFactory,
StreamFunctionProperties functionProperties) {
return new IntegrationFlowFunctionSupport(functionCatalog, functionInspector, messageConverterFactory,
functionProperties);
functionProperties);
}
@Bean
@@ -67,7 +66,8 @@ public class FunctionConfiguration {
* This configuration creates an instance of {@link IntegrationFlow} appropriate for binding declared using EnableBinding.
* At the moment only Source, Processor and Sink are supported.
*/
@ConditionalOnMissingBean // starter apps typically already provide and instance of IntegrationFlow, so we don't need this one.
@ConditionalOnMissingBean
// starter apps typically already provide and instance of IntegrationFlow, so we don't need this one.
@Bean
public IntegrationFlow integrationFlowCreator(IntegrationFlowFunctionSupport functionSupport) {
if (processor != null) {
@@ -77,10 +77,11 @@ public class FunctionConfiguration {
return functionSupport.integrationFlowForFunction(sink.input(), null).get();
}
else if (source != null) {
return functionSupport.containsFunction(Supplier.class)
? functionSupport.integrationFlowFromNamedSupplier().channel(this.source.output()).get()
: null;
return functionSupport.containsFunction(Supplier.class) ?
functionSupport.integrationFlowFromNamedSupplier().channel(this.source.output()).get() :
null;
}
throw new UnsupportedOperationException("Bindings other then Source, Processor and Sink are not currently supported");
throw new UnsupportedOperationException(
"Bindings other then Source, Processor and Sink are not currently supported");
}
}

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

@@ -21,17 +21,19 @@ import java.util.function.Function;
import java.util.function.Supplier;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
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.converter.CompositeMessageConverterFactory;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.messaging.Message;
@@ -44,10 +46,11 @@ import org.springframework.util.StringUtils;
*
* @author Oleg Zhurakousky
* @author David Turanski
* @author Ilayaperumal Gopinathan
*
* @since 2.1
*/
public class IntegrationFlowFunctionSupport {
public class IntegrationFlowFunctionSupport implements ApplicationContextAware, InitializingBean {
private final FunctionCatalogWrapper functionCatalog;
@@ -57,8 +60,8 @@ public class IntegrationFlowFunctionSupport {
private final StreamFunctionProperties functionProperties;
@Autowired
private MessageChannel errorChannel;
private ApplicationContext applicationContext;
/**
* @param functionCatalog
@@ -92,7 +95,8 @@ public class IntegrationFlowFunctionSupport {
}
public FunctionType getCurrentFunctionType() {
FunctionType functionType = functionInspector.getRegistration(functionCatalog.lookup(this.functionProperties.getDefinition())).getType();
FunctionType functionType = functionInspector.getRegistration(
functionCatalog.lookup(this.functionProperties.getDefinition())).getType();
return functionType;
}
@@ -161,33 +165,20 @@ public class IntegrationFlowFunctionSupport {
* @return true if {@link Function} was located and added and false if it wasn't.
*/
public <I,O> boolean andThenFunction(IntegrationFlowBuilder flowBuilder, MessageChannel outputChannel) {
if (StringUtils.hasText(this.functionProperties.getDefinition())) {
FunctionInvoker<I,O> functionInvoker =
new FunctionInvoker<>(this.functionProperties.getDefinition(), this.functionCatalog,
this.functionInspector, this.messageConverterFactory, this.errorChannel);
if (outputChannel != null) {
subscribeToInput(functionInvoker, flowBuilder.toReactivePublisher(), outputChannel::send);
}
else {
subscribeToInput(functionInvoker, flowBuilder.toReactivePublisher(), null);
}
return true;
}
return false;
return andThenFunction(flowBuilder.toReactivePublisher(), outputChannel);
}
public <I,O> boolean andThenFunction(FluxMessageChannel fluxChannel, MessageChannel outputChannel) {
public <I,O> boolean andThenFunction(Publisher<?> publisher, MessageChannel outputChannel) {
if (StringUtils.hasText(this.functionProperties.getDefinition())) {
FunctionInvoker<I,O> functionInvoker =
FunctionInvoker<I, O> functionInvoker =
new FunctionInvoker<>(this.functionProperties.getDefinition(), this.functionCatalog,
this.functionInspector, this.messageConverterFactory, this.errorChannel);
if (outputChannel != null) {
subscribeToInput(functionInvoker, fluxChannel, outputChannel::send);
subscribeToInput(functionInvoker, publisher, outputChannel::send);
}
else {
subscribeToInput(functionInvoker, fluxChannel, null);
subscribeToInput(functionInvoker, publisher, null);
}
return true;
}
@@ -211,4 +202,13 @@ public class IntegrationFlowFunctionSupport {
subscribeToOutput(outputProcessor, functionInvoker.apply((Flux<Message<I>>) inputPublisher)).subscribe();
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
@Override
public void afterPropertiesSet() throws Exception {
this.errorChannel = (MessageChannel) applicationContext.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
}
}

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

@@ -16,12 +16,11 @@
package org.springframework.cloud.stream.function;
import java.nio.charset.StandardCharsets;
import java.util.function.Consumer;
import java.util.function.Function;
import org.junit.After;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
@@ -34,34 +33,35 @@ import org.springframework.cloud.stream.binder.test.InputDestination;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat;
/**
*
* @author Oleg Zhurakousky
*
* @author David Turanski
* @since 2.1
*
*/
public class ProcessorToFunctionsSupportTests {
private ConfigurableApplicationContext context;
@After
public void cleanUp() {
context.close();
}
@Test
public void testPathThrough() {
ApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.jmx.enabled=false");
context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)).web(
WebApplicationType.NONE).run("--spring.jmx.enabled=false");
InputDestination source = context.getBean(InputDestination.class);
OutputDestination target = context.getBean(OutputDestination.class);
source.send(new GenericMessage<byte[]>("hello".getBytes(StandardCharsets.UTF_8)));
@@ -70,11 +70,10 @@ public class ProcessorToFunctionsSupportTests {
@Test
public void testSingleFunction() {
ApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=toUpperCase", "--spring.jmx.enabled=false");
context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)).web(
WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=toUpperCase", "--spring.jmx.enabled=false");
InputDestination source = context.getBean(InputDestination.class);
OutputDestination target = context.getBean(OutputDestination.class);
@@ -84,11 +83,11 @@ public class ProcessorToFunctionsSupportTests {
@Test
public void testComposedFunction() {
ApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf", "--spring.jmx.enabled=false");
context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)).web(
WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf",
"--spring.jmx" + ".enabled=false", "--logging.level.org.springframework.integration=DEBUG");
InputDestination source = context.getBean(InputDestination.class);
OutputDestination target = context.getBean(OutputDestination.class);
@@ -98,11 +97,9 @@ public class ProcessorToFunctionsSupportTests {
@Test
public void testConsumer() {
ApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(ConsumerConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=log", "--spring.jmx.enabled=false");
context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(ConsumerConfiguration.class)).web(
WebApplicationType.NONE).run("--spring.cloud.stream.function.definition=log", "--spring.jmx.enabled=false");
InputDestination source = context.getBean(InputDestination.class);
OutputDestination target = context.getBean(OutputDestination.class);
@@ -158,20 +155,10 @@ public class ProcessorToFunctionsSupportTests {
private Processor processor;
@Bean
public IntegrationFlow fromChannel(@Nullable IntegrationFlowFunctionSupport functionSupport) {
public IntegrationFlow fromChannel() {
IntegrationFlowBuilder flowBuilder = null;
if (functionSupport == null) {
flowBuilder = IntegrationFlows.from(processor.input()).bridge().channel(processor.output());
}
else {
flowBuilder = functionSupport.integrationFlowFromChannel(processor.input());
if (!functionSupport.andThenFunction(flowBuilder, processor.output())) {
flowBuilder = flowBuilder.channel(processor.output());
}
}
return flowBuilder.get();
return IntegrationFlows.from(processor.input())
.channel(processor.output()).get();
}
}

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

@@ -41,7 +41,6 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
@@ -84,7 +83,8 @@ public class SourceToFunctionsSupportTests {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)).web(
WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf", "--spring.jmx.enabled=false")) {
.run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf",
"--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(1000).getPayload()).isEqualTo(
"HELLO FUNCTION:HELLO FUNCTION".getBytes(StandardCharsets.UTF_8));
@@ -93,12 +93,11 @@ public class SourceToFunctionsSupportTests {
@Test
public void testFailedInputTypeConversion() {
try (ConfigurableApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfigurationNoConversionPossible.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf",
"--spring.jmx.enabled=false")) {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
FunctionsConfigurationNoConversionPossible.class)).web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf",
"--spring.jmx.enabled=false")) {
PollableChannel errorChannel = context.getBean("errorChannel", PollableChannel.class);
OutputDestination target = context.getBean(OutputDestination.class);
assertNull(target.receive(1000));
@@ -108,12 +107,11 @@ public class SourceToFunctionsSupportTests {
@Test
public void testComposedFunctionIsAppliedToExistingMessageSourceFailedTypeConversion() {
try (ConfigurableApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfigurationNoConversionPossible.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf",
"--spring.jmx.enabled=false")) {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
FunctionsConfigurationNoConversionPossible.class)).web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf",
"--spring.jmx.enabled=false")) {
PollableChannel errorChannel = context.getBean("errorChannel", PollableChannel.class);
OutputDestination target = context.getBean(OutputDestination.class);
assertNull(target.receive(1000));
@@ -125,7 +123,8 @@ public class SourceToFunctionsSupportTests {
public void testMessageSourceIsCreatedFromProvidedSupplier() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)).web(
WebApplicationType.NONE).run("--spring.cloud.stream.function.definition=number", "--spring.jmx.enabled=false")) {
WebApplicationType.NONE)
.run("--spring.cloud.stream.function.definition=number", "--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(10000).getPayload()).isEqualTo("1".getBytes(StandardCharsets.UTF_8));
@@ -257,13 +256,9 @@ public class SourceToFunctionsSupportTests {
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
.build();
IntegrationFlowBuilder flowBuilder = functionSupport.integrationFlowFromProvidedSupplier(messageSource);
if (!functionSupport.andThenFunction(flowBuilder, this.source.output())) {
flowBuilder = flowBuilder.channel(this.source.output());
}
return flowBuilder.get();
return functionSupport.integrationFlowFromProvidedSupplier(messageSource)
.channel(this.source.output())
.get();
}
}
@@ -277,15 +272,12 @@ public class SourceToFunctionsSupportTests {
@Bean
public IntegrationFlow messageSourceFlow(IntegrationFlowFunctionSupport functionSupport) {
Supplier<Message<String>> messageSource = () -> MessageBuilder.withPayload("hello function")
.setHeader(MessageHeaders.CONTENT_TYPE, "application/octet-stream").build();
.setHeader(MessageHeaders.CONTENT_TYPE, "application/octet-stream")
.build();
IntegrationFlowBuilder flowBuilder = functionSupport.integrationFlowFromProvidedSupplier(messageSource);
if (!functionSupport.andThenFunction(flowBuilder, this.source.output())) {
flowBuilder = flowBuilder.channel(this.source.output());
}
return flowBuilder.get();
return functionSupport.integrationFlowFromProvidedSupplier(messageSource)
.channel(this.source.output())
.get();
}
}