HttpConnection.h

// Copyright (c) Stefano Cristiano
// SPDX-License-Identifier: MIT
#pragma once
#include "HttpExport.h"
#include "HttpParser.h"
 
#include "../Async/Async.h"
#include "../AsyncStreams/AsyncRequestStreams.h"
#include "../Foundation/StringSpan.h"
#include "Internal/HttpFixedBufferWriter.h"
#include "Internal/HttpParsedHeaders.h"
 
namespace SC
{
 
enum class HttpBodyFramingKind : uint8_t
{
    None,
    ContentLength,
    Chunked,
    CloseDelimited,
};
 
struct HttpAsyncClient;
 
struct SC_HTTP_EXPORT HttpMultipartWriter
{
    struct Part
    {
        StringSpan partName;
        StringSpan fileName;
        StringSpan contentType;
 
        Span<const char> body;
    };
 
    static constexpr size_t MaxParts = 8;
 
    void reset();
 
    Result setBoundary(StringSpan boundaryValue);
    Result addField(StringSpan fieldName, StringSpan value);
    Result addFile(StringSpan fieldName, StringSpan fileName, Span<const char> body,
                   StringSpan contentType = "application/octet-stream");
 
    [[nodiscard]] StringSpan  getBoundary() const { return boundary; }
    [[nodiscard]] size_t      getNumParts() const { return numParts; }
    [[nodiscard]] const Part& getPart(size_t idx) const { return parts[idx]; }
    [[nodiscard]] size_t      getContentLength() const;
 
  private:
    Part       parts[MaxParts];
    size_t     numParts            = 0;
    char       boundaryStorage[71] = {0};
    StringSpan boundary;
};
 
struct SC_HTTP_EXPORT HttpConnectionBase
{
    using ReadableSocketStream = AsyncReadableSocketStream<AsyncEventLoop>;
    using WritableSocketStream = AsyncWritableSocketStream<AsyncEventLoop>;
 
    ReadableSocketStream readableSocketStream;
    WritableSocketStream writableSocketStream;
    AsyncBuffersPool     buffersPool;
    AsyncPipeline        pipeline;
    SocketDescriptor     socket;
 
    void setHeaderMemory(Span<char> memory) { headerMemory = memory; }
 
    [[nodiscard]] Span<char> getHeaderMemory() const { return headerMemory; }
 
    void reset();
 
  protected:
    Span<char> headerMemory;
};
 
struct SC_HTTP_EXPORT HttpIncomingMessage
{
    struct SC_HTTP_EXPORT BodyStream : public AsyncReadableStream
    {
        BodyStream();
 
        Result begin(AsyncBuffersPool& buffersPool, Function<Result()>&& onReadRequest);
        bool   pushBodyData(AsyncBufferView::ID bufferID, size_t sizeInBytes);
        void   finishBody();
        void   failBody(Result result);
 
        Function<Result()> onReadRequest;
        Request            queue[4];
 
      private:
        virtual Result asyncRead() override;
    };
 
    const HttpParser& getParser() const { return parsedHeaders.parser; }
 
    [[nodiscard]] bool getKeepAlive() const { return parsedHeaders.parser.connectionKeepAlive; }
 
    [[nodiscard]] HttpBodyFramingKind getBodyFramingKind() const { return bodyFramingKind; }
 
    [[nodiscard]] uint64_t getBodyBytesRemaining() const { return bodyBytesRemaining; }
 
    [[nodiscard]] bool isBodyComplete() const { return bodyComplete; }
 
    [[nodiscard]] bool isMultipart() const;
 
    [[nodiscard]] StringSpan getBoundary() const;
 
    [[nodiscard]] bool getHeader(StringSpan headerName, StringSpan& value) const;
 
    AsyncReadableStream& getReadableStream();
 
    const AsyncReadableStream& getReadableStream() const;
 
    Result consumeBodyBytes(size_t bytes);
 
  protected:
    void resetIncoming(HttpParser::Type type, Span<char> memory);
 
    [[nodiscard]] bool       hasReceivedHeaders() const { return parsedHeaders.headersEndReceived; }
    [[nodiscard]] Span<char> getUnusedHeaderMemory() const { return parsedHeaders.availableHeader; }
 
    Result initBodyStream(AsyncBuffersPool& buffersPool, Function<Result()>&& onReadRequest);
    Result startBodyStream();
    bool   pushBodyData(AsyncBufferView::ID bufferID, size_t sizeInBytes);
    void   finishBodyStream();
    void   failBodyStream(Result result);
    void   abortBodyStream();
    Result prepareBodyStream(AsyncBuffersPool& buffersPool, Function<Result()>&& onReadRequest,
                             bool allowCloseDelimited);
    Result processBodyData(AsyncReadableStream& sourceStream, AsyncBufferView::ID bufferID, Span<const char> readData,
                           bool allowTrailingData);
 
    void setBodyBytesRemaining(uint64_t value) { bodyBytesRemaining = value; }
    void setBodyFramingKind(HttpBodyFramingKind value) { bodyFramingKind = value; }
    void setBodyComplete(bool value) { bodyComplete = value; }
    void attachReadableStream(AsyncReadableStream& stream) { readableStream = &stream; }
 
    [[nodiscard]] bool findParserToken(HttpParser::Token token, StringSpan& res) const;
 
    Result writeHeaders(const uint32_t maxHeaderSize, Span<const char> readData, AsyncReadableStream& stream,
                        AsyncBufferView::ID bufferID, const char* outOfSpaceError, const char* sizeExceededError,
                        bool stopAtHeadersEnd, bool unshiftPendingBodyToStream = true);
 
    [[nodiscard]] size_t getHeadersLength() const;
 
    enum class ChunkedState : uint8_t
    {
        Size,
        SizeExtension,
        SizeLF,
        Data,
        DataCR,
        DataLF,
        TrailerLineStart,
        TrailerEndLF,
        Finished,
    };
 
    HttpParsedHeaders    parsedHeaders;
    Span<char>           headerMemory;
    AsyncReadableStream* readableStream     = nullptr;
    uint64_t             bodyBytesRemaining = 0;
    HttpBodyFramingKind  bodyFramingKind    = HttpBodyFramingKind::None;
    bool                 bodyComplete       = true;
    bool                 bodyStreamStarted  = false;
    BodyStream           bodyStream;
    ChunkedState         chunkedState          = ChunkedState::Size;
    uint64_t             chunkedChunkSize      = 0;
    uint64_t             chunkedBytesRemaining = 0;
    bool                 chunkedSizeHasDigits  = false;
};
 
struct SC_HTTP_EXPORT HttpRequest : public HttpIncomingMessage
{
    StringSpan getURL() const { return url; }
 
    void reset();
 
  private:
    friend struct HttpConnectionsPool;
    friend struct HttpResponse;
    friend struct HttpAsyncServer;
 
    void setHeaderMemory(Span<char> memory);
 
    Result writeHeaders(const uint32_t maxHeaderSize, Span<const char> readData, AsyncReadableStream& stream,
                        AsyncBufferView::ID bufferID);
 
    StringSpan url;
};
 
struct SC_HTTP_EXPORT HttpAsyncClientResponse : public HttpIncomingMessage
{
  private:
    friend struct HttpAsyncClient;
 
    void   reset(Span<char> memory);
    Result writeHeaders(uint32_t maxHeaderSize, Span<const char> readData, AsyncReadableStream& stream,
                        AsyncBufferView::ID bufferID);
};
 
struct SC_HTTP_EXPORT HttpOutgoingMessage
{
    struct SC_HTTP_EXPORT ChunkedWritableStream : public AsyncWritableStream
    {
        ChunkedWritableStream();
 
        Result init(AsyncBuffersPool& buffersPool, AsyncWritableStream& destination);
 
      protected:
        virtual Result asyncWrite(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)> cb) override;
        virtual bool   canEndWritable() override;
 
      private:
        void onChunkHeaderWritten(AsyncBufferView::ID);
        void onChunkBodyWritten(AsyncBufferView::ID);
        void onChunkTerminatorWritten(AsyncBufferView::ID);
        void onFinalChunkWritten(AsyncBufferView::ID);
 
        AsyncWritableStream* destination = nullptr;
 
        AsyncBufferView::ID                 currentBodyBufferID;
        Function<void(AsyncBufferView::ID)> currentBodyCallback;
 
        bool finalChunkStarted = false;
        bool finalChunkWritten = false;
 
        char headerStorage[20] = {0};
 
        Request queue[4];
    };
 
    Result addHeader(StringSpan headerName, StringSpan headerValue);
 
    Result setChunkedTransferEncoding();
 
    Result sendHeaders(Function<void(AsyncBufferView::ID)> callback = {});
 
    void reset();
 
    Result end();
 
    AsyncWritableStream& getWritableStream() { return *writableStream; }
 
    void setKeepAlive(bool keepAlive);
 
    [[nodiscard]] bool getKeepAlive() const { return forceDisableKeepAlive ? false : keepAlive; }
 
  protected:
    enum class KnownHeader : uint8_t
    {
        Connection,
        Host,
        UserAgent,
        ContentLength,
        ContentType,
        TransferEncoding,
    };
 
    void setHeaderMemory(Span<char> memory);
    void setWritableStream(AsyncWritableStream& stream)
    {
        destinationStream = &stream;
        writableStream    = &stream;
    }
 
    [[nodiscard]] bool hasHeader(KnownHeader header) const;
    [[nodiscard]] bool hasSentHeaders() const { return headersSent; }
    [[nodiscard]] bool hasEnded() const { return endCalled; }
    [[nodiscard]] bool isChunkedTransferEncodingEnabled() const { return chunkedTransferEncodingEnabled; }
 
    HttpFixedBufferWriter responseHeaders;
    Span<char>            headerMemory;
 
    bool headersSent = false;
    bool endCalled   = false;
 
    bool forceDisableKeepAlive          = false; 
    bool keepAlive                      = true;  
    bool connectionHeaderAdded          = false; 
    bool hostHeaderAdded                = false;
    bool userAgentHeaderAdded           = false;
    bool contentLengthAdded             = false;
    bool contentTypeAdded               = false;
    bool transferEncodingAdded          = false;
    bool chunkedTransferEncodingEnabled = false;
 
    AsyncWritableStream*  destinationStream = nullptr;
    AsyncWritableStream*  writableStream    = nullptr;
    ChunkedWritableStream chunkedWritableStream;
};
 
struct SC_HTTP_EXPORT HttpResponse : public HttpOutgoingMessage
{
    Result startResponse(int httpCode);
 
  private:
    friend struct HttpConnectionsPool;
    friend struct HttpAsyncServer;
 
    void grabUnusedHeaderMemory(HttpRequest& request);
};
 
struct SC_HTTP_EXPORT HttpAsyncClientRequest : public HttpOutgoingMessage
{
    enum class BodyType : uint8_t
    {
        None,
        Manual,
        Span,
        Stream,
        Multipart,
    };
 
    void reset();
 
    Result startRequest(HttpParser::Method method, StringSpan url);
    Result setExpectedBodyLength(uint64_t value);
    Result setBody(Span<const char> value);
    Result setBody(StringSpan value) { return setBody(value.toCharSpan()); }
    Result setBody(AsyncReadableStream& stream);
    void   setBody(AsyncReadableStream& stream, uint64_t contentLengthValue);
    void   setMultipart(HttpMultipartWriter& value);
    Result sendHeaders(Function<void(AsyncBufferView::ID)> callback = {});
 
    [[nodiscard]] HttpParser::Method getMethod() const { return method; }
 
    [[nodiscard]] StringSpan getURL() const { return url; }
    [[nodiscard]] BodyType   getBodyType() const { return bodyType; }
    [[nodiscard]] uint64_t   getContentLength() const { return contentLength; }
    [[nodiscard]] bool       usesChunkedTransferEncoding() const { return chunkedTransferEncodingEnabled; }
 
  private:
    friend struct HttpAsyncClient;
 
    void setDefaultHost(StringSpan value) { defaultHost = value; }
 
    [[nodiscard]] bool hasTransferEncodingHeader() const { return hasHeader(KnownHeader::TransferEncoding); }
 
    [[nodiscard]] AsyncReadableStream*       getBodyStream() const { return bodyStream; }
    [[nodiscard]] const HttpMultipartWriter* getMultipartWriter() const { return multipartWriter; }
    [[nodiscard]] Span<const char>           getBodySpan() const { return bodySpan; }
 
    HttpParser::Method   method = HttpParser::Method::HttpGET;
    StringSpan           url;
    AsyncReadableStream* bodyStream = nullptr;
    BodyType             bodyType   = BodyType::None;
    Span<const char>     bodySpan;
    uint64_t             contentLength   = 0;
    HttpMultipartWriter* multipartWriter = nullptr;
    StringSpan           defaultHost;
 
    Function<void(AsyncBufferView::ID)> userHeadersSentCallback;
    Function<void(AsyncBufferView::ID)> internalHeadersSentCallback;
};
 
struct SC_HTTP_EXPORT HttpConnection : public HttpConnectionBase
{
    HttpConnection();
 
    struct SC_HTTP_EXPORT ID
    {
        size_t getIndex() const { return index; }
 
      private:
        friend struct HttpConnectionsPool;
        size_t index = 0;
    };
 
    void reset();
 
    ID getConnectionID() const { return connectionID; }
 
    HttpRequest  request;
    HttpResponse response;
 
    uint32_t requestCount = 0; 
 
  protected:
    enum class State
    {
        Inactive,
        Active
    };
    friend struct HttpConnectionsPool;
    friend struct HttpAsyncServer;
 
    State state = State::Inactive;
    ID    connectionID;
};
 
template <typename Type>
struct SC_HTTP_EXPORT SpanWithStride
{
    constexpr SpanWithStride() : data(nullptr), sizeElements(0), strideInBytes(0) {}
 
    constexpr SpanWithStride(void* data, size_t sizeInElements, size_t strideInBytes)
        : data(data), sizeElements(sizeInElements), strideInBytes(strideInBytes)
    {}
 
    template <typename U>
    constexpr SpanWithStride(Span<U> span)
        : data(span.data()), sizeElements(span.sizeInElements()), strideInBytes(sizeof(U))
    {}
 
    [[nodiscard]] constexpr size_t sizeInElements() const { return sizeElements; }
    [[nodiscard]] constexpr bool   empty() const { return sizeElements == 0; }
 
    template <typename U>
    SpanWithStride<U> castTo()
    {
        return {static_cast<U*>(reinterpret_cast<Type*>(data)), sizeElements, strideInBytes};
    }
 
    [[nodiscard]] Type& operator[](size_t idx)
    {
        return *reinterpret_cast<Type*>(reinterpret_cast<char*>(data) + idx * strideInBytes);
    }
 
    [[nodiscard]] const Type& operator[](size_t idx) const
    {
        return *reinterpret_cast<const Type*>(reinterpret_cast<const char*>(data) + idx * strideInBytes);
    }
 
  private:
    void*  data;
    size_t sizeElements;
    size_t strideInBytes;
};
 
struct SC_HTTP_EXPORT HttpConnectionsPool
{
    struct Configuration
    {
        size_t readQueueSize     = 3;
        size_t writeQueueSize    = 3;
        size_t buffersQueueSize  = 6;
        size_t headerBytesLength = 8 * 1024;
        size_t streamBytesLength = 512 * 1024;
    };
 
    struct SC_HTTP_EXPORT Memory
    {
        Span<AsyncReadableStream::Request> allReadQueue;
        Span<AsyncWritableStream::Request> allWriteQueue;
 
        Span<AsyncBufferView> allBuffers;
 
        Span<char> allHeaders;
        Span<char> allStreams;
 
        Result assignTo(HttpConnectionsPool::Configuration conf, SpanWithStride<HttpConnection> connections);
    };
 
    Result init(SpanWithStride<HttpConnection> connectionsStorage);
 
    Result close();
 
    [[nodiscard]] size_t getNumActiveConnections() const { return numConnections; }
 
    [[nodiscard]] size_t getNumTotalConnections() const { return connections.sizeInElements(); }
 
    [[nodiscard]] size_t getHighestActiveConnection() const { return highestActiveConnection; }
 
    [[nodiscard]] HttpConnection& getConnection(HttpConnection::ID connectionID)
    {
        return connections[connectionID.index];
    }
 
    [[nodiscard]] HttpConnection& getConnectionAt(size_t idx) { return connections[idx]; }
 
    [[nodiscard]] bool activateNew(HttpConnection::ID& connectionID);
 
    [[nodiscard]] bool deactivate(HttpConnection::ID connectionID);
 
  private:
    SpanWithStride<HttpConnection> connections; 
 
    size_t numConnections = 0; 
 
    // Optimization for HttpAsyncServer::resize to avoid having to scan all connections
    size_t highestActiveConnection = 0;
};
 
template <int ReadQueue, int WriteQueue, int HeaderBytes, int StreamBytes, int ExtraBuffers, typename BaseClass>
struct SC_HTTP_EXPORT HttpStaticConnection : public BaseClass
{
    AsyncReadableStream::Request readQueue[ReadQueue];
    AsyncWritableStream::Request writeQueue[WriteQueue];
 
    AsyncBufferView buffers[ReadQueue + WriteQueue + ExtraBuffers];
 
    char headerStorage[HeaderBytes];
    char streamStorage[StreamBytes];
 
    constexpr HttpStaticConnection()
    {
        constexpr const size_t NumSlices   = ReadQueue;
        constexpr const size_t SliceLength = StreamBytes / NumSlices;
 
        Span<char> memory = streamStorage;
        for (size_t idx = 0; idx < NumSlices; ++idx)
        {
            Span<char> slice;
            (void)memory.sliceStartLength(idx * SliceLength, SliceLength, slice);
            buffers[idx] = slice;
            buffers[idx].setReusable(true);
        }
        this->setHeaderMemory(headerStorage);
        this->buffersPool.setBuffers(buffers);
        this->readableSocketStream.setReadQueue(readQueue);
        this->writableSocketStream.setWriteQueue(writeQueue);
    }
};
 
} // namespace SC