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 5dde18cc0f..5409e71489 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 @@ -123,9 +123,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 7e1b43ea20..81c68d847b 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 @@ -133,7 +133,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); } 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; }