On POST to a Consumer, reflect input and send 202

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

@@ -26,6 +26,8 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.cloud.function.registry.FunctionCatalog;
 import org.springframework.cloud.function.support.FluxSupplier;
 import org.springframework.cloud.function.support.FunctionUtils;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.PostMapping;
@@ -56,18 +58,19 @@ public class FunctionController {
 	}
 
 	@PostMapping(path = "/{name}")
-	public Flux<String> function(@PathVariable String name,
+	public ResponseEntity<Flux<String>> function(@PathVariable String name,
 			@RequestBody Flux<String> body) {
 		Function<Flux<?>, Flux<?>> function = functions.lookupFunction(name);
 		if (function != null) {
 			@SuppressWarnings("unchecked")
 			Flux<String> result = (Flux<String>) function.apply(body);
-			return debug ? result.log() : result;
+			return ResponseEntity.ok().body(debug ? result.log() : result);
 		}
 		Consumer<Flux<?>> consumer = functions.lookupConsumer(name);
 		if (consumer != null) {
+			body = body.cache(); // send a copy back to the caller
 			consumer.accept(body);
-			return null;
+			return ResponseEntity.status(HttpStatus.ACCEPTED).body(body);
 		}
 		throw new IllegalArgumentException("no such function: " + name);
 	}

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

@@ -19,6 +19,8 @@ package org.springframework.cloud.function.web.flux;
 import java.time.Duration;
 import java.util.List;
 
+import javax.servlet.http.HttpServletResponse;
+
 import org.reactivestreams.Publisher;
 
 import org.springframework.core.MethodParameter;
@@ -87,7 +89,10 @@ public class FluxReturnValueHandler implements AsyncHandlerMethodReturnValueHand
 			throws Exception {
 		Object adaptFrom = returnValue;
 		if (returnValue instanceof ResponseEntity) {
-			adaptFrom = ((ResponseEntity<?>) returnValue).getBody();
+			ResponseEntity<?> value = (ResponseEntity<?>) returnValue;
+			adaptFrom = value.getBody();
+			webRequest.getNativeResponse(HttpServletResponse.class)
+					.setStatus(value.getStatusCodeValue());
 		}
 		Publisher<?> flux = (Publisher<?>) adaptFrom;
 

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

@@ -61,6 +61,8 @@ public class RestApplicationTests {
 	private int port;
 	@Autowired
 	private TestRestTemplate rest;
+	@Autowired
+	private TestConfiguration test;
 
 	@Test
 	public void wordsSSE() throws Exception {
@@ -98,8 +100,9 @@ public class RestApplicationTests {
 	public void updates() throws Exception {
 		ResponseEntity<String> result = rest.exchange(
 				RequestEntity.post(new URI("/updates")).body("one\ntwo"), String.class);
-		assertThat(result.getStatusCode()).isEqualTo(HttpStatus.OK);
-		assertThat(result.getBody()).isNull();
+		assertThat(result.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED);
+		assertThat(test.list).hasSize(2);
+		assertThat(result.getBody()).isEqualTo("onetwo");
 	}
 
 	@Test