Initial Commit

Migrating the existing structure from the following location:
https://github.com/spring-cloud-stream-app-starters/stream-applications/tree/restructuring

function/filter-function/.gitignore

@@ -0,0 +1,28 @@
+/target/
+!.mvn/wrapper/maven-wrapper.jar
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+/build/
+
+### VS Code ###
+.vscode/

function/filter-function/pom.xml

@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<artifactId>filter-function</artifactId>
+	<version>1.0.0.BUILD-SNAPSHOT</version>
+	<name>filter-function</name>
+	<description>Spring Native Function for applying filter SpEL expressions</description>
+
+	<parent>
+		<groupId>org.springframework.cloud.fn</groupId>
+		<artifactId>spring-functions-parent</artifactId>
+		<version>1.0.0.BUILD-SNAPSHOT</version>
+		<relativePath>../../spring-functions-parent</relativePath>
+	</parent>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud.fn</groupId>
+			<artifactId>spel-function</artifactId>
+			<version>1.0.0.BUILD-SNAPSHOT</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>org.junit.vintage</groupId>
+					<artifactId>junit-vintage-engine</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+	</dependencies>
+
+</project>

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

@@ -0,0 +1,43 @@
+/*
+ * Copyright (c) 2020 Pivotal Software Inc, All Rights Reserved.
+ *
+ * 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.fn.filter;
+
+import java.util.Optional;
+import java.util.function.Function;
+
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.cloud.fn.spel.SpelFunctionConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.messaging.Message;
+
+@Configuration
+@Import(SpelFunctionConfiguration.class)
+public class FilterFunctionConfiguration {
+
+	@Bean
+	public Function<Message<?>, Message<?>> filterFunction(
+			@Qualifier("spelFunction") Function<Message<?>, Message<?>> spelFunction) {
+
+		return message ->
+				Optional.of(message)
+						.filter(m -> (Boolean) spelFunction.apply(m).getPayload())
+						.orElse(null);
+	}
+
+}

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

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

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

@@ -0,0 +1,53 @@
+/*
+ * Copyright (c) 2011-2020 Pivotal Software Inc, All Rights Reserved.
+ *
+ * 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.fn.filter;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.function.Function;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+
+@SpringBootTest(properties = "spel.function.expression=payload.length() > 5")
+@DirtiesContext
+public class FilterFunctionApplicationTests {
+
+	@Autowired
+	@Qualifier("filterFunction")
+	Function<Message<?>, 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");
+	}
+
+	@SpringBootApplication
+	static class TestApplication {}
+}

function/payload-converter-function/pom.xml

@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+		 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<artifactId>payload-converter-function</artifactId>
+	<version>1.0.0.BUILD-SNAPSHOT</version>
+	<name>payload-converter-function</name>
+	<description>Utility message conversion functions</description>
+
+	<parent>
+		<groupId>org.springframework.cloud.fn</groupId>
+		<artifactId>spring-functions-parent</artifactId>
+		<version>1.0.0.BUILD-SNAPSHOT</version>
+		<relativePath>../../spring-functions-parent</relativePath>
+	</parent>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-messaging</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>org.junit.vintage</groupId>
+					<artifactId>junit-vintage-engine</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+	</dependencies>
+
+</project>

function/payload-converter-function/src/main/java/functions/ByteArrayTextToString.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2020 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 functions;
+
+import java.util.function.Function;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.MimeTypeUtils;
+
+/**
+ *
+ * @author Christian Tzolov
+ */
+public class ByteArrayTextToString implements Function<Message<?>, Message<?>> {
+
+	@Override
+	public Message<?> apply(Message<?> message) {
+
+		if (message.getPayload() instanceof byte[]) {
+			final MessageHeaders headers = message.getHeaders();
+			String contentType = headers.containsKey(MessageHeaders.CONTENT_TYPE)
+					? headers.get(MessageHeaders.CONTENT_TYPE).toString()
+					: MimeTypeUtils.APPLICATION_JSON_VALUE;
+
+			if (contentType.contains("text") || contentType.contains("json") || contentType.contains("x-spring-tuple")) {
+				message = MessageBuilder.withPayload(new String(((byte[]) message.getPayload())))
+						.copyHeaders(message.getHeaders())
+						.build();
+			}
+		}
+
+		return message;
+	}
+}
+

function/payload-converter-function/src/test/java/functions/ByteArrayTextToStringTests.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) 2011-2020 Pivotal Software Inc, All Rights Reserved.
+ *
+ * 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 functions;
+
+import java.util.Collections;
+import java.util.function.Function;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+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;
+
+public class ByteArrayTextToStringTests {
+
+	private static final String MESSAGE = "hello world";
+	private static Function<Message<?>, Message<?>> converter;
+
+	@BeforeAll
+	static void before() {
+		converter = new ByteArrayTextToString();
+	}
+
+	@Test
+	public void testDefaultNoContentType() {
+		Message<?> converted = converter.apply(new GenericMessage<>(MESSAGE.getBytes()));
+		assertThat(converted).isNotNull().extracting(Message::getPayload).isEqualTo(MESSAGE);
+
+		converted = converter.apply(new GenericMessage<>(MESSAGE)); // String
+		assertThat(converted).isNotNull().extracting(Message::getPayload).isEqualTo(MESSAGE);
+	}
+
+	@Test
+	public void testApplicationJsonContentType() {
+		Message<?> converted = converter.apply(new GenericMessage<>(MESSAGE.getBytes(),
+				Collections.singletonMap(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE)));
+		assertThat(converted).isNotNull().extracting(Message::getPayload).isEqualTo("hello world");
+	}
+
+	@Test
+	public void testPlainTextContentType() {
+		Message<?> converted = converter.apply(new GenericMessage<>(MESSAGE.getBytes(),
+				Collections.singletonMap(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN_VALUE)));
+		assertThat(converted).isNotNull().extracting(Message::getPayload).isEqualTo(MESSAGE);
+	}
+
+	@Test
+	public void testOctetContentType() {
+		Message<?> converted = converter.apply(new GenericMessage<>(MESSAGE.getBytes(),
+				Collections.singletonMap(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE)));
+		assertThat(converted).isNotNull().extracting(Message::getPayload).isEqualTo(MESSAGE.getBytes());
+	}
+
+	@Test
+	public void testRandomNonTextContentType() {
+		Message<?> converted = converter.apply(new GenericMessage<>(MESSAGE.getBytes(),
+				Collections.singletonMap(MessageHeaders.CONTENT_TYPE, "Random Content Type")));
+		assertThat(converted).isNotNull().extracting(Message::getPayload).isEqualTo(MESSAGE.getBytes());
+	}
+
+}

function/spel-function/.gitignore

@@ -0,0 +1,28 @@
+/target/
+!.mvn/wrapper/maven-wrapper.jar
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+/build/
+
+### VS Code ###
+.vscode/

function/spel-function/pom.xml

@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<artifactId>spel-function</artifactId>
+	<version>1.0.0.BUILD-SNAPSHOT</version>
+	<name>spel-function</name>
+	<description>Spring Native Function for applying SpEL expressions</description>
+
+	<parent>
+		<groupId>org.springframework.cloud.fn</groupId>
+		<artifactId>spring-functions-parent</artifactId>
+		<version>1.0.0.BUILD-SNAPSHOT</version>
+		<relativePath>../../spring-functions-parent</relativePath>
+	</parent>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud.function</groupId>
+			<artifactId>payload-converter-function</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-integration</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>org.junit.vintage</groupId>
+					<artifactId>junit-vintage-engine</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-configuration-processor</artifactId>
+			<scope>provided</scope>
+		</dependency>
+	</dependencies>
+
+</project>

function/spel-function/src/main/java/org/springframework/cloud/fn/spel/SpelFunctionConfiguration.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright (c) 2020 Pivotal Software Inc, All Rights Reserved.
+ *
+ * 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.fn.spel;
+
+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;
+
+@Configuration
+@EnableConfigurationProperties(SpelFunctionProperties.class)
+public class SpelFunctionConfiguration {
+
+	@Bean
+	public Function<Message<?>, Message<?>> spelFunction(
+			ExpressionEvaluatingTransformer expressionEvaluatingTransformer) {
+
+		return message -> expressionEvaluatingTransformer.transform(message);
+	}
+
+	@Bean
+	public ExpressionEvaluatingTransformer expressionEvaluatingTransformer(
+			SpelFunctionProperties spelFunctionProperties) {
+
+		return new ExpressionEvaluatingTransformer(new SpelExpressionParser()
+				.parseExpression(spelFunctionProperties.getExpression()));
+	}
+
+}

function/spel-function/src/main/java/org/springframework/cloud/fn/spel/SpelFunctionProperties.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2020 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
+ *
+ *      http://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.fn.spel;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+
+/**
+ * Configuration properties for the SpEL function.
+ *
+ * @author Gary Russell
+ * @author Artem Bilan
+ */
+@ConfigurationProperties("spel.function")
+public class SpelFunctionProperties {
+
+	private static final Expression DEFAULT_EXPRESSION = new SpelExpressionParser().parseExpression("payload");
+
+	/**
+	 * A SpEL expression to apply.
+	 */
+	private String expression = DEFAULT_EXPRESSION.getExpressionString();
+
+	public void setExpression(String expression) {
+		this.expression = expression;
+	}
+
+	public String getExpression() {
+		return this.expression;
+	}
+
+}

function/spel-function/src/test/java/org/springframework/cloud/fn/spel/SpelFunctionApplicationTests.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright (c) 2011-2020 Pivotal Software Inc, All Rights Reserved.
+ *
+ * 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.fn.spel;
+
+import java.util.function.Function;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+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.test.annotation.DirtiesContext;
+import org.springframework.util.MimeTypeUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest(properties = "spel.function.expression=payload.toUpperCase()")
+@DirtiesContext
+public class SpelFunctionApplicationTests {
+
+	@Autowired
+	Function<Message<?>, Message<?>> transformer;
+
+	@Test
+	public void testTransform() {
+		final Message<?> transformed = this.transformer.apply(new GenericMessage<>("hello,world"));
+		assertThat(transformed.getPayload()).isEqualTo("HELLO,WORLD");
+	}
+
+	@Test
+	public void testJson() {
+		Message<?> message = MessageBuilder.withPayload("{\"foo\":\"bar\"}")
+				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build();
+		final Message<?> transformed = this.transformer.apply(message);
+		assertThat(transformed.getPayload()).isEqualTo("{\"FOO\":\"BAR\"}");
+	}
+
+	@SpringBootApplication
+	static class TestApplication {
+
+	}
+}

function/splitter-function/.gitignore

@@ -0,0 +1,28 @@
+/target/
+!.mvn/wrapper/maven-wrapper.jar
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+/build/
+
+### VS Code ###
+.vscode/

function/splitter-function/pom.xml

@@ -0,0 +1,50 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>splitter-function</artifactId>
+    <version>1.0.0.BUILD-SNAPSHOT</version>
+    <name>splitter-function</name>
+    <description>Spring Native Function for Splitter</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0.BUILD-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-file</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.junit.vintage</groupId>
+                    <artifactId>junit-vintage-engine</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
+</project>

function/splitter-function/src/main/java/org/springframework/cloud/fn/splitter/SplitterFunctionConfiguration.java

@@ -0,0 +1,125 @@
+/*
+ * Copyright (c) 2011-2020 Pivotal Software Inc, All Rights Reserved.
+ *
+ * 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.fn.splitter;
+
+import java.nio.charset.Charset;
+import java.util.List;
+import java.util.function.Function;
+
+import org.reactivestreams.Publisher;
+
+import org.springframework.boot.autoconfigure.condition.AnyNestedCondition;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
+import org.springframework.integration.file.splitter.FileSplitter;
+import org.springframework.integration.splitter.AbstractMessageSplitter;
+import org.springframework.integration.splitter.DefaultMessageSplitter;
+import org.springframework.integration.splitter.ExpressionEvaluatingSplitter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+
+import reactor.core.publisher.Flux;
+
+@Configuration
+@EnableConfigurationProperties(SplitterFunctionProperties.class)
+public class SplitterFunctionConfiguration {
+
+	@Bean
+	public Function<Message<?>, List<Message<?>>> splitterFunction(AbstractMessageSplitter messageSplitter,
+			SplitterFunctionProperties splitterFunctionProperties) {
+
+		messageSplitter.setApplySequence(splitterFunctionProperties.isApplySequence());
+		ThreadLocalFluxSinkMessageChannel outputChannel = new ThreadLocalFluxSinkMessageChannel();
+		messageSplitter.setOutputChannel(outputChannel);
+		return message -> {
+			messageSplitter.handleMessage(message);
+			return outputChannel.publisherThreadLocal.get();
+		};
+	}
+
+	@Bean
+	@ConditionalOnProperty(prefix = "splitter", name = "expression")
+	public AbstractMessageSplitter expressionSplitter(SplitterFunctionProperties splitterFunctionProperties) {
+		return new ExpressionEvaluatingSplitter(
+				new SpelExpressionParser()
+						.parseExpression(splitterFunctionProperties.getExpression()));
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	@Conditional(FileSplitterCondition.class)
+	public AbstractMessageSplitter fileSplitter(SplitterFunctionProperties splitterFunctionProperties) {
+		Boolean markers = splitterFunctionProperties.getFileMarkers();
+		String charset = splitterFunctionProperties.getCharset();
+		if (markers == null) {
+			markers = false;
+		}
+		FileSplitter fileSplitter = new FileSplitter(true, markers, splitterFunctionProperties.getMarkersJson());
+		if (charset != null) {
+			fileSplitter.setCharset(Charset.forName(charset));
+		}
+		return fileSplitter;
+	}
+
+
+	@Bean
+	@ConditionalOnMissingBean
+	public AbstractMessageSplitter defaultSplitter(SplitterFunctionProperties splitterFunctionProperties) {
+		DefaultMessageSplitter defaultMessageSplitter = new DefaultMessageSplitter();
+		defaultMessageSplitter.setDelimiters(splitterFunctionProperties.getDelimiters());
+		return defaultMessageSplitter;
+	}
+
+	static class FileSplitterCondition extends AnyNestedCondition {
+
+		FileSplitterCondition() {
+			super(ConfigurationPhase.REGISTER_BEAN);
+		}
+
+		@ConditionalOnProperty(prefix = "splitter", name = "charset")
+		static class Charset { }
+
+		@ConditionalOnProperty(prefix = "splitter", name = "fileMarkers")
+		static class FileMarkers { }
+
+	}
+
+	private static final class ThreadLocalFluxSinkMessageChannel
+			implements MessageChannel, ReactiveStreamsSubscribableChannel {
+
+		private final ThreadLocal<List<Message<?>>> publisherThreadLocal = new ThreadLocal<>();
+
+		@Override
+		@SuppressWarnings("unchecked")
+		public void subscribeTo(Publisher<? extends Message<?>> publisher) {
+			this.publisherThreadLocal.set(Flux.from(publisher).collectList().cast(List.class).block());
+		}
+
+		@Override
+		public boolean send(Message<?> message, long l) {
+			throw new UnsupportedOperationException("This channel only supports a reactive 'subscribeTo()' ");
+		}
+
+	}
+
+}

function/splitter-function/src/main/java/org/springframework/cloud/fn/splitter/SplitterFunctionProperties.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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.fn.splitter;
+
+import javax.validation.constraints.AssertTrue;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * Configuration properties for the Splitter Processor app.
+ *
+ * @author Gary Russell
+ * @author Artem Bilan
+ */
+@ConfigurationProperties("splitter")
+@Validated
+public class SplitterFunctionProperties {
+
+	/**
+	 * A SpEL expression for splitting payloads.
+	 */
+	private String expression;
+
+	/**
+	 * When expression is null, delimiters to use when tokenizing
+	 * {@link String} payloads.
+	 */
+	private String delimiters;
+
+	/**
+	 * Set to true or false to use a {@code FileSplitter} (to split
+	 * text-based files by line) that includes
+	 * (or not) beginning/end of file markers.
+	 */
+	private Boolean fileMarkers;
+
+	/**
+	 * When 'fileMarkers == true', specify if they should be produced
+	 * as FileSplitter.FileMarker objects or JSON.
+	 */
+	private boolean markersJson = true;
+
+	/**
+	 * The charset to use when converting bytes in text-based files
+	 * to String.
+	 */
+	private String charset;
+
+	/**
+	 * Add correlation/sequence information in headers to facilitate later
+	 * aggregation.
+	 */
+	private boolean applySequence = true;
+
+	public void setExpression(String expression) {
+		this.expression = expression;
+	}
+
+	public String getExpression() {
+		return this.expression;
+	}
+
+	public String getDelimiters() {
+		return this.delimiters;
+	}
+
+	public void setDelimiters(String delimiters) {
+		this.delimiters = delimiters;
+	}
+
+	public Boolean getFileMarkers() {
+		return this.fileMarkers;
+	}
+
+	public void setFileMarkers(Boolean fileMarkers) {
+		this.fileMarkers = fileMarkers;
+	}
+
+	public boolean getMarkersJson() {
+		return this.markersJson;
+	}
+
+	public void setMarkersJson(boolean markersJson) {
+		this.markersJson = markersJson;
+	}
+
+	public String getCharset() {
+		return this.charset;
+	}
+
+	public void setCharset(String charset) {
+		this.charset = charset;
+	}
+
+	public boolean isApplySequence() {
+		return this.applySequence;
+	}
+
+	public void setApplySequence(boolean applySequence) {
+		this.applySequence = applySequence;
+	}
+
+	@AssertTrue(message = "'delimiters' is not allowed when an 'expression' is provided")
+	public boolean isDelimitersAllowed() {
+		return this.expression == null || this.delimiters == null;
+	}
+
+	@AssertTrue(message = "File properties are not allowed when an 'expression' or 'delimiters' property is provided")
+	public boolean isFilePropsAllowed() {
+		return !(this.expression != null || this.delimiters != null) || this.fileMarkers == null && this.charset == null;
+	}
+
+}

function/splitter-function/src/test/java/org/springframework/cloud/fn/splitter/SplitterFunctionApplicationTests.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2011-2020 Pivotal Software Inc, All Rights Reserved.
+ *
+ * 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.fn.splitter;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.function.Function;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
+
+@SpringBootTest(properties = "splitter.expression=payload.split(',')")
+@DirtiesContext
+public class SplitterFunctionApplicationTests {
+
+	@Autowired
+	Function<Message<?>, List<Message<?>>> splitter;
+
+	@Test
+	public void testExpressionSplitter() {
+		List<Message<?>> messageList = this.splitter.apply(new GenericMessage<>("hello,world"));
+		assertThat(messageList).extracting(m -> m.getPayload().toString()).contains("hello", "world");
+	}
+
+	@SpringBootApplication
+	static class TestApplication {}
+}