From 97fdd0ccc56435fa749777637b944cb0602128fb Mon Sep 17 00:00:00 2001 From: Corneil du Plessis Date: Fri, 4 Nov 2022 17:40:01 +0200 Subject: [PATCH] [Issues] Change http-request to non-reactive (#402) Fixes #206 --- .../HttpRequestFunctionConfiguration.java | 31 ++- ...tpRequestFunctionTestApplicationTests.java | 194 ++++++++---------- 2 files changed, 95 insertions(+), 130 deletions(-) diff --git a/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java b/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java index fc307d14..1b5e6c86 100644 --- a/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java +++ b/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java @@ -20,8 +20,6 @@ import java.time.Duration; import java.util.Map; import java.util.function.Function; -import reactor.core.publisher.Flux; - import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -42,7 +40,7 @@ import org.springframework.web.util.UriBuilderFactory; * * @author David Turanski * @author Sunny Hemdev - * + * @author Corneil du Plessis **/ @Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(HttpRequestFunctionProperties.class) @@ -63,7 +61,7 @@ public class HttpRequestFunctionConfiguration { * Function that accepts a {@code Flux>} containing body and headers and * returns a {@code Flux>}. */ - public static class HttpRequestFunction implements Function>, Flux> { + public static class HttpRequestFunction implements Function, Object> { private final WebClient webClient; @@ -77,16 +75,17 @@ public class HttpRequestFunctionConfiguration { } @Override - public Flux apply(Flux> messageFlux) { - return messageFlux.flatMap(message -> this.webClient - .method(resolveHttpMethod(message)) - .uri(uriBuilderFactory.uriString(resolveUrl(message)).build()) - .bodyValue(resolveBody(message)) - .headers(httpHeaders -> httpHeaders.addAll(resolveHeaders(message))) - .retrieve() - .toEntity(properties.getExpectedResponseType()) - .map(responseEntity -> properties.getReplyExpression().getValue(responseEntity)) - .timeout(Duration.ofMillis(properties.getTimeout()))); + public Object apply(Message message) { + return this.webClient + .method(resolveHttpMethod(message)) + .uri(uriBuilderFactory.uriString(resolveUrl(message)).build()) + .bodyValue(resolveBody(message)) + .headers(httpHeaders -> httpHeaders.addAll(resolveHeaders(message))) + .retrieve() + .toEntity(properties.getExpectedResponseType()) + .map(responseEntity -> properties.getReplyExpression().getValue(responseEntity)) + .timeout(Duration.ofMillis(properties.getTimeout())) + .block(); } private String resolveUrl(Message message) { @@ -99,7 +98,7 @@ public class HttpRequestFunctionConfiguration { private Object resolveBody(Message message) { return properties.getBodyExpression() != null ? properties.getBodyExpression().getValue(message) - : message.getPayload(); + : message.getPayload(); } private HttpHeaders resolveHeaders(Message message) { @@ -109,7 +108,7 @@ public class HttpRequestFunctionConfiguration { for (Map.Entry header : headersMap.entrySet()) { if (header.getKey() != null && header.getValue() != null) { headers.add(header.getKey().toString(), - header.getValue().toString()); + header.getValue().toString()); } } } diff --git a/function/http-request-function/src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionTestApplicationTests.java b/function/http-request-function/src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionTestApplicationTests.java index bde2f390..3243c0b6 100644 --- a/function/http-request-function/src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionTestApplicationTests.java +++ b/function/http-request-function/src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionTestApplicationTests.java @@ -27,8 +27,6 @@ import okhttp3.mockwebserver.MockWebServer; import okhttp3.mockwebserver.RecordedRequest; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import reactor.core.publisher.Flux; -import reactor.test.StepVerifier; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.test.context.runner.ApplicationContextRunner; @@ -42,6 +40,7 @@ import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; public class HttpRequestFunctionTestApplicationTests { private MockWebServer server = new MockWebServer(); @@ -67,14 +66,9 @@ public class HttpRequestFunctionTestApplicationTests { runner.withPropertyValues("http.request.http-method-expression='POST'").run(context -> { HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); Message message = MessageBuilder.withPayload("").build(); - StepVerifier.create(httpRequestFunction.apply(Flux.just(message))) - .assertNext(o -> { - ResponseEntity r = (ResponseEntity) o; - assertThat(r.getBody()).isEqualTo("hello"); - assertThat(r.getStatusCode().is2xxSuccessful()).isTrue(); - }) - .expectComplete() - .verify(); + ResponseEntity r = (ResponseEntity) httpRequestFunction.apply(message); + assertThat(r.getBody()).isEqualTo("hello"); + assertThat(r.getStatusCode().is2xxSuccessful()).isTrue(); }); } @@ -86,29 +80,24 @@ public class HttpRequestFunctionTestApplicationTests { public MockResponse dispatch(RecordedRequest recordedRequest) { return new MockResponse().setHeader(HttpHeaders.CONTENT_TYPE, recordedRequest.getHeader(HttpHeaders.CONTENT_TYPE)) - .setBody(recordedRequest.getBody()) - .setResponseCode(HttpStatus.CREATED.value()); + .setBody(recordedRequest.getBody()) + .setResponseCode(HttpStatus.CREATED.value()); } }); runner.withPropertyValues("http.request.http-method-expression='POST'", - "http.request.headers-expression={'Content-Type':'application/json'}").run(context -> { - HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); - String json = "{\"hello\":\"world\"}"; - Message message = MessageBuilder.withPayload(json) - .build(); - StepVerifier.create(httpRequestFunction.apply(Flux.just(message))) - .assertNext(o -> { - ResponseEntity r = (ResponseEntity) o; - assertThat(r.getBody()).isEqualTo(json); - assertThat(r.getStatusCode().is2xxSuccessful()).isTrue(); - assertThat(r.getHeaders().getContentType()).isEqualTo(MediaType.APPLICATION_JSON); - }) - .expectComplete() - .verify(); - RecordedRequest request = server.takeRequest(100, TimeUnit.MILLISECONDS); - assertThat(request.getMethod()).isEqualTo("POST"); - }); + "http.request.headers-expression={'Content-Type':'application/json'}").run(context -> { + HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); + String json = "{\"hello\":\"world\"}"; + Message message = MessageBuilder.withPayload(json) + .build(); + ResponseEntity r = (ResponseEntity) httpRequestFunction.apply(message); + assertThat(r.getBody()).isEqualTo(json); + assertThat(r.getStatusCode().is2xxSuccessful()).isTrue(); + assertThat(r.getHeaders().getContentType()).isEqualTo(MediaType.APPLICATION_JSON); + RecordedRequest request = server.takeRequest(100, TimeUnit.MILLISECONDS); + assertThat(request.getMethod()).isEqualTo("POST"); + }); } @Test @@ -119,29 +108,24 @@ public class HttpRequestFunctionTestApplicationTests { public MockResponse dispatch(RecordedRequest recordedRequest) { return new MockResponse().setHeader(HttpHeaders.CONTENT_TYPE, recordedRequest.getHeader(HttpHeaders.CONTENT_TYPE)) - .setBody(recordedRequest.getBody()) - .setResponseCode(HttpStatus.CREATED.value()); + .setBody(recordedRequest.getBody()) + .setResponseCode(HttpStatus.CREATED.value()); } }); runner.withPropertyValues("http.request.http-method-expression='POST'", - "http.request.expected-response-type=" + Map.class.getName()).run(context -> { - HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); - Map json = Collections.singletonMap("hello", "world"); - Message message = MessageBuilder.withPayload(json) - .build(); - StepVerifier.create(httpRequestFunction.apply(Flux.just(message))) - .assertNext(o -> { - ResponseEntity r = (ResponseEntity) o; - assertThat(r.getBody()).isEqualTo(json); - assertThat(r.getStatusCode().is2xxSuccessful()).isTrue(); - assertThat(r.getHeaders().getContentType()).isEqualTo(MediaType.APPLICATION_JSON); - }) - .expectComplete() - .verify(); - RecordedRequest request = server.takeRequest(100, TimeUnit.MILLISECONDS); - assertThat(request.getMethod()).isEqualTo("POST"); - }); + "http.request.expected-response-type=" + Map.class.getName()).run(context -> { + HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); + Map json = Collections.singletonMap("hello", "world"); + Message message = MessageBuilder.withPayload(json) + .build(); + ResponseEntity r = (ResponseEntity) httpRequestFunction.apply(message); + assertThat(r.getBody()).isEqualTo(json); + assertThat(r.getStatusCode().is2xxSuccessful()).isTrue(); + assertThat(r.getHeaders().getContentType()).isEqualTo(MediaType.APPLICATION_JSON); + RecordedRequest request = server.takeRequest(100, TimeUnit.MILLISECONDS); + assertThat(request.getMethod()).isEqualTo("POST"); + }); } @Test @@ -157,21 +141,16 @@ public class HttpRequestFunctionTestApplicationTests { }); runner.withPropertyValues("http.request.http-method-expression='DELETE'", - "http.request.expected-response-type=" + Void.class.getName()).run(context -> { - HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); - Message message = MessageBuilder.withPayload("") - .build(); - StepVerifier.create(httpRequestFunction.apply(Flux.just(message))) - .assertNext(o -> { - ResponseEntity r = (ResponseEntity) o; - assertThat(r.getBody()).isNull(); - assertThat(r.getStatusCode().is2xxSuccessful()).isTrue(); - }) - .expectComplete() - .verify(); - RecordedRequest request = server.takeRequest(100, TimeUnit.MILLISECONDS); - assertThat(request.getMethod()).isEqualTo("DELETE"); - }); + "http.request.expected-response-type=" + Void.class.getName()).run(context -> { + HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); + Message message = MessageBuilder.withPayload("") + .build(); + ResponseEntity r = (ResponseEntity) httpRequestFunction.apply(message); + assertThat(r.getBody()).isNull(); + assertThat(r.getStatusCode().is2xxSuccessful()).isTrue(); + RecordedRequest request = server.takeRequest(100, TimeUnit.MILLISECONDS); + assertThat(request.getMethod()).isEqualTo("DELETE"); + }); } @Test @@ -179,9 +158,12 @@ public class HttpRequestFunctionTestApplicationTests { server.shutdown(); runner.run(context -> { HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); - StepVerifier.create(httpRequestFunction.apply(Flux.just(new GenericMessage("")))) - .expectErrorMatches(throwable -> throwable.getMessage().contains("Connection refused")) - .verify(); + try { + httpRequestFunction.apply(new GenericMessage("")); + fail("Expected exception"); + } catch (Throwable throwable) { + assertThat(throwable.getMessage()).contains("Connection refused"); + } }); } @@ -202,25 +184,20 @@ public class HttpRequestFunctionTestApplicationTests { "http.request.http-method-expression=headers['method']", "http.request.body-expression=headers['body']", "http.request.headers-expression={Accept:'application/json'}") - .run(context -> { - Message message = MessageBuilder.withPayload("") - .setHeader("url", url()) - .setHeader("method", "POST") - .setHeader("body", "{\"hello\":\"world\"}") - .build(); + .run(context -> { + Message message = MessageBuilder.withPayload("") + .setHeader("url", url()) + .setHeader("method", "POST") + .setHeader("body", "{\"hello\":\"world\"}") + .build(); - HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); - StepVerifier.create(httpRequestFunction.apply(Flux.just(message))) - .assertNext(o -> { - ResponseEntity responseEntity = (ResponseEntity) o; - assertThat(responseEntity.getStatusCode()).isEqualTo(HttpStatus.OK); - assertThat(responseEntity.getBody()).isEqualTo(message.getHeaders().get("body")); - assertThat(responseEntity.getHeaders().getContentType()) - .isEqualTo(MediaType.APPLICATION_JSON); - }) - .expectComplete() - .verify(); - }); + HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); + ResponseEntity responseEntity = (ResponseEntity) httpRequestFunction.apply(message); + assertThat(responseEntity.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(responseEntity.getBody()).isEqualTo(message.getHeaders().get("body")); + assertThat(responseEntity.getHeaders().getContentType()) + .isEqualTo(MediaType.APPLICATION_JSON); + }); } @Test @@ -238,21 +215,16 @@ public class HttpRequestFunctionTestApplicationTests { "http.request..http-method-expression='POST'", "http.request.headers-expression={'Content-Type':'application/octet-stream'}", "http.request.expected-response-type=byte[]") - .run(context -> { - Message message = MessageBuilder.withPayload("hello") - .build(); - HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); - StepVerifier.create(httpRequestFunction.apply(Flux.just(message))) - .assertNext(o -> { - ResponseEntity responseEntity = (ResponseEntity) o; - assertThat(responseEntity.getStatusCode()).isEqualTo(HttpStatus.OK); - assertThat(responseEntity.getBody()).isEqualTo("hello".getBytes()); - assertThat(responseEntity.getHeaders().getContentType()) - .isEqualTo(MediaType.APPLICATION_OCTET_STREAM); - }) - .expectComplete() - .verify(); - }); + .run(context -> { + Message message = MessageBuilder.withPayload("hello") + .build(); + HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); + ResponseEntity responseEntity = (ResponseEntity) httpRequestFunction.apply(message); + assertThat(responseEntity.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(responseEntity.getBody()).isEqualTo("hello".getBytes()); + assertThat(responseEntity.getHeaders().getContentType()) + .isEqualTo(MediaType.APPLICATION_OCTET_STREAM); + }); } @Test @@ -268,21 +240,15 @@ public class HttpRequestFunctionTestApplicationTests { }); runner.withPropertyValues( "http.request.http-method-expression=#jsonPath(payload,'$.myMethod')") - .run(context -> { - Message message = MessageBuilder - .withPayload("{\"name\":\"Fred\",\"age\":41, \"myMethod\":\"POST\"}") - .build(); - HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); - StepVerifier.create(httpRequestFunction.apply(Flux.just(message))) - .assertNext(o -> { - ResponseEntity responseEntity = (ResponseEntity) o; - assertThat(responseEntity.getStatusCode()).isEqualTo(HttpStatus.OK); - assertThat(responseEntity.getBody()).isEqualTo(message.getPayload()); - }) - .expectComplete() - .verify(); - - }); + .run(context -> { + Message message = MessageBuilder + .withPayload("{\"name\":\"Fred\",\"age\":41, \"myMethod\":\"POST\"}") + .build(); + HttpRequestFunction httpRequestFunction = context.getBean(HttpRequestFunction.class); + ResponseEntity responseEntity = (ResponseEntity) httpRequestFunction.apply(message); + assertThat(responseEntity.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(responseEntity.getBody()).isEqualTo(message.getPayload()); + }); } private String url() {