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 a85c033ced..c4d281080c 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 @@ -470,8 +470,7 @@ public abstract class DataBufferUtils { * any of the {@link InputStream#read} methods *

* 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 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 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/InputStreamSubscriber.java index 0dd94fd650..67c24d8cce 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/InputStreamSubscriber.java @@ -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; import java.io.IOException; import java.io.InputStream; -import java.nio.ByteBuffer; import java.util.ConcurrentModificationException; import java.util.Objects; import java.util.Queue; @@ -27,70 +42,83 @@ import org.springframework.util.Assert; * {@link org.springframework.http.client.InputStreamSubscriber}. * * @author Oleh Dokuka - * @since 6.1 + * @author Rossen Stoyanchev + * @since 6.2 */ final class InputStreamSubscriber extends InputStream implements Subscriber { - 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)); + private static final Object READY = new Object(); - final int prefetch; - final int limit; - final ReentrantLock lock; - final Queue queue; + private static final DataBuffer DONE = DefaultDataBufferFactory.sharedInstance.allocateBuffer(0); - final AtomicReference parkedThread = new AtomicReference<>(); - final AtomicInteger workAmount = new AtomicInteger(); + private static final DataBuffer CLOSED = DefaultDataBufferFactory.sharedInstance.allocateBuffer(0); - volatile boolean closed; - int consumed; + + private final int prefetch; + + private final int limit; + + private final ReentrantLock lock; + + private final Queue queue; + + private final AtomicReference parkedThread = new AtomicReference<>(); + + private final AtomicInteger workAmount = new AtomicInteger(); + + private volatile boolean closed; + + private int consumed; @Nullable - DataBuffer available; + private DataBuffer available; @Nullable - Subscription s; - boolean done; + private Subscription subscription; + + private boolean done; + @Nullable - Throwable error; + private Throwable error; + InputStreamSubscriber(int 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.lock = new ReentrantLock(false); } + @Override public void onSubscribe(Subscription subscription) { - if (this.s != null) { + if (this.subscription != null) { subscription.cancel(); return; } - this.s = subscription; - subscription.request(prefetch == Integer.MAX_VALUE ? Long.MAX_VALUE : prefetch); + this.subscription = subscription; + subscription.request(this.prefetch == Integer.MAX_VALUE ? Long.MAX_VALUE : this.prefetch); } @Override - public void onNext(DataBuffer t) { - Assert.notNull(t, "DataBuffer must not be null"); + public void onNext(DataBuffer buffer) { + Assert.notNull(buffer, "DataBuffer must not be null"); if (this.done) { - discard(t); + discard(buffer); return; } - if (!queue.offer(t)) { - discard(t); - error = new RuntimeException("Buffer overflow"); - done = true; + if (!this.queue.offer(buffer)) { + discard(buffer); + this.error = new RuntimeException("Buffer overflow"); + this.done = true; } int previousWorkState = addWork(); if (previousWorkState == Integer.MIN_VALUE) { - DataBuffer value = queue.poll(); + DataBuffer value = this.queue.poll(); if (value != null) { discard(value); } @@ -136,51 +164,62 @@ final class InputStreamSubscriber extends InputStream implements Subscriber input, List expectedOutput) { - super.bufferFactory = bufferFactory; - - Publisher 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); + void genericInputStreamSubscriberTest( + DataBufferFactory factory, int writeChunkSize, int readChunkSize, int bufferSize, + List input, List expectedOutput) { + super.bufferFactory = factory; + Publisher 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]; - ArrayList words = new ArrayList<>(); + List words = new ArrayList<>(); - try (InputStream inputStream = DataBufferUtils.subscribeAsInputStream(publisher, bufferSize)) { + try (InputStream in = DataBufferUtils.subscribeAsInputStream(publisher, bufferSize)) { int read; - while((read = inputStream.read(chunk)) > -1) { - String word = new String(chunk, 0, read, StandardCharsets.UTF_8); - words.add(word); + while ((read = in.read(chunk)) > -1) { + words.add(new String(chunk, 0, read, StandardCharsets.UTF_8)); } } catch (IOException e) { @@ -761,33 +767,34 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests { } @ParameterizedDataBufferAllocatingTest - void inputStreamSubscriberError(DataBufferFactory bufferFactory) throws InterruptedException { - super.bufferFactory = bufferFactory; + void inputStreamSubscriberError(DataBufferFactory factory) { + super.bufferFactory = factory; var input = List.of("foo ", "bar ", "baz"); - Publisher publisher = DataBufferUtils.outputStreamPublisher(outputStream -> { - try { - for (String word : input) { - outputStream.write(word.getBytes(StandardCharsets.UTF_8)); - } - throw new RuntimeException("boom"); - } - catch (IOException ex) { - fail(ex.getMessage(), ex); - } - }, super.bufferFactory, Executors.newSingleThreadExecutor(), 1); + Publisher publisher = DataBufferUtils.outputStreamPublisher( + out -> { + try { + for (String word : input) { + out.write(word.getBytes(StandardCharsets.UTF_8)); + } + throw new RuntimeException("boom"); + } + catch (IOException ex) { + fail(ex.getMessage(), ex); + } + }, + super.bufferFactory, Executors.newSingleThreadExecutor(), 1); RuntimeException error = null; byte[] chunk = new byte[4]; - ArrayList words = new ArrayList<>(); + List words = new ArrayList<>(); - try (InputStream inputStream = DataBufferUtils.subscribeAsInputStream(publisher, 1)) { + try (InputStream in = DataBufferUtils.subscribeAsInputStream(publisher, 1)) { int read; - while((read = inputStream.read(chunk)) > -1) { - String word = new String(chunk, 0, read, StandardCharsets.UTF_8); - words.add(word); + while ((read = in.read(chunk)) > -1) { + words.add(new String(chunk, 0, read, StandardCharsets.UTF_8)); } } catch (IOException e) { @@ -801,21 +808,23 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests { } @ParameterizedDataBufferAllocatingTest - void inputStreamSubscriberMixedReadMode(DataBufferFactory bufferFactory) throws InterruptedException { - super.bufferFactory = bufferFactory; + void inputStreamSubscriberMixedReadMode(DataBufferFactory factory) { + super.bufferFactory = factory; var input = List.of("foo ", "bar ", "baz"); - Publisher 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(), 1); + Publisher 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(), 1); byte[] chunk = new byte[3]; @@ -843,30 +852,34 @@ class DataBufferUtilsTests extends AbstractDataBufferAllocatingTests { var input = List.of("foo", "bar", "baz"); - Publisher publisher = DataBufferUtils.outputStreamPublisher(outputStream -> { - try { - assertThatIOException() - .isThrownBy(() -> { - for (String word : input) { - outputStream.write(word.getBytes(StandardCharsets.UTF_8)); - outputStream.flush(); - } - }) - .withMessage("Subscription has been terminated"); - } finally { - latch.countDown(); - } - }, super.bufferFactory, Executors.newSingleThreadExecutor(), 1); + Publisher publisher = DataBufferUtils.outputStreamPublisher( + out -> { + try { + assertThatIOException() + .isThrownBy(() -> { + for (String word : input) { + out.write(word.getBytes(StandardCharsets.UTF_8)); + out.flush(); + } + }) + .withMessage("Subscription has been terminated"); + } + finally { + latch.countDown(); + } + }, + super.bufferFactory, Executors.newSingleThreadExecutor(), 1); byte[] chunk = new byte[3]; ArrayList words = new ArrayList<>(); - try (InputStream inputStream = DataBufferUtils.subscribeAsInputStream(publisher, ThreadLocalRandom.current().nextInt(1, 4))) { - inputStream.read(chunk); + try (InputStream in = DataBufferUtils.subscribeAsInputStream(publisher, ThreadLocalRandom.current().nextInt(1, 4))) { + in.read(chunk); String word = new String(chunk, StandardCharsets.UTF_8); words.add(word); - } catch (IOException e) { + } + catch (IOException e) { throw new RuntimeException(e); } assertThat(words).containsExactlyElementsOf(List.of("foo")); diff --git a/spring-web/src/main/java/org/springframework/http/client/InputStreamSubscriber.java b/spring-web/src/main/java/org/springframework/http/client/InputStreamSubscriber.java index 21e4f25784..14c05411d8 100644 --- a/spring-web/src/main/java/org/springframework/http/client/InputStreamSubscriber.java +++ b/spring-web/src/main/java/org/springframework/http/client/InputStreamSubscriber.java @@ -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; import java.io.IOException; @@ -29,48 +45,65 @@ import org.springframework.util.Assert; * {@link org.springframework.core.io.buffer.InputStreamSubscriber}. * * @author Oleh Dokuka - * @since 6.1 + * @author Rossen Stoyanchev + * @since 6.2 + * @param the publisher byte buffer type */ final class InputStreamSubscriber extends InputStream implements Flow.Subscriber { private static final Log logger = LogFactory.getLog(InputStreamSubscriber.class); - static final Object READY = new Object(); - static final byte[] DONE = new byte[0]; - static final byte[] CLOSED = new byte[0]; + private static final Object READY = new Object(); - final int prefetch; - final int limit; - final ReentrantLock lock; - final Queue queue; - final Function mapper; - final Consumer onDiscardHandler; + private static final byte[] DONE = new byte[0]; - final AtomicReference parkedThread = new AtomicReference<>(); - final AtomicInteger workAmount = new AtomicInteger(); + private static final byte[] CLOSED = new byte[0]; + + + private final Function mapper; + + private final Consumer onDiscardHandler; + + private final int prefetch; + + private final int limit; + + private final ReentrantLock lock; + + private final Queue queue; + + private final AtomicReference parkedThread = new AtomicReference<>(); + + private final AtomicInteger workAmount = new AtomicInteger(); volatile boolean closed; - int consumed; + + private int consumed; @Nullable - byte[] available; - int position; + private byte[] available; + + private int position; @Nullable - Flow.Subscription s; - boolean done; + private Flow.Subscription subscription; + + private boolean done; + @Nullable - Throwable error; + private Throwable error; + private InputStreamSubscriber(Function mapper, Consumer onDiscardHandler, int prefetch) { - this.prefetch = prefetch; - this.limit = prefetch == Integer.MAX_VALUE ? Integer.MAX_VALUE : prefetch - (prefetch >> 2); 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.lock = new ReentrantLock(false); } + /** * Subscribes to given {@link Flow.Publisher} and returns subscription * as {@link InputStream} that allows reading all propagated {@link DataBuffer} messages via its imperative API. @@ -85,8 +118,7 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri * any of the {@link InputStream#read} methods *

* 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 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. @@ -107,33 +139,33 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri @Override public void onSubscribe(Flow.Subscription subscription) { - if (this.s != null) { + if (this.subscription != null) { subscription.cancel(); return; } - this.s = subscription; - subscription.request(prefetch == Integer.MAX_VALUE ? Long.MAX_VALUE : prefetch); + this.subscription = subscription; + subscription.request(this.prefetch == Integer.MAX_VALUE ? Long.MAX_VALUE : this.prefetch); } @Override - public void onNext(T t) { - Assert.notNull(t, "T value must not be null"); + public void onNext(T buffer) { + Assert.notNull(buffer, "Buffer must not be null"); if (this.done) { - discard(t); + discard(buffer); return; } - if (!queue.offer(t)) { - discard(t); - error = new RuntimeException("Buffer overflow"); - done = true; + if (!this.queue.offer(buffer)) { + discard(buffer); + this.error = new RuntimeException("Buffer overflow"); + this.done = true; } int previousWorkState = addWork(); if (previousWorkState == Integer.MIN_VALUE) { - T value = queue.poll(); + T value = this.queue.poll(); if (value != null) { discard(value); } @@ -179,51 +211,62 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri return Integer.MIN_VALUE; } - int nextProduced = produced == Integer.MAX_VALUE ? 1 : produced + 1; + int nextProduced = (produced == Integer.MAX_VALUE ? 1 : produced + 1); - - if (workAmount.weakCompareAndSetRelease(produced, nextProduced)) { + if (this.workAmount.weakCompareAndSetRelease(produced, nextProduced)) { 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 public int read() throws IOException { - if (!lock.tryLock()) { + if (!this.lock.tryLock()) { if (this.closed) { return -1; } - throw new ConcurrentModificationException("concurrent access is disallowed"); + throw new ConcurrentModificationException("Concurrent access is not allowed"); } try { - byte[] bytes = getBytesOrAwait(); + byte[] next = getNextOrAwait(); - if (bytes == DONE) { + if (next == DONE) { this.closed = true; cleanAndFinalize(); if (this.error == null) { return -1; } else { - throw Exceptions.propagate(error); + throw Exceptions.propagate(this.error); } - } else if (bytes == CLOSED) { + } + else if (next == CLOSED) { cleanAndFinalize(); return -1; } - return bytes[this.position++] & 0xFF; + return next[this.position++] & 0xFF; } - catch (Throwable t) { + catch (Throwable ex) { this.closed = true; requiredSubscriber().cancel(); cleanAndFinalize(); - throw Exceptions.propagate(t); + throw Exceptions.propagate(ex); } finally { - lock.unlock(); + this.lock.unlock(); } } @@ -234,7 +277,7 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri return 0; } - if (!lock.tryLock()) { + if (!this.lock.tryLock()) { if (this.closed) { return -1; } @@ -243,9 +286,9 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri try { for (int j = 0; j < len;) { - byte[] bytes = getBytesOrAwait(); + byte[] next = getNextOrAwait(); - if (bytes == DONE) { + if (next == DONE) { cleanAndFinalize(); if (this.error == null) { this.closed = true; @@ -254,37 +297,38 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri else { if (j == 0) { this.closed = true; - throw Exceptions.propagate(error); + throw Exceptions.propagate(this.error); } return j; } - } else if (bytes == CLOSED) { + } + else if (next == CLOSED) { requiredSubscriber().cancel(); cleanAndFinalize(); return -1; } int i = this.position; - for (; i < bytes.length && j < len; i++, j++) { - b[off + j] = bytes[i]; + for (; i < next.length && j < len; i++, j++) { + b[off + j] = next[i]; } this.position = i; } return len; } - catch (Throwable t) { + catch (Throwable ex) { this.closed = true; requiredSubscriber().cancel(); cleanAndFinalize(); - throw Exceptions.propagate(t); + throw Exceptions.propagate(ex); } finally { - lock.unlock(); + this.lock.unlock(); } } - byte[] getBytesOrAwait() { + byte[] getNextOrAwait() { if (this.available == null || this.available.length - this.position == 0) { this.available = null; @@ -294,12 +338,12 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri return CLOSED; } - boolean d = this.done; - T t = this.queue.poll(); - if (t != null) { + boolean done = this.done; + T buffer = this.queue.poll(); + if (buffer != null) { int consumed = ++this.consumed; this.position = 0; - this.available = Objects.requireNonNull(this.mapper.apply(t)); + this.available = Objects.requireNonNull(this.mapper.apply(buffer)); if (consumed == this.limit) { this.consumed = 0; requiredSubscriber().request(this.limit); @@ -307,11 +351,11 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri break; } - if (d) { + if (done) { return DONE; } - actualWorkAmount = workAmount.addAndGet(-actualWorkAmount); + actualWorkAmount = this.workAmount.addAndGet(-actualWorkAmount); if (actualWorkAmount == 0) { await(); } @@ -327,8 +371,7 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri for (;;) { int workAmount = this.workAmount.getPlain(); T value; - - while((value = queue.poll()) != null) { + while ((value = this.queue.poll()) != null) { discard(value); } @@ -338,16 +381,6 @@ final class InputStreamSubscriber 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 public void close() throws IOException { if (this.closed) { @@ -373,8 +406,19 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri } private Flow.Subscription requiredSubscriber() { - Assert.state(this.s != null, "Subscriber must be subscribed to use InputStream"); - return this.s; + Assert.state(this.subscription != null, "Subscriber must be subscribed to use InputStream"); + 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() { @@ -390,7 +434,7 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri throw new IllegalStateException("Only one (Virtual)Thread can await!"); } - if (parkedThread.compareAndSet( null, toUnpark)) { + if (this.parkedThread.compareAndSet( null, toUnpark)) { LockSupport.park(); // 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 @@ -400,13 +444,4 @@ final class InputStreamSubscriber extends InputStream implements Flow.Subscri this.parkedThread.lazySet(null); } - private void resume() { - if (this.parkedThread != READY) { - Object old = parkedThread.getAndSet(READY); - if (old != READY) { - LockSupport.unpark((Thread)old); - } - } - } - } diff --git a/spring-web/src/test/java/org/springframework/http/client/InputStreamSubscriberTests.java b/spring-web/src/test/java/org/springframework/http/client/InputStreamSubscriberTests.java index 66eaf2eced..6cb397b08d 100644 --- a/spring-web/src/test/java/org/springframework/http/client/InputStreamSubscriberTests.java +++ b/spring-web/src/test/java/org/springframework/http/client/InputStreamSubscriberTests.java @@ -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"); * 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.InputStream; import java.io.OutputStreamWriter; -import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; @@ -32,29 +31,33 @@ import org.reactivestreams.FlowAdapters; import reactor.core.publisher.Flux; 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.assertThatIOException; /** + * Unit tests for {@link InputStreamSubscriber}. + * * @author Arjen Poutsma * @author Oleh Dokuka */ 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 OutputStreamPublisher.ByteMapper byteMapper = new OutputStreamPublisher.ByteMapper<>() { + @Override public byte[] map(int b) { - return new byte[]{(byte) b}; + return new byte[] {(byte) b}; } @Override @@ -68,31 +71,36 @@ class InputStreamSubscriberTests { @Test void basic() { - Flow.Publisher flowPublisher = new OutputStreamPublisher<>(outputStream -> { - outputStream.write(FOO); - outputStream.write(BAR); - outputStream.write(BAZ); - }, this.byteMapper, this.executor, null); - Flux flux = toString(flowPublisher); + Flow.Publisher publisher = new OutputStreamPublisher<>( + out -> { + out.write(FOO); + out.write(BAR); + out.write(BAZ); + }, + this.byteMapper, this.executor, null); - StepVerifier.create(flux) + StepVerifier.create(toStringFlux(publisher)) .assertNext(s -> assertThat(s).isEqualTo("foobarbaz")) .verifyComplete(); } @Test - void flush() { - Flow.Publisher flowPublisher = new OutputStreamPublisher<>(outputStream -> { - outputStream.write(FOO); - outputStream.flush(); - outputStream.write(BAR); - outputStream.flush(); - outputStream.write(BAZ); - outputStream.flush(); - }, this.byteMapper, this.executor, null); - Flux flux = toString(flowPublisher); + void flush() throws IOException { + Flow.Publisher publisher = new OutputStreamPublisher<>( + out -> { + out.write(FOO); + out.flush(); + out.write(BAR); + out.flush(); + out.write(BAZ); + out.flush(); + }, + 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]; assertThat(is.read(chunk)).isEqualTo(3); @@ -103,38 +111,35 @@ class InputStreamSubscriberTests { assertThat(chunk).containsExactly(BAZ); assertThat(is.read(chunk)).isEqualTo(-1); } - catch (IOException e) { - throw new RuntimeException(e); - } - } + } @Test void chunkSize() { - Flow.Publisher flowPublisher = new OutputStreamPublisher<>(outputStream -> { - outputStream.write(FOO); - outputStream.write(BAR); - outputStream.write(BAZ); - }, this.byteMapper, this.executor, 2); - Flux flux = toString(flowPublisher); + Flow.Publisher publisher = new OutputStreamPublisher<>( + out -> { + out.write(FOO); + out.write(BAR); + out.write(BAZ); + }, + 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(); byte[] chunk = new byte[3]; + stringBuilder.append(new String(new byte[]{(byte)is.read()}, UTF_8)); + assertThat(is.read(chunk)).isEqualTo(3); - stringBuilder - .append(new String(new byte[]{(byte)is.read()}, 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(3); - stringBuilder - .append(new String(chunk, StandardCharsets.UTF_8)); + + stringBuilder.append(new String(chunk, UTF_8)); 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"); } catch (IOException e) { @@ -146,24 +151,25 @@ class InputStreamSubscriberTests { void cancel() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); - Flow.Publisher flowPublisher = new OutputStreamPublisher<>(outputStream -> { - assertThatIOException() - .isThrownBy(() -> { - outputStream.write(FOO); - outputStream.flush(); - outputStream.write(BAR); - outputStream.flush(); - outputStream.write(BAZ); - outputStream.flush(); - }) - .withMessage("Subscription has been terminated"); - latch.countDown(); + Flow.Publisher publisher = new OutputStreamPublisher<>( + out -> { + assertThatIOException().isThrownBy(() -> { + out.write(FOO); + out.flush(); + out.write(BAR); + out.flush(); + out.write(BAZ); + out.flush(); + }).withMessage("Subscription has been terminated"); + latch.countDown(); + + }, this.byteMapper, this.executor, null); - }, this.byteMapper, this.executor, null); - Flux flux = toString(flowPublisher); List 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]; assertThat(is.read(chunk)).isEqualTo(3); @@ -182,22 +188,23 @@ class InputStreamSubscriberTests { void closed() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); - Flow.Publisher flowPublisher = new OutputStreamPublisher<>(outputStream -> { - OutputStreamWriter writer = new OutputStreamWriter(outputStream, StandardCharsets.UTF_8); - writer.write("foo"); - writer.close(); - assertThatIOException().isThrownBy(() -> writer.write("bar")) - .withMessage("Stream closed"); - latch.countDown(); - }, this.byteMapper, this.executor, null); - Flux flux = toString(flowPublisher); + Flow.Publisher publisher = new OutputStreamPublisher<>( + out -> { + OutputStreamWriter writer = new OutputStreamWriter(out, UTF_8); + writer.write("foo"); + writer.close(); + assertThatIOException().isThrownBy(() -> writer.write("bar")).withMessage("Stream closed"); + latch.countDown(); + }, + 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]; assertThat(is.read(chunk)).isEqualTo(3); assertThat(chunk).containsExactly(FOO); - assertThat(is.read(chunk)).isEqualTo(-1); } catch (IOException e) { @@ -211,49 +218,52 @@ class InputStreamSubscriberTests { void mapperThrowsException() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); - Flow.Publisher flowPublisher = new OutputStreamPublisher<>(outputStream -> { - outputStream.write(FOO); - outputStream.flush(); - assertThatIOException().isThrownBy(() -> { - outputStream.write(BAR); - outputStream.flush(); - }).withMessage("Subscription has been terminated"); - latch.countDown(); - }, this.byteMapper, this.executor, null); - Throwable ex = null; + Flow.Publisher publisher = new OutputStreamPublisher<>( + out -> { + out.write(FOO); + out.flush(); + assertThatIOException().isThrownBy(() -> { + out.write(BAR); + out.flush(); + }) + .withMessage("Subscription has been terminated"); + 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]; - stringBuilder - .append(new String(new byte[]{(byte)is.read()}, StandardCharsets.UTF_8)); + sb.append(new String(new byte[]{(byte)is.read()}, UTF_8)); assertThat(is.read(chunk)).isEqualTo(3); - stringBuilder - .append(new String(chunk, StandardCharsets.UTF_8)); + sb.append(new String(chunk, UTF_8)); assertThat(is.read(chunk)).isEqualTo(3); - stringBuilder - .append(new String(chunk, StandardCharsets.UTF_8)); + sb.append(new String(chunk, UTF_8)); assertThat(is.read(chunk)).isEqualTo(2); - stringBuilder - .append(new String(chunk,0, 2, StandardCharsets.UTF_8)); + sb.append(new String(chunk,0, 2, UTF_8)); assertThat(is.read()).isEqualTo(-1); } - catch (Throwable e) { - ex = e; - } + catch (Throwable ex) { + savedEx = ex; + } - latch.await(); + latch.await(); - assertThat(stringBuilder.toString()).isEqualTo(""); - assertThat(ex).hasMessage("boom"); + assertThat(sb.toString()).isEqualTo(""); + assertThat(savedEx).hasMessage("boom"); } - private static Flux toString(Flow.Publisher flowPublisher) { - return Flux.from(FlowAdapters.toPublisher(flowPublisher)) - .map(bytes -> new String(bytes, StandardCharsets.UTF_8)); + private static Flow.Publisher toStringPublisher(Flow.Publisher publisher) { + return FlowAdapters.toFlowPublisher(toStringFlux(publisher)); + } + + private static Flux toStringFlux(Flow.Publisher publisher) { + return Flux.from(FlowAdapters.toPublisher(publisher)).map(bytes -> new String(bytes, UTF_8)); } }