From d8c75d5677e94be3ed5c95313be7fad3939dec12 Mon Sep 17 00:00:00 2001
From: Soby Chacko <schacko@pivotal.io>
Date: Mon, 5 Apr 2021 18:30:02 -0400
Subject: [PATCH] Fix filter processor issues

Update fiilter processor to add an extra processing at the end of the chain
to convert the receeived text data into byte[].

See the previous commit message for more details.
---
 .../processor/filter-processor/pom.xml        |  2 +-
 .../FilterPostProcessConfiguration.java       | 56 +++++++++++++++++++
 .../filter/FilterProcessorTests.java          |  2 +-
 3 files changed, 58 insertions(+), 2 deletions(-)
 create mode 100644 applications/processor/filter-processor/src/main/java/org/springframework/cloud/stream/app/processor/filter/FilterPostProcessConfiguration.java

diff --git a/applications/processor/filter-processor/pom.xml b/applications/processor/filter-processor/pom.xml
index 78a297d7..6877d4af 100644
--- a/applications/processor/filter-processor/pom.xml
+++ b/applications/processor/filter-processor/pom.xml
@@ -37,7 +37,7 @@
                         <type>processor</type>
                         <version>${project.version}</version>
                         <configClass>org.springframework.cloud.fn.filter.FilterFunctionConfiguration.class</configClass>
-                        <functionDefinition>byteArrayTextToString|filterFunction</functionDefinition>
+                        <functionDefinition>byteArrayTextToString|filterFunction|filterPostProcessFunction</functionDefinition>
                         <maven>
                             <dependencies>
                                 <dependency>
diff --git a/applications/processor/filter-processor/src/main/java/org/springframework/cloud/stream/app/processor/filter/FilterPostProcessConfiguration.java b/applications/processor/filter-processor/src/main/java/org/springframework/cloud/stream/app/processor/filter/FilterPostProcessConfiguration.java
new file mode 100644
index 00000000..2fefdcc2
--- /dev/null
+++ b/applications/processor/filter-processor/src/main/java/org/springframework/cloud/stream/app/processor/filter/FilterPostProcessConfiguration.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2021-2021 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.app.processor.filter;
+
+import java.util.function.Function;
+
+import reactor.core.publisher.Flux;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.CompositeMessageConverter;
+
+/**
+ * Since the commit mentioned below in Spring Cloud Stream removed the logic of
+ * converting the incoming text data back to byte[], we are adding an extra
+ * processing function to do that convrsion.
+ *
+ * https://github.com/spring-cloud/spring-cloud-stream/commit/5d9de8ad579d3464d1503d1a5d1390168bccbdb9
+ *
+ * @author Soby Chacko
+ */
+@Configuration
+public class FilterPostProcessConfiguration {
+
+	@Bean
+	public Function<Flux<Message<?>>, Flux<Message<?>>> filterPostProcessFunction(
+			CompositeMessageConverter messageConverter) {
+
+		return flux -> flux.map(message -> {
+			@SuppressWarnings("unchecked")
+			Message<byte[]> outboundMessage = message.getPayload() instanceof byte[]
+					? (Message<byte[]>) message : (Message<byte[]>) messageConverter
+					.toMessage(message.getPayload(), message.getHeaders());
+			if (outboundMessage == null) {
+				throw new IllegalStateException("Failed to convert message: '" + message
+						+ "' to outbound message.");
+			}
+			return outboundMessage;
+		});
+	}
+}
diff --git a/applications/processor/filter-processor/src/test/java/org/springframework/cloud/stream/app/processor/filter/FilterProcessorTests.java b/applications/processor/filter-processor/src/test/java/org/springframework/cloud/stream/app/processor/filter/FilterProcessorTests.java
index ea9c04a2..08867cb6 100644
--- a/applications/processor/filter-processor/src/test/java/org/springframework/cloud/stream/app/processor/filter/FilterProcessorTests.java
+++ b/applications/processor/filter-processor/src/test/java/org/springframework/cloud/stream/app/processor/filter/FilterProcessorTests.java
@@ -46,7 +46,7 @@ public class FilterProcessorTests {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
 				TestChannelBinderConfiguration.getCompleteConfiguration(FilterProcessorTestApplication.class))
 				.web(WebApplicationType.NONE)
-				.run("--spring.cloud.function.definition=byteArrayTextToString|filterFunction",
+				.run("--spring.cloud.function.definition=byteArrayTextToString|filterFunction|filterPostProcessFunction",
 						"--filter.function.expression=payload.length() > 5")) {
 
 			InputDestination processorInput = context.getBean(InputDestination.class);