SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[Issues] Change http-request to non-reactive (#402)

Browse Source 
Fixes #206
This commit is contained in:
Corneil du Plessis
2022-11-04 17:40:01 +02:00
committed by GitHub
parent 4f393ef9dc
commit 97fdd0ccc5
2 changed files with 95 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java
View File

@@ -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<Message<?>>} containing body and headers and
* returns a {@code Flux<ResponseEntity<?>>}.
*/
public static class HttpRequestFunction implements Function<Flux<Message<?>>, Flux<?>> {
public static class HttpRequestFunction implements Function<Message<?>, Object> {
private final WebClient webClient;
@@ -77,16 +75,17 @@ public class HttpRequestFunctionConfiguration {
}
@Override
public Flux<?> apply(Flux<Message<?>> 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());
}
}
}

194
function/http-request-function/src/test/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionTestApplicationTests.java
View File

@@ -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<String, String> 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<String, String> 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() {