GH-110: Fix filter function for emitting null (#111)

* GH-110: Fix filter function for emitting `null`

Fixes https://github.com/spring-cloud/stream-applications/issues/110

Turns out when functions are composed in the Spring Cloud Stream environment,
they are called via reactive wrappers which don't allow to emit `null` from the
`map()` operator.

* Make `filterFunction` fully reactive to reply on the `Flux.filter()` operator
* Add `config-common` for auto-conversion string configuration options into `Expression` instances
* Add `spring-boot-starter-json` dependency since it is required by the `SpelExpressionConverterConfiguration`
* Add `reactor-test` dependency to test the final solution
* Remove redundant dependencies from the `filter-processor`

* * Add `proxyBeanMethods = false` into `FilterFunctionConfiguration`
* Fix default expression to `true` instead of `payload`, which does not fit to filter logic
* Fix JavaDoc for `expression` property
* Remove redundant `application.properties` from the `filter-function`

functions/function/filter-function/pom.xml

@@ -15,6 +15,11 @@
     </parent>
 
     <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>config-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.springframework.cloud.fn</groupId>
             <artifactId>payload-converter-function</artifactId>
@@ -24,6 +29,10 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-integration</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-json</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-configuration-processor</artifactId>
@@ -40,6 +49,11 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

functions/function/filter-function/src/main/java/org/springframework/cloud/fn/filter/FilterFunctionConfiguration.java

@@ -16,39 +16,38 @@
 
 package org.springframework.cloud.fn.filter;
 
-import java.util.Optional;
 import java.util.function.Function;
 
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.transformer.ExpressionEvaluatingTransformer;
 import org.springframework.messaging.Message;
 
+import reactor.core.publisher.Flux;
+
 /**
  * @author Artem Bilan
  * @author David Turanski
  */
-@Configuration
+@Configuration(proxyBeanMethods = false)
 @EnableConfigurationProperties(FilterFunctionProperties.class)
 public class FilterFunctionConfiguration {
 
 	@Bean
-	public Function<Message<?>, Message<?>> filterFunction(
+	public Function<Flux<Message<?>>, Flux<Message<?>>> filterFunction(
 			ExpressionEvaluatingTransformer filterExpressionEvaluatingTransformer) {
 
-		return message -> Optional.of(message)
-				.filter(m -> (Boolean) filterExpressionEvaluatingTransformer.transform(m).getPayload())
-				.orElse(null);
+		return flux ->
+				flux.filter((message) ->
+						(Boolean) filterExpressionEvaluatingTransformer.transform(message).getPayload());
 	}
 
 	@Bean
 	public ExpressionEvaluatingTransformer filterExpressionEvaluatingTransformer(
 			FilterFunctionProperties filterFunctionProperties) {
 
-		return new ExpressionEvaluatingTransformer(new SpelExpressionParser()
-				.parseExpression(filterFunctionProperties.getExpression()));
+		return new ExpressionEvaluatingTransformer(filterFunctionProperties.getExpression());
 	}
 
 }

functions/function/filter-function/src/main/java/org/springframework/cloud/fn/filter/FilterFunctionProperties.java

@@ -18,7 +18,7 @@ package org.springframework.cloud.fn.filter;
 
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.expression.Expression;
-import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.expression.ValueExpression;
 
 /**
  * Configuration properties for the SpEL function.
@@ -29,18 +29,16 @@ import org.springframework.expression.spel.standard.SpelExpressionParser;
 @ConfigurationProperties("filter.function")
 public class FilterFunctionProperties {
 
-	private static final Expression DEFAULT_EXPRESSION = new SpelExpressionParser().parseExpression("payload");
-
 	/**
-	 * A SpEL expression to apply.
+	 * Boolean SpEL expression to apply against request message to filter.
 	 */
-	private String expression = DEFAULT_EXPRESSION.getExpressionString();
+	private Expression expression = new ValueExpression<>(true);
 
-	public String getExpression() {
+	public Expression getExpression() {
 		return this.expression;
 	}
 
-	public void setExpression(String expression) {
+	public void setExpression(Expression expression) {
 		this.expression = expression;
 	}
 

functions/function/filter-function/src/main/resources/application.properties

@@ -1 +0,0 @@
-spel.function.expression=true

functions/function/filter-function/src/test/java/org/springframework/cloud/fn/filter/FilterFunctionApplicationTests.java

@@ -30,25 +30,39 @@ import org.springframework.test.annotation.DirtiesContext;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+/**
+ * @author Artem Bilan
+ * @author David Turanski
+ */
 @SpringBootTest(properties = "filter.function.expression=payload.length() > 5")
 @DirtiesContext
 public class FilterFunctionApplicationTests {
 
 	@Autowired
 	@Qualifier("filterFunction")
-	Function<Message<?>, Message<?>> filter;
+	Function<Flux<Message<?>>, Flux<Message<?>>> filter;
 
 	@Test
 	public void testFilter() {
-		Message<?> filtered = this.filter.apply(new GenericMessage<>("hello"));
-		assertThat(filtered).isNull();
-		filtered = this.filter.apply(new GenericMessage<>("hello world"));
-		assertThat(filtered).isNotNull()
-				.extracting(Message::getPayload)
-				.isEqualTo("hello world");
+		Flux<Message<?>> messageFlux =
+				Flux.just("hello", "hello world")
+						.map(GenericMessage::new);
+		Flux<Message<?>> result = this.filter.apply(messageFlux);
+		result
+				.map(Message::getPayload)
+				.cast(String.class)
+				.as(StepVerifier::create)
+				.expectNext("hello world")
+				.expectComplete()
+				.verify();
 	}
 
 	@SpringBootApplication
 	static class FilterFunctionTestApplication {
+
 	}
+
 }