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

Fixed source and source-to-function functionality and tests

Browse Source
This commit is contained in:
Oleg Zhurakousky
2019-07-29 15:00:15 +02:00
parent 82530dd2cf
commit c8a4b40de0
3 changed files with 206 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,14 +18,17 @@ package org.springframework.cloud.stream.function;
import java.lang.reflect.Type;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
@@ -35,6 +38,7 @@ import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry.FunctionInvocationWrapper;
import org.springframework.cloud.function.context.catalog.FunctionInspector;
import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
import org.springframework.cloud.stream.binder.BindingCreatedEvent;
import org.springframework.cloud.stream.binding.BindableProxyFactory;
import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration;
import org.springframework.cloud.stream.config.BindingServiceConfiguration;
@@ -46,17 +50,20 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.channel.MessageChannelReactiveUtils;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.handler.ServiceActivatingHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.util.Assert;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.ObjectUtils;
/**
* @author Oleg Zhurakousky
* @author David Turanski
@@ -76,8 +83,68 @@ public class FunctionConfiguration {
ObjectUtils.isEmpty(bindableProxyFactory) ? null : bindableProxyFactory[0]);
}
@Bean
public IntegrationFlow standAloneSupplierFlow(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
StreamFunctionProperties functionProperties, GenericApplicationContext context) {
IntegrationFlow integrationFlow = null;
FunctionInvocationWrapper functionWrapper = functionCatalog.lookup(functionProperties.getDefinition());
if (functionWrapper != null) {
AtomicReference<MonoSink<Object>> triggerRef = new AtomicReference<>();
Publisher<Object> beginPublishingTrigger = Mono.create(emmiter -> {
triggerRef.set(emmiter);
});
context.addApplicationListener(event -> {
if (event instanceof BindingCreatedEvent) {
if (triggerRef.get() != null) {
triggerRef.get().success();
}
}
});
if (!functionProperties.isComposeFrom() && !functionProperties.isComposeTo() && functionWrapper.isSupplier()) {
integrationFlow = this.integrationFlowFromProvidedSupplier(functionWrapper, functionInspector, beginPublishingTrigger)
.channel("output").get();
}
}
return integrationFlow;
}
@SuppressWarnings({ "rawtypes", "unchecked" })
private IntegrationFlowBuilder integrationFlowFromProvidedSupplier(Supplier<?> supplier,
FunctionInspector inspector, Publisher<Object> beginPublishingTrigger) {
IntegrationFlowBuilder integrationFlowBuilder;
Type functionType = FunctionTypeUtils.getFunctionType(supplier, inspector);
if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0))) {
Publisher publisher = (Publisher) supplier.get();
publisher = publisher instanceof Mono
? ((Mono) publisher).delaySubscription(beginPublishingTrigger).map(this::wrapToMessageIfNecessary)
: ((Flux) publisher).delaySubscription(beginPublishingTrigger).map(this::wrapToMessageIfNecessary);
integrationFlowBuilder = IntegrationFlows.from(publisher);
}
else {
integrationFlowBuilder = IntegrationFlows.from(supplier);
}
return integrationFlowBuilder;
}
@SuppressWarnings("unchecked")
private <T> Message<T> wrapToMessageIfNecessary(T value) {
return value instanceof Message ? (Message<T>) value : MessageBuilder.withPayload(value).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build();
}
/**
*
* @author Oleg Zhurakousky
* @since 3.0
*/
private static class FunctionChannelBindingPostProcessor implements BeanPostProcessor, ApplicationContextAware {
private static Log logger = LogFactory.getLog(FunctionChannelBindingPostProcessor.class);
private final FunctionCatalog functionCatalog;
private final FunctionInspector functionInspector;
@@ -88,12 +155,14 @@ public class FunctionConfiguration {
private GenericApplicationContext context;
FunctionChannelBindingPostProcessor(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
StreamFunctionProperties functionProperties, BindableProxyFactory bindableProxyFactory) {
this.functionCatalog = functionCatalog;
this.functionInspector = functionInspector;
this.functionProperties = functionProperties;
this.bindableProxyFactory = bindableProxyFactory;
}
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
@@ -111,14 +180,13 @@ public class FunctionConfiguration {
}
private void doPostProcess(String channelName, SubscribableChannel messageChannel) {
//TODO there is something about moving channel interceptors in AMCB (not sure if it is still required)
if (functionProperties.isComposeTo() && messageChannel instanceof SubscribableChannel && "input".equals(channelName)) {
System.out.println("Composing at the tail");
throw new UnsupportedOperationException("Composing at tail is not currently supported");
}
else if (functionProperties.isComposeFrom() && "output".equals(channelName)) {
Assert.notNull(bindableProxyFactory, "Can not compose function into the existing app since `bindableProxyFactory` is null.");
System.out.println("Composing at the head");
Assert.notNull(this.bindableProxyFactory, "Can not compose function into the existing app since `bindableProxyFactory` is null.");
logger.info("Composing at the head of 'output' channel");
FunctionInvocationWrapper function = functionCatalog.lookup(functionProperties.getDefinition(), "application/json");
ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function));
handler.setBeanFactory(context);
@@ -133,20 +201,17 @@ public class FunctionConfiguration {
handler.setOutputChannelName("output.extended");
SubscribableChannel subscribeChannel = (SubscribableChannel) messageChannel;
subscribeChannel.subscribe(handler);
}
else {
FunctionInvocationWrapper function = functionCatalog.lookup(functionProperties.getDefinition(), "application/json");
if (function.getTarget() instanceof Supplier) {
System.out.println("Configuring supplier");
throw new UnsupportedOperationException("Standalone supplier are not currently supported");
}
else if (function.getTarget() instanceof Consumer) {
throw new UnsupportedOperationException("Consumers are not currently supported");
}
else {
if ("input".equals(channelName)) {
this.postProcessForStandAloneFunction(function, messageChannel);
if (!function.isSupplier()) {
if (function.isConsumer()) {
throw new UnsupportedOperationException("Consumers are not currently supported");
}
else if (function.isFunction()) {
if ("input".equals(channelName)) {
this.postProcessForStandAloneFunction(function, messageChannel);
}
}
}
}
@@ -170,12 +235,17 @@ public class FunctionConfiguration {
}
}
/*
* Enhance publisher to add error handling, retries etc.
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
private Publisher enhancePublisher(Publisher publisher) {
Flux flux = Flux.from(publisher)
.concatMap(message -> {
return Flux.just(message)
.doOnError(e -> e.printStackTrace())
.doOnError(e -> {
e.printStackTrace();
})
.retryBackoff(3, //this.consumerProperties.getMaxAttempts(),
Duration.ofMillis(1000),
//this.consumerProperties.getBackOffInitialInterval()),
@@ -190,6 +260,7 @@ public class FunctionConfiguration {
return flux;
}
@SuppressWarnings({ "unchecked", "rawtypes" })
private <I, O> void subscribeToInput(Function function,
Publisher<?> publisher, Consumer<Message<O>> outputProcessor) {

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

@@ -61,7 +61,7 @@ public class IntegrationFlowFunctionSupport {
private final StreamFunctionProperties functionProperties;
private final AtomicReference<MonoSink<Object>> triggerRef = new AtomicReference<>();
//private final AtomicReference<MonoSink<Object>> triggerRef = new AtomicReference<>();
private final Publisher<Object> trigger;
@@ -85,6 +85,7 @@ public class IntegrationFlowFunctionSupport {
this.functionProperties = functionProperties;
this.context = context;
this.functionProperties.setBindingServiceProperties(bindingServiceProperties);
AtomicReference<MonoSink<Object>> triggerRef = null;
trigger = Mono.create(emmiter -> {
triggerRef.set(emmiter);
});

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

@@ -17,17 +17,13 @@
package org.springframework.cloud.stream.function;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.function.Supplier;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import org.springframework.boot.WebApplicationType;
@@ -39,16 +35,11 @@ import org.springframework.cloud.stream.binder.test.TestChannelBinderConfigurati
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.context.ConfigurableApplicationContext;
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.IntegrationFlows;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.util.MimeTypeUtils;
import static org.assertj.core.api.Assertions.assertThat;
@@ -67,7 +58,7 @@ public class SourceToFunctionsSupportTests {
public void testFunctionIsAppliedToExistingMessageSource() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
FunctionsConfiguration.class)).web(WebApplicationType.NONE).run(
FunctionsConfiguration.class, ExistingMessageSourceConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=|toUpperCase",
"--spring.jmx.enabled=false")) {
@@ -78,165 +69,153 @@ public class SourceToFunctionsSupportTests {
}
@Test
public void testComposedFunctionIsAppliedToExistingMessageSource() {
public void testFunctionsAreAppliedToExistingMessageSource() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
FunctionsConfiguration.class)).web(WebApplicationType.NONE).run(
FunctionsConfiguration.class, ExistingMessageSourceConfiguration.class)).web(WebApplicationType.NONE).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));
assertThat(target.receive(1000).getPayload())
.isEqualTo("HELLO FUNCTION:HELLO FUNCTION".getBytes(StandardCharsets.UTF_8));
}
}
@Test
public void testFailedInputTypeConversion() {
public void testImperativeSupplier() {
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);
assertThat(target.receive(1000)).isNull();
assertThat(errorChannel.receive(10000)).isNotNull();
}
}
@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")) {
PollableChannel errorChannel = context.getBean("errorChannel",
PollableChannel.class);
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(1000)).isNull();
assertThat(errorChannel.receive(10000)).isNotNull();
}
}
@Test
@Ignore
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")) {
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(10000).getPayload())
.isEqualTo("1".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload())
.isEqualTo("2".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload())
.isEqualTo("3".getBytes(StandardCharsets.UTF_8));
// etc
}
}
@Test
@Ignore
public void testMessageSourceIsCreatedFromProvidedSupplierComposedWithSingleFunction() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
SupplierConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=number|concatWithSelf",
FunctionsConfiguration.class, SupplierConfiguration.class)).web(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("11".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload())
.isEqualTo("22".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload())
.isEqualTo("33".getBytes(StandardCharsets.UTF_8));
// etc
String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("1");
}
}
@Test
@Ignore
public void testMessageSourceIsCreatedFromProvidedSupplierComposedWithMultipleFunctions() {
public void testImperativeSupplierComposedWithFunctions() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
SupplierConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=number|concatWithSelf|multiplyByTwo",
FunctionsConfiguration.class, SupplierConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=number|toUpperCase|concatWithSelf",
"--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(10000).getPayload())
.isEqualTo("22".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload())
.isEqualTo("44".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload())
.isEqualTo("66".getBytes(StandardCharsets.UTF_8));
// etc
String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("1:1");
}
}
// @Test
// public void testMessageSourceIsCreatedFromProvidedStreamSupplier() {
// try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
// TestChannelBinderConfiguration.getCompleteConfiguration(
// StreamSupplierConfiguration.class)).web(WebApplicationType.NONE).run(
// "--spring.cloud.stream.function.definition=stream",
// "--spring.jmx.enabled=false")) {
//
// OutputDestination target = context.getBean(OutputDestination.class);
// assertThat(target.receive(1000).getPayload())
// .isEqualTo("0".getBytes(StandardCharsets.UTF_8));
// assertThat(target.receive(1000).getPayload())
// .isEqualTo("1".getBytes(StandardCharsets.UTF_8));
// assertThat(target.receive(1000).getPayload())
// .isEqualTo("2".getBytes(StandardCharsets.UTF_8));
// assertThat(target.receive(1000).getPayload())
// .isEqualTo("3".getBytes(StandardCharsets.UTF_8));
//
// // etc
// }
// }
@Test
public void testImperativeSupplierComposedWithMixedFunctions() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
FunctionsConfiguration.class, SupplierConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=number|toUpperCaseReactive|concatWithSelf",
"--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("1:1");
}
}
@Test
public void testReactiveSupplier() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=numberReactive",
"--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("0");
result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("1");
result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("2");
}
}
@Test
public void testReactiveSupplierComposedWithImperativeFunctions() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class,
SupplierConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=numberReactive|toUpperCase|concatWithSelf",
"--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("0:0");
result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("1:1");
result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("2:2");
}
}
@Test
public void testReactiveSupplierComposedWithMixedFunctions() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class,
SupplierConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=numberReactive|concatWithSelf|toUpperCaseReactive",
"--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("0:0");
result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("1:1");
result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("2:2");
}
}
@Test
public void testReactiveSupplierComposedWithMixedFunctions2() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class,
SupplierConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=numberReactive|toUpperCaseReactive|concatWithSelf",
"--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("0:0");
result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("1:1");
result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8);
assertThat(result).isEqualTo("2:2");
}
}
// @Test
// public void testFunctionDoesNotExist() {
//
// this.expectedException.expect(BeanCreationException.class);
//
// new SpringApplicationBuilder(TestChannelBinderConfiguration
// .getCompleteConfiguration(SupplierConfiguration.class))
// .web(WebApplicationType.NONE)
// .run("--spring.cloud.stream.function.definition=doesNotExist",
// "--spring.jmx.enabled=false");
// }
@EnableAutoConfiguration
//@Import(ProvidedMessageSourceConfiguration.class)
@EnableScheduling
public static class SupplierConfiguration {
AtomicInteger counter = new AtomicInteger();
@Bean
@Scheduled(fixedRate = 5000)
public Supplier<String> number() {
return () -> {
return String.valueOf(this.counter.incrementAndGet());
};
return () -> String.valueOf(this.counter.incrementAndGet());
}
@Bean
public Function<String, String> concatWithSelf() {
return x -> x + x;
public Supplier<Flux<String>> numberReactive() {
return () -> Flux.create(emitter -> {
for (int i = 0; i < 3; i++) {
emitter.next(String.valueOf(i));
}
});
}
@Bean
public Function<Flux<String>, Flux<String>> multiplyByTwo() {
return x -> x.map(i -> String.valueOf(Integer.valueOf(i) * 2));
@@ -245,7 +224,6 @@ public class SourceToFunctionsSupportTests {
}
@EnableAutoConfiguration
@Import(ExistingMessageSourceConfiguration.class)
public static class FunctionsConfiguration {
@Bean
@@ -253,6 +231,11 @@ public class SourceToFunctionsSupportTests {
return String::toUpperCase;
}
@Bean
public Function<Flux<String>, Flux<String>> toUpperCaseReactive() {
return flux -> flux.map(String::toUpperCase);
}
@Bean
public Function<String, String> concatWithSelf() {
return x -> x + ":" + x;
@@ -260,44 +243,6 @@ public class SourceToFunctionsSupportTests {
}
@EnableAutoConfiguration
public static class StreamSupplierConfiguration {
@Bean
public Supplier<Publisher<Object>> stream() {
ExecutorService executor = Executors.newFixedThreadPool(1);
return () -> Flux.create(emitter -> {
executor.execute(() -> {
for (int i = 0; i < 10; i++) {
emitter.next(MessageBuilder.withPayload(String.valueOf(i)).build());
}
});
});
}
}
@EnableAutoConfiguration
@Import(ExistingMessageSourceConfigurationNoContentTypeSet.class)
public static class FunctionsConfigurationNoConversionPossible {
@Bean
public PollableChannel errorChannel() {
return new QueueChannel(10);
}
@Bean
public Function<Boolean, Boolean> toUpperCase() {
return x -> true;
}
@Bean
public Function<Boolean, Integer> concatWithSelf() {
return x -> 1;
}
}
/**
* This configuration essentially emulates our existing app-starters for Sources and
* essentially demonstrates how a function(s) could be applied to an existing source
@@ -306,18 +251,12 @@ public class SourceToFunctionsSupportTests {
@EnableBinding(Source.class)
public static class ExistingMessageSourceConfiguration {
// @Autowired
// private Source source;
@Bean
public IntegrationFlow messageSourceFlow() {
Supplier<Message<String>> messageSource = () -> MessageBuilder
.withPayload("hello function")
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
.build();
// return functionSupport.integrationFlowFromProvidedSupplier(messageSource)
// .channel(this.source.output()).get();
return IntegrationFlows.from(messageSource).channel("output").get();
}
@@ -332,33 +271,9 @@ public class SourceToFunctionsSupportTests {
.withPayload("hello function")
.setHeader(MessageHeaders.CONTENT_TYPE, "application/octet-stream")
.build();
// return functionSupport.integrationFlowFromProvidedSupplier(messageSource)
// .channel(this.source.output()).get();
return IntegrationFlows.from(messageSource).channel("output").get();
}
}
// @EnableBinding(Source.class)
// public static class ProvidedMessageSourceConfiguration {
//
// @Autowired
// private Source source;
//
// @Autowired
// private StreamFunctionProperties functionProperties;
//
// @Bean
// public IntegrationFlow messageSourceFlow(
// IntegrationFlowFunctionSupport functionSupport) {
// Assert.hasText(this.functionProperties.getDefinition(),
// "Supplier name must be provided");
//
// return functionSupport.integrationFlowFromNamedSupplier()
// .channel(this.source.output()).get();
// }
//
// }
}