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
This commit is contained in:
@@ -42,16 +42,16 @@ import org.springframework.util.Assert;
|
||||
* Bridges between {@link Flow.Publisher Flow.Publisher<T>} and {@link InputStream}.
|
||||
*
|
||||
* <p>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 <T> the publisher byte buffer type
|
||||
*/
|
||||
final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscriber<T> {
|
||||
final class SubscriberInputStream<T> extends InputStream implements Flow.Subscriber<T> {
|
||||
|
||||
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<T> extends InputStream implements Flow.Subscri
|
||||
private Throwable error;
|
||||
|
||||
|
||||
private InputStreamSubscriber(Function<T, byte[]> mapper, Consumer<T> onDiscardHandler, int prefetch) {
|
||||
private SubscriberInputStream(Function<T, byte[]> mapper, Consumer<T> 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<T> 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 <T> InputStream subscribeTo(Flow.Publisher<T> publisher, Function<T, byte[]> mapper, Consumer<T> onDiscardHandler, int bufferSize) {
|
||||
|
||||
public static <T> InputStream subscribeTo(Flow.Publisher<T> publisher, Function<T, byte[]> mapper, Consumer<T> 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<T> iss = new InputStreamSubscriber<>(mapper, onDiscardHandler, bufferSize);
|
||||
SubscriberInputStream<T> iss = new SubscriberInputStream<>(mapper, onDiscardHandler, demand);
|
||||
publisher.subscribe(iss);
|
||||
return iss;
|
||||
}
|
||||
@@ -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<String> 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];
|
||||
Reference in New Issue
Block a user