package jdk.incubator.http;
import java.io.EOFException;
import java.lang.System.Logger.Level;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.Executor;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import jdk.incubator.http.ResponseContent.BodyParser;
import jdk.incubator.http.internal.common.Log;
import static jdk.incubator.http.HttpClient.Version.HTTP_1_1;
import jdk.incubator.http.internal.common.MinimalFuture;
import jdk.incubator.http.internal.common.Utils;
class Http1Response<T> {
private volatile ResponseContent content;
private final HttpRequestImpl request;
private Response response;
private final HttpConnection connection;
private HttpHeaders ;
private int responseCode;
private final Http1Exchange<T> exchange;
private boolean return2Cache;
private final HeadersReader ;
private final BodyReader bodyReader;
private final Http1AsyncReceiver asyncReceiver;
private volatile EOFException eof;
private final static int MAX_IGNORE = 1024;
static enum State {INITIAL, READING_HEADERS, READING_BODY, DONE}
private volatile State readProgress = State.INITIAL;
static final boolean DEBUG = Utils.DEBUG;
final System.Logger debug = Utils.getDebugLogger(this.getClass()::getSimpleName, DEBUG);
Http1Response(HttpConnection conn,
Http1Exchange<T> exchange,
Http1AsyncReceiver asyncReceiver) {
this.readProgress = State.INITIAL;
this.request = exchange.request();
this.exchange = exchange;
this.connection = conn;
this.asyncReceiver = asyncReceiver;
headersReader = new HeadersReader(this::advance);
bodyReader = new BodyReader(this::advance);
}
public CompletableFuture<Response> (Executor executor) {
debug.log(Level.DEBUG, () -> "Reading Headers: (remaining: "
+ asyncReceiver.remaining() +") " + readProgress);
asyncReceiver.unsubscribe(bodyReader);
bodyReader.reset();
Http1HeaderParser hd = new Http1HeaderParser();
readProgress = State.READING_HEADERS;
headersReader.start(hd);
asyncReceiver.subscribe(headersReader);
CompletableFuture<State> cf = headersReader.completion();
assert cf != null : "parsing not started";
Function<State, Response> lambda = (State completed) -> {
assert completed == State.READING_HEADERS;
debug.log(Level.DEBUG, () ->
"Reading Headers: creating Response object;"
+ " state is now " + readProgress);
asyncReceiver.unsubscribe(headersReader);
responseCode = hd.responseCode();
headers = hd.headers();
response = new Response(request,
exchange.getExchange(),
headers,
responseCode,
HTTP_1_1);
return response;
};
if (executor != null) {
return cf.thenApplyAsync(lambda, executor);
} else {
return cf.thenApply(lambda);
}
}
private boolean finished;
synchronized void completed() {
finished = true;
}
synchronized boolean finished() {
return finished;
}
int fixupContentLen(int clen) {
if (request.method().equalsIgnoreCase("HEAD")) {
return 0;
}
if (clen == -1) {
if (headers.firstValue("Transfer-encoding").orElse("")
.equalsIgnoreCase("chunked")) {
return -1;
}
return 0;
}
return clen;
}
public CompletableFuture<Void> ignoreBody(Executor executor) {
int clen = (int)headers.firstValueAsLong("Content-Length").orElse(-1);
if (clen == -1 || clen > MAX_IGNORE) {
connection.close();
return MinimalFuture.completedFuture(null);
} else {
return readBody(HttpResponse.BodySubscriber.discard((Void)null), true, executor);
}
}
public <U> CompletableFuture<U> readBody(HttpResponse.BodySubscriber<U> p,
boolean return2Cache,
Executor executor) {
this.return2Cache = return2Cache;
final HttpResponse.BodySubscriber<U> pusher = p;
final CompletionStage<U> bodyCF = p.getBody();
final CompletableFuture<U> cf = MinimalFuture.of(bodyCF);
int clen0 = (int)headers.firstValueAsLong("Content-Length").orElse(-1);
final int clen = fixupContentLen(clen0);
asyncReceiver.unsubscribe(headersReader);
headersReader.reset();
executor.execute(() -> {
try {
HttpClientImpl client = connection.client();
content = new ResponseContent(
connection, clen, headers, pusher,
this::onFinished
);
if (cf.isCompletedExceptionally()) {
connection.close();
return;
}
client.reference();
bodyReader.start(content.getBodyParser(
(t) -> {
try {
if (t != null) {
pusher.onError(t);
connection.close();
if (!cf.isDone())
cf.completeExceptionally(t);
}
} finally {
client.unreference();
bodyReader.onComplete(t);
}
}));
CompletableFuture<State> bodyReaderCF = bodyReader.completion();
asyncReceiver.subscribe(bodyReader);
assert bodyReaderCF != null : "parsing not started";
CompletableFuture<?> trailingOp = bodyReaderCF.whenComplete((s,t) -> {
t = Utils.getCompletionCause(t);
try {
if (t != null) {
debug.log(Level.DEBUG, () ->
"Finished reading body: " + s);
assert s == State.READING_BODY;
}
if (t != null && !cf.isDone()) {
pusher.onError(t);
cf.completeExceptionally(t);
}
} catch (Throwable x) {
asyncReceiver.onReadError(x);
}
});
connection.addTrailingOperation(trailingOp);
} catch (Throwable t) {
debug.log(Level.DEBUG, () -> "Failed reading body: " + t);
try {
if (!cf.isDone()) {
pusher.onError(t);
cf.completeExceptionally(t);
}
} finally {
asyncReceiver.onReadError(t);
}
}
});
return cf;
}
private void onFinished() {
asyncReceiver.clear();
if (return2Cache) {
Log.logTrace("Attempting to return connection to the pool: {0}", connection);
debug.log(Level.DEBUG, () -> connection.getConnectionFlow()
+ ": return to HTTP/1.1 pool");
connection.closeOrReturnToCache(eof == null ? headers : null);
}
}
HttpHeaders () {
return headers;
}
int responseCode() {
return responseCode;
}
void onReadError(Throwable t) {
Log.logError(t);
Receiver<?> receiver = receiver(readProgress);
if (t instanceof EOFException) {
debug.log(Level.DEBUG, "onReadError: received EOF");
eof = (EOFException) t;
}
CompletableFuture<?> cf = receiver == null ? null : receiver.completion();
debug.log(Level.DEBUG, () -> "onReadError: cf is "
+ (cf == null ? "null"
: (cf.isDone() ? "already completed"
: "not yet completed")));
if (cf != null && !cf.isDone()) cf.completeExceptionally(t);
else { debug.log(Level.DEBUG, "onReadError", t); }
debug.log(Level.DEBUG, () -> "closing connection: cause is " + t);
connection.close();
}
private State advance(State previous) {
assert readProgress == previous;
switch(previous) {
case READING_HEADERS:
asyncReceiver.unsubscribe(headersReader);
return readProgress = State.READING_BODY;
case READING_BODY:
asyncReceiver.unsubscribe(bodyReader);
return readProgress = State.DONE;
default:
throw new InternalError("can't advance from " + previous);
}
}
Receiver<?> receiver(State state) {
switch(state) {
case READING_HEADERS: return headersReader;
case READING_BODY: return bodyReader;
default: return null;
}
}
static abstract class Receiver<T>
implements Http1AsyncReceiver.Http1AsyncDelegate {
abstract void start(T parser);
abstract CompletableFuture<State> completion();
public abstract boolean tryAsyncReceive(ByteBuffer buffer);
public abstract void onReadError(Throwable t);
abstract void handle(ByteBuffer buf, T parser,
CompletableFuture<State> cf);
abstract void reset();
final boolean accept(ByteBuffer buf, T parser,
CompletableFuture<State> cf) {
if (cf == null || parser == null || cf.isDone()) return false;
handle(buf, parser, cf);
return !cf.isDone();
}
public abstract void onSubscribe(AbstractSubscription s);
public abstract AbstractSubscription subscription();
}
final class extends Receiver<Http1HeaderParser> {
final Consumer<State> ;
volatile Http1HeaderParser ;
volatile CompletableFuture<State> ;
volatile long ;
volatile AbstractSubscription ;
(Consumer<State> onComplete) {
this.onComplete = onComplete;
}
@Override
public AbstractSubscription () {
return subscription;
}
@Override
public void (AbstractSubscription s) {
this.subscription = s;
s.request(1);
}
@Override
void () {
cf = null;
parser = null;
count = 0;
subscription = null;
}
@Override
final void (Http1HeaderParser hp) {
count = 0;
cf = new MinimalFuture<>();
parser = hp;
}
@Override
CompletableFuture<State> () {
return cf;
}
@Override
public final boolean (ByteBuffer ref) {
boolean hasDemand = subscription.demand().tryDecrement();
assert hasDemand;
boolean needsMore = accept(ref, parser, cf);
if (needsMore) subscription.request(1);
return needsMore;
}
@Override
public final void (Throwable t) {
Http1Response.this.onReadError(t);
}
@Override
final void handle(ByteBuffer b,
Http1HeaderParser parser,
CompletableFuture<State> cf) {
assert cf != null : "parsing not started";
assert parser != null : "no parser";
try {
count += b.remaining();
debug.log(Level.DEBUG, () -> "Sending " + b.remaining()
+ "/" + b.capacity() + " bytes to header parser");
if (parser.parse(b)) {
count -= b.remaining();
debug.log(Level.DEBUG, () ->
"Parsing headers completed. bytes=" + count);
onComplete.accept(State.READING_HEADERS);
cf.complete(State.READING_HEADERS);
}
} catch (Throwable t) {
debug.log(Level.DEBUG,
() -> "Header parser failed to handle buffer: " + t);
cf.completeExceptionally(t);
}
}
}
final class BodyReader extends Receiver<BodyParser> {
final Consumer<State> onComplete;
volatile BodyParser parser;
volatile CompletableFuture<State> cf;
volatile AbstractSubscription subscription;
BodyReader(Consumer<State> onComplete) {
this.onComplete = onComplete;
}
@Override
void reset() {
parser = null;
cf = null;
subscription = null;
}
@Override
final void start(BodyParser parser) {
cf = new MinimalFuture<>();
this.parser = parser;
}
@Override
CompletableFuture<State> completion() {
return cf;
}
@Override
public final boolean tryAsyncReceive(ByteBuffer b) {
return accept(b, parser, cf);
}
@Override
public final void onReadError(Throwable t) {
Http1Response.this.onReadError(t);
}
@Override
public AbstractSubscription subscription() {
return subscription;
}
@Override
public void onSubscribe(AbstractSubscription s) {
this.subscription = s;
parser.onSubscribe(s);
}
@Override
final void handle(ByteBuffer b,
BodyParser parser,
CompletableFuture<State> cf) {
assert cf != null : "parsing not started";
assert parser != null : "no parser";
try {
debug.log(Level.DEBUG, () -> "Sending " + b.remaining()
+ "/" + b.capacity() + " bytes to body parser");
parser.accept(b);
} catch (Throwable t) {
debug.log(Level.DEBUG,
() -> "Body parser failed to handle buffer: " + t);
if (!cf.isDone()) {
cf.completeExceptionally(t);
}
}
}
final void onComplete(Throwable closedExceptionally) {
if (cf.isDone()) return;
if (closedExceptionally != null) {
cf.completeExceptionally(closedExceptionally);
} else {
onComplete.accept(State.READING_BODY);
cf.complete(State.READING_BODY);
}
}
@Override
public String toString() {
return super.toString() + "/parser=" + String.valueOf(parser);
}
}
}