AsyncStreams.h

// Copyright (c) Stefano Cristiano
// SPDX-License-Identifier: MIT
#pragma once
 
#include "../Foundation/AlignedStorage.h"
#include "../Foundation/Function.h"
#include "../Foundation/Internal/IGrowableBuffer.h"
#include "../Foundation/Result.h"
#include "../Foundation/Span.h"
#include "Internal/CircularQueue.h"
#include "Internal/Event.h"
 
 
namespace SC
{
struct AsyncBufferView
{
    struct ID
    {
        using NumericType = int32_t;
        NumericType                  identifier;
        static constexpr NumericType InvalidValue = -1;
 
        constexpr ID() : identifier(InvalidValue) {}
        explicit constexpr ID(int32_t value) : identifier(value) {}
 
        [[nodiscard]] constexpr bool operator==(ID other) const { return identifier == other.identifier; }
    };
    enum class Type : uint8_t
    {
        Empty,
        Writable,
        ReadOnly,
        Growable,
    };
 
    AsyncBufferView() { type = Type::Empty; }
    AsyncBufferView(Span<char> data) : writableData(data) { type = Type::Writable; }
    AsyncBufferView(Span<const char> data) : readonlyData(data) { type = Type::ReadOnly; }
 
    void setReusable(bool reusable) { reUse = reusable; }
 
    template <typename T>
    AsyncBufferView(T&& t) // universal reference, it can capture both lvalue and rvalue
    {
        type = Type::Growable;
        // Here we're type-erasing T in our own inline storage provided by a slightly oversized Function<>
        // that it will be able to construct (and destruct) the right GrowableBuffer<T> from just a piece of storage
        // and return a pointer to the corresponding IGrowableBuffer* interface
        getGrowableBuffer = [t = forward<T>(t)](GrowableStorage& storage, bool construct) mutable -> IGrowableBuffer*
        {
            using Type = typename TypeTraits::RemoveReference<T>::type;
            if (construct)
            {
                placementNew(storage.reinterpret_as<GrowableBuffer<Type>>(), t);
                return &storage.reinterpret_as<GrowableBuffer<Type>>();
            }
            else
            {
                dtor(storage.reinterpret_as<GrowableBuffer<Type>>());
                return nullptr;
            }
        };
    }
 
    template <int N>
    AsyncBufferView(const char (&literal)[N])
    {
        readonlyData = {literal, N - 1};
        type         = Type::ReadOnly;
    }
 
    Type getType() const { return type; }
 
  private:
    static constexpr int TypeErasedCaptureSize  = sizeof(void*) * 3; // This is enough to hold String / Buffer by copy
    static constexpr int TypeErasedGrowableSize = sizeof(void*) * 6;
 
    using GrowableStorage = AlignedStorage<TypeErasedGrowableSize>;
    Function<IGrowableBuffer*(GrowableStorage&, bool), TypeErasedCaptureSize> getGrowableBuffer;
 
    union
    {
        Span<char>       writableData;
        Span<const char> readonlyData;
    };
 
    union
    {
        Span<char>       originalWritableData;
        Span<const char> originalReadonlyData;
    };
    friend struct AsyncBuffersPool;
 
    int32_t refs = 0;      // Counts AsyncReadable (single) or AsyncWritable (multiple) using it
    Type    type;          // If it's Empty, Writable, ReadOnly or Growable
    bool    reUse = false; // If it can be re-used after refs == 0
};
 
struct AsyncBuffersPool
{
    Span<AsyncBufferView> buffers;
 
    void refBuffer(AsyncBufferView::ID bufferID);
 
    void unrefBuffer(AsyncBufferView::ID bufferID);
 
    Result getReadableData(AsyncBufferView::ID bufferID, Span<const char>& data);
 
    Result getWritableData(AsyncBufferView::ID bufferID, Span<char>& data);
 
    AsyncBufferView* getBuffer(AsyncBufferView::ID bufferID);
 
    Result requestNewBuffer(size_t minimumSizeInBytes, AsyncBufferView::ID& bufferID, Span<char>& data);
 
    void setNewBufferSize(AsyncBufferView::ID bufferID, size_t newSizeInBytes);
 
    Result pushBuffer(AsyncBufferView&& buffer, AsyncBufferView::ID& bufferID);
};
 
struct AsyncReadableStream
{
    struct Request
    {
        AsyncBufferView::ID bufferID;
    };
    Function<Result()> asyncRead;
 
    static constexpr int MaxListeners = 8;
 
    Event<MaxListeners, Result>              eventError; 
    Event<MaxListeners, AsyncBufferView::ID> eventData;  
    Event<MaxListeners>                      eventEnd;   
    Event<MaxListeners>                      eventClose; 
 
    Result init(AsyncBuffersPool& buffersPool, Span<Request> requests);
 
    Result start();
 
    void pause();
 
    void resumeReading();
 
    void destroy();
 
    [[nodiscard]] bool isEnded() const { return state == State::Ended; }
 
    AsyncBuffersPool& getBuffersPool();
 
    [[nodiscard]] bool push(AsyncBufferView::ID bufferID, size_t newSize);
 
    void pushEnd();
 
    void reactivate(bool doReactivate);
 
    void emitError(Result error);
 
    [[nodiscard]] bool getBufferOrPause(size_t minumumSizeInBytes, AsyncBufferView::ID& bufferID, Span<char>& data);
 
  private:
    void emitOnData();
    void executeRead();
 
    enum class State
    {
        Stopped,      // Stream must be inited
        CanRead,      // Stream is ready to issue a read ( AsyncReadableStream::start / AsyncReadableStream::resume)
        Reading,      // A read is being issued (may be sync or async)
        SyncPushing,  // One or multiple AsyncReadableStream::push have been received (sync)
        SyncReadMore, // SyncPushing + AsyncReadableStream::reactivate(true)
        AsyncReading, // An async read is in flight
        AsyncPushing, // AsyncReading + AsyncReadableStream::push
        Pausing,      // Pause requested while read in flight
        Paused,       // Actually paused with no read in flight
        Ended,        // Emitted all data, no more data will be emitted
        Destroyed,    // Readable has been destroyed before emitting all data
        Errored,      // Error occurred
    };
    State state = State::Stopped;
 
    AsyncBuffersPool* buffers = nullptr;
 
    CircularQueue<Request> readQueue;
};
 
struct AsyncWritableStream
{
    Function<Result(AsyncBufferView::ID, Function<void(AsyncBufferView::ID)>)> asyncWrite;
 
    struct Request
    {
        AsyncBufferView::ID bufferID;
 
        Function<void(AsyncBufferView::ID)> cb;
    };
    static constexpr int MaxListeners = 8;
 
    Event<MaxListeners, Result> eventError; 
 
    Event<MaxListeners> eventDrain;  
    Event<MaxListeners> eventFinish; 
 
    Result init(AsyncBuffersPool& buffersPool, Span<Request> requests);
 
    Result write(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)> cb = {});
 
    Result write(AsyncBufferView&& bufferView, Function<void(AsyncBufferView::ID)> cb = {});
 
    void end();
 
    AsyncBuffersPool& getBuffersPool();
 
    void finishedWriting(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)>&& cb, Result res);
 
    void resumeWriting();
 
    Result unshift(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)>&& cb);
 
    void emitError(Result error);
 
    Function<bool()> canEndWritable;
 
    void tryAsync(Result potentialError);
 
    void stop() { state = State::Stopped; }
 
  private:
    enum class State
    {
        Stopped,
        Writing,
        Ending,
        Ended
    };
    State state = State::Stopped;
 
    AsyncBuffersPool* buffers = nullptr;
 
    CircularQueue<Request> writeQueue;
};
 
struct AsyncDuplexStream : public AsyncReadableStream, public AsyncWritableStream
{
    AsyncDuplexStream();
 
    Result init(AsyncBuffersPool& buffersPool, Span<AsyncReadableStream::Request> readableRequests,
                Span<AsyncWritableStream::Request> writableRequests);
};
 
struct AsyncTransformStream : public AsyncDuplexStream
{
    AsyncTransformStream();
 
    void afterProcess(Span<const char> inputAfter, Span<char> outputAfter);
    void afterFinalize(Span<char> outputAfter, bool streamEnded);
 
    Function<Result(Span<const char>, Span<char>)> onProcess;
    Function<Result(Span<char>)>                   onFinalize;
 
  private:
    Function<void(AsyncBufferView::ID)> inputCallback;
 
    Span<const char> inputData;
    Span<char>       outputData;
 
    AsyncBufferView::ID inputBufferID;
    AsyncBufferView::ID outputBufferID;
 
    Result transform(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)> cb);
    Result prepare(AsyncBufferView::ID bufferID, Function<void(AsyncBufferView::ID)> cb);
 
    bool canEndTransform();
    void tryFinalize();
 
    enum class State
    {
        None,
        Paused,
        Processing,
        Finalizing,
        Finalized,
    };
    State state = State::None;
};
 
struct AsyncPipeline
{
    static constexpr int MaxListeners  = 8;
    static constexpr int MaxTransforms = 8;
    static constexpr int MaxSinks      = 8;
 
    AsyncPipeline()                                = default;
    AsyncPipeline(const AsyncPipeline&)            = delete;
    AsyncPipeline(AsyncPipeline&&)                 = delete;
    AsyncPipeline& operator=(const AsyncPipeline&) = delete;
    AsyncPipeline& operator=(AsyncPipeline&&)      = delete;
    ~AsyncPipeline();
 
    AsyncReadableStream* source                    = nullptr;   
    AsyncDuplexStream*   transforms[MaxTransforms] = {nullptr}; 
    AsyncWritableStream* sinks[MaxSinks]           = {nullptr}; 
    Event<MaxListeners, Result> eventError         = {};        
 
    Result pipe();
 
    [[nodiscard]] bool unpipe();
 
    Result start();
 
    // TODO: Add a pause and cancel/step
  private:
    void   emitError(Result res);
    Result checkBuffersPool();
    Result chainTransforms(AsyncReadableStream*& readable);
    Result validate();
 
    void asyncWriteWritable(AsyncBufferView::ID bufferID, AsyncWritableStream& writable);
    void dispatchToPipes(AsyncBufferView::ID bufferID);
    void endPipes();
    void afterWrite(AsyncBufferView::ID bufferID);
    bool listenToEventData(AsyncReadableStream& readable, AsyncDuplexStream& transform, bool listen);
};
} // namespace SC