Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use write aggregator from DataBufferUtils

Browse Source 
This commit uses the newly introduced write aggregator from
DataBufferUtils throughout WebFlux.

Issue: SPR-16351
This commit is contained in:
Arjen Poutsma
2018-01-10 11:15:47 +01:00
parent 69ccba30e9
commit 8a129ef3da
10 changed files with 27 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-core/src/main/java/org/springframework/core/codec/AbstractDataBufferDecoder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -24,6 +24,7 @@ import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.lang.Nullable;
import org.springframework.util.MimeType;
@@ -63,7 +64,7 @@ public abstract class AbstractDataBufferDecoder<T> extends AbstractDecoder<T> {
@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
return Flux.from(inputStream)
.reduce(DataBuffer::write)
.reduce(DataBufferUtils.writeAggregator())
.map(buffer -> decodeDataBuffer(buffer, elementType, mimeType, hints));
}

6
spring-web/src/main/java/org/springframework/http/codec/FormHttpMessageReader.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -29,7 +29,6 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.MediaType;
import org.springframework.http.ReactiveHttpInputMessage;
@@ -97,8 +96,7 @@ public class FormHttpMessageReader implements HttpMessageReader<MultiValueMap<St
MediaType contentType = message.getHeaders().getContentType();
Charset charset = getMediaTypeCharset(contentType);
return message.getBody()
.reduce(DataBuffer::write)
return message.getBody().reduce(DataBufferUtils.writeAggregator())
.map(buffer -> {
CharBuffer charBuffer = charset.decode(buffer.asByteBuffer());
String body = charBuffer.toString();

4
spring-web/src/main/java/org/springframework/http/codec/xml/XmlEventDecoder.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -103,7 +103,7 @@ public class XmlEventDecoder extends AbstractDecoder<XMLEvent> {
.doFinally(signalType -> aaltoMapper.endOfInput());
}
else {
Mono<DataBuffer> singleBuffer = flux.reduce(DataBuffer::write);
Mono<DataBuffer> singleBuffer = flux.reduce(DataBufferUtils.writeAggregator());
return singleBuffer.
flatMapMany(dataBuffer -> {
try {

9
spring-web/src/test/java/org/springframework/http/codec/multipart/SynchronossPartHttpMessageReaderTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -26,6 +26,7 @@ import reactor.test.StepVerifier;
import org.springframework.core.ResolvableType;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
@@ -38,9 +39,7 @@ import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import static java.util.Collections.emptyMap;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.*;
import static org.springframework.core.ResolvableType.forClassWithGenerics;
import static org.springframework.http.HttpHeaders.CONTENT_LENGTH;
import static org.springframework.http.HttpHeaders.CONTENT_TYPE;
@@ -90,7 +89,7 @@ public class SynchronossPartHttpMessageReaderTests {
assertTrue(part instanceof FilePart);
assertEquals("fooPart", part.name());
assertEquals("foo.txt", ((FilePart) part).filename());
DataBuffer buffer = part.content().reduce(DataBuffer::write).block();
DataBuffer buffer = part.content().reduce(DataBufferUtils.writeAggregator()).block();
assertEquals(12, buffer.readableByteCount());
byte[] byteContent = new byte[12];
buffer.read(byteContent);

5
spring-web/src/test/java/org/springframework/http/server/reactive/MultipartIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -23,6 +23,7 @@ import reactor.core.publisher.Mono;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
@@ -104,7 +105,7 @@ public class MultipartIntegrationTests extends AbstractHttpHandlerIntegrationTes
assertEquals("foo.txt", ((FilePart) part).filename());
DataBuffer buffer = part
.content()
.reduce(DataBuffer::write)
.reduce(DataBufferUtils.writeAggregator())
.block();
assertEquals(12, buffer.readableByteCount());
byte[] byteContent = new byte[12];

5
spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -36,7 +36,6 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
@@ -459,7 +458,7 @@ class DefaultWebClient implements WebClient {
private static Mono<WebClientResponseException> createResponseException(ClientResponse response) {
return response.body(BodyExtractors.toDataBuffers())
.reduce(DataBuffer::write)
.reduce(DataBufferUtils.writeAggregator())
.map(dataBuffer -> {
byte[] bytes = new byte[dataBuffer.readableByteCount()];
dataBuffer.read(bytes);

5
spring-webflux/src/main/java/org/springframework/web/reactive/resource/AppCacheManifestTransformer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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,6 @@ import reactor.core.publisher.Mono;
import reactor.core.publisher.SynchronousSink;
import org.springframework.core.io.Resource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.lang.Nullable;
@@ -111,7 +110,7 @@ public class AppCacheManifestTransformer extends ResourceTransformerSupport {
}
DataBufferFactory bufferFactory = exchange.getResponse().bufferFactory();
return DataBufferUtils.read(outputResource, bufferFactory, StreamUtils.BUFFER_SIZE)
.reduce(DataBuffer::write)
.reduce(DataBufferUtils.writeAggregator())
.flatMap(dataBuffer -> {
CharBuffer charBuffer = DEFAULT_CHARSET.decode(dataBuffer.asByteBuffer());
DataBufferUtils.release(dataBuffer);

7
spring-webflux/src/main/java/org/springframework/web/reactive/resource/ContentVersionStrategy.java
View File

@@ -1,11 +1,11 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
@@ -19,7 +19,6 @@ package org.springframework.web.reactive.resource;
import reactor.core.publisher.Mono;
import org.springframework.core.io.Resource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
@@ -44,7 +43,7 @@ public class ContentVersionStrategy extends AbstractFileNameVersionStrategy {
@Override
public Mono<String> getResourceVersion(Resource resource) {
return DataBufferUtils.read(resource, dataBufferFactory, StreamUtils.BUFFER_SIZE)
.reduce(DataBuffer::write)
.reduce(DataBufferUtils.writeAggregator())
.map(buffer -> {
byte[] result = new byte[buffer.readableByteCount()];
buffer.read(result);

5
spring-webflux/src/main/java/org/springframework/web/reactive/resource/CssLinkResourceTransformer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -33,7 +33,6 @@ import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.core.io.Resource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.lang.Nullable;
@@ -88,7 +87,7 @@ public class CssLinkResourceTransformer extends ResourceTransformerSupport {
DataBufferFactory bufferFactory = exchange.getResponse().bufferFactory();
return DataBufferUtils.read(ouptputResource, bufferFactory, StreamUtils.BUFFER_SIZE)
.reduce(DataBuffer::write)
.reduce(DataBufferUtils.writeAggregator())
.flatMap(dataBuffer -> {
CharBuffer charBuffer = DEFAULT_CHARSET.decode(dataBuffer.asByteBuffer());
DataBufferUtils.release(dataBuffer);

7
spring-webflux/src/test/java/org/springframework/web/reactive/function/BodyInsertersTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -65,7 +65,7 @@ import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.hamcrest.Matchers.*;
import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.*;
import static org.springframework.http.codec.json.Jackson2CodecSupport.JSON_VIEW_HINT;
@@ -332,7 +332,8 @@ public class BodyInsertersTests {
Mono<Void> result = inserter.insert(request, this.context);
StepVerifier.create(result).expectComplete().verify();
StepVerifier.create(request.getBody().reduce(DataBuffer::write))
StepVerifier.create(request.getBody()
.reduce(DataBufferUtils.writeAggregator()))
.consumeNextWith(dataBuffer -> {
byte[] resultBytes = new byte[dataBuffer.readableByteCount()];
dataBuffer.read(resultBytes);