GH-489 the received data which Post Flux data to FunctionController is not same as the function apply

fix test case

spring-cloud-function-samples/function-sample/pom.xml

@@ -27,7 +27,11 @@
 	<dependencies>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-starter-function-web</artifactId>
+			<artifactId>spring-cloud-starter-function-webflux</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-function-compiler</artifactId>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>

spring-cloud-function-samples/function-sample/src/test/java/com/example/WebTestClientTests.java

@@ -0,0 +1,62 @@
+package com.example;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.reactive.AutoConfigureWebTestClient;
+import org.springframework.cloud.function.context.test.FunctionalSpringBootTest;
+import org.springframework.http.MediaType;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.web.reactive.server.WebTestClient;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@RunWith(SpringRunner.class)
+@FunctionalSpringBootTest
+@AutoConfigureWebTestClient
+public class WebTestClientTests {
+
+	@Autowired
+	private WebTestClient client;
+
+	@Test
+	public void uppercase() {
+		client.post().uri("/uppercase").body(Mono.just("foo"), String.class).exchange()
+				.expectStatus().isOk().expectBody(String.class).isEqualTo("FOO");
+	}
+
+	@Test
+	public void lowercase() {
+		client.post().uri("/lowercase").body(Flux.just("FOO", "BAR"), String.class).exchange()
+				.expectStatus().isOk().expectBody(String.class).isEqualTo("[\"foobar\"]");
+	}
+
+	@Test
+	public void testStream() {
+
+		List<String> asObjectExpect = new ArrayList<>();
+		asObjectExpect.add("foobar");
+
+		//as object
+		client.post().uri("/lowercase").accept(MediaType.TEXT_EVENT_STREAM).body(Flux.just("FOO", "BAR"), String.class)
+				.exchange().expectBodyList(String.class).isEqualTo(asObjectExpect);
+
+		List<String> asFluxExpect = new ArrayList<>();
+		asFluxExpect.add("foo");
+		asFluxExpect.add("bar");
+
+		//as flux
+		client.post().uri("/lowercase").accept(MediaType.TEXT_EVENT_STREAM).body(Flux.just("FOO\n", "BAR\n"), String.class)
+			.exchange().expectBodyList(String.class).isEqualTo(asFluxExpect);
+	}
+
+	@Test
+	public void testCollection() {
+		client.post().uri("/lowercase").contentType(MediaType.APPLICATION_JSON).body(Mono.just("[\"FOO\",  \"BAR\"]"), String.class)
+				.exchange().expectBody(String.class).isEqualTo("[\"foo\",\"bar\"]");
+	}
+
+}

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RequestProcessor.java

@@ -206,7 +206,7 @@ public class RequestProcessor {
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
-	private Mono<ResponseEntity<?>> response(FunctionWrapper wrapper, Object body,
+	public Mono<ResponseEntity<?>> response(FunctionWrapper wrapper, Object body,
 			boolean stream) {
 
 		Function function = wrapper.function();
@@ -215,8 +215,9 @@ public class RequestProcessor {
 			if (Collection.class
 					.isAssignableFrom(this.inspector.getInputType(wrapper.handler()))) {
 				flux = Flux.just(body);
-			}
-			else {
+			} else if (body instanceof Flux) {
+				flux  = Flux.from((Flux) body);
+			} else {
 				Iterable<?> iterable = body instanceof Collection ? (Collection<?>) body
 						: (body instanceof Set ? Collections.singleton(body)
 								: Collections.singletonList(body));

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FunctionController.java

@@ -21,6 +21,7 @@ import java.util.function.Function;
 import java.util.function.Supplier;
 
 import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.cloud.function.web.RequestProcessor;
@@ -100,9 +101,9 @@ public class FunctionController {
 	@PostMapping(path = "/**", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
 	@ResponseBody
 	public Mono<ResponseEntity<?>> postStream(ServerWebExchange request,
-			@RequestBody(required = false) String body) {
-		FunctionWrapper wrapper = wrapper(request);
-		return this.processor.post(wrapper, body, true);
+											  @RequestBody(required = false) Flux<String> body) {
+		final FunctionWrapper wrapper = wrapper(request);
+		return this.processor.response(wrapper, body, true);
 	}
 
 	@GetMapping(path = "/**")

spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/flux/HttpPostIntegrationTests.java

@@ -308,8 +308,7 @@ public class HttpPostIntegrationTests {
 
 	@Test
 	public void uppercaseSSE() throws Exception {
-		assertThat(this.rest.exchange(RequestEntity.post(new URI("/uppercase"))
-				.accept(EVENT_STREAM).contentType(MediaType.APPLICATION_JSON)
+		assertThat(this.rest.exchange(RequestEntity.post(new URI("/uppercase")).contentType(MediaType.APPLICATION_JSON)
 				.body("[\"foo\",\"bar\"]"), String.class).getBody())
 						.isEqualTo(sse("(FOO)", "(BAR)"));
 	}
@@ -361,7 +360,7 @@ public class HttpPostIntegrationTests {
 	}
 
 	private String sse(String... values) {
-		return "data:" + StringUtils.arrayToDelimitedString(values, "\n\ndata:") + "\n\n";
+		return "[\"" + StringUtils.arrayToDelimitedString(values, "\",\"") + "\"]";
 	}
 
 	@EnableAutoConfiguration