HttpAsyncServer.h

// Copyright (c) Stefano Cristiano
// SPDX-License-Identifier: MIT
#pragma once
#include "../AsyncStreams/AsyncRequestStreams.h"
#include "HttpConnection.h"
 
namespace SC
{
namespace TypeTraits
{
template <typename Base, typename Derived>
struct IsBaseOf
{
    static constexpr bool value = __is_base_of(Base, Derived);
};
 
} // namespace TypeTraits
 
struct SC_COMPILER_EXPORT HttpAsyncConnectionBase : public HttpConnection
{
    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_COMPILER_EXPORT Memory
    {
        Span<AsyncReadableStream::Request> allReadQueue;
        Span<AsyncWritableStream::Request> allWriteQueue;
        Span<AsyncBufferView>              allBuffers;
        Span<char>                         allHeaders;
        Span<char>                         allStreams;
 
        Result assignTo(HttpAsyncConnectionBase::Configuration  conf,
                        SpanWithStride<HttpAsyncConnectionBase> connections);
    };
    ReadableSocketStream readableSocketStream;
    WritableSocketStream writableSocketStream;
    SocketDescriptor     socket;
    uint32_t             requestCount = 0; 
};
 
template <int ReadQueue, int WriteQueue, int HeaderBytes, int StreamBytes>
struct SC_COMPILER_EXPORT HttpAsyncConnection : public HttpAsyncConnectionBase
{
    AsyncReadableStream::Request readQueue[ReadQueue];
    AsyncWritableStream::Request writeQueue[WriteQueue];
    AsyncBufferView              buffers[ReadQueue + WriteQueue];
 
    char headerStorage[HeaderBytes];
    char streamStorage[StreamBytes];
 
    constexpr HttpAsyncConnection()
    {
        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);
        }
        HttpConnection::setHeaderMemory(headerStorage);
        HttpConnection::buffersPool.setBuffers(buffers);
        readableSocketStream.setReadQueue(readQueue);
        writableSocketStream.setWriteQueue(writeQueue);
    }
};
 
struct SC_COMPILER_EXPORT HttpAsyncServer
{
    template <typename T,
              typename = typename TypeTraits::EnableIf<TypeTraits::IsBaseOf<HttpAsyncConnectionBase, T>::value>::type>
    Result init(Span<T> clients)
    {
        return initInternal({clients.data(), clients.sizeInElements(), sizeof(T)});
    }
 
    template <typename T,
              typename = typename TypeTraits::EnableIf<TypeTraits::IsBaseOf<HttpAsyncConnectionBase, T>::value>::type>
    Result resize(Span<T> clients)
    {
        return resizeInternal({clients.data(), clients.sizeInElements(), sizeof(T)});
    }
 
    Result close();
 
    Result start(AsyncEventLoop& loop, StringSpan address, uint16_t port);
 
    Result stop();
 
    [[nodiscard]] bool isStarted() const { return state == State::Started; }
 
    [[nodiscard]] const HttpConnectionsPool& getConnections() const { return connections; }
 
    Function<void(HttpConnection&)> onRequest;
 
    void setDefaultKeepAlive(bool enabled) { defaultKeepAlive = enabled; }
 
    [[nodiscard]] bool getDefaultKeepAlive() const { return defaultKeepAlive; }
 
    void setMaxRequestsPerConnection(uint32_t maxRequests) { maxRequestsPerConnection = maxRequests; }
 
    [[nodiscard]] uint32_t getMaxRequestsPerConnection() const { return maxRequestsPerConnection; }
 
  private:
    HttpConnectionsPool connections;
 
    uint32_t maxHeaderSize = 8 * 1024;
 
    enum class State
    {
        Stopped,  // Server was not started at all, or it was stopped and wait(ed)ForStopToFinish
        Started,  // Server has been started (successfully)
        Stopping, // Server has stop() called and needs waitForStopToFinish() call
    };
 
    State state = State::Stopped;
 
    bool     defaultKeepAlive         = true; 
    uint32_t maxRequestsPerConnection = 0;    
 
    void onNewClient(AsyncSocketAccept::Result& result);
    void closeAsync(HttpAsyncConnectionBase& requestClient);
    void deactivateConnection(HttpAsyncConnectionBase& requestClient);
    void onStreamReceive(HttpAsyncConnectionBase& client, AsyncBufferView::ID bufferID);
 
    Result waitForStopToFinish();
    Result initInternal(SpanWithStride<HttpAsyncConnectionBase> connections);
    Result resizeInternal(SpanWithStride<HttpAsyncConnectionBase> connections);
 
    void onAsyncRequestStopped(AsyncResult& result);
 
    Function<void(AsyncResult&)> onAsyncRequestStoppedFunc;
 
    AsyncEventLoop*   eventLoop = nullptr;
    SocketDescriptor  serverSocket;
    AsyncSocketAccept asyncServerAccept;
 
    struct EventDataListener;
    struct EventEndListener;
};
} // namespace SC