From 65b46079a2d58899b1d1c146cd347d3701be6d13 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 22 Mar 2019 20:24:26 -0400 Subject: [PATCH 1/2] Replace stream with loop in RequestCondition's --- .../result/condition/HeadersRequestCondition.java | 8 +++++++- .../result/condition/ParamsRequestCondition.java | 8 +++++++- .../result/condition/PatternsRequestCondition.java | 13 ++++++++----- .../mvc/condition/HeadersRequestCondition.java | 8 +++++++- .../mvc/condition/ParamsRequestCondition.java | 8 +++++++- 5 files changed, 36 insertions(+), 9 deletions(-) diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/result/condition/HeadersRequestCondition.java b/spring-webflux/src/main/java/org/springframework/web/reactive/result/condition/HeadersRequestCondition.java index f486cd2933..9136c6d9a1 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/result/condition/HeadersRequestCondition.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/result/condition/HeadersRequestCondition.java @@ -142,7 +142,13 @@ public final class HeadersRequestCondition extends AbstractRequestCondition expressions) { - return expressions.stream().filter(e -> e.getValue() != null && !e.isNegated()).count(); + long count = 0; + for (HeaderExpression e : expressions) { + if (e.getValue() != null && !e.isNegated()) { + count++; + } + } + return count; } diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/result/condition/ParamsRequestCondition.java b/spring-webflux/src/main/java/org/springframework/web/reactive/result/condition/ParamsRequestCondition.java index be35f2b1cf..544726cafc 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/result/condition/ParamsRequestCondition.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/result/condition/ParamsRequestCondition.java @@ -124,7 +124,13 @@ public final class ParamsRequestCondition extends AbstractRequestCondition expressions) { - return expressions.stream().filter(e -> e.getValue() != null && !e.isNegated()).count(); + long count = 0; + for (ParamExpression e : expressions) { + if (e.getValue() != null && !e.isNegated()) { + count++; + } + } + return count; } diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/result/condition/PatternsRequestCondition.java b/spring-webflux/src/main/java/org/springframework/web/reactive/result/condition/PatternsRequestCondition.java index 0dd91b2b73..75c09038f6 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/result/condition/PatternsRequestCondition.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/result/condition/PatternsRequestCondition.java @@ -24,7 +24,6 @@ import java.util.List; import java.util.Set; import java.util.SortedSet; import java.util.TreeSet; -import java.util.stream.Collectors; import org.springframework.http.server.PathContainer; import org.springframework.lang.Nullable; @@ -53,7 +52,7 @@ public final class PatternsRequestCondition extends AbstractRequestCondition patterns) { this(new TreeSet<>(patterns)); @@ -136,9 +135,13 @@ public final class PatternsRequestCondition extends AbstractRequestCondition getMatchingPatterns(ServerWebExchange exchange) { PathContainer lookupPath = exchange.getRequest().getPath().pathWithinApplication(); - return this.patterns.stream() - .filter(pattern -> pattern.matches(lookupPath)) - .collect(Collectors.toCollection(TreeSet::new)); + TreeSet pathPatterns = new TreeSet<>(); + for (PathPattern pattern : this.patterns) { + if (pattern.matches(lookupPath)) { + pathPatterns.add(pattern); + } + } + return pathPatterns; } /** diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/condition/HeadersRequestCondition.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/condition/HeadersRequestCondition.java index b30c53aeab..061305dd42 100644 --- a/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/condition/HeadersRequestCondition.java +++ b/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/condition/HeadersRequestCondition.java @@ -142,7 +142,13 @@ public final class HeadersRequestCondition extends AbstractRequestCondition expressions) { - return expressions.stream().filter(e -> e.getValue() != null && !e.isNegated()).count(); + long count = 0; + for (HeaderExpression e : expressions) { + if (e.getValue() != null && !e.isNegated()) { + count++; + } + } + return count; } diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/condition/ParamsRequestCondition.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/condition/ParamsRequestCondition.java index c55678dc93..d93d6b9922 100644 --- a/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/condition/ParamsRequestCondition.java +++ b/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/condition/ParamsRequestCondition.java @@ -127,7 +127,13 @@ public final class ParamsRequestCondition extends AbstractRequestCondition expressions) { - return expressions.stream().filter(e -> e.getValue() != null && !e.isNegated()).count(); + long count = 0; + for (ParamExpression e : expressions) { + if (e.getValue() != null && !e.isNegated()) { + count++; + } + } + return count; } From c54355784e5868c41341abed83e246709e318474 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Tue, 26 Mar 2019 21:11:19 -0400 Subject: [PATCH 2/2] Buffer leak fixes Address issues where buffers are allocated (and cached somehow) at or before subscription, and before explicit demand. The commit adds tests proving the leaks and fixes. The common thread for all tests is a "zero demand" subscriber that subscribes but does not request, and then cancels without consuming anything. Closes gh-22107 --- .../codec/AbstractSingleValueEncoder.java | 10 +- .../core/codec/ResourceRegionEncoder.java | 25 +-- .../core/io/buffer/DataBufferUtils.java | 99 +++++----- .../codec/ResourceRegionEncoderTests.java | 51 ++++- .../AbstractDataBufferAllocatingTestCase.java | 22 ++- .../core/io/buffer/DataBufferUtilsTests.java | 26 ++- .../io/buffer/LeakAwareDataBufferFactory.java | 38 +++- .../http/codec/EncoderHttpMessageWriter.java | 4 +- .../http/codec/FormHttpMessageWriter.java | 4 +- .../ServerSentEventHttpMessageWriter.java | 33 ++-- .../codec/json/AbstractJackson2Encoder.java | 19 +- .../multipart/MultipartHttpMessageWriter.java | 138 +++++++------ .../http/codec/xml/Jaxb2XmlEncoder.java | 50 ++--- .../http/codec/CodecDataBufferLeakTests.java | 186 ++++++++++++++++++ .../MultipartHttpMessageWriterTests.java | 7 +- .../reactive/test/MockServerHttpResponse.java | 3 +- 16 files changed, 504 insertions(+), 211 deletions(-) create mode 100644 spring-web/src/test/java/org/springframework/http/codec/CodecDataBufferLeakTests.java diff --git a/spring-core/src/main/java/org/springframework/core/codec/AbstractSingleValueEncoder.java b/spring-core/src/main/java/org/springframework/core/codec/AbstractSingleValueEncoder.java index 9691deb748..3a45507b03 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/AbstractSingleValueEncoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/AbstractSingleValueEncoder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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.Flux; import org.springframework.core.ResolvableType; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferFactory; +import org.springframework.core.io.buffer.PooledDataBuffer; import org.springframework.lang.Nullable; import org.springframework.util.MimeType; @@ -47,9 +48,10 @@ public abstract class AbstractSingleValueEncoder extends AbstractEncoder { public final Flux encode(Publisher inputStream, DataBufferFactory bufferFactory, ResolvableType elementType, @Nullable MimeType mimeType, @Nullable Map hints) { - return Flux.from(inputStream). - take(1). - concatMap(t -> encode(t, bufferFactory, elementType, mimeType, hints)); + return Flux.from(inputStream) + .take(1) + .concatMap(value -> encode(value, bufferFactory, elementType, mimeType, hints)) + .doOnDiscard(PooledDataBuffer.class, PooledDataBuffer::release); } /** diff --git a/spring-core/src/main/java/org/springframework/core/codec/ResourceRegionEncoder.java b/spring-core/src/main/java/org/springframework/core/codec/ResourceRegionEncoder.java index b382ae6a79..3695a5514f 100644 --- a/spring-core/src/main/java/org/springframework/core/codec/ResourceRegionEncoder.java +++ b/spring-core/src/main/java/org/springframework/core/codec/ResourceRegionEncoder.java @@ -17,7 +17,6 @@ package org.springframework.core.codec; import java.io.IOException; -import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.util.Map; import java.util.OptionalLong; @@ -89,24 +88,22 @@ public class ResourceRegionEncoder extends AbstractEncoder { return Mono.from(inputStream) .flatMapMany(region -> { if (!region.getResource().isReadable()) { - return Flux.error(new EncodingException("Resource " + - region.getResource() + " is not readable")); + return Flux.error(new EncodingException( + "Resource " + region.getResource() + " is not readable")); } - return writeResourceRegion(region, bufferFactory, hints); }); } else { final String boundaryString = Hints.getRequiredHint(hints, BOUNDARY_STRING_HINT); byte[] startBoundary = getAsciiBytes("\r\n--" + boundaryString + "\r\n"); - byte[] contentType = - (mimeType != null ? getAsciiBytes("Content-Type: " + mimeType + "\r\n") : new byte[0]); + byte[] contentType = mimeType != null ? getAsciiBytes("Content-Type: " + mimeType + "\r\n") : new byte[0]; return Flux.from(inputStream). concatMap(region -> { if (!region.getResource().isReadable()) { - return Flux.error(new EncodingException("Resource " + - region.getResource() + " is not readable")); + return Flux.error(new EncodingException( + "Resource " + region.getResource() + " is not readable")); } else { return Flux.concat( @@ -121,11 +118,10 @@ public class ResourceRegionEncoder extends AbstractEncoder { private Flux getRegionPrefix(DataBufferFactory bufferFactory, byte[] startBoundary, byte[] contentType, ResourceRegion region) { - return Flux.defer(() -> Flux.just( - bufferFactory.allocateBuffer(startBoundary.length).write(startBoundary), - bufferFactory.allocateBuffer(contentType.length).write(contentType), - bufferFactory.wrap(ByteBuffer.wrap(getContentRangeHeader(region)))) - ); + return Flux.just( + bufferFactory.wrap(startBoundary), + bufferFactory.wrap(contentType), + bufferFactory.wrap(getContentRangeHeader(region))); // only wrapping, no allocation } private Flux writeResourceRegion( @@ -146,8 +142,7 @@ public class ResourceRegionEncoder extends AbstractEncoder { private Flux getRegionSuffix(DataBufferFactory bufferFactory, String boundaryString) { byte[] endBoundary = getAsciiBytes("\r\n--" + boundaryString + "--"); - return Flux.defer(() -> Flux.just( - bufferFactory.allocateBuffer(endBoundary.length).write(endBoundary))); + return Flux.just(bufferFactory.wrap(endBoundary)); } private byte[] getAsciiBytes(String in) { diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java index 902a9a91b8..8ed89edff3 100644 --- a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java +++ b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -82,40 +82,36 @@ public abstract class DataBufferUtils { * Obtain a {@link ReadableByteChannel} from the given supplier, and read it into a * {@code Flux} of {@code DataBuffer}s. Closes the channel when the flux is terminated. * @param channelSupplier the supplier for the channel to read from - * @param dataBufferFactory the factory to create data buffers with + * @param bufferFactory the factory to create data buffers with * @param bufferSize the maximum size of the data buffers * @return a flux of data buffers read from the given channel */ public static Flux readByteChannel( - Callable channelSupplier, DataBufferFactory dataBufferFactory, int bufferSize) { + Callable channelSupplier, DataBufferFactory bufferFactory, int bufferSize) { Assert.notNull(channelSupplier, "'channelSupplier' must not be null"); - Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null"); + Assert.notNull(bufferFactory, "'dataBufferFactory' must not be null"); Assert.isTrue(bufferSize > 0, "'bufferSize' must be > 0"); return Flux.using(channelSupplier, - channel -> { - ReadableByteChannelGenerator generator = - new ReadableByteChannelGenerator(channel, dataBufferFactory, - bufferSize); - return Flux.generate(generator); - }, - DataBufferUtils::closeChannel) - .doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release); + channel -> Flux.generate(new ReadableByteChannelGenerator(channel, bufferFactory, bufferSize)), + DataBufferUtils::closeChannel); + + // No doOnDiscard as operators used do not cache } /** * Obtain a {@code AsynchronousFileChannel} from the given supplier, and read it into a * {@code Flux} of {@code DataBuffer}s. Closes the channel when the flux is terminated. * @param channelSupplier the supplier for the channel to read from - * @param dataBufferFactory the factory to create data buffers with + * @param bufferFactory the factory to create data buffers with * @param bufferSize the maximum size of the data buffers * @return a flux of data buffers read from the given channel */ public static Flux readAsynchronousFileChannel( - Callable channelSupplier, DataBufferFactory dataBufferFactory, int bufferSize) { + Callable channelSupplier, DataBufferFactory bufferFactory, int bufferSize) { - return readAsynchronousFileChannel(channelSupplier, 0, dataBufferFactory, bufferSize); + return readAsynchronousFileChannel(channelSupplier, 0, bufferFactory, bufferSize); } /** @@ -124,32 +120,30 @@ public abstract class DataBufferUtils { * channel when the flux is terminated. * @param channelSupplier the supplier for the channel to read from * @param position the position to start reading from - * @param dataBufferFactory the factory to create data buffers with + * @param bufferFactory the factory to create data buffers with * @param bufferSize the maximum size of the data buffers * @return a flux of data buffers read from the given channel */ public static Flux readAsynchronousFileChannel(Callable channelSupplier, - long position, DataBufferFactory dataBufferFactory, int bufferSize) { + long position, DataBufferFactory bufferFactory, int bufferSize) { Assert.notNull(channelSupplier, "'channelSupplier' must not be null"); - Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null"); + Assert.notNull(bufferFactory, "'dataBufferFactory' must not be null"); Assert.isTrue(position >= 0, "'position' must be >= 0"); Assert.isTrue(bufferSize > 0, "'bufferSize' must be > 0"); - DataBuffer dataBuffer = dataBufferFactory.allocateBuffer(bufferSize); - ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, bufferSize); - - Flux result = Flux.using(channelSupplier, + Flux flux = Flux.using(channelSupplier, channel -> Flux.create(sink -> { - AsynchronousFileChannelReadCompletionHandler completionHandler = - new AsynchronousFileChannelReadCompletionHandler(channel, - sink, position, dataBufferFactory, bufferSize); - channel.read(byteBuffer, position, dataBuffer, completionHandler); - sink.onDispose(completionHandler::dispose); + ReadCompletionHandler handler = + new ReadCompletionHandler(channel, sink, position, bufferFactory, bufferSize); + DataBuffer dataBuffer = bufferFactory.allocateBuffer(bufferSize); + ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, bufferSize); + channel.read(byteBuffer, position, dataBuffer, handler); + sink.onDispose(handler::dispose); }), DataBufferUtils::closeChannel); - return result.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release); + return flux.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release); } /** @@ -246,8 +240,7 @@ public abstract class DataBufferUtils { Flux flux = Flux.from(source); return Flux.create(sink -> { - WritableByteChannelSubscriber subscriber = - new WritableByteChannelSubscriber(sink, channel); + WritableByteChannelSubscriber subscriber = new WritableByteChannelSubscriber(sink, channel); sink.onDispose(subscriber); flux.subscribe(subscriber); }); @@ -292,10 +285,9 @@ public abstract class DataBufferUtils { Flux flux = Flux.from(source); return Flux.create(sink -> { - AsynchronousFileChannelWriteCompletionHandler completionHandler = - new AsynchronousFileChannelWriteCompletionHandler(sink, channel, position); - sink.onDispose(completionHandler); - flux.subscribe(completionHandler); + WriteCompletionHandler handler = new WriteCompletionHandler(sink, channel, position); + sink.onDispose(handler); + flux.subscribe(handler); }); } @@ -326,21 +318,21 @@ public abstract class DataBufferUtils { Assert.notNull(publisher, "Publisher must not be null"); Assert.isTrue(maxByteCount >= 0, "'maxByteCount' must be a positive number"); - return Flux.defer(() -> { - AtomicLong countDown = new AtomicLong(maxByteCount); - return Flux.from(publisher) - .map(buffer -> { - long remainder = countDown.addAndGet(-buffer.readableByteCount()); - if (remainder < 0) { - int length = buffer.readableByteCount() + (int) remainder; - return buffer.slice(0, length); - } - else { - return buffer; - } - }) - .takeUntil(buffer -> countDown.get() <= 0); - }); // no doOnDiscard necessary, as this method does not drop buffers + AtomicLong countDown = new AtomicLong(maxByteCount); + return Flux.from(publisher) + .map(buffer -> { + long remainder = countDown.addAndGet(-buffer.readableByteCount()); + if (remainder < 0) { + int length = buffer.readableByteCount() + (int) remainder; + return buffer.slice(0, length); + } + else { + return buffer; + } + }) + .takeUntil(buffer -> countDown.get() <= 0); + + // No doOnDiscard as operators used do not cache (and drop) buffers } /** @@ -487,8 +479,7 @@ public abstract class DataBufferUtils { } - private static class AsynchronousFileChannelReadCompletionHandler - implements CompletionHandler { + private static class ReadCompletionHandler implements CompletionHandler { private final AsynchronousFileChannel channel; @@ -502,7 +493,7 @@ public abstract class DataBufferUtils { private final AtomicBoolean disposed = new AtomicBoolean(); - public AsynchronousFileChannelReadCompletionHandler(AsynchronousFileChannel channel, + public ReadCompletionHandler(AsynchronousFileChannel channel, FluxSink sink, long position, DataBufferFactory dataBufferFactory, int bufferSize) { this.channel = channel; @@ -586,7 +577,7 @@ public abstract class DataBufferUtils { } - private static class AsynchronousFileChannelWriteCompletionHandler extends BaseSubscriber + private static class WriteCompletionHandler extends BaseSubscriber implements CompletionHandler { private final FluxSink sink; @@ -601,7 +592,7 @@ public abstract class DataBufferUtils { private final AtomicReference dataBuffer = new AtomicReference<>(); - public AsynchronousFileChannelWriteCompletionHandler( + public WriteCompletionHandler( FluxSink sink, AsynchronousFileChannel channel, long position) { this.sink = sink; diff --git a/spring-core/src/test/java/org/springframework/core/codec/ResourceRegionEncoderTests.java b/spring-core/src/test/java/org/springframework/core/codec/ResourceRegionEncoderTests.java index 316e288c49..112fd6a7c5 100644 --- a/spring-core/src/test/java/org/springframework/core/codec/ResourceRegionEncoderTests.java +++ b/spring-core/src/test/java/org/springframework/core/codec/ResourceRegionEncoderTests.java @@ -20,6 +20,8 @@ import java.util.Collections; import java.util.function.Consumer; import org.junit.Test; +import org.reactivestreams.Subscription; +import reactor.core.publisher.BaseSubscriber; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; @@ -28,7 +30,6 @@ import org.springframework.core.ResolvableType; import org.springframework.core.io.ClassPathResource; 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.LeakAwareDataBufferFactory; import org.springframework.core.io.buffer.support.DataBufferTestUtils; @@ -36,19 +37,18 @@ import org.springframework.core.io.support.ResourceRegion; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; -import static java.nio.charset.StandardCharsets.UTF_8; +import static java.nio.charset.StandardCharsets.*; import static org.junit.Assert.*; /** * Test cases for {@link ResourceRegionEncoder} class. - * * @author Brian Clozel */ public class ResourceRegionEncoderTests { private ResourceRegionEncoder encoder = new ResourceRegionEncoder(); - private DataBufferFactory bufferFactory = new LeakAwareDataBufferFactory(); + private LeakAwareDataBufferFactory bufferFactory = new LeakAwareDataBufferFactory(); @Test @@ -79,10 +79,13 @@ public class ResourceRegionEncoderTests { .consumeNextWith(stringConsumer("Spring")) .expectComplete() .verify(); + + // TODO: https://github.com/reactor/reactor-core/issues/1634 + // this.bufferFactory.checkForLeaks(); } @Test - public void shouldEncodeMultipleResourceRegionsFileResource() throws Exception { + public void shouldEncodeMultipleResourceRegionsFileResource() { Resource resource = new ClassPathResource("ResourceRegionEncoderTests.txt", getClass()); Flux regions = Flux.just( new ResourceRegion(resource, 0, 6), @@ -118,6 +121,33 @@ public class ResourceRegionEncoderTests { .consumeNextWith(stringConsumer("\r\n--" + boundary + "--")) .expectComplete() .verify(); + + // TODO: https://github.com/reactor/reactor-core/issues/1634 + // this.bufferFactory.checkForLeaks(); + } + + @Test // gh- + public void cancelWithoutDemandForMultipleResourceRegions() { + Resource resource = new ClassPathResource("ResourceRegionEncoderTests.txt", getClass()); + Flux regions = Flux.just( + new ResourceRegion(resource, 0, 6), + new ResourceRegion(resource, 7, 9), + new ResourceRegion(resource, 17, 4), + new ResourceRegion(resource, 22, 17) + ); + String boundary = MimeTypeUtils.generateMultipartBoundaryString(); + + Flux flux = this.encoder.encode(regions, this.bufferFactory, + ResolvableType.forClass(ResourceRegion.class), + MimeType.valueOf("text/plain"), + Collections.singletonMap(ResourceRegionEncoder.BOUNDARY_STRING_HINT, boundary) + ); + + ZeroDemandSubscriber subscriber = new ZeroDemandSubscriber(); + flux.subscribe(subscriber); + subscriber.cancel(); + + this.bufferFactory.checkForLeaks(); } @Test @@ -142,6 +172,9 @@ public class ResourceRegionEncoderTests { .consumeNextWith(stringConsumer("Spring")) .expectError(EncodingException.class) .verify(); + + // TODO: https://github.com/reactor/reactor-core/issues/1634 + // this.bufferFactory.checkForLeaks(); } protected Consumer stringConsumer(String expected) { @@ -154,4 +187,12 @@ public class ResourceRegionEncoderTests { } + private static class ZeroDemandSubscriber extends BaseSubscriber { + + @Override + protected void hookOnSubscribe(Subscription subscription) { + // Just subscribe without requesting + } + } + } diff --git a/spring-core/src/test/java/org/springframework/core/io/buffer/AbstractDataBufferAllocatingTestCase.java b/spring-core/src/test/java/org/springframework/core/io/buffer/AbstractDataBufferAllocatingTestCase.java index 733bda3662..a7e21feefc 100644 --- a/spring-core/src/test/java/org/springframework/core/io/buffer/AbstractDataBufferAllocatingTestCase.java +++ b/spring-core/src/test/java/org/springframework/core/io/buffer/AbstractDataBufferAllocatingTestCase.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -121,9 +121,23 @@ public abstract class AbstractDataBufferAllocatingTestCase { if (this.bufferFactory instanceof NettyDataBufferFactory) { ByteBufAllocator allocator = ((NettyDataBufferFactory) this.bufferFactory).getByteBufAllocator(); if (allocator instanceof PooledByteBufAllocator) { - PooledByteBufAllocatorMetric metric = ((PooledByteBufAllocator) allocator).metric(); - long total = getAllocations(metric.directArenas()) + getAllocations(metric.heapArenas()); - assertEquals("ByteBuf Leak: " + total + " unreleased allocations", 0, total); + Instant start = Instant.now(); + while (true) { + PooledByteBufAllocatorMetric metric = ((PooledByteBufAllocator) allocator).metric(); + long total = getAllocations(metric.directArenas()) + getAllocations(metric.heapArenas()); + if (total == 0) { + return; + } + if (Instant.now().isBefore(start.plus(Duration.ofSeconds(5)))) { + try { + Thread.sleep(50); + } + catch (InterruptedException ex) { + // ignore + } + } + assertEquals("ByteBuf Leak: " + total + " unreleased allocations", 0, total); + } } } } diff --git a/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java b/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java index b0d3a334fe..314788bdcd 100644 --- a/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java +++ b/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -35,8 +35,11 @@ import java.util.concurrent.CountDownLatch; import io.netty.buffer.ByteBuf; import org.junit.Before; +import org.junit.Ignore; import org.junit.Test; import org.mockito.stubbing.Answer; +import org.reactivestreams.Subscription; +import reactor.core.publisher.BaseSubscriber; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; @@ -184,6 +187,20 @@ public class DataBufferUtilsTests extends AbstractDataBufferAllocatingTestCase { .verify(); } + // TODO: Remove ignore after https://github.com/reactor/reactor-core/issues/1634 + @Ignore + @Test // gh-22107 + public void readAsynchronousFileChannelCancelWithoutDemand() throws Exception { + URI uri = this.resource.getURI(); + Flux flux = DataBufferUtils.readAsynchronousFileChannel( + () -> AsynchronousFileChannel.open(Paths.get(uri), StandardOpenOption.READ), + this.bufferFactory, 3); + + BaseSubscriber subscriber = new ZeroDemandSubscriber(); + flux.subscribe(subscriber); + subscriber.cancel(); + } + @Test public void readResource() throws Exception { Flux flux = DataBufferUtils.read(this.resource, this.bufferFactory, 3); @@ -735,5 +752,12 @@ public class DataBufferUtilsTests extends AbstractDataBufferAllocatingTestCase { } + private static class ZeroDemandSubscriber extends BaseSubscriber { + + @Override + protected void hookOnSubscribe(Subscription subscription) { + // Just subscribe without requesting + } + } } diff --git a/spring-core/src/test/java/org/springframework/core/io/buffer/LeakAwareDataBufferFactory.java b/spring-core/src/test/java/org/springframework/core/io/buffer/LeakAwareDataBufferFactory.java index 73337191ad..e270533af8 100644 --- a/spring-core/src/test/java/org/springframework/core/io/buffer/LeakAwareDataBufferFactory.java +++ b/spring-core/src/test/java/org/springframework/core/io/buffer/LeakAwareDataBufferFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -17,9 +17,14 @@ package org.springframework.core.io.buffer; import java.nio.ByteBuffer; +import java.time.Duration; +import java.time.Instant; import java.util.ArrayList; import java.util.List; +import java.util.stream.Collectors; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.jetbrains.annotations.NotNull; import org.junit.After; @@ -37,6 +42,9 @@ import org.springframework.util.Assert; */ public class LeakAwareDataBufferFactory implements DataBufferFactory { + private static final Log logger = LogFactory.getLog(LeakAwareDataBufferFactory.class); + + private final DataBufferFactory delegate; private final List created = new ArrayList<>(); @@ -65,13 +73,27 @@ public class LeakAwareDataBufferFactory implements DataBufferFactory { * method. */ public void checkForLeaks() { - this.created.stream() - .filter(LeakAwareDataBuffer::isAllocated) - .findFirst() - .map(LeakAwareDataBuffer::leakError) - .ifPresent(leakError -> { - throw leakError; - }); + Instant start = Instant.now(); + while (true) { + if (this.created.stream().noneMatch(LeakAwareDataBuffer::isAllocated)) { + return; + } + if (Instant.now().isBefore(start.plus(Duration.ofSeconds(5)))) { + try { + Thread.sleep(50); + } + catch (InterruptedException ex) { + // ignore + } + } + List errors = this.created.stream() + .filter(LeakAwareDataBuffer::isAllocated) + .map(LeakAwareDataBuffer::leakError) + .collect(Collectors.toList()); + + errors.forEach(it -> logger.error("Leaked error: ", it)); + throw new AssertionError(errors.size() + " buffer leaks detected (see logs above)"); + } } @Override diff --git a/spring-web/src/main/java/org/springframework/http/codec/EncoderHttpMessageWriter.java b/spring-web/src/main/java/org/springframework/http/codec/EncoderHttpMessageWriter.java index bc9cab2190..d1067e29ae 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/EncoderHttpMessageWriter.java +++ b/spring-web/src/main/java/org/springframework/http/codec/EncoderHttpMessageWriter.java @@ -29,6 +29,7 @@ import org.springframework.core.codec.AbstractEncoder; import org.springframework.core.codec.Encoder; import org.springframework.core.codec.Hints; import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.PooledDataBuffer; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpLogging; import org.springframework.http.MediaType; @@ -124,7 +125,8 @@ public class EncoderHttpMessageWriter implements HttpMessageWriter { })) .flatMap(buffer -> { headers.setContentLength(buffer.readableByteCount()); - return message.writeWith(Mono.just(buffer)); + return message.writeWith(Mono.fromCallable(() -> buffer) + .doOnDiscard(PooledDataBuffer.class, PooledDataBuffer::release)); }); } diff --git a/spring-web/src/main/java/org/springframework/http/codec/FormHttpMessageWriter.java b/spring-web/src/main/java/org/springframework/http/codec/FormHttpMessageWriter.java index eb5ec682b1..c982dab0b7 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/FormHttpMessageWriter.java +++ b/spring-web/src/main/java/org/springframework/http/codec/FormHttpMessageWriter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -134,7 +134,7 @@ public class FormHttpMessageWriter extends LoggingCodecSupport logFormData(form, hints); String value = serializeForm(form, charset); ByteBuffer byteBuffer = charset.encode(value); - DataBuffer buffer = message.bufferFactory().wrap(byteBuffer); + DataBuffer buffer = message.bufferFactory().wrap(byteBuffer); // wrapping only, no allocation message.getHeaders().setContentLength(byteBuffer.remaining()); return message.writeWith(Mono.just(buffer)); }); diff --git a/spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageWriter.java b/spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageWriter.java index 9470cfb69c..2a15257393 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageWriter.java +++ b/spring-web/src/main/java/org/springframework/http/codec/ServerSentEventHttpMessageWriter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -143,30 +143,32 @@ public class ServerSentEventHttpMessageWriter implements HttpMessageWriter flux = Flux.concat( + encodeText(sb, mediaType, factory), encodeData(data, valueType, mediaType, factory, hints), - encodeText("\n", mediaType, factory)) - .doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release); + encodeText("\n", mediaType, factory)); + + return flux.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release); }); } - private void writeField(String fieldName, Object fieldValue, StringBuilder stringBuilder) { - stringBuilder.append(fieldName); - stringBuilder.append(':'); - stringBuilder.append(fieldValue.toString()); - stringBuilder.append("\n"); + private void writeField(String fieldName, Object fieldValue, StringBuilder sb) { + sb.append(fieldName); + sb.append(':'); + sb.append(fieldValue.toString()); + sb.append("\n"); } @SuppressWarnings("unchecked") - private Flux encodeData(@Nullable T data, ResolvableType valueType, + private Flux encodeData(@Nullable T dataValue, ResolvableType valueType, MediaType mediaType, DataBufferFactory factory, Map hints) { - if (data == null) { + if (dataValue == null) { return Flux.empty(); } - if (data instanceof String) { - String text = (String) data; + if (dataValue instanceof String) { + String text = (String) dataValue; return Flux.from(encodeText(StringUtils.replace(text, "\n", "\ndata:") + "\n", mediaType, factory)); } @@ -175,15 +177,14 @@ public class ServerSentEventHttpMessageWriter implements HttpMessageWriter) this.encoder) - .encode(Mono.just(data), factory, valueType, mediaType, hints) + .encode(Mono.just(dataValue), factory, valueType, mediaType, hints) .concatWith(encodeText("\n", mediaType, factory)); } private Mono encodeText(CharSequence text, MediaType mediaType, DataBufferFactory bufferFactory) { Assert.notNull(mediaType.getCharset(), "Expected MediaType with charset"); byte[] bytes = text.toString().getBytes(mediaType.getCharset()); - return Mono.defer(() -> - Mono.just(bufferFactory.allocateBuffer(bytes.length).write(bytes))); + return Mono.fromCallable(() -> bufferFactory.wrap(bytes)); // wrapping, not allocating } @Override diff --git a/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java b/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java index e237cc031d..652ecb5d28 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Encoder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -126,12 +126,10 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple .filter(mediaType -> mediaType.isCompatibleWith(mimeType)) .findFirst() .map(mediaType -> { - byte[] separator = - STREAM_SEPARATORS.getOrDefault(mediaType, NEWLINE_SEPARATOR); + byte[] separator = STREAM_SEPARATORS.getOrDefault(mediaType, NEWLINE_SEPARATOR); return Flux.from(inputStream).map(value -> { - DataBuffer buffer = - encodeValue(value, mimeType, bufferFactory, elementType, hints, - encoding); + DataBuffer buffer = encodeValue( + value, mimeType, bufferFactory, elementType, hints, encoding); if (separator != null) { buffer.write(separator); } @@ -139,11 +137,9 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple }); }) .orElseGet(() -> { - ResolvableType listType = - ResolvableType.forClassWithGenerics(List.class, elementType); + ResolvableType listType = ResolvableType.forClassWithGenerics(List.class, elementType); return Flux.from(inputStream).collectList().map(list -> - encodeValue(list, mimeType, bufferFactory, listType, hints, - encoding)).flux(); + encodeValue(list, mimeType, bufferFactory, listType, hints, encoding)).flux(); }); } } @@ -174,8 +170,7 @@ public abstract class AbstractJackson2Encoder extends Jackson2CodecSupport imple OutputStream outputStream = buffer.asOutputStream(); try { - JsonGenerator generator = - getObjectMapper().getFactory().createGenerator(outputStream, encoding); + JsonGenerator generator = getObjectMapper().getFactory().createGenerator(outputStream, encoding); writer.writeValue(generator, value); release = false; } diff --git a/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartHttpMessageWriter.java b/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartHttpMessageWriter.java index 39f02ac81c..1118b808b5 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartHttpMessageWriter.java +++ b/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartHttpMessageWriter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -40,7 +40,7 @@ import org.springframework.core.codec.Hints; 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.DefaultDataBufferFactory; +import org.springframework.core.io.buffer.PooledDataBuffer; import org.springframework.core.log.LogFormatUtils; import org.springframework.http.HttpEntity; import org.springframework.http.HttpHeaders; @@ -99,8 +99,6 @@ public class MultipartHttpMessageWriter extends LoggingCodecSupport private final List supportedMediaTypes; - private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory(); - /** * Constructor with a default list of part writers (String and Resource). @@ -187,17 +185,17 @@ public class MultipartHttpMessageWriter extends LoggingCodecSupport ResolvableType elementType, @Nullable MediaType mediaType, ReactiveHttpOutputMessage outputMessage, Map hints) { - return Mono.from(inputStream).flatMap(map -> { - if (this.formWriter == null || isMultipart(map, mediaType)) { - return writeMultipart(map, outputMessage, hints); - } - else { - @SuppressWarnings("unchecked") - MultiValueMap formData = (MultiValueMap) map; - return this.formWriter.write(Mono.just(formData), elementType, mediaType, outputMessage, hints); - } - - }); + return Mono.from(inputStream) + .flatMap(map -> { + if (this.formWriter == null || isMultipart(map, mediaType)) { + return writeMultipart(map, outputMessage, hints); + } + else { + @SuppressWarnings("unchecked") + Mono> input = Mono.just((MultiValueMap) map); + return this.formWriter.write(input, elementType, mediaType, outputMessage, hints); + } + }); } private boolean isMultipart(MultiValueMap map, @Nullable MediaType contentType) { @@ -230,9 +228,12 @@ public class MultipartHttpMessageWriter extends LoggingCodecSupport LogFormatUtils.formatValue(map, !traceOn) : "parts " + map.keySet() + " (content masked)")); + DataBufferFactory bufferFactory = outputMessage.bufferFactory(); + Flux body = Flux.fromIterable(map.entrySet()) - .concatMap(entry -> encodePartValues(boundary, entry.getKey(), entry.getValue())) - .concatWith(Mono.just(generateLastLine(boundary))); + .concatMap(entry -> encodePartValues(boundary, entry.getKey(), entry.getValue(), bufferFactory)) + .concatWith(generateLastLine(boundary, bufferFactory)) + .doOnDiscard(PooledDataBuffer.class, PooledDataBuffer::release); return outputMessage.writeWith(body); } @@ -245,14 +246,16 @@ public class MultipartHttpMessageWriter extends LoggingCodecSupport return MimeTypeUtils.generateMultipartBoundary(); } - private Flux encodePartValues(byte[] boundary, String name, List values) { + private Flux encodePartValues( + byte[] boundary, String name, List values, DataBufferFactory bufferFactory) { + return Flux.concat(values.stream().map(v -> - encodePart(boundary, name, v)).collect(Collectors.toList())); + encodePart(boundary, name, v, bufferFactory)).collect(Collectors.toList())); } @SuppressWarnings("unchecked") - private Flux encodePart(byte[] boundary, String name, T value) { - MultipartHttpOutputMessage outputMessage = new MultipartHttpOutputMessage(this.bufferFactory, getCharset()); + private Flux encodePart(byte[] boundary, String name, T value, DataBufferFactory bufferFactory) { + MultipartHttpOutputMessage outputMessage = new MultipartHttpOutputMessage(bufferFactory, getCharset()); HttpHeaders outputHeaders = outputMessage.getHeaders(); T body; @@ -314,37 +317,46 @@ public class MultipartHttpMessageWriter extends LoggingCodecSupport Flux partContent = partContentReady.thenMany(Flux.defer(outputMessage::getBody)); - return Flux.concat(Mono.just(generateBoundaryLine(boundary)), partContent, Mono.just(generateNewLine())); + return Flux.concat( + generateBoundaryLine(boundary, bufferFactory), + partContent, + generateNewLine(bufferFactory)); } - private DataBuffer generateBoundaryLine(byte[] boundary) { - DataBuffer buffer = this.bufferFactory.allocateBuffer(boundary.length + 4); - buffer.write((byte)'-'); - buffer.write((byte)'-'); - buffer.write(boundary); - buffer.write((byte)'\r'); - buffer.write((byte)'\n'); - return buffer; + private Mono generateBoundaryLine(byte[] boundary, DataBufferFactory bufferFactory) { + return Mono.fromCallable(() -> { + DataBuffer buffer = bufferFactory.allocateBuffer(boundary.length + 4); + buffer.write((byte)'-'); + buffer.write((byte)'-'); + buffer.write(boundary); + buffer.write((byte)'\r'); + buffer.write((byte)'\n'); + return buffer; + }); } - private DataBuffer generateNewLine() { - DataBuffer buffer = this.bufferFactory.allocateBuffer(2); - buffer.write((byte)'\r'); - buffer.write((byte)'\n'); - return buffer; + private Mono generateNewLine(DataBufferFactory bufferFactory) { + return Mono.fromCallable(() -> { + DataBuffer buffer = bufferFactory.allocateBuffer(2); + buffer.write((byte)'\r'); + buffer.write((byte)'\n'); + return buffer; + }); } - private DataBuffer generateLastLine(byte[] boundary) { - DataBuffer buffer = this.bufferFactory.allocateBuffer(boundary.length + 6); - buffer.write((byte)'-'); - buffer.write((byte)'-'); - buffer.write(boundary); - buffer.write((byte)'-'); - buffer.write((byte)'-'); - buffer.write((byte)'\r'); - buffer.write((byte)'\n'); - return buffer; + private Mono generateLastLine(byte[] boundary, DataBufferFactory bufferFactory) { + return Mono.fromCallable(() -> { + DataBuffer buffer = bufferFactory.allocateBuffer(boundary.length + 6); + buffer.write((byte)'-'); + buffer.write((byte)'-'); + buffer.write(boundary); + buffer.write((byte)'-'); + buffer.write((byte)'-'); + buffer.write((byte)'\r'); + buffer.write((byte)'\n'); + return buffer; + }); } @@ -391,29 +403,31 @@ public class MultipartHttpMessageWriter extends LoggingCodecSupport if (this.body != null) { return Mono.error(new IllegalStateException("Multiple calls to writeWith() not supported")); } - this.body = Flux.just(generateHeaders()).concatWith(body); + this.body = generateHeaders().concatWith(body); // We don't actually want to write (just save the body Flux) return Mono.empty(); } - private DataBuffer generateHeaders() { - DataBuffer buffer = this.bufferFactory.allocateBuffer(); - for (Map.Entry> entry : this.headers.entrySet()) { - byte[] headerName = entry.getKey().getBytes(this.charset); - for (String headerValueString : entry.getValue()) { - byte[] headerValue = headerValueString.getBytes(this.charset); - buffer.write(headerName); - buffer.write((byte)':'); - buffer.write((byte)' '); - buffer.write(headerValue); - buffer.write((byte)'\r'); - buffer.write((byte)'\n'); + private Mono generateHeaders() { + return Mono.fromCallable(() -> { + DataBuffer buffer = this.bufferFactory.allocateBuffer(); + for (Map.Entry> entry : this.headers.entrySet()) { + byte[] headerName = entry.getKey().getBytes(this.charset); + for (String headerValueString : entry.getValue()) { + byte[] headerValue = headerValueString.getBytes(this.charset); + buffer.write(headerName); + buffer.write((byte)':'); + buffer.write((byte)' '); + buffer.write(headerValue); + buffer.write((byte)'\r'); + buffer.write((byte)'\n'); + } } - } - buffer.write((byte)'\r'); - buffer.write((byte)'\n'); - return buffer; + buffer.write((byte)'\r'); + buffer.write((byte)'\n'); + return buffer; + }); } @Override diff --git a/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlEncoder.java b/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlEncoder.java index e7e054e2b9..c112d369ea 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlEncoder.java +++ b/spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlEncoder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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 javax.xml.bind.annotation.XmlRootElement; import javax.xml.bind.annotation.XmlType; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import org.springframework.core.ResolvableType; import org.springframework.core.codec.AbstractSingleValueEncoder; @@ -97,7 +98,7 @@ public class Jaxb2XmlEncoder extends AbstractSingleValueEncoder { } @Override - protected Flux encode(Object value, DataBufferFactory dataBufferFactory, + protected Flux encode(Object value, DataBufferFactory bufferFactory, ResolvableType type, @Nullable MimeType mimeType, @Nullable Map hints) { if (!Hints.isLoggingSuppressed(hints)) { @@ -107,29 +108,30 @@ public class Jaxb2XmlEncoder extends AbstractSingleValueEncoder { }); } - boolean release = true; - DataBuffer buffer = dataBufferFactory.allocateBuffer(1024); - OutputStream outputStream = buffer.asOutputStream(); - Class clazz = ClassUtils.getUserClass(value); - - try { - Marshaller marshaller = initMarshaller(clazz); - marshaller.marshal(value, outputStream); - release = false; - return Flux.just(buffer); - } - catch (MarshalException ex) { - return Flux.error(new EncodingException( - "Could not marshal " + value.getClass() + " to XML", ex)); - } - catch (JAXBException ex) { - return Flux.error(new CodecException("Invalid JAXB configuration", ex)); - } - finally { - if (release) { - DataBufferUtils.release(buffer); + return Flux.defer(() -> { + boolean release = true; + DataBuffer buffer = bufferFactory.allocateBuffer(1024); + OutputStream outputStream = buffer.asOutputStream(); + Class clazz = ClassUtils.getUserClass(value); + try { + Marshaller marshaller = initMarshaller(clazz); + marshaller.marshal(value, outputStream); + release = false; + return Mono.fromCallable(() -> buffer); // Rely on doOnDiscard in base class } - } + catch (MarshalException ex) { + return Flux.error(new EncodingException( + "Could not marshal " + value.getClass() + " to XML", ex)); + } + catch (JAXBException ex) { + return Flux.error(new CodecException("Invalid JAXB configuration", ex)); + } + finally { + if (release) { + DataBufferUtils.release(buffer); + } + } + }); } private Marshaller initMarshaller(Class clazz) throws JAXBException { diff --git a/spring-web/src/test/java/org/springframework/http/codec/CodecDataBufferLeakTests.java b/spring-web/src/test/java/org/springframework/http/codec/CodecDataBufferLeakTests.java new file mode 100644 index 0000000000..484020cb40 --- /dev/null +++ b/spring-web/src/test/java/org/springframework/http/codec/CodecDataBufferLeakTests.java @@ -0,0 +1,186 @@ +/* + * Copyright 2002-2019 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 + * + * https://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, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.http.codec; + +import java.time.Duration; +import java.util.Collections; +import java.util.List; +import java.util.function.Supplier; + +import org.junit.Test; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscription; +import reactor.core.publisher.BaseSubscriber; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.core.ResolvableType; +import org.springframework.core.codec.CharSequenceEncoder; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferFactory; +import org.springframework.core.io.buffer.LeakAwareDataBufferFactory; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.http.ReactiveHttpOutputMessage; +import org.springframework.http.client.MultipartBodyBuilder; +import org.springframework.http.codec.json.Jackson2JsonEncoder; +import org.springframework.http.codec.multipart.MultipartHttpMessageWriter; +import org.springframework.http.codec.xml.Jaxb2XmlEncoder; + +/** + * Test scenarios for data buffer leaks. + * @author Rossen Stoyanchev + * @since 5.2 + */ +public class CodecDataBufferLeakTests { + + private final LeakAwareDataBufferFactory bufferFactory = new LeakAwareDataBufferFactory(); + + + @Test // gh-22107 + public void cancelWithEncoderHttpMessageWriterAndSingleValue() { + CharSequenceEncoder encoder = CharSequenceEncoder.allMimeTypes(); + HttpMessageWriter writer = new EncoderHttpMessageWriter<>(encoder); + CancellingOutputMessage outputMessage = new CancellingOutputMessage(this.bufferFactory); + + writer.write(Mono.just("foo"), ResolvableType.forType(String.class), MediaType.TEXT_PLAIN, + outputMessage, Collections.emptyMap()).block(Duration.ofSeconds(5)); + + this.bufferFactory.checkForLeaks(); + } + + @Test // gh-22107 + public void cancelWithJackson() { + Jackson2JsonEncoder encoder = new Jackson2JsonEncoder(); + + Flux flux = encoder.encode(Flux.just(new Pojo("foofoo", "barbar"), new Pojo("bar", "baz")), + this.bufferFactory, ResolvableType.forClass(Pojo.class), + MediaType.APPLICATION_JSON, Collections.emptyMap()); + + BaseSubscriber subscriber = new ZeroDemandSubscriber(); + flux.subscribe(subscriber); // Assume sync execution (e.g. encoding with Flux.just).. + subscriber.cancel(); + + this.bufferFactory.checkForLeaks(); + } + + @Test // gh-22107 + public void cancelWithJaxb2() { + Jaxb2XmlEncoder encoder = new Jaxb2XmlEncoder(); + + Flux flux = encoder.encode(Mono.just(new Pojo("foo", "bar")), + this.bufferFactory, ResolvableType.forClass(Pojo.class), + MediaType.APPLICATION_XML, Collections.emptyMap()); + + BaseSubscriber subscriber = new ZeroDemandSubscriber(); + flux.subscribe(subscriber); // Assume sync execution (e.g. encoding with Flux.just).. + subscriber.cancel(); + + this.bufferFactory.checkForLeaks(); + } + + @Test // gh-22107 + public void cancelWithMultipartContent() { + MultipartBodyBuilder builder = new MultipartBodyBuilder(); + builder.part("part1", "value1"); + builder.part("part2", "value2"); + + List> writers = ClientCodecConfigurer.create().getWriters(); + MultipartHttpMessageWriter writer = new MultipartHttpMessageWriter(writers); + CancellingOutputMessage outputMessage = new CancellingOutputMessage(this.bufferFactory); + + writer.write(Mono.just(builder.build()), null, MediaType.MULTIPART_FORM_DATA, + outputMessage, Collections.emptyMap()).block(Duration.ofSeconds(5)); + + this.bufferFactory.checkForLeaks(); + } + + @Test // gh-22107 + public void cancelWithSse() { + ServerSentEvent event = ServerSentEvent.builder().data("bar").id("c42").event("foo").build(); + ServerSentEventHttpMessageWriter writer = new ServerSentEventHttpMessageWriter(new Jackson2JsonEncoder()); + CancellingOutputMessage outputMessage = new CancellingOutputMessage(this.bufferFactory); + + writer.write(Mono.just(event), ResolvableType.forClass(ServerSentEvent.class), MediaType.TEXT_EVENT_STREAM, + outputMessage, Collections.emptyMap()).block(Duration.ofSeconds(5)); + + this.bufferFactory.checkForLeaks(); + } + + + + private static class CancellingOutputMessage implements ReactiveHttpOutputMessage { + + private final DataBufferFactory bufferFactory; + + + public CancellingOutputMessage(DataBufferFactory bufferFactory) { + this.bufferFactory = bufferFactory; + } + + + @Override + public DataBufferFactory bufferFactory() { + return this.bufferFactory; + } + + @Override + public void beforeCommit(Supplier> action) { + } + + @Override + public boolean isCommitted() { + return false; + } + + @Override + public Mono writeWith(Publisher body) { + Flux flux = Flux.from(body); + BaseSubscriber subscriber = new ZeroDemandSubscriber(); + flux.subscribe(subscriber); // Assume sync execution (e.g. encoding with Flux.just).. + subscriber.cancel(); + return Mono.empty(); + } + + @Override + public Mono writeAndFlushWith(Publisher> body) { + Flux flux = Flux.from(body).concatMap(Flux::from); + BaseSubscriber subscriber = new ZeroDemandSubscriber(); + flux.subscribe(subscriber); // Assume sync execution (e.g. encoding with Flux.just).. + subscriber.cancel(); + return Mono.empty(); + } + + @Override + public Mono setComplete() { + throw new UnsupportedOperationException(); + } + + @Override + public HttpHeaders getHeaders() { + return new HttpHeaders(); + } + } + + + private static class ZeroDemandSubscriber extends BaseSubscriber { + + @Override + protected void hookOnSubscribe(Subscription subscription) { + // Just subscribe without requesting + } + } +} diff --git a/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartHttpMessageWriterTests.java b/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartHttpMessageWriterTests.java index a9cf78dfef..e82cae764f 100644 --- a/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartHttpMessageWriterTests.java +++ b/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartHttpMessageWriterTests.java @@ -197,8 +197,11 @@ public class MultipartHttpMessageWriterTests extends AbstractLeakCheckingTestCas Mono>> result = Mono.just(bodyBuilder.build()); - Map hints = Collections.emptyMap(); - this.writer.write(result, null, MediaType.MULTIPART_FORM_DATA, this.response, hints).block(); + this.writer.write(result, null, MediaType.MULTIPART_FORM_DATA, this.response, Collections.emptyMap()) + .block(Duration.ofSeconds(5)); + + // Make sure body is consumed to avoid leak reports + this.response.getBodyAsString().block(Duration.ofSeconds(5)); } @Test // SPR-16376 diff --git a/spring-web/src/test/java/org/springframework/mock/http/server/reactive/test/MockServerHttpResponse.java b/spring-web/src/test/java/org/springframework/mock/http/server/reactive/test/MockServerHttpResponse.java index 94f88122b0..76049ee9e6 100644 --- a/spring-web/src/test/java/org/springframework/mock/http/server/reactive/test/MockServerHttpResponse.java +++ b/spring-web/src/test/java/org/springframework/mock/http/server/reactive/test/MockServerHttpResponse.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2019 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. @@ -147,6 +147,7 @@ public class MockServerHttpResponse extends AbstractServerHttpResponse { Assert.notNull(charset, "'charset' must not be null"); byte[] bytes = new byte[buffer.readableByteCount()]; buffer.read(bytes); + DataBufferUtils.release(buffer); return new String(bytes, charset); }