diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/FunctionController.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/FunctionController.java index 5242ba702..6c43d968c 100644 --- a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/FunctionController.java +++ b/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 function(@PathVariable String name, + public ResponseEntity> function(@PathVariable String name, @RequestBody Flux body) { Function, Flux> function = functions.lookupFunction(name); if (function != null) { @SuppressWarnings("unchecked") Flux result = (Flux) function.apply(body); - return debug ? result.log() : result; + return ResponseEntity.ok().body(debug ? result.log() : result); } Consumer> 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); } diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FluxReturnValueHandler.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FluxReturnValueHandler.java index 8f5702bfb..592e56613 100644 --- a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FluxReturnValueHandler.java +++ b/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; diff --git a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/RestApplicationTests.java b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/RestApplicationTests.java index bff7128d1..e742b2975 100644 --- a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/RestApplicationTests.java +++ b/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 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