From dfaf7a0c6a84aa1d1fa174cfcd120cdc9d80fad7 Mon Sep 17 00:00:00 2001 From: rstoyanchev Date: Fri, 25 Oct 2024 17:39:55 +0100 Subject: [PATCH] Rename InputStreamSubscriber to SubscriberInputStream It is both, but InputStream is what's exposed for public use, in effect an InputStream backed by a Subscriber source. See gh-31677 --- .../core/io/buffer/DataBufferUtils.java | 12 +++++----- ...criber.java => SubscriberInputStream.java} | 12 +++++----- .../core/io/buffer/DataBufferUtilsTests.java | 8 +++---- ...criber.java => SubscriberInputStream.java} | 23 +++++++++---------- ...s.java => SubscriberInputStreamTests.java} | 14 +++++------ 5 files changed, 34 insertions(+), 35 deletions(-) rename spring-core/src/main/java/org/springframework/core/io/buffer/{InputStreamSubscriber.java => SubscriberInputStream.java} (96%) rename spring-web/src/main/java/org/springframework/http/client/{InputStreamSubscriber.java => SubscriberInputStream.java} (92%) rename spring-web/src/test/java/org/springframework/http/client/{InputStreamSubscriberTests.java => SubscriberInputStreamTests.java} (94%) 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 c4d281080c..1fd2c1a8f3 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 @@ -472,16 +472,16 @@ public abstract class DataBufferUtils { * Note: {@link Subscription#request(long)} happens eagerly for the first time upon subscription * and then repeats every time {@code bufferSize - (bufferSize >> 2)} consumed. * @param publisher the source of {@link DataBuffer} which should be represented as an {@link InputStream} - * @param bufferSize the maximum amount of {@link DataBuffer} prefetched in advance and stored inside {@link InputStream} + * @param demand the maximum number of buffers to request from the Publisher and buffer on an ongoing basis * @return an {@link InputStream} instance representing given {@link Publisher} messages */ - public static InputStream subscribeAsInputStream(Publisher publisher, int bufferSize) { + public static InputStream subscriberInputStream(Publisher publisher, int demand) { Assert.notNull(publisher, "Publisher must not be null"); - Assert.isTrue(bufferSize > 0, "Buffer size must be > 0"); + Assert.isTrue(demand > 0, "maxBufferCount must be > 0"); - InputStreamSubscriber inputStreamSubscriber = new InputStreamSubscriber(bufferSize); - publisher.subscribe(inputStreamSubscriber); - return inputStreamSubscriber; + SubscriberInputStream subscriber = new SubscriberInputStream(demand); + publisher.subscribe(subscriber); + return subscriber; } diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/InputStreamSubscriber.java b/spring-core/src/main/java/org/springframework/core/io/buffer/SubscriberInputStream.java similarity index 96% rename from spring-core/src/main/java/org/springframework/core/io/buffer/InputStreamSubscriber.java rename to spring-core/src/main/java/org/springframework/core/io/buffer/SubscriberInputStream.java index 67c24d8cce..a4d9e465f4 100644 --- a/spring-core/src/main/java/org/springframework/core/io/buffer/InputStreamSubscriber.java +++ b/spring-core/src/main/java/org/springframework/core/io/buffer/SubscriberInputStream.java @@ -39,13 +39,13 @@ import org.springframework.util.Assert; * Bridges between {@link Publisher Publisher<DataBuffer>} and {@link InputStream}. * *

Note that this class has a near duplicate in - * {@link org.springframework.http.client.InputStreamSubscriber}. + * {@link org.springframework.http.client.SubscriberInputStream}. * * @author Oleh Dokuka * @author Rossen Stoyanchev * @since 6.2 */ -final class InputStreamSubscriber extends InputStream implements Subscriber { +final class SubscriberInputStream extends InputStream implements Subscriber { private static final Object READY = new Object(); @@ -82,10 +82,10 @@ final class InputStreamSubscriber extends InputStream implements Subscriber> 2)); - this.queue = new ArrayBlockingQueue<>(prefetch); + SubscriberInputStream(int demand) { + this.prefetch = demand; + this.limit = (demand == Integer.MAX_VALUE ? Integer.MAX_VALUE : demand - (demand >> 2)); + this.queue = new ArrayBlockingQueue<>(demand); this.lock = new ReentrantLock(false); } 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 d561ec7cfa..cbea649a74 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 @@ -754,7 +754,7 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests { byte[] chunk = new byte[readChunkSize]; List words = new ArrayList<>(); - try (InputStream in = DataBufferUtils.subscribeAsInputStream(publisher, bufferSize)) { + try (InputStream in = DataBufferUtils.subscriberInputStream(publisher, bufferSize)) { int read; while ((read = in.read(chunk)) > -1) { words.add(new String(chunk, 0, read, StandardCharsets.UTF_8)); @@ -791,7 +791,7 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests { byte[] chunk = new byte[4]; List words = new ArrayList<>(); - try (InputStream in = DataBufferUtils.subscribeAsInputStream(publisher, 1)) { + try (InputStream in = DataBufferUtils.subscriberInputStream(publisher, 1)) { int read; while ((read = in.read(chunk)) > -1) { words.add(new String(chunk, 0, read, StandardCharsets.UTF_8)); @@ -830,7 +830,7 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests { byte[] chunk = new byte[3]; ArrayList words = new ArrayList<>(); - try (InputStream inputStream = DataBufferUtils.subscribeAsInputStream(publisher, 1)) { + try (InputStream inputStream = DataBufferUtils.subscriberInputStream(publisher, 1)) { words.add(new String(chunk,0, inputStream.read(chunk), StandardCharsets.UTF_8)); assertThat(inputStream.read()).isEqualTo(' ' & 0xFF); words.add(new String(chunk,0, inputStream.read(chunk), StandardCharsets.UTF_8)); @@ -874,7 +874,7 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests { byte[] chunk = new byte[3]; ArrayList words = new ArrayList<>(); - try (InputStream in = DataBufferUtils.subscribeAsInputStream(publisher, ThreadLocalRandom.current().nextInt(1, 4))) { + try (InputStream in = DataBufferUtils.subscriberInputStream(publisher, ThreadLocalRandom.current().nextInt(1, 4))) { in.read(chunk); String word = new String(chunk, StandardCharsets.UTF_8); words.add(word); diff --git a/spring-web/src/main/java/org/springframework/http/client/InputStreamSubscriber.java b/spring-web/src/main/java/org/springframework/http/client/SubscriberInputStream.java similarity index 92% rename from spring-web/src/main/java/org/springframework/http/client/InputStreamSubscriber.java rename to spring-web/src/main/java/org/springframework/http/client/SubscriberInputStream.java index 14c05411d8..c114c83584 100644 --- a/spring-web/src/main/java/org/springframework/http/client/InputStreamSubscriber.java +++ b/spring-web/src/main/java/org/springframework/http/client/SubscriberInputStream.java @@ -42,16 +42,16 @@ import org.springframework.util.Assert; * Bridges between {@link Flow.Publisher Flow.Publisher<T>} and {@link InputStream}. * *

Note that this class has a near duplicate in - * {@link org.springframework.core.io.buffer.InputStreamSubscriber}. + * {@link org.springframework.core.io.buffer.SubscriberInputStream}. * * @author Oleh Dokuka * @author Rossen Stoyanchev * @since 6.2 * @param the publisher byte buffer type */ -final class InputStreamSubscriber extends InputStream implements Flow.Subscriber { +final class SubscriberInputStream extends InputStream implements Flow.Subscriber { - private static final Log logger = LogFactory.getLog(InputStreamSubscriber.class); + private static final Log logger = LogFactory.getLog(SubscriberInputStream.class); private static final Object READY = new Object(); @@ -94,12 +94,12 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri private Throwable error; - private InputStreamSubscriber(Function mapper, Consumer onDiscardHandler, int prefetch) { + private SubscriberInputStream(Function mapper, Consumer onDiscardHandler, int demand) { this.mapper = mapper; this.onDiscardHandler = onDiscardHandler; - this.prefetch = prefetch; - this.limit = (prefetch == Integer.MAX_VALUE ? Integer.MAX_VALUE : prefetch - (prefetch >> 2)); - this.queue = new ArrayBlockingQueue<>(prefetch); + this.prefetch = demand; + this.limit = (demand == Integer.MAX_VALUE ? Integer.MAX_VALUE : demand - (demand >> 2)); + this.queue = new ArrayBlockingQueue<>(demand); this.lock = new ReentrantLock(false); } @@ -122,17 +122,16 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri * @param publisher the source of {@link DataBuffer} which should be represented as an {@link InputStream} * @param mapper function to transform <T> element to {@code byte[]}. Note, <T> should be released during the mapping if needed. * @param onDiscardHandler <T> element consumer if returned {@link InputStream} is closed prematurely. - * @param bufferSize the maximum amount of <T> elements prefetched in advance and stored inside {@link InputStream} + * @param demand the maximum number of buffers to request from the Publisher and buffer on an ongoing basis * @return an {@link InputStream} instance representing given {@link Flow.Publisher} messages */ - public static InputStream subscribeTo(Flow.Publisher publisher, Function mapper, Consumer onDiscardHandler, int bufferSize) { - + public static InputStream subscribeTo(Flow.Publisher publisher, Function mapper, Consumer onDiscardHandler, int demand) { Assert.notNull(publisher, "Flow.Publisher must not be null"); Assert.notNull(mapper, "mapper must not be null"); Assert.notNull(onDiscardHandler, "onDiscardHandler must not be null"); - Assert.isTrue(bufferSize > 0, "bufferSize must be greater than 0"); + Assert.isTrue(demand > 0, "demand must be greater than 0"); - InputStreamSubscriber iss = new InputStreamSubscriber<>(mapper, onDiscardHandler, bufferSize); + SubscriberInputStream iss = new SubscriberInputStream<>(mapper, onDiscardHandler, demand); publisher.subscribe(iss); return iss; } diff --git a/spring-web/src/test/java/org/springframework/http/client/InputStreamSubscriberTests.java b/spring-web/src/test/java/org/springframework/http/client/SubscriberInputStreamTests.java similarity index 94% rename from spring-web/src/test/java/org/springframework/http/client/InputStreamSubscriberTests.java rename to spring-web/src/test/java/org/springframework/http/client/SubscriberInputStreamTests.java index 6cb397b08d..634885da00 100644 --- a/spring-web/src/test/java/org/springframework/http/client/InputStreamSubscriberTests.java +++ b/spring-web/src/test/java/org/springframework/http/client/SubscriberInputStreamTests.java @@ -36,12 +36,12 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatIOException; /** - * Unit tests for {@link InputStreamSubscriber}. + * Unit tests for {@link SubscriberInputStream}. * * @author Arjen Poutsma * @author Oleh Dokuka */ -class InputStreamSubscriberTests { +class SubscriberInputStreamTests { private static final byte[] FOO = "foo".getBytes(UTF_8); @@ -98,7 +98,7 @@ class InputStreamSubscriberTests { this.byteMapper, this.executor, null); - try (InputStream is = InputStreamSubscriber.subscribeTo( + try (InputStream is = SubscriberInputStream.subscribeTo( toStringPublisher(publisher), s -> s.getBytes(UTF_8), s -> {}, 1)) { byte[] chunk = new byte[3]; @@ -123,7 +123,7 @@ class InputStreamSubscriberTests { }, this.byteMapper, this.executor, 2); - try (InputStream is = InputStreamSubscriber.subscribeTo( + try (InputStream is = SubscriberInputStream.subscribeTo( toStringPublisher(publisher), s -> s.getBytes(UTF_8), s -> {}, 1)) { StringBuilder stringBuilder = new StringBuilder(); @@ -167,7 +167,7 @@ class InputStreamSubscriberTests { List discarded = new ArrayList<>(); - try (InputStream is = InputStreamSubscriber.subscribeTo( + try (InputStream is = SubscriberInputStream.subscribeTo( toStringPublisher(publisher), s -> s.getBytes(UTF_8), discarded::add, 1)) { byte[] chunk = new byte[3]; @@ -198,7 +198,7 @@ class InputStreamSubscriberTests { }, this.byteMapper, this.executor, null); - try (InputStream is = InputStreamSubscriber.subscribeTo( + try (InputStream is = SubscriberInputStream.subscribeTo( toStringPublisher(publisher), s -> s.getBytes(UTF_8), s -> {}, 1)) { byte[] chunk = new byte[3]; @@ -234,7 +234,7 @@ class InputStreamSubscriberTests { Throwable savedEx = null; StringBuilder sb = new StringBuilder(); - try (InputStream is = InputStreamSubscriber.subscribeTo( + try (InputStream is = SubscriberInputStream.subscribeTo( publisher, s -> { throw new NullPointerException("boom"); }, s -> {}, 1)) { byte[] chunk = new byte[3];