HttpAsyncClient.h

// Copyright (c) Stefano Cristiano
// SPDX-License-Identifier: MIT
#pragma once
#include "HttpConnection.h"
#include "HttpExport.h"
#include "HttpURLParser.h"
 
namespace SC
{
 
struct HttpWebSocketTransportView;
struct SC_HTTP_EXPORT HttpAsyncClientTransportSetup
{
    HttpConnectionBase* connection = nullptr;
    AsyncEventLoop*     eventLoop  = nullptr;
 
    const HttpURLParser* url = nullptr;
 
    SocketDescriptor::Handle nativeSocket = SocketDescriptor::Invalid;
 
    Function<void(Result)> complete;
    Function<void(Result)> fail;
};
 
struct SC_HTTP_EXPORT HttpAsyncClientExternalConnection
{
    HttpConnectionBase* connection = nullptr;
    AsyncEventLoop*     eventLoop  = nullptr;
 
    const HttpURLParser* url = nullptr;
 
    Function<void(Result)> complete;
    Function<void(Result)> fail;
};
 
template <int ReadQueue, int WriteQueue, int HeaderBytes, int StreamBytes>
struct SC_HTTP_EXPORT HttpAsyncClientConnection
    : public HttpStaticConnection<ReadQueue, WriteQueue, HeaderBytes, StreamBytes, 8, HttpConnectionBase>
{
    static constexpr int ExtraBuffers = 8;
 
    HttpAsyncClientConnection()
    {
        this->readableSocketStream.setAutoDestroy(false);
        this->writableSocketStream.setAutoDestroy(false);
    }
};
 
struct SC_HTTP_EXPORT HttpAsyncClient
{
    struct Header
    {
        StringSpan name;
        StringSpan value;
    };
 
    struct RequestOptions
    {
        enum class BodyMode : uint8_t
        {
            None,
            Span,
            Stream,
            Multipart,
        };
 
        HttpParser::Method method = HttpParser::Method::HttpGET;
        StringSpan         url;
 
        Span<const Header> headers;
 
        Span<const char>     body;
        AsyncReadableStream* bodyStream      = nullptr;
        HttpMultipartWriter* multipartWriter = nullptr;
 
        uint64_t bodyLength = 0;
 
        BodyMode bodyMode  = BodyMode::None;
        bool     keepAlive = false;
 
        RequestOptions& setRequest(HttpParser::Method newMethod, StringSpan newURL, bool newKeepAlive = false)
        {
            method    = newMethod;
            url       = newURL;
            keepAlive = newKeepAlive;
            return *this;
        }
 
        RequestOptions& setHeaders(Span<const Header> newHeaders)
        {
            headers = newHeaders;
            return *this;
        }
 
        RequestOptions& setKeepAlive(bool newKeepAlive = true)
        {
            keepAlive = newKeepAlive;
            return *this;
        }
 
        RequestOptions& clearBody()
        {
            body            = {};
            bodyStream      = nullptr;
            multipartWriter = nullptr;
            bodyLength      = 0;
            bodyMode        = BodyMode::None;
            return *this;
        }
 
        RequestOptions& setBody(Span<const char> newBody)
        {
            clearBody();
            body       = newBody;
            bodyLength = newBody.sizeInBytes();
            bodyMode   = BodyMode::Span;
            return *this;
        }
 
        RequestOptions& setBody(StringSpan newBody) { return setBody(newBody.toCharSpan()); }
 
        RequestOptions& setBody(const char* newBody)
        {
            return setBody(StringSpan::fromNullTerminated(newBody, StringEncoding::Ascii));
        }
 
        RequestOptions& setBody(AsyncReadableStream& newBodyStream, uint64_t newBodyLength)
        {
            clearBody();
            bodyStream = &newBodyStream;
            bodyLength = newBodyLength;
            bodyMode   = BodyMode::Stream;
            return *this;
        }
 
        RequestOptions& setMultipart(HttpMultipartWriter& newMultipartWriter)
        {
            clearBody();
            multipartWriter = &newMultipartWriter;
            bodyMode        = BodyMode::Multipart;
            return *this;
        }
    };
 
    Result init(HttpConnectionBase& storage);
 
    Result close();
 
    void setResponseDecompression(SyncZLibTransformStream& decoder) { responseDecoder = &decoder; }
 
    void clearResponseDecompression() { responseDecoder = nullptr; }
 
    Result setTransportSetup(Function<Result(HttpAsyncClientTransportSetup&)>&& setup, const void* owner = nullptr)
    {
        SC_TRY(checkTransportMutation(owner));
        transportSetup = move(setup);
        return Result(true);
    }
 
    Result setTransportPreflight(Function<Result(const HttpURLParser&)>&& preflight, const void* owner = nullptr)
    {
        SC_TRY(checkTransportMutation(owner));
        transportPreflight = move(preflight);
        return Result(true);
    }
 
    Result setExternalConnector(Function<Result(HttpAsyncClientExternalConnection&)>&& connector,
                                const void*                                            owner = nullptr)
    {
        SC_TRY(checkTransportMutation(owner));
        externalConnector = move(connector);
        return Result(true);
    }
 
    Result setTransportClose(Function<void()>&& close, const void* owner = nullptr)
    {
        SC_TRY(checkTransportMutation(owner));
        transportClose = move(close);
        return Result(true);
    }
 
    Result setTransportShutdown(Function<Result(Function<void(Result)>)>&& shutdown, const void* owner = nullptr)
    {
        SC_TRY(checkTransportMutation(owner));
        transportShutdown = move(shutdown);
        return Result(true);
    }
 
    Result clearTransportSetup(const void* owner = nullptr);
 
    Result claimTransport(const void* owner);
 
    Result releaseTransport(const void* owner);
 
    Result detachWebSocketTransport(HttpWebSocketTransportView& transport);
 
    Result start(AsyncEventLoop& loop, HttpParser::Method method, StringSpan url, bool keepAlive = false);
 
    Result sendRequest(AsyncEventLoop& loop, const RequestOptions& options);
 
    Result get(AsyncEventLoop& loop, StringSpan url, bool keepAlive = false);
 
    Result head(AsyncEventLoop& loop, StringSpan url, bool keepAlive = false);
 
    Result options(AsyncEventLoop& loop, StringSpan url, bool keepAlive = false);
 
    Result deleteRequest(AsyncEventLoop& loop, StringSpan url, bool keepAlive = false);
 
    Result put(AsyncEventLoop& loop, StringSpan url, Span<const char> body, bool keepAlive = false);
    Result put(AsyncEventLoop& loop, StringSpan url, StringSpan body, bool keepAlive = false)
    {
        return put(loop, url, body.toCharSpan(), keepAlive);
    }
 
    Result post(AsyncEventLoop& loop, StringSpan url, Span<const char> body, bool keepAlive = false);
    Result post(AsyncEventLoop& loop, StringSpan url, StringSpan body, bool keepAlive = false)
    {
        return post(loop, url, body.toCharSpan(), keepAlive);
    }
 
    Result patch(AsyncEventLoop& loop, StringSpan url, Span<const char> body, bool keepAlive = false);
    Result patch(AsyncEventLoop& loop, StringSpan url, StringSpan body, bool keepAlive = false)
    {
        return patch(loop, url, body.toCharSpan(), keepAlive);
    }
 
    Result postMultipart(AsyncEventLoop& loop, StringSpan url, HttpMultipartWriter& writer, bool keepAlive = false);
 
    [[nodiscard]] HttpAsyncClientResponse&       getResponse() { return response; }
    [[nodiscard]] const HttpAsyncClientResponse& getResponse() const { return response; }
 
    Function<void(HttpAsyncClientRequest&)> onPrepareRequest;
 
    Function<void(HttpAsyncClientResponse&)> onResponse;
 
    Function<void(Result)> onError;
 
  private:
    Result checkTransportMutation(const void* owner) const;
    Result checkTransportQuiescence() const;
 
    struct RequestPreset
    {
        enum class BodyMode : uint8_t
        {
            None,
            Span,
            Stream,
            Multipart,
        };
 
        HttpParser::Method method = HttpParser::Method::HttpGET;
 
        StringSpan url;
 
        bool keepAlive = false;
        bool autoSend  = false;
 
        Span<const char> bodySpan;
 
        BodyMode bodyMode      = BodyMode::None;
        uint64_t contentLength = 0;
 
        Span<const Header> headers;
 
        AsyncReadableStream* bodyStream      = nullptr;
        HttpMultipartWriter* multipartWriter = nullptr;
    };
 
    enum class State : uint8_t
    {
        Idle,
        Connecting,
        Sending,
        WaitingResponse,
        StreamingResponse,
    };
 
    Result startRequest(AsyncEventLoop& loop, const RequestPreset& preset);
    Result prepareRequest(const RequestPreset& preset);
    Result startPreparedRequest(const RequestPreset& preset);
    Result ensureConnected();
    Result closeConnectionForReconnect();
    Result beginReconnectClose();
    Result continueReconnectAfterClose();
    void   clearReconnectCloseListeners();
    Result beginExternalConnection();
    Result beginSocketConnection();
    Result beginResponseRead();
    Result beginRequestSend();
    Result onResponseBodyStreamRead();
    Result validateActiveRequest() const;
 
    void                 completeTransportSetup(Result result);
    [[nodiscard]] Result rememberConnectedOrigin();
 
    void finalizeResponse(bool finishBodyStream);
    void closeConnection(bool preserveReconnectListeners = false);
    void finishResponse();
    void fail(Result error);
 
    void onReconnectReadableClosed();
    void onReconnectWritableClosed();
    void onReconnectSocketReadableClosed();
    void onReconnectSocketWritableClosed();
    void onReconnectTransportShutdown(Result result);
 
    void onConnected(AsyncSocketConnect::Result& result);
    void onReadableError(Result result);
    void onWritableError(Result result);
    void onPipelineError(Result result);
    void onReadableEnd();
    void onHeadersBufferWritten(AsyncBufferView::ID bufferID);
    void onResponseData(AsyncBufferView::ID bufferID);
    void onResponseBodyData(AsyncBufferView::ID bufferID);
    void onCompressedResponseBodyData(AsyncBufferView::ID bufferID);
    void onCompressedResponseBodyWritten(AsyncBufferView::ID bufferID);
    void onCompressedResponseBodyEnd();
    void onCompressedResponseError(Result result);
 
    [[nodiscard]] bool   canReuseConnectionFor(StringSpan protocol, StringSpan host, uint16_t port) const;
    [[nodiscard]] bool   responseMustNotHaveBody() const;
    [[nodiscard]] bool   responseHasKnownLength() const;
    [[nodiscard]] Result prepareResponseDecompression();
    [[nodiscard]] Result startResponseStreams();
    void                 detachResponseDecompression();
 
    HttpConnectionBase* connection = nullptr;
 
    AsyncEventLoop*         eventLoop      = nullptr;
    HttpAsyncClientRequest* currentRequest = nullptr;
 
    HttpAsyncClientRequest  request;
    HttpAsyncClientResponse response;
    AsyncSocketConnect      connectAsync;
    RequestPreset           currentPreset;
 
    SyncZLibTransformStream* responseDecoder       = nullptr;
    bool                     responseDecoderActive = false;
 
    Function<Result(const HttpURLParser&)>               transportPreflight;
    Function<Result(HttpAsyncClientExternalConnection&)> externalConnector;
    Function<Result(HttpAsyncClientTransportSetup&)>     transportSetup;
    Function<Result(Function<void(Result)>)>             transportShutdown;
    Function<void()>                                     transportClose;
 
    const void* transportOwner = nullptr;
 
    uint32_t transportCallbackDepth = 0;
 
    State state = State::Idle;
 
    StringSpan currentProtocol;
    char       currentProtocolStorage[16] = {0};
 
    StringSpan currentHost;
    char       currentHostStorage[256] = {0};
    uint16_t   currentPort             = 0;
 
    HttpURLParser currentURL;
    uint32_t      requestCount = 0;
 
    bool hasOpenConnection = false;
    bool responseDelivered = false;
    bool responseFinalized = false;
    bool webSocketUpgraded = false;
 
    uint8_t reconnectClosuresPending = 0;
    bool    reconnectCloseInProgress = false;
    bool    reconnectPending         = false;
 
    AsyncReadableStream* reconnectReadableStream = nullptr;
    AsyncWritableStream* reconnectWritableStream = nullptr;
 
    AsyncReadableStream* reconnectSocketReadableStream = nullptr;
    AsyncWritableStream* reconnectSocketWritableStream = nullptr;
};
 
} // namespace SC