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

InputStreamSubscriber/Tests conform to style

Browse Source 
See gh-31677
This commit is contained in:
rstoyanchev
2024-10-25 17:31:28 +01:00
parent d4b31fd4b2
commit 86a42db002
5 changed files with 438 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java
View File

@@ -470,8 +470,7 @@ public abstract class DataBufferUtils {
* any of the {@link InputStream#read} methods * any of the {@link InputStream#read} methods
* <p> * <p>
* Note: {@link Subscription#request(long)} happens eagerly for the first time upon subscription * Note: {@link Subscription#request(long)} happens eagerly for the first time upon subscription
* and then repeats every time {@code bufferSize - (bufferSize >> 2)} consumed * 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 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 bufferSize the maximum amount of {@link DataBuffer} prefetched in advance and stored inside {@link InputStream}
* @return an {@link InputStream} instance representing given {@link Publisher} messages * @return an {@link InputStream} instance representing given {@link Publisher} messages

189
spring-core/src/main/java/org/springframework/core/io/buffer/InputStreamSubscriber.java
View File

@@ -1,8 +1,23 @@
/*
* Copyright 2002-2024 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.core.io.buffer; package org.springframework.core.io.buffer;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.ConcurrentModificationException; import java.util.ConcurrentModificationException;
import java.util.Objects; import java.util.Objects;
import java.util.Queue; import java.util.Queue;
@@ -27,70 +42,83 @@ import org.springframework.util.Assert;
* {@link org.springframework.http.client.InputStreamSubscriber}. * {@link org.springframework.http.client.InputStreamSubscriber}.
* *
* @author Oleh Dokuka * @author Oleh Dokuka
* @since 6.1 * @author Rossen Stoyanchev
* @since 6.2
*/ */
final class InputStreamSubscriber extends InputStream implements Subscriber<DataBuffer> { final class InputStreamSubscriber extends InputStream implements Subscriber<DataBuffer> {
static final Object READY = new Object(); private static final Object READY = new Object();
static final DataBuffer DONE = DefaultDataBuffer.fromEmptyByteBuffer(DefaultDataBufferFactory.sharedInstance, ByteBuffer.allocate(0));
static final DataBuffer CLOSED = DefaultDataBuffer.fromEmptyByteBuffer(DefaultDataBufferFactory.sharedInstance, ByteBuffer.allocate(0));
final int prefetch; private static final DataBuffer DONE = DefaultDataBufferFactory.sharedInstance.allocateBuffer(0);
final int limit;
final ReentrantLock lock;
final Queue<DataBuffer> queue;
final AtomicReference<Object> parkedThread = new AtomicReference<>(); private static final DataBuffer CLOSED = DefaultDataBufferFactory.sharedInstance.allocateBuffer(0);
final AtomicInteger workAmount = new AtomicInteger();
volatile boolean closed;
int consumed; private final int prefetch;
private final int limit;
private final ReentrantLock lock;
private final Queue<DataBuffer> queue;
private final AtomicReference<Object> parkedThread = new AtomicReference<>();
private final AtomicInteger workAmount = new AtomicInteger();
private volatile boolean closed;
private int consumed;
@Nullable @Nullable
DataBuffer available; private DataBuffer available;
@Nullable @Nullable
Subscription s; private Subscription subscription;
boolean done;
private boolean done;
@Nullable @Nullable
Throwable error; private Throwable error;
InputStreamSubscriber(int prefetch) { InputStreamSubscriber(int prefetch) {
this.prefetch = prefetch; this.prefetch = prefetch;
this.limit = prefetch == Integer.MAX_VALUE ? Integer.MAX_VALUE : prefetch - (prefetch >> 2); this.limit = (prefetch == Integer.MAX_VALUE ? Integer.MAX_VALUE : prefetch - (prefetch >> 2));
this.queue = new ArrayBlockingQueue<>(prefetch); this.queue = new ArrayBlockingQueue<>(prefetch);
this.lock = new ReentrantLock(false); this.lock = new ReentrantLock(false);
} }
@Override @Override
public void onSubscribe(Subscription subscription) { public void onSubscribe(Subscription subscription) {
if (this.s != null) { if (this.subscription != null) {
subscription.cancel(); subscription.cancel();
return; return;
} }
this.s = subscription; this.subscription = subscription;
subscription.request(prefetch == Integer.MAX_VALUE ? Long.MAX_VALUE : prefetch); subscription.request(this.prefetch == Integer.MAX_VALUE ? Long.MAX_VALUE : this.prefetch);
} }
@Override @Override
public void onNext(DataBuffer t) { public void onNext(DataBuffer buffer) {
Assert.notNull(t, "DataBuffer must not be null"); Assert.notNull(buffer, "DataBuffer must not be null");
if (this.done) { if (this.done) {
discard(t); discard(buffer);
return; return;
} }
if (!queue.offer(t)) { if (!this.queue.offer(buffer)) {
discard(t); discard(buffer);
error = new RuntimeException("Buffer overflow"); this.error = new RuntimeException("Buffer overflow");
done = true; this.done = true;
} }
int previousWorkState = addWork(); int previousWorkState = addWork();
if (previousWorkState == Integer.MIN_VALUE) { if (previousWorkState == Integer.MIN_VALUE) {
DataBuffer value = queue.poll(); DataBuffer value = this.queue.poll();
if (value != null) { if (value != null) {
discard(value); discard(value);
} }
@@ -136,51 +164,62 @@ final class InputStreamSubscriber extends InputStream implements Subscriber<Data
return Integer.MIN_VALUE; return Integer.MIN_VALUE;
} }
int nextProduced = produced == Integer.MAX_VALUE ? 1 : produced + 1; int nextProduced = (produced == Integer.MAX_VALUE ? 1 : produced + 1);
if (this.workAmount.weakCompareAndSetRelease(produced, nextProduced)) {
if (workAmount.weakCompareAndSetRelease(produced, nextProduced)) {
return produced; return produced;
} }
} }
} }
private void resume() {
if (this.parkedThread != READY) {
Object old = this.parkedThread.getAndSet(READY);
if (old != READY) {
LockSupport.unpark((Thread)old);
}
}
}
/* InputStream implementation */
@Override @Override
public int read() throws IOException { public int read() throws IOException {
if (!lock.tryLock()) { if (!this.lock.tryLock()) {
if (this.closed) { if (this.closed) {
return -1; return -1;
} }
throw new ConcurrentModificationException("concurrent access is disallowed"); throw new ConcurrentModificationException("Concurrent access is not allowed");
} }
try { try {
DataBuffer bytes = getBytesOrAwait(); DataBuffer next = getNextOrAwait();
if (bytes == DONE) { if (next == DONE) {
this.closed = true; this.closed = true;
cleanAndFinalize(); cleanAndFinalize();
if (this.error == null) { if (this.error == null) {
return -1; return -1;
} }
else { else {
throw Exceptions.propagate(error); throw Exceptions.propagate(this.error);
} }
} else if (bytes == CLOSED) { }
else if (next == CLOSED) {
cleanAndFinalize(); cleanAndFinalize();
return -1; return -1;
} }
return bytes.read() & 0xFF; return next.read() & 0xFF;
} }
catch (Throwable t) { catch (Throwable ex) {
this.closed = true; this.closed = true;
requiredSubscriber().cancel(); requiredSubscriber().cancel();
cleanAndFinalize(); cleanAndFinalize();
throw Exceptions.propagate(t); throw Exceptions.propagate(ex);
} }
finally { finally {
lock.unlock(); this.lock.unlock();
} }
} }
@@ -191,7 +230,7 @@ final class InputStreamSubscriber extends InputStream implements Subscriber<Data
return 0; return 0;
} }
if (!lock.tryLock()) { if (!this.lock.tryLock()) {
if (this.closed) { if (this.closed) {
return -1; return -1;
} }
@@ -200,9 +239,9 @@ final class InputStreamSubscriber extends InputStream implements Subscriber<Data
try { try {
for (int j = 0; j < len;) { for (int j = 0; j < len;) {
DataBuffer bytes = getBytesOrAwait(); DataBuffer next = getNextOrAwait();
if (bytes == DONE) { if (next == DONE) {
cleanAndFinalize(); cleanAndFinalize();
if (this.error == null) { if (this.error == null) {
this.closed = true; this.closed = true;
@@ -211,37 +250,37 @@ final class InputStreamSubscriber extends InputStream implements Subscriber<Data
else { else {
if (j == 0) { if (j == 0) {
this.closed = true; this.closed = true;
throw Exceptions.propagate(error); throw Exceptions.propagate(this.error);
} }
return j; return j;
} }
} else if (bytes == CLOSED) { }
else if (next == CLOSED) {
requiredSubscriber().cancel(); requiredSubscriber().cancel();
cleanAndFinalize(); cleanAndFinalize();
return -1; return -1;
} }
int initialReadPosition = bytes.readPosition(); int initialReadPosition = next.readPosition();
bytes.read(b, off + j, Math.min(len - j, bytes.readableByteCount())); next.read(b, off + j, Math.min(len - j, next.readableByteCount()));
j += bytes.readPosition() - initialReadPosition; j += next.readPosition() - initialReadPosition;
} }
return len; return len;
} }
catch (Throwable t) { catch (Throwable ex) {
this.closed = true; this.closed = true;
requiredSubscriber().cancel(); requiredSubscriber().cancel();
cleanAndFinalize(); cleanAndFinalize();
throw Exceptions.propagate(t); throw Exceptions.propagate(ex);
} }
finally { finally {
lock.unlock(); this.lock.unlock();
} }
} }
DataBuffer getBytesOrAwait() { private DataBuffer getNextOrAwait() {
if (this.available == null || this.available.readableByteCount() == 0) { if (this.available == null || this.available.readableByteCount() == 0) {
discard(this.available); discard(this.available);
this.available = null; this.available = null;
@@ -251,11 +290,11 @@ final class InputStreamSubscriber extends InputStream implements Subscriber<Data
return CLOSED; return CLOSED;
} }
boolean d = this.done; boolean done = this.done;
DataBuffer t = this.queue.poll(); DataBuffer buffer = this.queue.poll();
if (t != null) { if (buffer != null) {
int consumed = ++this.consumed; int consumed = ++this.consumed;
this.available = t; this.available = buffer;
if (consumed == this.limit) { if (consumed == this.limit) {
this.consumed = 0; this.consumed = 0;
requiredSubscriber().request(this.limit); requiredSubscriber().request(this.limit);
@@ -263,11 +302,11 @@ final class InputStreamSubscriber extends InputStream implements Subscriber<Data
break; break;
} }
if (d) { if (done) {
return DONE; return DONE;
} }
actualWorkAmount = workAmount.addAndGet(-actualWorkAmount); actualWorkAmount = this.workAmount.addAndGet(-actualWorkAmount);
if (actualWorkAmount == 0) { if (actualWorkAmount == 0) {
await(); await();
} }
@@ -277,15 +316,14 @@ final class InputStreamSubscriber extends InputStream implements Subscriber<Data
return this.available; return this.available;
} }
void cleanAndFinalize() { private void cleanAndFinalize() {
discard(this.available); discard(this.available);
this.available = null; this.available = null;
for (;;) { for (;;) {
int workAmount = this.workAmount.getPlain(); int workAmount = this.workAmount.getPlain();
DataBuffer value; DataBuffer value;
while ((value = this.queue.poll()) != null) {
while((value = queue.poll()) != null) {
discard(value); discard(value);
} }
@@ -295,10 +333,6 @@ final class InputStreamSubscriber extends InputStream implements Subscriber<Data
} }
} }
void discard(@Nullable DataBuffer value) {
DataBufferUtils.release(value);
}
@Override @Override
public void close() throws IOException { public void close() throws IOException {
if (this.closed) { if (this.closed) {
@@ -324,8 +358,12 @@ final class InputStreamSubscriber extends InputStream implements Subscriber<Data
} }
private Subscription requiredSubscriber() { private Subscription requiredSubscriber() {
Assert.state(this.s != null, "Subscriber must be subscribed to use InputStream"); Assert.state(this.subscription != null, "Subscriber must be subscribed to use InputStream");
return this.s; return this.subscription;
}
private void discard(@Nullable DataBuffer buffer) {
DataBufferUtils.release(buffer);
} }
private void await() { private void await() {
@@ -341,7 +379,7 @@ final class InputStreamSubscriber extends InputStream implements Subscriber<Data
throw new IllegalStateException("Only one (Virtual)Thread can await!"); throw new IllegalStateException("Only one (Virtual)Thread can await!");
} }
if (parkedThread.compareAndSet( null, toUnpark)) { if (this.parkedThread.compareAndSet( null, toUnpark)) {
LockSupport.park(); LockSupport.park();
// we don't just break here because park() can wake up spuriously // we don't just break here because park() can wake up spuriously
// if we got a proper resume, get() == READY and the loop will quit above // if we got a proper resume, get() == READY and the loop will quit above
@@ -351,13 +389,4 @@ final class InputStreamSubscriber extends InputStream implements Subscriber<Data
this.parkedThread.lazySet(null); this.parkedThread.lazySet(null);
} }
private void resume() {
if (this.parkedThread != READY) {
Object old = parkedThread.getAndSet(READY);
if (old != READY) {
LockSupport.unpark((Thread)old);
}
}
}
} }

175
spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java
View File

@@ -691,67 +691,73 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests {
latch.await(); latch.await();
} }
@ParameterizedDataBufferAllocatingTest @ParameterizedDataBufferAllocatingTest
void inputStreamSubscriberChunkSize(DataBufferFactory bufferFactory) { void inputStreamSubscriberChunkSize(DataBufferFactory factory) {
genericInputStreamSubscriberTest(bufferFactory, 3, 3, 64, List.of("foo", "bar", "baz"), List.of("foo", "bar", "baz")); genericInputStreamSubscriberTest(
factory, 3, 3, 64, List.of("foo", "bar", "baz"), List.of("foo", "bar", "baz"));
} }
@ParameterizedDataBufferAllocatingTest @ParameterizedDataBufferAllocatingTest
void inputStreamSubscriberChunkSize2(DataBufferFactory bufferFactory) { void inputStreamSubscriberChunkSize2(DataBufferFactory factory) {
genericInputStreamSubscriberTest(bufferFactory, 3, 3, 1, List.of("foo", "bar", "baz"), List.of("foo", "bar", "baz")); genericInputStreamSubscriberTest(
factory, 3, 3, 1, List.of("foo", "bar", "baz"), List.of("foo", "bar", "baz"));
} }
@ParameterizedDataBufferAllocatingTest @ParameterizedDataBufferAllocatingTest
void inputStreamSubscriberChunkSize3(DataBufferFactory bufferFactory) { void inputStreamSubscriberChunkSize3(DataBufferFactory factory) {
genericInputStreamSubscriberTest(bufferFactory, 3, 12, 1, List.of("foo", "bar", "baz"), List.of("foobarbaz")); genericInputStreamSubscriberTest(factory, 3, 12, 1, List.of("foo", "bar", "baz"), List.of("foobarbaz"));
} }
@ParameterizedDataBufferAllocatingTest @ParameterizedDataBufferAllocatingTest
void inputStreamSubscriberChunkSize4(DataBufferFactory bufferFactory) { void inputStreamSubscriberChunkSize4(DataBufferFactory factory) {
genericInputStreamSubscriberTest(bufferFactory, 3, 1, 1, List.of("foo", "bar", "baz"), List.of("f", "o", "o", "b", "a", "r", "b", "a", "z")); genericInputStreamSubscriberTest(
factory, 3, 1, 1, List.of("foo", "bar", "baz"), List.of("f", "o", "o", "b", "a", "r", "b", "a", "z"));
} }
@ParameterizedDataBufferAllocatingTest @ParameterizedDataBufferAllocatingTest
void inputStreamSubscriberChunkSize5(DataBufferFactory bufferFactory) { void inputStreamSubscriberChunkSize5(DataBufferFactory factory) {
genericInputStreamSubscriberTest(bufferFactory, 3, 2, 1, List.of("foo", "bar", "baz"), List.of("fo", "ob", "ar", "ba", "z")); genericInputStreamSubscriberTest(
factory, 3, 2, 1, List.of("foo", "bar", "baz"), List.of("fo", "ob", "ar", "ba", "z"));
} }
@ParameterizedDataBufferAllocatingTest @ParameterizedDataBufferAllocatingTest
void inputStreamSubscriberChunkSize6(DataBufferFactory bufferFactory) { void inputStreamSubscriberChunkSize6(DataBufferFactory factory) {
genericInputStreamSubscriberTest(bufferFactory, 1, 3, 1, List.of("foo", "bar", "baz"), List.of("foo", "bar", "baz")); genericInputStreamSubscriberTest(
factory, 1, 3, 1, List.of("foo", "bar", "baz"), List.of("foo", "bar", "baz"));
} }
@ParameterizedDataBufferAllocatingTest @ParameterizedDataBufferAllocatingTest
void inputStreamSubscriberChunkSize7(DataBufferFactory bufferFactory) { void inputStreamSubscriberChunkSize7(DataBufferFactory factory) {
genericInputStreamSubscriberTest(bufferFactory, 1, 3, 64, List.of("foo", "bar", "baz"), List.of("foo", "bar", "baz")); genericInputStreamSubscriberTest(
factory, 1, 3, 64, List.of("foo", "bar", "baz"), List.of("foo", "bar", "baz"));
} }
void genericInputStreamSubscriberTest(DataBufferFactory bufferFactory, int writeChunkSize, int readChunkSize, int bufferSize, List<String> input, List<String> expectedOutput) { void genericInputStreamSubscriberTest(
super.bufferFactory = bufferFactory; DataBufferFactory factory, int writeChunkSize, int readChunkSize, int bufferSize,
List<String> input, List<String> expectedOutput) {
Publisher<DataBuffer> publisher = DataBufferUtils.outputStreamPublisher(outputStream -> {
try {
for (String word : input) {
outputStream.write(word.getBytes(StandardCharsets.UTF_8));
}
}
catch (IOException ex) {
fail(ex.getMessage(), ex);
}
}, super.bufferFactory, Executors.newSingleThreadExecutor(), writeChunkSize);
super.bufferFactory = factory;
Publisher<DataBuffer> publisher = DataBufferUtils.outputStreamPublisher(
out -> {
try {
for (String word : input) {
out.write(word.getBytes(StandardCharsets.UTF_8));
}
}
catch (IOException ex) {
fail(ex.getMessage(), ex);
}
},
super.bufferFactory, Executors.newSingleThreadExecutor(), writeChunkSize);
byte[] chunk = new byte[readChunkSize]; byte[] chunk = new byte[readChunkSize];
ArrayList<String> words = new ArrayList<>(); List<String> words = new ArrayList<>();
try (InputStream inputStream = DataBufferUtils.subscribeAsInputStream(publisher, bufferSize)) { try (InputStream in = DataBufferUtils.subscribeAsInputStream(publisher, bufferSize)) {
int read; int read;
while((read = inputStream.read(chunk)) > -1) { while ((read = in.read(chunk)) > -1) {
String word = new String(chunk, 0, read, StandardCharsets.UTF_8); words.add(new String(chunk, 0, read, StandardCharsets.UTF_8));
words.add(word);
} }
} }
catch (IOException e) { catch (IOException e) {
@@ -761,33 +767,34 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests {
} }
@ParameterizedDataBufferAllocatingTest @ParameterizedDataBufferAllocatingTest
void inputStreamSubscriberError(DataBufferFactory bufferFactory) throws InterruptedException { void inputStreamSubscriberError(DataBufferFactory factory) {
super.bufferFactory = bufferFactory; super.bufferFactory = factory;
var input = List.of("foo ", "bar ", "baz"); var input = List.of("foo ", "bar ", "baz");
Publisher<DataBuffer> publisher = DataBufferUtils.outputStreamPublisher(outputStream -> { Publisher<DataBuffer> publisher = DataBufferUtils.outputStreamPublisher(
try { out -> {
for (String word : input) { try {
outputStream.write(word.getBytes(StandardCharsets.UTF_8)); for (String word : input) {
} out.write(word.getBytes(StandardCharsets.UTF_8));
throw new RuntimeException("boom"); }
} throw new RuntimeException("boom");
catch (IOException ex) { }
fail(ex.getMessage(), ex); catch (IOException ex) {
} fail(ex.getMessage(), ex);
}, super.bufferFactory, Executors.newSingleThreadExecutor(), 1); }
},
super.bufferFactory, Executors.newSingleThreadExecutor(), 1);
RuntimeException error = null; RuntimeException error = null;
byte[] chunk = new byte[4]; byte[] chunk = new byte[4];
ArrayList<String> words = new ArrayList<>(); List<String> words = new ArrayList<>();
try (InputStream inputStream = DataBufferUtils.subscribeAsInputStream(publisher, 1)) { try (InputStream in = DataBufferUtils.subscribeAsInputStream(publisher, 1)) {
int read; int read;
while((read = inputStream.read(chunk)) > -1) { while ((read = in.read(chunk)) > -1) {
String word = new String(chunk, 0, read, StandardCharsets.UTF_8); words.add(new String(chunk, 0, read, StandardCharsets.UTF_8));
words.add(word);
} }
} }
catch (IOException e) { catch (IOException e) {
@@ -801,21 +808,23 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests {
} }
@ParameterizedDataBufferAllocatingTest @ParameterizedDataBufferAllocatingTest
void inputStreamSubscriberMixedReadMode(DataBufferFactory bufferFactory) throws InterruptedException { void inputStreamSubscriberMixedReadMode(DataBufferFactory factory) {
super.bufferFactory = bufferFactory; super.bufferFactory = factory;
var input = List.of("foo ", "bar ", "baz"); var input = List.of("foo ", "bar ", "baz");
Publisher<DataBuffer> publisher = DataBufferUtils.outputStreamPublisher(outputStream -> { Publisher<DataBuffer> publisher = DataBufferUtils.outputStreamPublisher(
try { out -> {
for (String word : input) { try {
outputStream.write(word.getBytes(StandardCharsets.UTF_8)); for (String word : input) {
} out.write(word.getBytes(StandardCharsets.UTF_8));
} }
catch (IOException ex) { }
fail(ex.getMessage(), ex); catch (IOException ex) {
} fail(ex.getMessage(), ex);
}, super.bufferFactory, Executors.newSingleThreadExecutor(), 1); }
},
super.bufferFactory, Executors.newSingleThreadExecutor(), 1);
byte[] chunk = new byte[3]; byte[] chunk = new byte[3];
@@ -843,30 +852,34 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests {
var input = List.of("foo", "bar", "baz"); var input = List.of("foo", "bar", "baz");
Publisher<DataBuffer> publisher = DataBufferUtils.outputStreamPublisher(outputStream -> { Publisher<DataBuffer> publisher = DataBufferUtils.outputStreamPublisher(
try { out -> {
assertThatIOException() try {
.isThrownBy(() -> { assertThatIOException()
for (String word : input) { .isThrownBy(() -> {
outputStream.write(word.getBytes(StandardCharsets.UTF_8)); for (String word : input) {
outputStream.flush(); out.write(word.getBytes(StandardCharsets.UTF_8));
} out.flush();
}) }
.withMessage("Subscription has been terminated"); })
} finally { .withMessage("Subscription has been terminated");
latch.countDown(); }
} finally {
}, super.bufferFactory, Executors.newSingleThreadExecutor(), 1); latch.countDown();
}
},
super.bufferFactory, Executors.newSingleThreadExecutor(), 1);
byte[] chunk = new byte[3]; byte[] chunk = new byte[3];
ArrayList<String> words = new ArrayList<>(); ArrayList<String> words = new ArrayList<>();
try (InputStream inputStream = DataBufferUtils.subscribeAsInputStream(publisher, ThreadLocalRandom.current().nextInt(1, 4))) { try (InputStream in = DataBufferUtils.subscribeAsInputStream(publisher, ThreadLocalRandom.current().nextInt(1, 4))) {
inputStream.read(chunk); in.read(chunk);
String word = new String(chunk, StandardCharsets.UTF_8); String word = new String(chunk, StandardCharsets.UTF_8);
words.add(word); words.add(word);
} catch (IOException e) { }
catch (IOException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
assertThat(words).containsExactlyElementsOf(List.of("foo")); assertThat(words).containsExactlyElementsOf(List.of("foo"));

209
spring-web/src/main/java/org/springframework/http/client/InputStreamSubscriber.java
View File

@@ -1,3 +1,19 @@
/*
* Copyright 2002-2024 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.client; package org.springframework.http.client;
import java.io.IOException; import java.io.IOException;
@@ -29,48 +45,65 @@ import org.springframework.util.Assert;
* {@link org.springframework.core.io.buffer.InputStreamSubscriber}. * {@link org.springframework.core.io.buffer.InputStreamSubscriber}.
* *
* @author Oleh Dokuka * @author Oleh Dokuka
* @since 6.1 * @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 InputStreamSubscriber<T> extends InputStream implements Flow.Subscriber<T> {
private static final Log logger = LogFactory.getLog(InputStreamSubscriber.class); private static final Log logger = LogFactory.getLog(InputStreamSubscriber.class);
static final Object READY = new Object(); private static final Object READY = new Object();
static final byte[] DONE = new byte[0];
static final byte[] CLOSED = new byte[0];
final int prefetch; private static final byte[] DONE = new byte[0];
final int limit;
final ReentrantLock lock;
final Queue<T> queue;
final Function<T, byte[]> mapper;
final Consumer<T> onDiscardHandler;
final AtomicReference<Object> parkedThread = new AtomicReference<>(); private static final byte[] CLOSED = new byte[0];
final AtomicInteger workAmount = new AtomicInteger();
private final Function<T, byte[]> mapper;
private final Consumer<T> onDiscardHandler;
private final int prefetch;
private final int limit;
private final ReentrantLock lock;
private final Queue<T> queue;
private final AtomicReference<Object> parkedThread = new AtomicReference<>();
private final AtomicInteger workAmount = new AtomicInteger();
volatile boolean closed; volatile boolean closed;
int consumed;
private int consumed;
@Nullable @Nullable
byte[] available; private byte[] available;
int position;
private int position;
@Nullable @Nullable
Flow.Subscription s; private Flow.Subscription subscription;
boolean done;
private boolean done;
@Nullable @Nullable
Throwable error; private Throwable error;
private InputStreamSubscriber(Function<T, byte[]> mapper, Consumer<T> onDiscardHandler, int prefetch) { private InputStreamSubscriber(Function<T, byte[]> mapper, Consumer<T> onDiscardHandler, int prefetch) {
this.prefetch = prefetch;
this.limit = prefetch == Integer.MAX_VALUE ? Integer.MAX_VALUE : prefetch - (prefetch >> 2);
this.mapper = mapper; this.mapper = mapper;
this.onDiscardHandler = onDiscardHandler; this.onDiscardHandler = onDiscardHandler;
this.prefetch = prefetch;
this.limit = (prefetch == Integer.MAX_VALUE ? Integer.MAX_VALUE : prefetch - (prefetch >> 2));
this.queue = new ArrayBlockingQueue<>(prefetch); this.queue = new ArrayBlockingQueue<>(prefetch);
this.lock = new ReentrantLock(false); this.lock = new ReentrantLock(false);
} }
/** /**
* Subscribes to given {@link Flow.Publisher} and returns subscription * Subscribes to given {@link Flow.Publisher} and returns subscription
* as {@link InputStream} that allows reading all propagated {@link DataBuffer} messages via its imperative API. * as {@link InputStream} that allows reading all propagated {@link DataBuffer} messages via its imperative API.
@@ -85,8 +118,7 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
* any of the {@link InputStream#read} methods * any of the {@link InputStream#read} methods
* <p> * <p>
* Note: {@link Flow.Subscription#request(long)} happens eagerly for the first time upon subscription * Note: {@link Flow.Subscription#request(long)} happens eagerly for the first time upon subscription
* and then repeats every time {@code bufferSize - (bufferSize >> 2)} consumed * 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 publisher the source of {@link DataBuffer} which should be represented as an {@link InputStream}
* @param mapper function to transform &lt;T&gt; element to {@code byte[]}. Note, &lt;T&gt; should be released during the mapping if needed. * @param mapper function to transform &lt;T&gt; element to {@code byte[]}. Note, &lt;T&gt; should be released during the mapping if needed.
* @param onDiscardHandler &lt;T&gt; element consumer if returned {@link InputStream} is closed prematurely. * @param onDiscardHandler &lt;T&gt; element consumer if returned {@link InputStream} is closed prematurely.
@@ -107,33 +139,33 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
@Override @Override
public void onSubscribe(Flow.Subscription subscription) { public void onSubscribe(Flow.Subscription subscription) {
if (this.s != null) { if (this.subscription != null) {
subscription.cancel(); subscription.cancel();
return; return;
} }
this.s = subscription; this.subscription = subscription;
subscription.request(prefetch == Integer.MAX_VALUE ? Long.MAX_VALUE : prefetch); subscription.request(this.prefetch == Integer.MAX_VALUE ? Long.MAX_VALUE : this.prefetch);
} }
@Override @Override
public void onNext(T t) { public void onNext(T buffer) {
Assert.notNull(t, "T value must not be null"); Assert.notNull(buffer, "Buffer must not be null");
if (this.done) { if (this.done) {
discard(t); discard(buffer);
return; return;
} }
if (!queue.offer(t)) { if (!this.queue.offer(buffer)) {
discard(t); discard(buffer);
error = new RuntimeException("Buffer overflow"); this.error = new RuntimeException("Buffer overflow");
done = true; this.done = true;
} }
int previousWorkState = addWork(); int previousWorkState = addWork();
if (previousWorkState == Integer.MIN_VALUE) { if (previousWorkState == Integer.MIN_VALUE) {
T value = queue.poll(); T value = this.queue.poll();
if (value != null) { if (value != null) {
discard(value); discard(value);
} }
@@ -179,51 +211,62 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
return Integer.MIN_VALUE; return Integer.MIN_VALUE;
} }
int nextProduced = produced == Integer.MAX_VALUE ? 1 : produced + 1; int nextProduced = (produced == Integer.MAX_VALUE ? 1 : produced + 1);
if (this.workAmount.weakCompareAndSetRelease(produced, nextProduced)) {
if (workAmount.weakCompareAndSetRelease(produced, nextProduced)) {
return produced; return produced;
} }
} }
} }
private void resume() {
if (this.parkedThread != READY) {
Object old = this.parkedThread.getAndSet(READY);
if (old != READY) {
LockSupport.unpark((Thread)old);
}
}
}
/* InputStream implementation */
@Override @Override
public int read() throws IOException { public int read() throws IOException {
if (!lock.tryLock()) { if (!this.lock.tryLock()) {
if (this.closed) { if (this.closed) {
return -1; return -1;
} }
throw new ConcurrentModificationException("concurrent access is disallowed"); throw new ConcurrentModificationException("Concurrent access is not allowed");
} }
try { try {
byte[] bytes = getBytesOrAwait(); byte[] next = getNextOrAwait();
if (bytes == DONE) { if (next == DONE) {
this.closed = true; this.closed = true;
cleanAndFinalize(); cleanAndFinalize();
if (this.error == null) { if (this.error == null) {
return -1; return -1;
} }
else { else {
throw Exceptions.propagate(error); throw Exceptions.propagate(this.error);
} }
} else if (bytes == CLOSED) { }
else if (next == CLOSED) {
cleanAndFinalize(); cleanAndFinalize();
return -1; return -1;
} }
return bytes[this.position++] & 0xFF; return next[this.position++] & 0xFF;
} }
catch (Throwable t) { catch (Throwable ex) {
this.closed = true; this.closed = true;
requiredSubscriber().cancel(); requiredSubscriber().cancel();
cleanAndFinalize(); cleanAndFinalize();
throw Exceptions.propagate(t); throw Exceptions.propagate(ex);
} }
finally { finally {
lock.unlock(); this.lock.unlock();
} }
} }
@@ -234,7 +277,7 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
return 0; return 0;
} }
if (!lock.tryLock()) { if (!this.lock.tryLock()) {
if (this.closed) { if (this.closed) {
return -1; return -1;
} }
@@ -243,9 +286,9 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
try { try {
for (int j = 0; j < len;) { for (int j = 0; j < len;) {
byte[] bytes = getBytesOrAwait(); byte[] next = getNextOrAwait();
if (bytes == DONE) { if (next == DONE) {
cleanAndFinalize(); cleanAndFinalize();
if (this.error == null) { if (this.error == null) {
this.closed = true; this.closed = true;
@@ -254,37 +297,38 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
else { else {
if (j == 0) { if (j == 0) {
this.closed = true; this.closed = true;
throw Exceptions.propagate(error); throw Exceptions.propagate(this.error);
} }
return j; return j;
} }
} else if (bytes == CLOSED) { }
else if (next == CLOSED) {
requiredSubscriber().cancel(); requiredSubscriber().cancel();
cleanAndFinalize(); cleanAndFinalize();
return -1; return -1;
} }
int i = this.position; int i = this.position;
for (; i < bytes.length && j < len; i++, j++) { for (; i < next.length && j < len; i++, j++) {
b[off + j] = bytes[i]; b[off + j] = next[i];
} }
this.position = i; this.position = i;
} }
return len; return len;
} }
catch (Throwable t) { catch (Throwable ex) {
this.closed = true; this.closed = true;
requiredSubscriber().cancel(); requiredSubscriber().cancel();
cleanAndFinalize(); cleanAndFinalize();
throw Exceptions.propagate(t); throw Exceptions.propagate(ex);
} }
finally { finally {
lock.unlock(); this.lock.unlock();
} }
} }
byte[] getBytesOrAwait() { byte[] getNextOrAwait() {
if (this.available == null || this.available.length - this.position == 0) { if (this.available == null || this.available.length - this.position == 0) {
this.available = null; this.available = null;
@@ -294,12 +338,12 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
return CLOSED; return CLOSED;
} }
boolean d = this.done; boolean done = this.done;
T t = this.queue.poll(); T buffer = this.queue.poll();
if (t != null) { if (buffer != null) {
int consumed = ++this.consumed; int consumed = ++this.consumed;
this.position = 0; this.position = 0;
this.available = Objects.requireNonNull(this.mapper.apply(t)); this.available = Objects.requireNonNull(this.mapper.apply(buffer));
if (consumed == this.limit) { if (consumed == this.limit) {
this.consumed = 0; this.consumed = 0;
requiredSubscriber().request(this.limit); requiredSubscriber().request(this.limit);
@@ -307,11 +351,11 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
break; break;
} }
if (d) { if (done) {
return DONE; return DONE;
} }
actualWorkAmount = workAmount.addAndGet(-actualWorkAmount); actualWorkAmount = this.workAmount.addAndGet(-actualWorkAmount);
if (actualWorkAmount == 0) { if (actualWorkAmount == 0) {
await(); await();
} }
@@ -327,8 +371,7 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
for (;;) { for (;;) {
int workAmount = this.workAmount.getPlain(); int workAmount = this.workAmount.getPlain();
T value; T value;
while ((value = this.queue.poll()) != null) {
while((value = queue.poll()) != null) {
discard(value); discard(value);
} }
@@ -338,16 +381,6 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
} }
} }
void discard(T value) {
try {
this.onDiscardHandler.accept(value);
} catch (Throwable t) {
if (logger.isDebugEnabled()) {
logger.debug("Failed to release " + value.getClass().getSimpleName() + ": " + value, t);
}
}
}
@Override @Override
public void close() throws IOException { public void close() throws IOException {
if (this.closed) { if (this.closed) {
@@ -373,8 +406,19 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
} }
private Flow.Subscription requiredSubscriber() { private Flow.Subscription requiredSubscriber() {
Assert.state(this.s != null, "Subscriber must be subscribed to use InputStream"); Assert.state(this.subscription != null, "Subscriber must be subscribed to use InputStream");
return this.s; return this.subscription;
}
void discard(T buffer) {
try {
this.onDiscardHandler.accept(buffer);
}
catch (Throwable ex) {
if (logger.isDebugEnabled()) {
logger.debug("Failed to release " + buffer.getClass().getSimpleName() + ": " + buffer, ex);
}
}
} }
private void await() { private void await() {
@@ -390,7 +434,7 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
throw new IllegalStateException("Only one (Virtual)Thread can await!"); throw new IllegalStateException("Only one (Virtual)Thread can await!");
} }
if (parkedThread.compareAndSet( null, toUnpark)) { if (this.parkedThread.compareAndSet( null, toUnpark)) {
LockSupport.park(); LockSupport.park();
// we don't just break here because park() can wake up spuriously // we don't just break here because park() can wake up spuriously
// if we got a proper resume, get() == READY and the loop will quit above // if we got a proper resume, get() == READY and the loop will quit above
@@ -400,13 +444,4 @@ final class InputStreamSubscriber<T> extends InputStream implements Flow.Subscri
this.parkedThread.lazySet(null); this.parkedThread.lazySet(null);
} }
private void resume() {
if (this.parkedThread != READY) {
Object old = parkedThread.getAndSet(READY);
if (old != READY) {
LockSupport.unpark((Thread)old);
}
}
}
} }

214
spring-web/src/test/java/org/springframework/http/client/InputStreamSubscriberTests.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2023 the original author or authors. * Copyright 2002-2024 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -19,7 +19,6 @@ package org.springframework.http.client;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.io.OutputStreamWriter; import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
@@ -32,29 +31,33 @@ import org.reactivestreams.FlowAdapters;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.test.StepVerifier; import reactor.test.StepVerifier;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIOException; import static org.assertj.core.api.Assertions.assertThatIOException;
/** /**
* Unit tests for {@link InputStreamSubscriber}.
*
* @author Arjen Poutsma * @author Arjen Poutsma
* @author Oleh Dokuka * @author Oleh Dokuka
*/ */
class InputStreamSubscriberTests { class InputStreamSubscriberTests {
private static final byte[] FOO = "foo".getBytes(StandardCharsets.UTF_8); private static final byte[] FOO = "foo".getBytes(UTF_8);
private static final byte[] BAR = "bar".getBytes(StandardCharsets.UTF_8); private static final byte[] BAR = "bar".getBytes(UTF_8);
private static final byte[] BAZ = "baz".getBytes(StandardCharsets.UTF_8); private static final byte[] BAZ = "baz".getBytes(UTF_8);
private final Executor executor = Executors.newSingleThreadExecutor(); private final Executor executor = Executors.newSingleThreadExecutor();
private final OutputStreamPublisher.ByteMapper<byte[]> byteMapper = private final OutputStreamPublisher.ByteMapper<byte[]> byteMapper =
new OutputStreamPublisher.ByteMapper<>() { new OutputStreamPublisher.ByteMapper<>() {
@Override @Override
public byte[] map(int b) { public byte[] map(int b) {
return new byte[]{(byte) b}; return new byte[] {(byte) b};
} }
@Override @Override
@@ -68,31 +71,36 @@ class InputStreamSubscriberTests {
@Test @Test
void basic() { void basic() {
Flow.Publisher<byte[]> flowPublisher = new OutputStreamPublisher<>(outputStream -> { Flow.Publisher<byte[]> publisher = new OutputStreamPublisher<>(
outputStream.write(FOO); out -> {
outputStream.write(BAR); out.write(FOO);
outputStream.write(BAZ); out.write(BAR);
}, this.byteMapper, this.executor, null); out.write(BAZ);
Flux<String> flux = toString(flowPublisher); },
this.byteMapper, this.executor, null);
StepVerifier.create(flux) StepVerifier.create(toStringFlux(publisher))
.assertNext(s -> assertThat(s).isEqualTo("foobarbaz")) .assertNext(s -> assertThat(s).isEqualTo("foobarbaz"))
.verifyComplete(); .verifyComplete();
} }
@Test @Test
void flush() { void flush() throws IOException {
Flow.Publisher<byte[]> flowPublisher = new OutputStreamPublisher<>(outputStream -> { Flow.Publisher<byte[]> publisher = new OutputStreamPublisher<>(
outputStream.write(FOO); out -> {
outputStream.flush(); out.write(FOO);
outputStream.write(BAR); out.flush();
outputStream.flush(); out.write(BAR);
outputStream.write(BAZ); out.flush();
outputStream.flush(); out.write(BAZ);
}, this.byteMapper, this.executor, null); out.flush();
Flux<String> flux = toString(flowPublisher); },
this.byteMapper, this.executor, null);
try (InputStream is = InputStreamSubscriber.subscribeTo(
toStringPublisher(publisher), s -> s.getBytes(UTF_8), s -> {}, 1)) {
try (InputStream is = InputStreamSubscriber.subscribeTo(FlowAdapters.toFlowPublisher(flux), (s) -> s.getBytes(StandardCharsets.UTF_8), (ignore) -> {}, 1)) {
byte[] chunk = new byte[3]; byte[] chunk = new byte[3];
assertThat(is.read(chunk)).isEqualTo(3); assertThat(is.read(chunk)).isEqualTo(3);
@@ -103,38 +111,35 @@ class InputStreamSubscriberTests {
assertThat(chunk).containsExactly(BAZ); assertThat(chunk).containsExactly(BAZ);
assertThat(is.read(chunk)).isEqualTo(-1); assertThat(is.read(chunk)).isEqualTo(-1);
} }
catch (IOException e) { }
throw new RuntimeException(e);
}
}
@Test @Test
void chunkSize() { void chunkSize() {
Flow.Publisher<byte[]> flowPublisher = new OutputStreamPublisher<>(outputStream -> { Flow.Publisher<byte[]> publisher = new OutputStreamPublisher<>(
outputStream.write(FOO); out -> {
outputStream.write(BAR); out.write(FOO);
outputStream.write(BAZ); out.write(BAR);
}, this.byteMapper, this.executor, 2); out.write(BAZ);
Flux<String> flux = toString(flowPublisher); },
this.byteMapper, this.executor, 2);
try (InputStream is = InputStreamSubscriber.subscribeTo(
toStringPublisher(publisher), s -> s.getBytes(UTF_8), s -> {}, 1)) {
try (InputStream is = InputStreamSubscriber.subscribeTo(FlowAdapters.toFlowPublisher(flux), (s) -> s.getBytes(StandardCharsets.UTF_8), (ignore) -> {}, 1)) {
StringBuilder stringBuilder = new StringBuilder(); StringBuilder stringBuilder = new StringBuilder();
byte[] chunk = new byte[3]; byte[] chunk = new byte[3];
stringBuilder.append(new String(new byte[]{(byte)is.read()}, UTF_8));
assertThat(is.read(chunk)).isEqualTo(3);
stringBuilder stringBuilder.append(new String(chunk, UTF_8));
.append(new String(new byte[]{(byte)is.read()}, StandardCharsets.UTF_8));
assertThat(is.read(chunk)).isEqualTo(3); assertThat(is.read(chunk)).isEqualTo(3);
stringBuilder
.append(new String(chunk, StandardCharsets.UTF_8)); stringBuilder.append(new String(chunk, UTF_8));
assertThat(is.read(chunk)).isEqualTo(3);
stringBuilder
.append(new String(chunk, StandardCharsets.UTF_8));
assertThat(is.read(chunk)).isEqualTo(2); assertThat(is.read(chunk)).isEqualTo(2);
stringBuilder
.append(new String(chunk,0, 2, StandardCharsets.UTF_8));
assertThat(is.read()).isEqualTo(-1);
stringBuilder.append(new String(chunk,0, 2, UTF_8));
assertThat(is.read()).isEqualTo(-1);
assertThat(stringBuilder.toString()).isEqualTo("foobarbaz"); assertThat(stringBuilder.toString()).isEqualTo("foobarbaz");
} }
catch (IOException e) { catch (IOException e) {
@@ -146,24 +151,25 @@ class InputStreamSubscriberTests {
void cancel() throws InterruptedException { void cancel() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(1);
Flow.Publisher<byte[]> flowPublisher = new OutputStreamPublisher<>(outputStream -> { Flow.Publisher<byte[]> publisher = new OutputStreamPublisher<>(
assertThatIOException() out -> {
.isThrownBy(() -> { assertThatIOException().isThrownBy(() -> {
outputStream.write(FOO); out.write(FOO);
outputStream.flush(); out.flush();
outputStream.write(BAR); out.write(BAR);
outputStream.flush(); out.flush();
outputStream.write(BAZ); out.write(BAZ);
outputStream.flush(); out.flush();
}) }).withMessage("Subscription has been terminated");
.withMessage("Subscription has been terminated"); latch.countDown();
latch.countDown();
}, this.byteMapper, this.executor, null);
}, this.byteMapper, this.executor, null);
Flux<String> flux = toString(flowPublisher);
List<String> discarded = new ArrayList<>(); List<String> discarded = new ArrayList<>();
try (InputStream is = InputStreamSubscriber.subscribeTo(FlowAdapters.toFlowPublisher(flux), (s) -> s.getBytes(StandardCharsets.UTF_8), discarded::add, 1)) { try (InputStream is = InputStreamSubscriber.subscribeTo(
toStringPublisher(publisher), s -> s.getBytes(UTF_8), discarded::add, 1)) {
byte[] chunk = new byte[3]; byte[] chunk = new byte[3];
assertThat(is.read(chunk)).isEqualTo(3); assertThat(is.read(chunk)).isEqualTo(3);
@@ -182,22 +188,23 @@ class InputStreamSubscriberTests {
void closed() throws InterruptedException { void closed() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(1);
Flow.Publisher<byte[]> flowPublisher = new OutputStreamPublisher<>(outputStream -> { Flow.Publisher<byte[]> publisher = new OutputStreamPublisher<>(
OutputStreamWriter writer = new OutputStreamWriter(outputStream, StandardCharsets.UTF_8); out -> {
writer.write("foo"); OutputStreamWriter writer = new OutputStreamWriter(out, UTF_8);
writer.close(); writer.write("foo");
assertThatIOException().isThrownBy(() -> writer.write("bar")) writer.close();
.withMessage("Stream closed"); assertThatIOException().isThrownBy(() -> writer.write("bar")).withMessage("Stream closed");
latch.countDown(); latch.countDown();
}, this.byteMapper, this.executor, null); },
Flux<String> flux = toString(flowPublisher); this.byteMapper, this.executor, null);
try (InputStream is = InputStreamSubscriber.subscribeTo(
toStringPublisher(publisher), s -> s.getBytes(UTF_8), s -> {}, 1)) {
try (InputStream is = InputStreamSubscriber.subscribeTo(FlowAdapters.toFlowPublisher(flux), (s) -> s.getBytes(StandardCharsets.UTF_8), ig -> {}, 1)) {
byte[] chunk = new byte[3]; byte[] chunk = new byte[3];
assertThat(is.read(chunk)).isEqualTo(3); assertThat(is.read(chunk)).isEqualTo(3);
assertThat(chunk).containsExactly(FOO); assertThat(chunk).containsExactly(FOO);
assertThat(is.read(chunk)).isEqualTo(-1); assertThat(is.read(chunk)).isEqualTo(-1);
} }
catch (IOException e) { catch (IOException e) {
@@ -211,49 +218,52 @@ class InputStreamSubscriberTests {
void mapperThrowsException() throws InterruptedException { void mapperThrowsException() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(1);
Flow.Publisher<byte[]> flowPublisher = new OutputStreamPublisher<>(outputStream -> { Flow.Publisher<byte[]> publisher = new OutputStreamPublisher<>(
outputStream.write(FOO); out -> {
outputStream.flush(); out.write(FOO);
assertThatIOException().isThrownBy(() -> { out.flush();
outputStream.write(BAR); assertThatIOException().isThrownBy(() -> {
outputStream.flush(); out.write(BAR);
}).withMessage("Subscription has been terminated"); out.flush();
latch.countDown(); })
}, this.byteMapper, this.executor, null); .withMessage("Subscription has been terminated");
Throwable ex = null; latch.countDown();
},
this.byteMapper, this.executor, null);
Throwable savedEx = null;
StringBuilder sb = new StringBuilder();
try (InputStream is = InputStreamSubscriber.subscribeTo(
publisher, s -> { throw new NullPointerException("boom"); }, s -> {}, 1)) {
StringBuilder stringBuilder = new StringBuilder();
try (InputStream is = InputStreamSubscriber.subscribeTo(flowPublisher, (s) -> {
throw new NullPointerException("boom");
}, ig -> {}, 1)) {
byte[] chunk = new byte[3]; byte[] chunk = new byte[3];
stringBuilder sb.append(new String(new byte[]{(byte)is.read()}, UTF_8));
.append(new String(new byte[]{(byte)is.read()}, StandardCharsets.UTF_8));
assertThat(is.read(chunk)).isEqualTo(3); assertThat(is.read(chunk)).isEqualTo(3);
stringBuilder sb.append(new String(chunk, UTF_8));
.append(new String(chunk, StandardCharsets.UTF_8));
assertThat(is.read(chunk)).isEqualTo(3); assertThat(is.read(chunk)).isEqualTo(3);
stringBuilder sb.append(new String(chunk, UTF_8));
.append(new String(chunk, StandardCharsets.UTF_8));
assertThat(is.read(chunk)).isEqualTo(2); assertThat(is.read(chunk)).isEqualTo(2);
stringBuilder sb.append(new String(chunk,0, 2, UTF_8));
.append(new String(chunk,0, 2, StandardCharsets.UTF_8));
assertThat(is.read()).isEqualTo(-1); assertThat(is.read()).isEqualTo(-1);
} }
catch (Throwable e) { catch (Throwable ex) {
ex = e; savedEx = ex;
} }
latch.await(); latch.await();
assertThat(stringBuilder.toString()).isEqualTo(""); assertThat(sb.toString()).isEqualTo("");
assertThat(ex).hasMessage("boom"); assertThat(savedEx).hasMessage("boom");
} }
private static Flux<String> toString(Flow.Publisher<byte[]> flowPublisher) { private static Flow.Publisher<String> toStringPublisher(Flow.Publisher<byte[]> publisher) {
return Flux.from(FlowAdapters.toPublisher(flowPublisher)) return FlowAdapters.toFlowPublisher(toStringFlux(publisher));
.map(bytes -> new String(bytes, StandardCharsets.UTF_8)); }
private static Flux<String> toStringFlux(Flow.Publisher<byte[]> publisher) {
return Flux.from(FlowAdapters.toPublisher(publisher)).map(bytes -> new String(bytes, UTF_8));
} }
} }