From 1d56f143d0781b05e8685e341d964739b53effa7 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 27 Aug 2019 13:10:03 +0200 Subject: [PATCH] GH-1797 Added support for PollableSupplier annotation Resolves #1797 --- .../function/FunctionConfiguration.java | 27 ++++++-- .../function/StreamFunctionProperties.java | 5 ++ .../SourceToFunctionsSupportTests.java | 68 +++++++++++++++++++ 3 files changed, 96 insertions(+), 4 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index b7122ab57..62227815b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -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 beginPublishingTrigger) { + FunctionInspector inspector, Publisher 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; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java index 939727f53..bc2879207 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java @@ -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("|"); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java index 5fafe292b..44d88a70a 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java @@ -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>> messageStreamSupplier() { + return () -> { + Message m1 = new GenericMessage<>(String.valueOf(counter.incrementAndGet())); + Message m2 = new GenericMessage<>(String.valueOf(counter.incrementAndGet())); + Message 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> simpleStreamSupplier() { + return () -> { + return Flux.just(String.valueOf(counter.incrementAndGet()), + String.valueOf(counter.incrementAndGet()), + String.valueOf(counter.incrementAndGet())); + }; + } + } @EnableAutoConfiguration public static class SupplierConfiguration {