From 42a5319c998d8d4085ea650537478561ba2353f6 Mon Sep 17 00:00:00 2001 From: abilan Date: Wed, 17 May 2023 13:59:52 -0400 Subject: [PATCH] * Fix typos in docs * Add `spring-cloud-function-integration` into `spring-cloud-function-dependencies` * Fix "memoize" logic in the `FunctionLookupHelper` and cover it in the `FunctionFlowTests` against `@SpyBean FunctionCatalog` --- .../src/main/asciidoc/spring-integration.adoc | 2 +- spring-cloud-function-dependencies/pom.xml | 5 +++ .../integration/dsl/FunctionLookupHelper.java | 22 +++++----- .../integration/dsl/FunctionFlowTests.java | 42 +++++++++++++------ 4 files changed, 46 insertions(+), 25 deletions(-) diff --git a/docs/src/main/asciidoc/spring-integration.adoc b/docs/src/main/asciidoc/spring-integration.adoc index 1c3edf455..52c6d1499 100644 --- a/docs/src/main/asciidoc/spring-integration.adoc +++ b/docs/src/main/asciidoc/spring-integration.adoc @@ -20,7 +20,7 @@ The following example demonstrates the `FunctionFlowBuilder` in action alongside [source,java] ---- @Configuration -static class IntegrationConfiguration { +public class IntegrationConfiguration { @Bean Supplier simpleByteArraySupplier() { diff --git a/spring-cloud-function-dependencies/pom.xml b/spring-cloud-function-dependencies/pom.xml index 3cda57926..099ae4140 100644 --- a/spring-cloud-function-dependencies/pom.xml +++ b/spring-cloud-function-dependencies/pom.xml @@ -61,6 +61,11 @@ spring-cloud-function-adapter-gcp ${project.version} + + org.springframework.cloud + spring-cloud-function-integration + ${project.version} + org.springframework.cloud spring-cloud-function-kotlin diff --git a/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionLookupHelper.java b/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionLookupHelper.java index 60e965982..d436b417b 100644 --- a/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionLookupHelper.java +++ b/spring-cloud-function-integration/src/main/java/org/springframework/cloud/function/integration/dsl/FunctionLookupHelper.java @@ -40,24 +40,22 @@ public class FunctionLookupHelper { }

Supplier

lookupSupplier(String functionDefinition) { - return () -> - memoize(() -> this.functionCatalog.>lookup(Supplier.class, functionDefinition)) - .get() - .get(); + Supplier> memoizedSupplier = lazyLookup(Supplier.class, functionDefinition); + return () -> memoizedSupplier.get().get(); }

Function lookupFunction(String functionDefinition) { - return (p) -> - memoize(() -> this.functionCatalog.>lookup(Function.class, functionDefinition)) - .get() - .apply(p); + Supplier> memoizedFunction = lazyLookup(Function.class, functionDefinition); + return (p) -> memoizedFunction.get().apply(p); }

Consumer

lookupConsumer(String consumerDefinition) { - return (p) -> - memoize(() -> this.functionCatalog.>lookup(Consumer.class, consumerDefinition)) - .get() - .accept(p); + Supplier> memoizedConsumer = lazyLookup(Consumer.class, consumerDefinition); + return (p) -> memoizedConsumer.get().accept(p); + } + + private Supplier lazyLookup(Class functionType, String functionDefinition) { + return memoize(() -> this.functionCatalog.lookup(functionType, functionDefinition)); } /** diff --git a/spring-cloud-function-integration/src/test/java/org/springframework/cloud/function/integration/dsl/FunctionFlowTests.java b/spring-cloud-function-integration/src/test/java/org/springframework/cloud/function/integration/dsl/FunctionFlowTests.java index 861c7510c..eb5eb0f40 100644 --- a/spring-cloud-function-integration/src/test/java/org/springframework/cloud/function/integration/dsl/FunctionFlowTests.java +++ b/spring-cloud-function-integration/src/test/java/org/springframework/cloud/function/integration/dsl/FunctionFlowTests.java @@ -28,12 +28,15 @@ import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.mock.mockito.SpyBean; +import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.PollerSpec; import org.springframework.integration.dsl.Pollers; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.handler.LoggingHandler; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.test.util.OnlyOnceTrigger; @@ -43,6 +46,7 @@ import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.verify; /** * @author Artem Bilan @@ -53,31 +57,45 @@ import static org.assertj.core.api.Assertions.assertThat; @DirtiesContext public class FunctionFlowTests { - @Autowired - QueueChannel wireTapChannel; + // To verify cached lookups + @SpyBean + FunctionCatalog functionCatalog; @Autowired BlockingQueue results; @Test - void fromSupplierOverFunctionToConsumer() throws InterruptedException { - String result = results.poll(10, TimeUnit.SECONDS); + void fromSupplierOverFunctionToConsumer(@Autowired SourcePollingChannelAdapter supplierEndpoint, + @Autowired QueueChannel wireTapChannel) throws InterruptedException { + + supplierEndpoint.start(); + + String result = this.results.poll(10, TimeUnit.SECONDS); assertThat(result).isEqualTo("SIMPLE TEST DATA"); Message receive = wireTapChannel.receive(10_000); - assertThat(receive).isNotNull() + assertThat(receive) .extracting(Message::getPayload) .isEqualTo("simple test data".getBytes()); + + supplierEndpoint.stop(); } - @Autowired - MessageChannel functionCompositionInput; - @Test - void fromChannelToFunctionComposition() throws InterruptedException { - this.functionCompositionInput.send(new GenericMessage<>("compose this")); + void fromChannelToFunctionComposition(@Autowired MessageChannel functionCompositionInput) + throws InterruptedException { - String result = results.poll(10, TimeUnit.SECONDS); + functionCompositionInput.send(new GenericMessage<>("compose this")); + + String result = this.results.poll(10, TimeUnit.SECONDS); assertThat(result).isEqualTo("COMPOSE THIS"); + + functionCompositionInput.send(new GenericMessage<>("compose again")); + + result = this.results.poll(10, TimeUnit.SECONDS); + assertThat(result).isEqualTo("COMPOSE AGAIN"); + + // Ensure that FunctionLookupHelper.memoize() does its trick calling FunctionCatalog.lookup() only once + verify(this.functionCatalog).lookup(Consumer.class, "upperCaseFunction|simpleStringConsumer"); } @EnableAutoConfiguration @@ -117,7 +135,7 @@ public class FunctionFlowTests { @Bean IntegrationFlow someFunctionFlow(FunctionFlowBuilder functionFlowBuilder) { return functionFlowBuilder - .fromSupplier("simpleByteArraySupplier") + .fromSupplier("simpleByteArraySupplier", e -> e.id("supplierEndpoint").autoStartup(false)) .wireTap("wireTapChannel") .apply("upperCaseFunction") .log(LoggingHandler.Level.WARN, FunctionFlowTests.class.getName())