package jdk.incubator.http;
import java.security.AccessControlContext;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import jdk.incubator.http.HttpResponse.BodyHandler;
import jdk.incubator.http.HttpResponse.UntrustedBodyHandler;
import jdk.incubator.http.internal.common.MinimalFuture;
import jdk.incubator.http.internal.common.Log;
class PushGroup<U,T> {
final CompletableFuture<Void> resultCF;
final CompletableFuture<Void> noMorePushesCF;
volatile Throwable error;
final CompletableFuture<HttpResponse<T>> mainResponse;
final HttpResponse.MultiSubscriber<U, T> multiSubscriber;
final HttpResponse.BodyHandler<T> mainBodyHandler;
private final AccessControlContext acc;
int numberOfPushes;
int remainingPushes;
boolean noMorePushes = false;
PushGroup(HttpResponse.MultiSubscriber<U, T> multiSubscriber,
HttpRequestImpl req,
AccessControlContext acc) {
this(multiSubscriber, req, new MinimalFuture<>(), acc);
}
private PushGroup(HttpResponse.MultiSubscriber<U, T> multiSubscriber,
HttpRequestImpl req,
CompletableFuture<HttpResponse<T>> mainResponse,
AccessControlContext acc) {
this(multiSubscriber,
mainResponse,
multiSubscriber.onRequest(req),
acc);
}
private PushGroup(HttpResponse.MultiSubscriber<U, T> multiSubscriber,
CompletableFuture<HttpResponse<T>> mainResponse,
HttpResponse.BodyHandler<T> mainBodyHandler,
AccessControlContext acc) {
assert mainResponse != null;
assert mainBodyHandler != null;
this.resultCF = new MinimalFuture<>();
this.noMorePushesCF = new MinimalFuture<>();
this.multiSubscriber = multiSubscriber;
this.mainResponse = mainResponse.thenApply(r -> {
multiSubscriber.onResponse(r);
return r;
});
this.mainBodyHandler = mainBodyHandler;
if (acc != null) {
if (mainBodyHandler instanceof UntrustedBodyHandler)
((UntrustedBodyHandler)this.mainBodyHandler).setAccessControlContext(acc);
}
this.acc = acc;
}
CompletableFuture<Void> groupResult() {
return resultCF;
}
HttpResponse.MultiSubscriber<U, T> subscriber() {
return multiSubscriber;
}
Optional<BodyHandler<T>> handlerForPushRequest(HttpRequest ppRequest) {
Optional<BodyHandler<T>> bh = multiSubscriber.onPushPromise(ppRequest);
if (acc != null && bh.isPresent()) {
BodyHandler<T> x = bh.get();
if (x instanceof UntrustedBodyHandler)
((UntrustedBodyHandler)x).setAccessControlContext(acc);
bh = Optional.of(x);
}
return bh;
}
HttpResponse.BodyHandler<T> mainResponseHandler() {
return mainBodyHandler;
}
synchronized void setMainResponse(CompletableFuture<HttpResponse<T>> r) {
r.whenComplete((HttpResponse<T> response, Throwable t) -> {
if (t != null)
mainResponse.completeExceptionally(t);
else
mainResponse.complete(response);
});
}
synchronized void addPush() {
numberOfPushes++;
remainingPushes++;
}
synchronized void noMorePushes(boolean noMore) {
noMorePushes = noMore;
checkIfCompleted();
noMorePushesCF.complete(null);
}
CompletableFuture<Void> pushesCF() {
return noMorePushesCF;
}
synchronized boolean noMorePushes() {
return noMorePushes;
}
synchronized void pushCompleted() {
remainingPushes--;
checkIfCompleted();
}
synchronized void checkIfCompleted() {
if (Log.trace()) {
Log.logTrace("PushGroup remainingPushes={0} error={1} noMorePushes={2}",
remainingPushes,
(error==null)?error:error.getClass().getSimpleName(),
noMorePushes);
}
if (remainingPushes == 0 && error == null && noMorePushes) {
if (Log.trace()) {
Log.logTrace("push completed");
}
resultCF.complete(null);
}
}
synchronized void pushError(Throwable t) {
if (t == null) {
return;
}
this.error = t;
resultCF.completeExceptionally(t);
}
}