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

Fix HTTP Supplier for auto-configured codecs

Browse Source 
SO: https://stackoverflow.com/questions/70242824/spring-cloud-data-flow-http-source-kafka-application-request-body-memory-size-is

The `WebFlux.inboundChannelAdapter()` creates internally `ServerCodecConfigurer.create()` if not provided.
And with that any customizations for codes are not possible from end-user perspective

* Fix `HttpSupplierConfiguration` to autowire `ServerCodecConfigurer`
which is auto-configured by Spring Boot.
This way end-user can use external configuration properties to customize his/her requirements,
e.g. `spring.codec.max-in-memory-size`
* Cover the feature with unit test to make sure that `spring.codec.max-in-memory-size=10MB`
is propagated properly to the `WebFluxInboundEndpoint.codecConfigurer`
This commit is contained in:
Artem Bilan
2021-12-08 13:12:24 -05:00
committed by Soby Chacko
parent fb4926f66e
commit 9828120f34
2 changed files with 30 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
supplier/http-supplier/src/main/java/org/springframework/cloud/fn/supplier/http/HttpSupplierConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2011-2020 the original author or authors.
* Copyright 2011-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -27,6 +27,7 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.http.support.DefaultHttpHeaderMapper;
@@ -46,12 +47,15 @@ import org.springframework.messaging.MessageHeaders;
public class HttpSupplierConfiguration {
@Bean
public Publisher<Message<byte[]>> httpSupplierFlow(HttpSupplierProperties httpSupplierProperties) {
public Publisher<Message<byte[]>> httpSupplierFlow(HttpSupplierProperties httpSupplierProperties,
ServerCodecConfigurer serverCodecConfigurer) {
return IntegrationFlows.from(
WebFlux.inboundChannelAdapter(httpSupplierProperties.getPathPattern())
.requestPayloadType(byte[].class)
.statusCodeExpression(new ValueExpression<>(HttpStatus.ACCEPTED))
.mappedRequestHeaders(httpSupplierProperties.getMappedRequestHeaders())
.codecConfigurer(serverCodecConfigurer)
.crossOrigin(crossOrigin ->
crossOrigin.origin(httpSupplierProperties.getCors().getAllowedOrigins())
.allowedHeaders(httpSupplierProperties.getCors().getAllowedHeaders())

26
supplier/http-supplier/src/test/java/org/springframework/cloud/fn/supplier/http/HttpSupplierApplicationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -34,7 +34,11 @@ import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.web.server.LocalServerPort;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.http.codec.json.AbstractJackson2Decoder;
import org.springframework.integration.http.HttpHeaders;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.webflux.inbound.WebFluxInboundEndpoint;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.web.reactive.function.client.WebClient;
@@ -52,18 +56,36 @@ import static org.assertj.core.api.Assertions.assertThat;
"server.ssl.key-store=classpath:test.jks",
"server.ssl.key-password=password",
"server.ssl.trust-store=classpath:test.jks",
"server.ssl.client-auth=want"
"server.ssl.client-auth=want",
"spring.codec.max-in-memory-size=10MB"
})
public class HttpSupplierApplicationTests {
@Autowired
private Supplier<Flux<Message<byte[]>>> httpSupplier;
@Autowired
private WebFluxInboundEndpoint webFluxInboundEndpoint;
@LocalServerPort
private int port;
@Test
public void testHttpSupplier() {
ServerCodecConfigurer codecConfigurer =
TestUtils.getPropertyValue(this.webFluxInboundEndpoint, "codecConfigurer", ServerCodecConfigurer.class);
final ServerCodecConfigurer.ServerDefaultCodecs serverDefaultCodecs = codecConfigurer.defaultCodecs();
assertThat(TestUtils.getPropertyValue(serverDefaultCodecs, "maxInMemorySize", Integer.class))
.isEqualTo(1024 * 1024 * 10);
AbstractJackson2Decoder jackson2JsonDecoder =
TestUtils.getPropertyValue(serverDefaultCodecs, "jackson2JsonDecoder", AbstractJackson2Decoder.class);
assertThat(jackson2JsonDecoder).isNotNull()
.extracting("maxInMemorySize")
.isEqualTo(1024 * 1024 * 10);
Flux<Message<byte[]>> messageFlux = this.httpSupplier.get();
StepVerifier stepVerifier =