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

GH-1797 Added support for PollableSupplier annotation

Browse Source 
Resolves #1797
This commit is contained in:
Oleg Zhurakousky
2019-08-27 13:10:03 +02:00
parent 6bf1714374
commit 1d56f143d0
3 changed files with 96 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
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.lang.reflect.Method;
import java.lang.reflect.Type;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicReference;
@@ -32,9 +33,11 @@ import reactor.core.publisher.MonoSink;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.PollableSupplier;
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;
@@ -49,6 +52,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.channel.MessageChannelReactiveUtils;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
@@ -101,8 +105,15 @@ public class FunctionConfiguration {
}
});
RootBeanDefinition bd = (RootBeanDefinition) context.getBeanDefinition(functionProperties.getParsedDefinition()[0]);
Method factoryMethod = bd.getResolvedFactoryMethod();
PollableSupplier pollable = factoryMethod.getReturnType().isAssignableFrom(Supplier.class)
? AnnotationUtils.findAnnotation(factoryMethod, PollableSupplier.class)
: null;
if (!functionProperties.isComposeFrom() && !functionProperties.isComposeTo() && functionWrapper.isSupplier()) {
integrationFlow = this.integrationFlowFromProvidedSupplier(functionWrapper, functionInspector, beginPublishingTrigger)
integrationFlow = this.integrationFlowFromProvidedSupplier(functionWrapper, functionInspector, beginPublishingTrigger, pollable)
.channel("output").get();
}
}
@@ -112,11 +123,15 @@ public class FunctionConfiguration {
@SuppressWarnings({ "rawtypes", "unchecked" })
private IntegrationFlowBuilder integrationFlowFromProvidedSupplier(Supplier<?> supplier,
FunctionInspector inspector, Publisher<Object> beginPublishingTrigger) {
FunctionInspector inspector, Publisher<Object> beginPublishingTrigger, PollableSupplier pollable) {
IntegrationFlowBuilder integrationFlowBuilder;
Type functionType = FunctionTypeUtils.getFunctionType(supplier, inspector);
if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0))) {
boolean splittable = pollable != null && (boolean) AnnotationUtils.getAnnotationAttributes(pollable).get("splittable");
if (pollable == null && FunctionTypeUtils.isReactive(FunctionTypeUtils.getInputType(functionType, 0))) {
Publisher publisher = (Publisher) supplier.get();
publisher = publisher instanceof Mono
? ((Mono) publisher).delaySubscription(beginPublishingTrigger).map(this::wrapToMessageIfNecessary)
@@ -124,9 +139,13 @@ public class FunctionConfiguration {
integrationFlowBuilder = IntegrationFlows.from(publisher);
}
else {
else { // implies pollable
integrationFlowBuilder = IntegrationFlows.from(supplier);
if (splittable) {
integrationFlowBuilder = integrationFlowBuilder.split();
}
}
return integrationFlowBuilder;
}

5
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java
View File

@@ -23,6 +23,7 @@ import java.util.Map;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.util.StringUtils;
/**
* @author Oleg Zhurakousky
@@ -67,6 +68,10 @@ public class StreamFunctionProperties {
return this.definition;
}
public String[] getParsedDefinition() {
return StringUtils.delimitedListToStringArray(this.getDefinition().replaceAll(",", "|").trim(), "|");
}
public void setDefinition(String definition) {
this.composeFrom = definition.startsWith("|");
this.composeTo = definition.endsWith("|");

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

@@ -29,6 +29,7 @@ import reactor.core.publisher.Flux;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.function.context.PollableSupplier;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
@@ -40,6 +41,7 @@ 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.support.GenericMessage;
import org.springframework.util.MimeTypeUtils;
import static org.assertj.core.api.Assertions.assertThat;
@@ -195,6 +197,72 @@ public class SourceToFunctionsSupportTests {
}
}
@Test
public void testFiniteFluxSupplierMessage() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class,
MessageFluxSupplierConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=messageStreamSupplier",
"--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("1");
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("2");
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("3");
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("4");
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("5");
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("6");
}
}
@Test
public void testFiniteFluxSupplierSimple() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class,
SimpleFluxSupplierConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.function.definition=simpleStreamSupplier",
"--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("1");
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("2");
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("3");
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("4");
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("5");
assertThat(new String(target.receive(2000).getPayload())).isEqualTo("6");
}
}
@EnableAutoConfiguration
public static class MessageFluxSupplierConfiguration {
AtomicInteger counter = new AtomicInteger();
@PollableSupplier(splittable = true)
public Supplier<Flux<Message<?>>> messageStreamSupplier() {
return () -> {
Message<String> m1 = new GenericMessage<>(String.valueOf(counter.incrementAndGet()));
Message<String> m2 = new GenericMessage<>(String.valueOf(counter.incrementAndGet()));
Message<String> m3 = new GenericMessage<>(String.valueOf(counter.incrementAndGet()));
return Flux.just(m1, m2, m3);
};
}
}
@EnableAutoConfiguration
public static class SimpleFluxSupplierConfiguration {
AtomicInteger counter = new AtomicInteger();
@PollableSupplier(splittable = true)
public Supplier<Flux<String>> simpleStreamSupplier() {
return () -> {
return Flux.just(String.valueOf(counter.incrementAndGet()),
String.valueOf(counter.incrementAndGet()),
String.valueOf(counter.incrementAndGet()));
};
}
}
@EnableAutoConfiguration
public static class SupplierConfiguration {