| /* |
| * Copyright (c) 2016, 2024, Oracle and/or its affiliates. All rights reserved. |
| * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. |
| * |
| * This code is free software; you can redistribute it and/or modify it |
| * under the terms of the GNU General Public License version 2 only, as |
| * published by the Free Software Foundation. Oracle designates this |
| * particular file as subject to the "Classpath" exception as provided |
| * by Oracle in the LICENSE file that accompanied this code. |
| * |
| * This code is distributed in the hope that it will be useful, but WITHOUT |
| * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or |
| * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License |
| * version 2 for more details (a copy is included in the LICENSE file that |
| * accompanied this code). |
| * |
| * You should have received a copy of the GNU General Public License version |
| * 2 along with this work; if not, write to the Free Software Foundation, |
| * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. |
| * |
| * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA |
| * or visit www.oracle.com if you need additional information or have any |
| * questions. |
| */ |
| |
| package jdk.internal.net.http; |
| |
| import java.io.BufferedReader; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.InputStreamReader; |
| import java.nio.ByteBuffer; |
| import java.nio.channels.FileChannel; |
| import java.nio.charset.Charset; |
| import java.nio.file.OpenOption; |
| import java.nio.file.Path; |
| import java.util.ArrayList; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Objects; |
| import java.util.Optional; |
| import java.util.concurrent.ArrayBlockingQueue; |
| import java.util.concurrent.BlockingQueue; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.CompletionStage; |
| import java.util.concurrent.Executor; |
| import java.util.concurrent.Flow; |
| import java.util.concurrent.Flow.Subscriber; |
| import java.util.concurrent.Flow.Subscription; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicReference; |
| import java.util.function.Consumer; |
| import java.util.function.Function; |
| import java.util.stream.Stream; |
| import java.net.http.HttpResponse.BodySubscriber; |
| import jdk.internal.net.http.common.Log; |
| import jdk.internal.net.http.common.Logger; |
| import jdk.internal.net.http.common.MinimalFuture; |
| import jdk.internal.net.http.common.Utils; |
| import jdk.internal.net.http.HttpClientImpl.DelegatingExecutor; |
| import static java.nio.charset.StandardCharsets.UTF_8; |
| |
| public class ResponseSubscribers { |
| |
| /** |
| * This interface is used by our BodySubscriber implementations to |
| * declare whether calling getBody() inline is safe, or whether |
| * it needs to be called asynchronously in an executor thread. |
| * Calling getBody() inline is usually safe except when it |
| * might block - which can be the case if the BodySubscriber |
| * is provided by custom code, or if it uses a finisher that |
| * might be called and might block before the last bit is |
| * received (for instance, if a mapping subscriber is used with |
| * a mapper function that maps an InputStream to a GZIPInputStream, |
| * as the constructor of GZIPInputStream calls read()). |
| * @param <T> The response type. |
| */ |
| public interface TrustedSubscriber<T> extends BodySubscriber<T> { |
| /** |
| * Returns true if getBody() should be called asynchronously. |
| * @implSpec The default implementation of this method returns |
| * false. |
| * @return true if getBody() should be called asynchronously. |
| */ |
| default boolean needsExecutor() { return false;} |
| |
| /** |
| * Returns true if calling {@code bs::getBody} might block |
| * and requires an executor. |
| * |
| * @implNote |
| * In particular this method returns |
| * true if {@code bs} is not a {@code TrustedSubscriber}. |
| * If it is a {@code TrustedSubscriber}, it returns |
| * {@code ((TrustedSubscriber) bs).needsExecutor()}. |
| * |
| * @param bs A BodySubscriber. |
| * @return true if calling {@code bs::getBody} requires using |
| * an executor. |
| */ |
| static boolean needsExecutor(BodySubscriber<?> bs) { |
| if (bs instanceof TrustedSubscriber) { |
| return ((TrustedSubscriber) bs).needsExecutor(); |
| } else return true; |
| } |
| } |
| |
| public static class ConsumerSubscriber implements TrustedSubscriber<Void> { |
| private final Consumer<Optional<byte[]>> consumer; |
| private Flow.Subscription subscription; |
| private final CompletableFuture<Void> result = new MinimalFuture<>(); |
| private final AtomicBoolean subscribed = new AtomicBoolean(); |
| |
| public ConsumerSubscriber(Consumer<Optional<byte[]>> consumer) { |
| this.consumer = Objects.requireNonNull(consumer); |
| } |
| |
| @Override |
| public CompletionStage<Void> getBody() { |
| return result; |
| } |
| |
| @Override |
| public void onSubscribe(Flow.Subscription subscription) { |
| Objects.requireNonNull(subscription); |
| if (!subscribed.compareAndSet(false, true)) { |
| subscription.cancel(); |
| } else { |
| this.subscription = subscription; |
| subscription.request(1); |
| } |
| } |
| |
| @Override |
| public void onNext(List<ByteBuffer> items) { |
| Objects.requireNonNull(items); |
| for (ByteBuffer item : items) { |
| byte[] buf = new byte[item.remaining()]; |
| item.get(buf); |
| consumer.accept(Optional.of(buf)); |
| } |
| subscription.request(1); |
| } |
| |
| @Override |
| public void onError(Throwable throwable) { |
| Objects.requireNonNull(throwable); |
| result.completeExceptionally(throwable); |
| } |
| |
| @Override |
| public void onComplete() { |
| consumer.accept(Optional.empty()); |
| result.complete(null); |
| } |
| |
| } |
| |
| /** |
| * A Subscriber that writes the flow of data to a given file. |
| */ |
| public static class PathSubscriber implements TrustedSubscriber<Path> { |
| private final Path file; |
| private final OpenOption[] options; |
| private final CompletableFuture<Path> result = new MinimalFuture<>(); |
| |
| private final AtomicBoolean subscribed = new AtomicBoolean(); |
| private volatile Flow.Subscription subscription; |
| private volatile FileChannel out; |
| |
| /** |
| * Factory for creating PathSubscriber. |
| */ |
| public static PathSubscriber create(Path file, |
| List<OpenOption> options) { |
| return new PathSubscriber(file, options); |
| } |
| |
| // pp so handler implementations in the same package can construct |
| /*package-private*/ PathSubscriber(Path file, |
| List<OpenOption> options) { |
| this.file = file; |
| this.options = options.stream().toArray(OpenOption[]::new); |
| } |
| |
| @Override |
| public void onSubscribe(Flow.Subscription subscription) { |
| Objects.requireNonNull(subscription); |
| if (!subscribed.compareAndSet(false, true)) { |
| subscription.cancel(); |
| return; |
| } |
| |
| this.subscription = subscription; |
| try { |
| out = FileChannel.open(file, options); |
| } catch (IOException ioe) { |
| result.completeExceptionally(ioe); |
| subscription.cancel(); |
| return; |
| } |
| subscription.request(1); |
| } |
| |
| @Override |
| public void onNext(List<ByteBuffer> items) { |
| try { |
| ByteBuffer[] buffers = items.toArray(Utils.EMPTY_BB_ARRAY); |
| do { |
| out.write(buffers); |
| } while (Utils.hasRemaining(buffers)); |
| } catch (IOException ex) { |
| close(); |
| subscription.cancel(); |
| result.completeExceptionally(ex); |
| } |
| subscription.request(1); |
| } |
| |
| @Override |
| public void onError(Throwable e) { |
| result.completeExceptionally(e); |
| close(); |
| } |
| |
| @Override |
| public void onComplete() { |
| close(); |
| result.complete(file); |
| } |
| |
| @Override |
| public CompletionStage<Path> getBody() { |
| return result; |
| } |
| |
| private void close() { |
| Utils.close(out); |
| } |
| } |
| |
| public static class ByteArraySubscriber<T> implements TrustedSubscriber<T> { |
| private final Function<byte[], T> finisher; |
| private final CompletableFuture<T> result = new MinimalFuture<>(); |
| private final List<ByteBuffer> received = new ArrayList<>(); |
| |
| private volatile Flow.Subscription subscription; |
| |
| public ByteArraySubscriber(Function<byte[],T> finisher) { |
| this.finisher = finisher; |
| } |
| |
| @Override |
| public void onSubscribe(Flow.Subscription subscription) { |
| if (this.subscription != null) { |
| subscription.cancel(); |
| return; |
| } |
| this.subscription = subscription; |
| // We can handle whatever you've got |
| subscription.request(Long.MAX_VALUE); |
| } |
| |
| @Override |
| public void onNext(List<ByteBuffer> items) { |
| // incoming buffers are allocated by http client internally, |
| // and won't be used anywhere except this place. |
| // So it's free simply to store them for further processing. |
| Objects.requireNonNull(items); // ensure NPE is thrown before assert |
| assert Utils.hasRemaining(items); |
| received.addAll(items); |
| } |
| |
| @Override |
| public void onError(Throwable throwable) { |
| received.clear(); |
| result.completeExceptionally(throwable); |
| } |
| |
| private static byte[] join(List<ByteBuffer> bytes) { |
| int size = Utils.remaining(bytes, Integer.MAX_VALUE); |
| byte[] res = new byte[size]; |
| int from = 0; |
| for (ByteBuffer b : bytes) { |
| int l = b.remaining(); |
| b.get(res, from, l); |
| from += l; |
| } |
| return res; |
| } |
| |
| @Override |
| public void onComplete() { |
| try { |
| result.complete(finisher.apply(join(received))); |
| received.clear(); |
| } catch (IllegalArgumentException e) { |
| result.completeExceptionally(e); |
| } |
| } |
| |
| @Override |
| public CompletionStage<T> getBody() { |
| return result; |
| } |
| } |
| |
| /** |
| * An InputStream built on top of the Flow API. |
| */ |
| public static class HttpResponseInputStream extends InputStream |
| implements TrustedSubscriber<InputStream> |
| { |
| static final int MAX_BUFFERS_IN_QUEUE = 1; // lock-step with the producer |
| |
| // An immutable ByteBuffer sentinel to mark that the last byte was received. |
| private static final ByteBuffer LAST_BUFFER = ByteBuffer.wrap(new byte[0]); |
| private static final List<ByteBuffer> LAST_LIST = List.of(LAST_BUFFER); |
| private static final Logger debug = |
| Utils.getDebugLogger("HttpResponseInputStream"::toString, Utils.DEBUG); |
| |
| // A queue of yet unprocessed ByteBuffers received from the flow API. |
| private final BlockingQueue<List<ByteBuffer>> buffers; |
| private volatile Flow.Subscription subscription; |
| private volatile boolean closed; |
| private volatile Throwable failed; |
| private volatile Iterator<ByteBuffer> currentListItr; |
| private volatile ByteBuffer currentBuffer; |
| private final AtomicBoolean subscribed = new AtomicBoolean(); |
| |
| public HttpResponseInputStream() { |
| this(MAX_BUFFERS_IN_QUEUE); |
| } |
| |
| HttpResponseInputStream(int maxBuffers) { |
| int capacity = (maxBuffers <= 0 ? MAX_BUFFERS_IN_QUEUE : maxBuffers); |
| // 1 additional slot needed for LAST_LIST added by onComplete |
| this.buffers = new ArrayBlockingQueue<>(capacity + 1); |
| } |
| |
| @Override |
| public CompletionStage<InputStream> getBody() { |
| // Returns the stream immediately, before the |
| // response body is received. |
| // This makes it possible for sendAsync().get().body() |
| // to complete before the response body is received. |
| return CompletableFuture.completedStage(this); |
| } |
| |
| // Returns the current byte buffer to read from. |
| // If the current buffer has no remaining data, this method will take the |
| // next buffer from the buffers queue, possibly blocking until |
| // a new buffer is made available through the Flow API, or the |
| // end of the flow has been reached. |
| private ByteBuffer current() throws IOException { |
| while (currentBuffer == null || !currentBuffer.hasRemaining()) { |
| // Check whether the stream is closed or exhausted |
| if (closed || failed != null) { |
| throw new IOException("closed", failed); |
| } |
| if (currentBuffer == LAST_BUFFER) break; |
| |
| try { |
| if (currentListItr == null || !currentListItr.hasNext()) { |
| // Take a new list of buffers from the queue, blocking |
| // if none is available yet... |
| |
| if (debug.on()) debug.log("Taking list of Buffers"); |
| List<ByteBuffer> lb = buffers.take(); |
| currentListItr = lb.iterator(); |
| if (debug.on()) debug.log("List of Buffers Taken"); |
| |
| // Check whether an exception was encountered upstream |
| if (closed || failed != null) |
| throw new IOException("closed", failed); |
| |
| // Check whether we're done. |
| if (lb == LAST_LIST) { |
| currentListItr = null; |
| currentBuffer = LAST_BUFFER; |
| break; |
| } |
| |
| // Request another upstream item ( list of buffers ) |
| Flow.Subscription s = subscription; |
| if (s != null) { |
| if (debug.on()) debug.log("Increased demand by 1"); |
| s.request(1); |
| } |
| assert currentListItr != null; |
| if (lb.isEmpty()) continue; |
| } |
| assert currentListItr != null; |
| assert currentListItr.hasNext(); |
| if (debug.on()) debug.log("Next Buffer"); |
| currentBuffer = currentListItr.next(); |
| } catch (InterruptedException ex) { |
| try { |
| close(); |
| } catch (IOException ignored) { |
| } |
| Thread.currentThread().interrupt(); |
| throw new IOException(ex); |
| } |
| } |
| assert currentBuffer == LAST_BUFFER || currentBuffer.hasRemaining(); |
| return currentBuffer; |
| } |
| |
| @Override |
| public int read(byte[] bytes, int off, int len) throws IOException { |
| Objects.checkFromIndexSize(off, len, bytes.length); |
| if (len == 0) { |
| return 0; |
| } |
| // get the buffer to read from, possibly blocking if |
| // none is available |
| ByteBuffer buffer; |
| if ((buffer = current()) == LAST_BUFFER) return -1; |
| |
| // don't attempt to read more than what is available |
| // in the current buffer. |
| int read = Math.min(buffer.remaining(), len); |
| assert read > 0 && read <= buffer.remaining(); |
| |
| // buffer.get() will do the boundary check for us. |
| buffer.get(bytes, off, read); |
| return read; |
| } |
| |
| @Override |
| public int read() throws IOException { |
| ByteBuffer buffer; |
| if ((buffer = current()) == LAST_BUFFER) return -1; |
| return buffer.get() & 0xFF; |
| } |
| |
| @Override |
| public int available() throws IOException { |
| // best effort: returns the number of remaining bytes in |
| // the current buffer if any, or 1 if the current buffer |
| // is null or empty but the queue or current buffer list |
| // are not empty. Returns 0 otherwise. |
| if (closed) return 0; |
| int available = 0; |
| ByteBuffer current = currentBuffer; |
| if (current == LAST_BUFFER) return 0; |
| if (current != null) available = current.remaining(); |
| if (available != 0) return available; |
| Iterator<?> iterator = currentListItr; |
| if (iterator != null && iterator.hasNext()) return 1; |
| if (!buffers.isEmpty() && buffers.peek() != LAST_LIST ) return 1; |
| return available; |
| } |
| |
| @Override |
| public void onSubscribe(Flow.Subscription s) { |
| Objects.requireNonNull(s); |
| if (debug.on()) debug.log("onSubscribe called"); |
| try { |
| if (!subscribed.compareAndSet(false, true)) { |
| if (debug.on()) debug.log("Already subscribed: canceling"); |
| s.cancel(); |
| } else { |
| // check whether the stream is already closed. |
| // if so, we should cancel the subscription |
| // immediately. |
| boolean closed; |
| synchronized (this) { |
| closed = this.closed; |
| if (!closed) { |
| this.subscription = s; |
| // should contain at least 2, unless closed or failed. |
| assert buffers.remainingCapacity() > 1 || failed != null |
| : "buffers capacity: " + buffers.remainingCapacity() |
| + ", closed: " + closed + ", terminated: " |
| + buffers.contains(LAST_LIST) |
| + ", failed: " + failed; |
| } |
| } |
| if (closed) { |
| if (debug.on()) debug.log("Already closed: canceling"); |
| s.cancel(); |
| return; |
| } |
| if (debug.on()) |
| debug.log("onSubscribe: requesting " |
| + Math.max(1, buffers.remainingCapacity() - 1)); |
| s.request(Math.max(1, buffers.remainingCapacity() - 1)); |
| } |
| } catch (Throwable t) { |
| failed = t; |
| if (debug.on()) |
| debug.log("onSubscribe failed", t); |
| try { |
| close(); |
| } catch (IOException x) { |
| // OK |
| } finally { |
| onError(t); |
| } |
| } |
| } |
| |
| @Override |
| public void onNext(List<ByteBuffer> t) { |
| Objects.requireNonNull(t); |
| try { |
| if (debug.on()) debug.log("next item received"); |
| if (!buffers.offer(t)) { |
| throw new IllegalStateException("queue is full"); |
| } |
| if (debug.on()) debug.log("item offered"); |
| } catch (Throwable ex) { |
| failed = ex; |
| try { |
| close(); |
| } catch (IOException ex1) { |
| // OK |
| } finally { |
| onError(ex); |
| } |
| } |
| } |
| |
| @Override |
| public void onError(Throwable thrwbl) { |
| if (debug.on()) |
| debug.log("onError called: " + thrwbl); |
| subscription = null; |
| failed = Objects.requireNonNull(thrwbl); |
| // The client process that reads the input stream might |
| // be blocked in queue.take(). |
| // Tries to offer LAST_LIST to the queue. If the queue is |
| // full we don't care if we can't insert this buffer, as |
| // the client can't be blocked in queue.take() in that case. |
| // Adding LAST_LIST to the queue is harmless, as the client |
| // should find failed != null before handling LAST_LIST. |
| buffers.offer(LAST_LIST); |
| } |
| |
| @Override |
| public void onComplete() { |
| if (debug.on()) |
| debug.log("onComplete called"); |
| subscription = null; |
| onNext(LAST_LIST); |
| } |
| |
| @Override |
| public void close() throws IOException { |
| Flow.Subscription s; |
| synchronized (this) { |
| if (closed) return; |
| closed = true; |
| s = subscription; |
| subscription = null; |
| } |
| if (debug.on()) |
| debug.log("close called"); |
| // s will be null if already completed |
| try { |
| if (s != null) { |
| s.cancel(); |
| } |
| } finally { |
| buffers.offer(LAST_LIST); |
| super.close(); |
| } |
| } |
| |
| } |
| |
| public static BodySubscriber<Stream<String>> createLineStream() { |
| return createLineStream(UTF_8); |
| } |
| |
| public static BodySubscriber<Stream<String>> createLineStream(Charset charset) { |
| Objects.requireNonNull(charset); |
| BodySubscriber<InputStream> s = new HttpResponseInputStream(); |
| // Creates a MappingSubscriber with a trusted finisher that is |
| // trusted not to block. |
| return new MappingSubscriber<InputStream,Stream<String>>(s, |
| (InputStream stream) -> { |
| return new BufferedReader(new InputStreamReader(stream, charset)) |
| .lines().onClose(() -> Utils.close(stream)); |
| }, true); |
| } |
| |
| /** |
| * Currently this consumes all of the data and ignores it |
| */ |
| public static class NullSubscriber<T> implements TrustedSubscriber<T> { |
| |
| private final CompletableFuture<T> cf = new MinimalFuture<>(); |
| private final Optional<T> result; |
| private final AtomicBoolean subscribed = new AtomicBoolean(); |
| |
| public NullSubscriber(Optional<T> result) { |
| this.result = result; |
| } |
| |
| @Override |
| public void onSubscribe(Flow.Subscription subscription) { |
| Objects.requireNonNull(subscription); |
| if (!subscribed.compareAndSet(false, true)) { |
| subscription.cancel(); |
| } else { |
| subscription.request(Long.MAX_VALUE); |
| } |
| } |
| |
| @Override |
| public void onNext(List<ByteBuffer> items) { |
| Objects.requireNonNull(items); |
| } |
| |
| @Override |
| public void onError(Throwable throwable) { |
| Objects.requireNonNull(throwable); |
| cf.completeExceptionally(throwable); |
| } |
| |
| @Override |
| public void onComplete() { |
| if (result.isPresent()) { |
| cf.complete(result.get()); |
| } else { |
| cf.complete(null); |
| } |
| } |
| |
| @Override |
| public CompletionStage<T> getBody() { |
| return cf; |
| } |
| } |
| |
| /** An adapter between {@code BodySubscriber} and {@code Flow.Subscriber}. */ |
| public static final class SubscriberAdapter<S extends Subscriber<? super List<ByteBuffer>>,R> |
| implements TrustedSubscriber<R> |
| { |
| private final CompletableFuture<R> cf = new MinimalFuture<>(); |
| private final S subscriber; |
| private final Function<? super S,? extends R> finisher; |
| private volatile Subscription subscription; |
| |
| // The finisher isn't called until all bytes have been received, |
| // and so shouldn't need an executor. No need to override |
| // TrustedSubscriber::needsExecutor |
| public SubscriberAdapter(S subscriber, Function<? super S,? extends R> finisher) { |
| this.subscriber = Objects.requireNonNull(subscriber); |
| this.finisher = Objects.requireNonNull(finisher); |
| } |
| |
| @Override |
| public void onSubscribe(Subscription subscription) { |
| Objects.requireNonNull(subscription); |
| if (this.subscription != null) { |
| subscription.cancel(); |
| } else { |
| this.subscription = subscription; |
| subscriber.onSubscribe(subscription); |
| } |
| } |
| |
| @Override |
| public void onNext(List<ByteBuffer> item) { |
| Objects.requireNonNull(item); |
| try { |
| subscriber.onNext(item); |
| } catch (Throwable throwable) { |
| subscription.cancel(); |
| onError(throwable); |
| } |
| } |
| |
| @Override |
| public void onError(Throwable throwable) { |
| Objects.requireNonNull(throwable); |
| try { |
| subscriber.onError(throwable); |
| } finally { |
| cf.completeExceptionally(throwable); |
| } |
| } |
| |
| @Override |
| public void onComplete() { |
| try { |
| subscriber.onComplete(); |
| } finally { |
| try { |
| cf.completeAsync(() -> finisher.apply(subscriber)); |
| } catch (Throwable throwable) { |
| cf.completeExceptionally(throwable); |
| } |
| } |
| } |
| |
| @Override |
| public CompletionStage<R> getBody() { |
| return cf; |
| } |
| } |
| |
| /** |
| * A body subscriber which receives input from an upstream subscriber |
| * and maps that subscriber's body type to a new type. The upstream subscriber |
| * delegates all flow operations directly to this object. The |
| * {@link CompletionStage} returned by {@link #getBody()}} takes the output |
| * of the upstream {@code getBody()} and applies the mapper function to |
| * obtain the new {@code CompletionStage} type. |
| * |
| * @param <T> the upstream body type |
| * @param <U> this subscriber's body type |
| */ |
| public static class MappingSubscriber<T,U> implements TrustedSubscriber<U> { |
| private final BodySubscriber<T> upstream; |
| private final Function<? super T,? extends U> mapper; |
| private final boolean trusted; |
| |
| public MappingSubscriber(BodySubscriber<T> upstream, |
| Function<? super T,? extends U> mapper) { |
| this(upstream, mapper, false); |
| } |
| |
| // creates a MappingSubscriber with a mapper that is trusted |
| // to not block when called. |
| MappingSubscriber(BodySubscriber<T> upstream, |
| Function<? super T,? extends U> mapper, |
| boolean trusted) { |
| this.upstream = Objects.requireNonNull(upstream); |
| this.mapper = Objects.requireNonNull(mapper); |
| this.trusted = trusted; |
| } |
| |
| // There is no way to know whether a custom mapper function |
| // might block or not - so we should return true unless the |
| // mapper is implemented and trusted by our own code not to |
| // block. |
| @Override |
| public boolean needsExecutor() { |
| return !trusted || TrustedSubscriber.needsExecutor(upstream); |
| } |
| |
| // If upstream.getBody() is already completed (case of InputStream), |
| // then calling upstream.getBody().thenApply(mapper) might block |
| // if the mapper blocks. We should probably add a variant of |
| // MappingSubscriber that calls thenApplyAsync instead, but this |
| // needs a new public API point. See needsExecutor() above. |
| @Override |
| public CompletionStage<U> getBody() { |
| return upstream.getBody().thenApply(mapper); |
| } |
| |
| @Override |
| public void onSubscribe(Flow.Subscription subscription) { |
| upstream.onSubscribe(subscription); |
| } |
| |
| @Override |
| public void onNext(List<ByteBuffer> item) { |
| upstream.onNext(item); |
| } |
| |
| @Override |
| public void onError(Throwable throwable) { |
| upstream.onError(throwable); |
| } |
| |
| @Override |
| public void onComplete() { |
| upstream.onComplete(); |
| } |
| } |
| |
| // A BodySubscriber that returns a Publisher<List<ByteBuffer>> |
| static class PublishingBodySubscriber |
| implements TrustedSubscriber<Flow.Publisher<List<ByteBuffer>>> { |
| private final MinimalFuture<Flow.Subscription> |
| subscriptionCF = new MinimalFuture<>(); |
| private final MinimalFuture<SubscriberRef> |
| subscribedCF = new MinimalFuture<>(); |
| private AtomicReference<SubscriberRef> |
| subscriberRef = new AtomicReference<>(); |
| private final CompletionStage<Flow.Publisher<List<ByteBuffer>>> body = |
| subscriptionCF.thenCompose( |
| (s) -> MinimalFuture.completedFuture(this::subscribe)); |
| |
| // We use the completionCF to ensure that only one of |
| // onError or onComplete is ever called. |
| private final MinimalFuture<Void> completionCF; |
| private PublishingBodySubscriber() { |
| completionCF = new MinimalFuture<>(); |
| completionCF.whenComplete( |
| (r,t) -> subscribedCF.thenAccept( s -> complete(s, t))); |
| } |
| |
| // An object that holds a reference to a Flow.Subscriber. |
| // The reference is cleared when the subscriber is completed - either |
| // normally or exceptionally, or when the subscription is cancelled. |
| static final class SubscriberRef { |
| volatile Flow.Subscriber<? super List<ByteBuffer>> ref; |
| SubscriberRef(Flow.Subscriber<? super List<ByteBuffer>> subscriber) { |
| ref = subscriber; |
| } |
| Flow.Subscriber<? super List<ByteBuffer>> get() { |
| return ref; |
| } |
| Flow.Subscriber<? super List<ByteBuffer>> clear() { |
| Flow.Subscriber<? super List<ByteBuffer>> res = ref; |
| ref = null; |
| return res; |
| } |
| } |
| |
| // A subscription that wraps an upstream subscription and |
| // holds a reference to a subscriber. The subscriber reference |
| // is cleared when the subscription is cancelled |
| static final class SubscriptionRef implements Flow.Subscription { |
| final Flow.Subscription subscription; |
| final SubscriberRef subscriberRef; |
| SubscriptionRef(Flow.Subscription subscription, |
| SubscriberRef subscriberRef) { |
| this.subscription = subscription; |
| this.subscriberRef = subscriberRef; |
| } |
| @Override |
| public void request(long n) { |
| if (subscriberRef.get() != null) { |
| subscription.request(n); |
| } |
| } |
| @Override |
| public void cancel() { |
| subscription.cancel(); |
| subscriberRef.clear(); |
| } |
| |
| void subscribe() { |
| Subscriber<?> subscriber = subscriberRef.get(); |
| if (subscriber != null) { |
| subscriber.onSubscribe(this); |
| } |
| } |
| |
| @Override |
| public String toString() { |
| return "SubscriptionRef/" |
| + subscription.getClass().getName() |
| + "@" |
| + System.identityHashCode(subscription); |
| } |
| } |
| |
| // This is a callback for the subscribedCF. |
| // Do not call directly! |
| private void complete(SubscriberRef ref, Throwable t) { |
| assert ref != null; |
| Subscriber<?> s = ref.clear(); |
| // maybe null if subscription was cancelled |
| if (s == null) return; |
| if (t == null) { |
| try { |
| s.onComplete(); |
| } catch (Throwable x) { |
| s.onError(x); |
| } |
| } else { |
| s.onError(t); |
| } |
| } |
| |
| private void signalError(Throwable err) { |
| if (err == null) { |
| err = new NullPointerException("null throwable"); |
| } |
| completionCF.completeExceptionally(err); |
| } |
| |
| private void signalComplete() { |
| completionCF.complete(null); |
| } |
| |
| private void subscribe(Flow.Subscriber<? super List<ByteBuffer>> subscriber) { |
| Objects.requireNonNull(subscriber, "subscriber must not be null"); |
| SubscriberRef ref = new SubscriberRef(subscriber); |
| if (subscriberRef.compareAndSet(null, ref)) { |
| subscriptionCF.thenAccept((s) -> { |
| SubscriptionRef subscription = new SubscriptionRef(s,ref); |
| try { |
| subscription.subscribe(); |
| subscribedCF.complete(ref); |
| } catch (Throwable t) { |
| if (Log.errors()) { |
| Log.logError("Failed to call onSubscribe: " + |
| "cancelling subscription: " + t); |
| Log.logError(t); |
| } |
| subscription.cancel(); |
| } |
| }); |
| } else { |
| subscriber.onSubscribe(new Flow.Subscription() { |
| @Override public void request(long n) { } |
| @Override public void cancel() { } |
| }); |
| subscriber.onError(new IllegalStateException( |
| "This publisher has already one subscriber")); |
| } |
| } |
| |
| private final AtomicBoolean subscribed = new AtomicBoolean(); |
| |
| @Override |
| public void onSubscribe(Flow.Subscription subscription) { |
| Objects.requireNonNull(subscription); |
| if (!subscribed.compareAndSet(false, true)) { |
| subscription.cancel(); |
| } else { |
| subscriptionCF.complete(subscription); |
| } |
| } |
| |
| @Override |
| public void onNext(List<ByteBuffer> item) { |
| Objects.requireNonNull(item); |
| try { |
| // cannot be called before onSubscribe() |
| assert subscriptionCF.isDone(); |
| SubscriberRef ref = subscriberRef.get(); |
| // cannot be called before subscriber calls request(1) |
| assert ref != null; |
| Flow.Subscriber<? super List<ByteBuffer>> |
| subscriber = ref.get(); |
| if (subscriber != null) { |
| // may be null if subscription was cancelled. |
| subscriber.onNext(item); |
| } |
| } catch (Throwable err) { |
| signalError(err); |
| subscriptionCF.thenAccept(s -> s.cancel()); |
| } |
| } |
| |
| @Override |
| public void onError(Throwable throwable) { |
| // cannot be called before onSubscribe(); |
| assert suppress(subscriptionCF.isDone(), |
| "onError called before onSubscribe", |
| throwable); |
| // onError can be called before request(1), and therefore can |
| // be called before subscriberRef is set. |
| signalError(throwable); |
| Objects.requireNonNull(throwable); |
| } |
| |
| @Override |
| public void onComplete() { |
| // cannot be called before onSubscribe() |
| if (!subscriptionCF.isDone()) { |
| signalError(new InternalError( |
| "onComplete called before onSubscribed")); |
| } else { |
| // onComplete can be called before request(1), |
| // and therefore can be called before subscriberRef |
| // is set. |
| signalComplete(); |
| } |
| } |
| |
| @Override |
| public CompletionStage<Flow.Publisher<List<ByteBuffer>>> getBody() { |
| return body; |
| } |
| |
| private boolean suppress(boolean condition, |
| String assertion, |
| Throwable carrier) { |
| if (!condition) { |
| if (carrier != null) { |
| carrier.addSuppressed(new AssertionError(assertion)); |
| } else if (Log.errors()) { |
| Log.logError(new AssertionError(assertion)); |
| } |
| } |
| return true; |
| } |
| |
| } |
| |
| public static BodySubscriber<Flow.Publisher<List<ByteBuffer>>> |
| createPublisher() { |
| return new PublishingBodySubscriber(); |
| } |
| |
| |
| /** |
| * Tries to determine whether bs::getBody must be invoked asynchronously, |
| * and if so, uses the provided executor to do it. |
| * If the executor is a {@link HttpClientImpl.DelegatingExecutor}, |
| * uses the executor's delegate. |
| * @param e The executor to use if an executor is required. |
| * @param bs The BodySubscriber (trusted or not) |
| * @param <T> The type of the response. |
| * @return A completion stage that completes when the completion |
| * stage returned by bs::getBody completes. This may, or |
| * may not, be the same completion stage. |
| */ |
| public static <T> CompletionStage<T> getBodyAsync(Executor e, BodySubscriber<T> bs) { |
| if (TrustedSubscriber.needsExecutor(bs)) { |
| // getBody must be called in the executor |
| return getBodyAsync(e, bs, new MinimalFuture<>()); |
| } else { |
| // No executor needed |
| return bs.getBody(); |
| } |
| } |
| |
| /** |
| * Invokes bs::getBody using the provided executor. |
| * If invoking bs::getBody requires an executor, and the given executor |
| * is a {@link HttpClientImpl.DelegatingExecutor}, then the executor's |
| * delegate is used. If an error occurs anywhere then the given {@code cf} |
| * is completed exceptionally (this method does not throw). |
| * @param e The executor that should be used to call bs::getBody |
| * @param bs The BodySubscriber |
| * @param cf A completable future that this function will set up |
| * to complete when the completion stage returned by |
| * bs::getBody completes. |
| * In case of any error while trying to set up the |
| * completion chain, {@code cf} will be completed |
| * exceptionally with that error. |
| * @param <T> The response type. |
| * @return The provided {@code cf}. |
| */ |
| public static <T> CompletableFuture<T> getBodyAsync(Executor e, |
| BodySubscriber<T> bs, |
| CompletableFuture<T> cf) { |
| return getBodyAsync(e, bs, cf, cf::completeExceptionally); |
| } |
| |
| /** |
| * Invokes bs::getBody using the provided executor. |
| * If invoking bs::getBody requires an executor, and the given executor |
| * is a {@link HttpClientImpl.DelegatingExecutor}, then the executor's |
| * delegate is used. |
| * The provided {@code cf} is completed with the result (exceptional |
| * or not) of the completion stage returned by bs::getBody. |
| * If an error occurs when trying to set up the |
| * completion chain, the provided {@code errorHandler} is invoked, |
| * but {@code cf} is not necessarily affected. |
| * This method does not throw. |
| * @param e The executor that should be used to call bs::getBody |
| * @param bs The BodySubscriber |
| * @param cf A completable future that this function will set up |
| * to complete when the completion stage returned by |
| * bs::getBody completes. |
| * In case of any error while trying to set up the |
| * completion chain, {@code cf} will be completed |
| * exceptionally with that error. |
| * @param errorHandler The handler to invoke if an error is raised |
| * while trying to set up the completion chain. |
| * @param <T> The response type. |
| * @return The provide {@code cf}. If the {@code errorHandler} is |
| * invoked, it is the responsibility of the {@code errorHandler} to |
| * complete the {@code cf}, if needed. |
| */ |
| public static <T> CompletableFuture<T> getBodyAsync(Executor e, |
| BodySubscriber<T> bs, |
| CompletableFuture<T> cf, |
| Consumer<Throwable> errorHandler) { |
| assert errorHandler != null; |
| try { |
| assert e != null; |
| assert cf != null; |
| |
| if (TrustedSubscriber.needsExecutor(bs)) { |
| e = (e instanceof DelegatingExecutor exec) |
| ? exec::ensureExecutedAsync : e; |
| } |
| |
| e.execute(() -> { |
| try { |
| bs.getBody().whenComplete((r, t) -> { |
| if (t != null) { |
| cf.completeExceptionally(t); |
| } else { |
| cf.complete(r); |
| } |
| }); |
| } catch (Throwable t) { |
| // the errorHandler will complete the CF |
| errorHandler.accept(t); |
| } |
| }); |
| return cf; |
| |
| } catch (Throwable t) { |
| // the errorHandler will complete the CF |
| errorHandler.accept(t); |
| } |
| return cf; |
| } |
| } |
