From 134e884361ff417abd7825014f33dabe0a14ca4f Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 8 Dec 2021 13:12:24 -0500 Subject: [PATCH] Fix HTTP Supplier for auto-configured codecs 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` --- .../http/HttpSupplierConfiguration.java | 8 ++++-- .../http/HttpSupplierApplicationTests.java | 26 +++++++++++++++++-- 2 files changed, 30 insertions(+), 4 deletions(-) diff --git a/functions/supplier/http-supplier/src/main/java/org/springframework/cloud/fn/supplier/http/HttpSupplierConfiguration.java b/functions/supplier/http-supplier/src/main/java/org/springframework/cloud/fn/supplier/http/HttpSupplierConfiguration.java index f2c9a80e..a7ccdc95 100644 --- a/functions/supplier/http-supplier/src/main/java/org/springframework/cloud/fn/supplier/http/HttpSupplierConfiguration.java +++ b/functions/supplier/http-supplier/src/main/java/org/springframework/cloud/fn/supplier/http/HttpSupplierConfiguration.java @@ -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> httpSupplierFlow(HttpSupplierProperties httpSupplierProperties) { + public Publisher> 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()) diff --git a/functions/supplier/http-supplier/src/test/java/org/springframework/cloud/fn/supplier/http/HttpSupplierApplicationTests.java b/functions/supplier/http-supplier/src/test/java/org/springframework/cloud/fn/supplier/http/HttpSupplierApplicationTests.java index 374735a3..4d749888 100644 --- a/functions/supplier/http-supplier/src/test/java/org/springframework/cloud/fn/supplier/http/HttpSupplierApplicationTests.java +++ b/functions/supplier/http-supplier/src/test/java/org/springframework/cloud/fn/supplier/http/HttpSupplierApplicationTests.java @@ -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>> 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> messageFlux = this.httpSupplier.get(); StepVerifier stepVerifier =