From 8e4c20dc773dffbe751d19a0c90f22db2a47edba Mon Sep 17 00:00:00 2001 From: Corneil du Plessis Date: Tue, 14 Feb 2023 18:23:31 +0200 Subject: [PATCH] GH-430: Change Filter function to non-reactive Fixes https://github.com/spring-cloud/stream-applications/issues/430 Fix greenmail exclusion. Added -X for github.debug --- applications/source/mail-source/pom.xml | 2 ++ .../filter/FilterFunctionConfiguration.java | 19 +++++++----- .../FilterFunctionApplicationTests.java | 30 ++++++++++--------- .../http-request-function/README.adoc | 2 +- 4 files changed, 30 insertions(+), 23 deletions(-) diff --git a/applications/source/mail-source/pom.xml b/applications/source/mail-source/pom.xml index 3cd70a18..bfd6c5e1 100644 --- a/applications/source/mail-source/pom.xml +++ b/applications/source/mail-source/pom.xml @@ -32,9 +32,11 @@ com.sun.mail + jakarta.mail jakarta.activation + jakarta.activation-api diff --git a/functions/function/filter-function/src/main/java/org/springframework/cloud/fn/filter/FilterFunctionConfiguration.java b/functions/function/filter-function/src/main/java/org/springframework/cloud/fn/filter/FilterFunctionConfiguration.java index 89179722..cef6f671 100644 --- a/functions/function/filter-function/src/main/java/org/springframework/cloud/fn/filter/FilterFunctionConfiguration.java +++ b/functions/function/filter-function/src/main/java/org/springframework/cloud/fn/filter/FilterFunctionConfiguration.java @@ -18,8 +18,6 @@ package org.springframework.cloud.fn.filter; import java.util.function.Function; -import reactor.core.publisher.Flux; - import org.springframework.boot.autoconfigure.AutoConfiguration; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; @@ -35,17 +33,22 @@ import org.springframework.messaging.Message; public class FilterFunctionConfiguration { @Bean - public Function>, Flux>> filterFunction( - ExpressionEvaluatingTransformer filterExpressionEvaluatingTransformer) { + public Function, Message> filterFunction( + ExpressionEvaluatingTransformer filterExpressionEvaluatingTransformer) { - return flux -> - flux.filter((message) -> - (Boolean) filterExpressionEvaluatingTransformer.transform(message).getPayload()); + return message -> { + if ((Boolean) filterExpressionEvaluatingTransformer.transform(message).getPayload()) { + return message; + } + else { + return null; + } + }; } @Bean public ExpressionEvaluatingTransformer filterExpressionEvaluatingTransformer( - FilterFunctionProperties filterFunctionProperties) { + FilterFunctionProperties filterFunctionProperties) { return new ExpressionEvaluatingTransformer(filterFunctionProperties.getExpression()); } diff --git a/functions/function/filter-function/src/test/java/org/springframework/cloud/fn/filter/FilterFunctionApplicationTests.java b/functions/function/filter-function/src/test/java/org/springframework/cloud/fn/filter/FilterFunctionApplicationTests.java index 163c3ca4..6cf98aa6 100644 --- a/functions/function/filter-function/src/test/java/org/springframework/cloud/fn/filter/FilterFunctionApplicationTests.java +++ b/functions/function/filter-function/src/test/java/org/springframework/cloud/fn/filter/FilterFunctionApplicationTests.java @@ -16,11 +16,15 @@ package org.springframework.cloud.fn.filter; +import java.util.List; import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.Stream; + import org.junit.jupiter.api.Test; -import reactor.core.publisher.Flux; -import reactor.test.StepVerifier; + + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -30,6 +34,8 @@ import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; +import static org.assertj.core.api.Assertions.assertThat; + /** * @author Artem Bilan * @author David Turanski @@ -40,21 +46,17 @@ public class FilterFunctionApplicationTests { @Autowired @Qualifier("filterFunction") - Function>, Flux>> filter; + Function, Message> filter; @Test public void testFilter() { - Flux> messageFlux = - Flux.just("hello", "hello world") - .map(GenericMessage::new); - Flux> result = this.filter.apply(messageFlux); - result - .map(Message::getPayload) - .cast(String.class) - .as(StepVerifier::create) - .expectNext("hello world") - .expectComplete() - .verify(); + Stream> messages = List.of("hello", "hello world") + .stream() + .map(GenericMessage::new); + List> result = messages.filter(message -> this.filter.apply(message) != null).collect(Collectors.toList()); + assertThat(result.size()).isEqualTo(1); + assertThat(result.get(0).getPayload()).isNotNull(); + assertThat(result.get(0).getPayload()).isEqualTo("hello world"); } @SpringBootApplication diff --git a/functions/function/http-request-function/README.adoc b/functions/function/http-request-function/README.adoc index d0a0ad41..e3f4bf01 100644 --- a/functions/function/http-request-function/README.adoc +++ b/functions/function/http-request-function/README.adoc @@ -2,7 +2,7 @@ This module provides an HTTP request function that can be reused and composed in other applications. The `Function` uses the reactive `WebClient` from `Spring WebFlux` and is implemented as a `java.util.function.Function`. -This function gives you a reactive stream of `ResponseEntity` given a stream of request messages as the function a signature of `Function,Flux>`. +This function gives you a reactive stream of `ResponseEntity` given a stream of request messages as the function a signature of `Function,ResponseEntity>`. Users have to subscribe to the returned `Flux` to receive the data. ## Beans for injection