src/java.net.http/share/classes/jdk/internal/net/http/Http1Response.java

/*
 * Copyright (c) 2015, 2025, 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.EOFException;
import java.lang.System.Logger.Level;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.net.http.HttpResponse.BodySubscriber;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.Function;
import java.net.http.HttpHeaders;
import java.net.http.HttpResponse;
import jdk.internal.net.http.ResponseContent.BodyParser;
import jdk.internal.net.http.ResponseContent.UnknownLengthBodyParser;
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 static java.net.http.HttpClient.Version.HTTP_1_1;
import static java.net.http.HttpResponse.BodySubscribers.discarding;
import static jdk.internal.net.http.common.Utils.wrapWithExtraDetail;
import static jdk.internal.net.http.RedirectFilter.HTTP_NOT_MODIFIED;

/**
 * Handles a HTTP/1.1 response (headers + body).
 * There can be more than one of these per Http exchange.
 */
class Http1Response<T> {

    private volatile ResponseContent content;
    private final HttpRequestImpl request;
    private Response response;
    private final HttpConnection connection;
    private HttpHeaders headers;
    private int responseCode;
    private final Http1Exchange<T> exchange;
    private boolean return2Cache; // return connection to cache when finished
    private final HeadersReader headersReader; // used to read the headers
    private final BodyReader bodyReader; // used to read the body
    private final Http1AsyncReceiver asyncReceiver;
    private volatile EOFException eof;
    private volatile BodyParser bodyParser;
    private volatile boolean closeWhenFinished;
    // max number of bytes of (fixed length) body to ignore on redirect
    private static final int MAX_IGNORE = 1024;

    // Revisit: can we get rid of this?
    enum State {INITIAL, READING_HEADERS, READING_BODY, DONE}
    private volatile State readProgress = State.INITIAL;

    final Logger debug = Utils.getDebugLogger(this::dbgString, Utils.DEBUG);
    static final AtomicLong responseCount = new AtomicLong();
    final long id = responseCount.incrementAndGet();
    private Http1HeaderParser hd;

    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);

        hd = new Http1HeaderParser();
        readProgress = State.READING_HEADERS;
        headersReader.start(hd);
        asyncReceiver.subscribe(headersReader);
    }

    String dbgTag;
    private String dbgString() {
        String dbg = dbgTag;
        if (dbg == null) {
            String cdbg = connection.dbgTag;
            if (cdbg != null) {
                dbgTag = dbg = "Http1Response(id=" + id + ", " + cdbg + ")";
            } else {
                dbg = "Http1Response(id=" + id + ")";
            }
        }
        return dbg;
    }

    // The ClientRefCountTracker is used to track the state
    // of a pending operation. Although there usually is a single
    // point where the operation starts, it may terminate at
    // different places.
    private static final class ClientRefCountTracker {
        final HttpClientImpl client;
        final Logger debug;
        // state & 0x01 != 0 => acquire called
        // state & 0x02 != 0 => tryRelease called
        volatile byte state;

        ClientRefCountTracker(HttpClientImpl client, Logger logger) {
            this.client = client;
            this.debug = logger;
        }

        public boolean acquire() {
            if (STATE.compareAndSet(this, (byte) 0, (byte) 0x01)) {
                // increment the reference count on the HttpClientImpl
                // to prevent the SelectorManager thread from exiting
                // until our operation is complete.
                if (debug.on())
                    debug.log("Operation started: incrementing ref count for %s", client);
                client.reference();
                return true;
            } else {
                if (debug.on())
                    debug.log("Operation ref count for %s is already %s",
                              client, ((state & 0x2) == 0x2) ? "released." : "incremented!" );
                assert (state & 0x01) == 0 : "reference count already incremented";
                return false;
            }
        }

        public void tryRelease() {
            if (STATE.compareAndSet(this, (byte) 0x01, (byte) 0x03)) {
                // decrement the reference count on the HttpClientImpl
                // to allow the SelectorManager thread to exit if no
                // other operation is pending and the facade is no
                // longer referenced.
                if (debug.on())
                    debug.log("Operation finished: decrementing ref count for %s", client);
                client.unreference();
            } else if (state == 0) {
                if (debug.on())
                    debug.log("Operation not started: releasing ref count for %s", client);
            } else if ((state & 0x02) == 0x02) {
                if (debug.on())
                    debug.log("ref count for %s already released", client);
            }
        }

        private static final VarHandle STATE;
        static {
            try {
                STATE = MethodHandles.lookup().findVarHandle(
                        ClientRefCountTracker.class, "state", byte.class);
            } catch (NoSuchFieldException | IllegalAccessException e) {
                throw new ExceptionInInitializerError(e);
            }
        }
    }

    private volatile boolean firstTimeAround = true;

    public CompletableFuture<Response> readHeadersAsync(Executor executor) {
        if (debug.on())
            debug.log("Reading Headers: (remaining: "
                      + asyncReceiver.remaining() +") "  + readProgress);

        if (firstTimeAround) {
            if (debug.on()) debug.log("First time around");
            firstTimeAround = false;
        } else {
            // with expect continue we will resume reading headers + body.
            asyncReceiver.unsubscribe(bodyReader);
            bodyReader.reset();

            hd = new Http1HeaderParser();
            readProgress = State.READING_HEADERS;
            headersReader.reset();
            headersReader.start(hd);
            asyncReceiver.subscribe(headersReader);
        }

        CompletableFuture<State> cf = headersReader.completion();
        assert cf != null : "parsing not started";
        if (debug.on()) {
            debug.log("headersReader is %s",
                    cf == null ? "not yet started"
                            : cf.isDone() ? "already completed"
                            : "not yet completed");
        }

        Function<State, Response> lambda = (State completed) -> {
                assert completed == State.READING_HEADERS;
                if (debug.on())
                    debug.log("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,
                                        connection,
                                        responseCode,
                                        HTTP_1_1);

                if (Log.headers()) {
                    StringBuilder sb = new StringBuilder("RESPONSE HEADERS:\n");
                    sb.append("  %s %s %s\n".formatted(request.method(), request.uri(), responseCode));
                    Log.dumpHeaders(sb, "    ", headers);
                    Log.logHeaders(sb.toString());
                }

                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;
    }

    /**
     * Return known fixed content length or -1 if chunked, or -2 if no content-length
     * information in which case, connection termination delimits the response body
     */
    long fixupContentLen(long clen) {
        if (request.method().equalsIgnoreCase("HEAD") || responseCode == HTTP_NOT_MODIFIED) {
            return 0L;
        }
        if (clen == -1L) {
            if (headers.firstValue("Transfer-encoding").orElse("")
                       .equalsIgnoreCase("chunked")) {
                return -1L;
            }
            if (responseCode == 101) {
                // this is a h2c or websocket upgrade, contentlength must be zero
                return 0L;
            }
            return -2L;
        }
        return clen;
    }

    /**
     * Read up to MAX_IGNORE bytes discarding
     */
    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); // not treating as error
        } else {
            return readBody(discarding(), !request.isWebSocket(), executor);
        }
    }

    // Used for those response codes that have no body associated
    public void nullBody(HttpResponse<T> resp, Throwable t) {
        if (t != null) connection.close(t);
        else {
            return2Cache = !request.isWebSocket();
            onFinished();
        }
    }


    public <U> CompletableFuture<U> readBody(HttpResponse.BodySubscriber<U> p,
                                         boolean return2Cache,
                                         Executor executor) {
        if (debug.on()) {
            debug.log("readBody: return2Cache: " + return2Cache);
            if (request.isWebSocket() && return2Cache && connection != null) {
                debug.log("websocket connection will be returned to cache: "
                        + connection.getClass() + "/" + connection);
            }
        }
        assert !return2Cache || !request.isWebSocket();
        this.return2Cache = return2Cache;
        final BodySubscriber<U> subscriber = p;


        final CompletableFuture<U> cf = new MinimalFuture<>();

        long clen0 = headers.firstValueAsLong("Content-Length").orElse(-1L);
        final long clen = fixupContentLen(clen0);

        // expect-continue reads headers and body twice.
        // if we reach here, we must reset the headersReader state.
        asyncReceiver.unsubscribe(headersReader);
        headersReader.reset();
        ClientRefCountTracker refCountTracker = new ClientRefCountTracker(connection.client(), debug);

        // We need to keep hold on the client facade until the
        // tracker has been incremented.
        connection.client().reference();
        executor.execute(() -> {
            boolean acquired = false;
            try {
                content = new ResponseContent(
                        connection, clen, headers, subscriber,
                        this::onFinished
                );
                if (cf.isCompletedExceptionally()) {
                    // if an error occurs during subscription
                    connection.close(Utils.exceptionNow(cf));
                    return;
                }
                // increment the reference count on the HttpClientImpl
                // to prevent the SelectorManager thread from exiting until
                // the body is fully read.
                acquired = refCountTracker.acquire();
                assert acquired == true;
                bodyParser = content.getBodyParser(
                    (t) -> {
                        try {
                            if (t != null) {
                                try {
                                    subscriber.onError(t);
                                } finally {
                                    cf.completeExceptionally(t);
                                }
                            }
                        } finally {
                            bodyReader.onComplete(t);
                            if (t != null) {
                                connection.close(t);
                            }
                        }
                    });
                bodyReader.start(bodyParser);
                CompletableFuture<State> bodyReaderCF = bodyReader.completion();
                asyncReceiver.subscribe(bodyReader);
                assert bodyReaderCF != null : "parsing not started";
                // Make sure to keep a reference to asyncReceiver from
                // within this
                CompletableFuture<?> trailingOp = bodyReaderCF.whenComplete((s, t) -> {
                    t = Utils.getCompletionCause(t);
                    try {
                        if (t == null) {
                            if (debug.on()) debug.log("Finished reading body: " + s);
                            assert s == State.READING_BODY;
                        }
                        if (t != null) {
                            subscriber.onError(t);
                            cf.completeExceptionally(t);
                        }
                    } catch (Throwable x) {
                        // not supposed to happen
                        asyncReceiver.onReadError(x);
                    } finally {
                        // we're done: release the ref count for
                        // the current operation.
                        refCountTracker.tryRelease();
                    }
                });
                connection.addTrailingOperation(trailingOp);
            } catch (Throwable t) {
                if (debug.on()) debug.log("Failed reading body: " + t);
                try {
                    subscriber.onError(t);
                    cf.completeExceptionally(t);
                } finally {
                    if (acquired) refCountTracker.tryRelease();
                    asyncReceiver.onReadError(t);
                }
            } finally {
                connection.client().unreference();
            }
        });

        ResponseSubscribers.getBodyAsync(executor, p, cf, (t) -> {
            subscriber.onError(t);
            cf.completeExceptionally(t);
            asyncReceiver.setRetryOnError(false);
            asyncReceiver.onReadError(t);
        });

        return cf.whenComplete((s,t) -> {
            if (t != null) {
                // If an exception occurred, release the
                // ref count for the current operation, as
                // it may never be triggered otherwise
                // (BodySubscriber ofInputStream)
                // If there was no exception then the
                // ref count will be/have been released when
                // the last byte of the response is/was received
                refCountTracker.tryRelease();
            }
        });
    }


    private void onFinished() {
        asyncReceiver.clear();
        if (closeWhenFinished) {
            if (debug.on())
                debug.log("Closing Connection when finished");
            connection.close();
        } else if (return2Cache) {
            Log.logTrace("Attempting to return connection to the pool: {0}", connection);
            // TODO: need to do something here?
            // connection.setAsyncCallbacks(null, null, null);

            // don't return the connection to the cache if EOF happened.
            if (debug.on())
                debug.log(connection.getConnectionFlow() + ": return to HTTP/1.1 pool");
            connection.closeOrReturnToCache(eof == null ? headers : null);
        }
    }

    void closeWhenFinished() {
        closeWhenFinished = true;
    }

    HttpHeaders responseHeaders() {
        return headers;
    }

    int responseCode() {
        return responseCode;
    }

// ================ Support for plugging into Http1Receiver   =================
// ============================================================================

    // Callback: Error receiver: Consumer of Throwable.
    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.completeExceptionally(t);
        } else {
            debug.log(Level.DEBUG, "onReadError", t);
        }
        debug.log(Level.DEBUG, () -> "closing connection: cause is " + t);
        connection.close(t);
    }

    // ========================================================================

    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) {
        return switch (state) {
            case READING_HEADERS    -> headersReader;
            case READING_BODY       -> bodyReader;

            default -> null;
        };

    }

    abstract static class Receiver<T>
            implements Http1AsyncReceiver.Http1AsyncDelegate {
        abstract void start(T parser);
        abstract CompletableFuture<State> completion();
        // accepts a buffer from upstream.
        // this should be implemented as a simple call to
        // accept(ref, parser, cf)
        public abstract boolean tryAsyncReceive(ByteBuffer buffer);
        public abstract void onReadError(Throwable t);
        // handle a byte buffer received from upstream.
        // this method should set the value of Http1Response.buffer
        // to ref.get() before beginning parsing.
        abstract void handle(ByteBuffer buf, T parser,
                             CompletableFuture<State> cf);
        // resets this objects state so that it can be reused later on
        // typically puts the reference to parser and completion to null
        abstract void reset();

        // accepts a byte buffer received from upstream
        // returns true if the buffer is fully parsed and more data can
        // be accepted, false otherwise.
        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();

    }

    // Invoked with each new ByteBuffer when reading headers...
    final class HeadersReader extends Receiver<Http1HeaderParser> {
        final Consumer<State> onComplete;
        volatile Http1HeaderParser parser;
        volatile CompletableFuture<State> cf;
        volatile long count; // bytes parsed (for debug)
        volatile AbstractSubscription subscription;

        HeadersReader(Consumer<State> onComplete) {
            this.onComplete = onComplete;
        }

        @Override
        public AbstractSubscription subscription() {
            return subscription;
        }

        @Override
        public void onSubscribe(AbstractSubscription s) {
            this.subscription = s;
            s.request(1);
        }

        @Override
        void reset() {
            cf = null;
            parser = null;
            count = 0;
            subscription = null;
        }

        // Revisit: do we need to support restarting?
        @Override
        final void start(Http1HeaderParser hp) {
            count = 0;
            cf = new MinimalFuture<>();
            parser = hp;
        }

        @Override
        CompletableFuture<State> completion() {
            return cf;
        }

        @Override
        public final boolean tryAsyncReceive(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 onReadError(Throwable t) {
            t = wrapWithExtraDetail(t, parser::currentStateMessage);
            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();
                if (debug.on())
                    debug.log("Sending " + b.remaining() + "/" + b.capacity()
                              + " bytes to header parser");
                if (parser.parse(b)) {
                    count -= b.remaining();
                    if (debug.on())
                        debug.log("Parsing headers completed. bytes=" + count);
                    onComplete.accept(State.READING_HEADERS);
                    cf.complete(State.READING_HEADERS);
                }
            } catch (Throwable t) {
                if (debug.on())
                    debug.log("Header parser failed to handle buffer: " + t);
                cf.completeExceptionally(t);
            }
        }

        @Override
        public void close(Throwable error) {
            // if there's no error nothing to do: the cf should/will
            // be completed.
            if (error != null) {
                CompletableFuture<State> cf = this.cf;
                if (cf != null) {
                    if (debug.on())
                        debug.log("close: completing header parser CF with " + error);
                    cf.completeExceptionally(error);
                }
            }
        }
    }

    // Invoked with each new ByteBuffer when reading bodies...
    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;
        }

        // Revisit: do we need to support restarting?
        @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) {
            BodyParser parser = bodyParser;
            if (t instanceof EOFException && parser != null &&
                    parser instanceof UnknownLengthBodyParser ulBodyParser) {
                ulBodyParser.complete();
                return;
            }
            t = wrapWithExtraDetail(t, parser::currentStateMessage);
            parser.onError(t);
            Http1Response.this.onReadError(t);
        }

        @Override
        public AbstractSubscription subscription() {
            return subscription;
        }

        @Override
        public void onSubscribe(AbstractSubscription s) {
            this.subscription = s;
            try {
                parser.onSubscribe(s);
            } catch (Throwable t) {
                cf.completeExceptionally(t);
                throw t;
            }
        }

        @Override
        final void handle(ByteBuffer b,
                          BodyParser parser,
                          CompletableFuture<State> cf) {
            assert cf != null : "parsing not started";
            assert parser != null : "no parser";
            try {
                if (debug.on())
                    debug.log("Sending " + b.remaining() + "/" + b.capacity()
                              + " bytes to body parser");
                parser.accept(b);
            } catch (Throwable t) {
                if (debug.on())
                    debug.log("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 final void close(Throwable error) {
            CompletableFuture<State> cf = this.cf;
            if (cf != null && !cf.isDone()) {
                // we want to make sure dependent actions are triggered
                // in order to make sure the client reference count
                // is decremented
                if (error != null) {
                    if (debug.on())
                        debug.log("close: completing body parser CF with " + error);
                    cf.completeExceptionally(error);
                } else {
                    if (debug.on())
                        debug.log("close: completing body parser CF");
                    cf.complete(State.READING_BODY);
                }
            }
            if (error != null) {
                // makes sure the parser gets the error
                BodyParser parser = this.parser;
                if (parser != null) {
                    if (debug.on()) {
                        debug.log("propagating error to parser: " + error);
                    }
                    parser.onError(error);
                } else {
                    if (debug.on()) {
                        debug.log("no parser - error not propagated: " + error);
                    }
                }
            }
        }

        @Override
        public String toString() {
            return super.toString() + "/parser=" + parser;
        }
    }
}