AsyncStreams.h

// Copyright (c) Stefano Cristiano
// SPDX-License-Identifier: MIT
#pragma once
 
#include "../Foundation/Function.h"
#include "../Foundation/Result.h"
#include "../Foundation/Span.h"
#include "../Foundation/StrongID.h"
#include "Internal/CircularQueue.h"
#include "Internal/Event.h"
 
 
namespace SC
{
 
struct AsyncBufferView
{
    struct Tag
    {
    };
    using ID = StrongID<Tag>;
    Span<char> data;
 
  private:
    Span<char> originalData;
    friend struct AsyncBuffersPool;
 
    int32_t refs = 0; // Counts AsyncReadable (single) or AsyncWritable (multiple) using it
};
 
struct AsyncBuffersPool
{
    Span<AsyncBufferView> buffers;
 
    void refBuffer(AsyncBufferView::ID bufferID);
 
    void unrefBuffer(AsyncBufferView::ID bufferID);
 
    Result getData(AsyncBufferView::ID bufferID, Span<const char>& data);
 
    Result getData(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);
};
 
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();
 
    void 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
        Destroying,   // Readable is waiting for async call before
        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(Span<const char> data, Function<void(AsyncBufferView::ID)> cb = {});
 
    template <size_t N>
    [[nodiscard]] Result write(const char (&str)[N])
    {
        return write(Span<const char>(str, N - 1));
    }
 
    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;
 
    AsyncPipeline()                                = default;
    AsyncPipeline(const AsyncPipeline&)            = delete;
    AsyncPipeline(AsyncPipeline&&)                 = delete;
    AsyncPipeline& operator=(const AsyncPipeline&) = delete;
    AsyncPipeline& operator=(AsyncPipeline&&)      = delete;
    ~AsyncPipeline();
    Event<MaxListeners, Result> eventError; 
 
    Result pipe(AsyncReadableStream& asyncSource, Span<AsyncWritableStream*> asyncSinks);
 
    Result pipe(AsyncReadableStream& asyncSource, Span<AsyncDuplexStream*> asyncTransforms,
                Span<AsyncWritableStream*> asyncSinks);
 
    [[nodiscard]] bool unpipe();
 
    Result start();
 
    // TODO: Add a pause and cancel/step
  private:
    AsyncReadableStream* source = nullptr; 
 
    Span<AsyncWritableStream*> sinks; 
 
    Span<AsyncDuplexStream*> transforms;
 
    void   emitError(Result res);
    Result checkBuffersPool();
    Result chainTransforms(AsyncReadableStream*& readable);
 
    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