package jdk.incubator.http;
import java.io.IOException;
import java.io.InputStream;
import java.lang.System.Logger.Level;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.OpenOption;
import java.nio.file.Path;
import java.security.AccessControlContext;
import java.security.AccessController;
import java.security.PrivilegedActionException;
import java.security.PrivilegedExceptionAction;
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.ConcurrentHashMap;
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.function.Consumer;
import java.util.function.Function;
import jdk.incubator.http.internal.common.MinimalFuture;
import jdk.incubator.http.internal.common.Utils;
class ResponseSubscribers {
static class ConsumerSubscriber implements HttpResponse.BodySubscriber<Void> {
private final Consumer<Optional<byte[]>> consumer;
private Flow.Subscription subscription;
private final CompletableFuture<Void> result = new MinimalFuture<>();
private final AtomicBoolean subscribed = new AtomicBoolean();
ConsumerSubscriber(Consumer<Optional<byte[]>> consumer) {
this.consumer = Objects.requireNonNull(consumer);
}
@Override
public CompletionStage<Void> getBody() {
return result;
}
@Override
public void onSubscribe(Flow.Subscription subscription) {
if (!subscribed.compareAndSet(false, true)) {
subscription.cancel();
} else {
this.subscription = subscription;
subscription.request(1);
}
}
@Override
public void onNext(List<ByteBuffer> 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) {
result.completeExceptionally(throwable);
}
@Override
public void onComplete() {
consumer.accept(Optional.empty());
result.complete(null);
}
}
static class PathSubscriber implements HttpResponse.BodySubscriber<Path> {
private final Path file;
private final CompletableFuture<Path> result = new MinimalFuture<>();
private volatile Flow.Subscription subscription;
private volatile FileChannel out;
private volatile AccessControlContext acc;
private final OpenOption[] options;
PathSubscriber(Path file, OpenOption... options) {
this.file = file;
this.options = options;
}
void setAccessControlContext(AccessControlContext acc) {
this.acc = acc;
}
@Override
public void onSubscribe(Flow.Subscription subscription) {
if (System.getSecurityManager() != null && acc == null)
throw new InternalError(
"Unexpected null acc when security manager has been installed");
this.subscription = subscription;
try {
PrivilegedExceptionAction<FileChannel> pa =
() -> FileChannel.open(file, options);
out = AccessController.doPrivileged(pa, acc);
} catch (PrivilegedActionException pae) {
Throwable t = pae.getCause() != null ? pae.getCause() : pae;
result.completeExceptionally(t);
subscription.cancel();
return;
}
subscription.request(1);
}
@Override
public void onNext(List<ByteBuffer> items) {
try {
out.write(items.toArray(Utils.EMPTY_BB_ARRAY));
} catch (IOException ex) {
Utils.close(out);
subscription.cancel();
result.completeExceptionally(ex);
}
subscription.request(1);
}
@Override
public void onError(Throwable e) {
result.completeExceptionally(e);
Utils.close(out);
}
@Override
public void onComplete() {
Utils.close(out);
result.complete(file);
}
@Override
public CompletionStage<Path> getBody() {
return result;
}
}
static class ByteArraySubscriber<T> implements HttpResponse.BodySubscriber<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;
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;
subscription.request(Long.MAX_VALUE);
}
@Override
public void onNext(List<ByteBuffer> items) {
assert Utils.hasRemaining(items);
Utils.accumulateBuffers(received, items);
}
@Override
public void onError(Throwable throwable) {
received.clear();
result.completeExceptionally(throwable);
}
static private 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;
}
}
static class HttpResponseInputStream extends InputStream
implements HttpResponse.BodySubscriber<InputStream>
{
final static boolean DEBUG = Utils.DEBUG;
final static int MAX_BUFFERS_IN_QUEUE = 1;
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 System.Logger DEBUG_LOGGER =
Utils.getDebugLogger("HttpResponseInputStream"::toString, DEBUG);
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();
HttpResponseInputStream() {
this(MAX_BUFFERS_IN_QUEUE);
}
HttpResponseInputStream(int maxBuffers) {
int capacity = (maxBuffers <= 0 ? MAX_BUFFERS_IN_QUEUE : maxBuffers);
this.buffers = new ArrayBlockingQueue<>(capacity + 1);
}
@Override
public CompletionStage<InputStream> getBody() {
return CompletableFuture.completedStage(this);
}
private ByteBuffer current() throws IOException {
while (currentBuffer == null || !currentBuffer.hasRemaining()) {
if (closed || failed != null) {
throw new IOException("closed", failed);
}
if (currentBuffer == LAST_BUFFER) break;
try {
if (currentListItr == null || !currentListItr.hasNext()) {
DEBUG_LOGGER.log(Level.DEBUG, "Taking list of Buffers");
List<ByteBuffer> lb = buffers.take();
currentListItr = lb.iterator();
DEBUG_LOGGER.log(Level.DEBUG, "List of Buffers Taken");
if (closed || failed != null)
throw new IOException("closed", failed);
if (lb == LAST_LIST) {
currentListItr = null;
currentBuffer = LAST_BUFFER;
break;
}
Flow.Subscription s = subscription;
if (s != null) {
DEBUG_LOGGER.log(Level.DEBUG, "Increased demand by 1");
s.request(1);
}
assert currentListItr != null;
if (lb.isEmpty()) continue;
}
assert currentListItr != null;
assert currentListItr.hasNext();
DEBUG_LOGGER.log(Level.DEBUG, "Next Buffer");
currentBuffer = currentListItr.next();
} catch (InterruptedException ex) {
}
}
assert currentBuffer == LAST_BUFFER || currentBuffer.hasRemaining();
return currentBuffer;
}
@Override
public int read(byte[] bytes, int off, int len) throws IOException {
ByteBuffer buffer;
if ((buffer = current()) == LAST_BUFFER) return -1;
int read = Math.min(buffer.remaining(), len);
assert read > 0 && read <= buffer.remaining();
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 void onSubscribe(Flow.Subscription s) {
try {
if (!subscribed.compareAndSet(false, true)) {
s.cancel();
} else {
boolean closed;
synchronized (this) {
closed = this.closed;
if (!closed) {
this.subscription = s;
}
}
if (closed) {
s.cancel();
return;
}
assert buffers.remainingCapacity() > 1;
DEBUG_LOGGER.log(Level.DEBUG, () -> "onSubscribe: requesting "
+ Math.max(1, buffers.remainingCapacity() - 1));
s.request(Math.max(1, buffers.remainingCapacity() - 1));
}
} catch (Throwable t) {
failed = t;
try {
close();
} catch (IOException x) {
} finally {
onError(t);
}
}
}
@Override
public void onNext(List<ByteBuffer> t) {
Objects.requireNonNull(t);
try {
DEBUG_LOGGER.log(Level.DEBUG, "next item received");
if (!buffers.offer(t)) {
throw new IllegalStateException("queue is full");
}
DEBUG_LOGGER.log(Level.DEBUG, "item offered");
} catch (Throwable ex) {
failed = ex;
try {
close();
} catch (IOException ex1) {
} finally {
onError(ex);
}
}
}
@Override
public void onError(Throwable thrwbl) {
subscription = null;
failed = Objects.requireNonNull(thrwbl);
buffers.offer(LAST_LIST);
}
@Override
public void onComplete() {
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;
}
try {
if (s != null) {
s.cancel();
}
} finally {
buffers.offer(LAST_LIST);
super.close();
}
}
}
static class MultiSubscriberImpl<V>
implements HttpResponse.MultiSubscriber<MultiMapResult<V>,V>
{
private final MultiMapResult<V> results;
private final Function<HttpRequest,Optional<HttpResponse.BodyHandler<V>>> pushHandler;
private final Function<HttpRequest,HttpResponse.BodyHandler<V>> requestHandler;
private final boolean completion;
MultiSubscriberImpl(
Function<HttpRequest,HttpResponse.BodyHandler<V>> requestHandler,
Function<HttpRequest,Optional<HttpResponse.BodyHandler<V>>> pushHandler, boolean completion) {
this.results = new MultiMapResult<>(new ConcurrentHashMap<>());
this.requestHandler = requestHandler;
this.pushHandler = pushHandler;
this.completion = completion;
}
@Override
public HttpResponse.BodyHandler<V> onRequest(HttpRequest request) {
CompletableFuture<HttpResponse<V>> cf = MinimalFuture.newMinimalFuture();
results.put(request, cf);
return requestHandler.apply(request);
}
@Override
public Optional<HttpResponse.BodyHandler<V>> onPushPromise(HttpRequest push) {
CompletableFuture<HttpResponse<V>> cf = MinimalFuture.newMinimalFuture();
results.put(push, cf);
return pushHandler.apply(push);
}
@Override
public void onResponse(HttpResponse<V> response) {
CompletableFuture<HttpResponse<V>> cf = results.get(response.request());
cf.complete(response);
}
@Override
public void onError(HttpRequest request, Throwable t) {
CompletableFuture<HttpResponse<V>> cf = results.get(request);
cf.completeExceptionally(t);
}
@Override
public CompletableFuture<MultiMapResult<V>> completion(
CompletableFuture<Void> onComplete, CompletableFuture<Void> onFinalPushPromise) {
if (completion)
return onComplete.thenApply((ignored)-> results);
else
return onFinalPushPromise.thenApply((ignored) -> results);
}
}
static class NullSubscriber<T> implements HttpResponse.BodySubscriber<T> {
private final CompletableFuture<T> cf = new MinimalFuture<>();
private final Optional<T> result;
private final AtomicBoolean subscribed = new AtomicBoolean();
NullSubscriber(Optional<T> result) {
this.result = result;
}
@Override
public void onSubscribe(Flow.Subscription 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) {
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;
}
}
static final class SubscriberAdapter<S extends Subscriber<? super List<ByteBuffer>>,R>
implements HttpResponse.BodySubscriber<R>
{
private final CompletableFuture<R> cf = new MinimalFuture<>();
private final S subscriber;
private final Function<S,R> finisher;
private volatile Subscription subscription;
SubscriberAdapter(S subscriber, Function<S,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.complete(finisher.apply(subscriber));
} catch (Throwable throwable) {
cf.completeExceptionally(throwable);
}
}
}
@Override
public CompletionStage<R> getBody() {
return cf;
}
}
}