AuroraRuntime/Source/IO/AuIOBufferedProcessor.cpp

/***
    Copyright (C) 2022 J Reece Wilson (a/k/a "Reece"). All rights reserved.

    File: AuIOBufferedProcessor.cpp
    Date: 2022-6-6
    Author: Reece
***/
#include <Source/RuntimeInternal.hpp>
#include <Aurora/IO/IOExperimental.hpp>
#include "AuIOBufferedProcessor.hpp"

namespace Aurora::IO
{
    struct IOBufferedProcessor : IIOBufferedProcessor
    {
        AuSPtr<IStreamReader> pSource;
        AuSPtr<IStreamWriter> pDrain;
        AuSPtr<IIOPipeInterceptor> pProcessor;
        AuUInt32 uBufferSize {};
        AuByteBuffer buffer;

        AU_DEFINE_FOR_VA(IOBufferedProcessor,
                        (AU_DEFINE_CTOR_VA,             // initializer-list-like ctor (extending a struct or adding a ctor will break initializer lists)
                         AU_DEFINE_THIS_MOVE_CTOR_VA,   // add move `Object(Object &&)`
                         AU_DEFINE_EQUALS_VA,           // add equals operator
                         AU_DEFINE_MOVE_VA,             // add move assignment operator
                         AU_DEFINE_COPY_VA),            // add copy assignment operator
                        (pSource, pDrain, pProcessor, uBufferSize));


        AuUInt32 TryProcessBuffered() override;
        AuUInt32 GetRawBytesBuffered() override;
        AuUInt32 GetRawBytesLimit() override;

        AuUInt32 TryPump();
    };

    AuUInt32 IOBufferedProcessor::TryProcessBuffered()
    {
        if (this->buffer.IsEmpty())
        {
            this->buffer.Allocate(this->uBufferSize);
            this->buffer.flagCircular = true; // !!!
        }

        if (this->buffer.IsEmpty())
        {
            return TryPump();
        }

        AuUInt canBuffer = this->buffer.RemainingWrite();
        canBuffer = AuMin(canBuffer, AuUInt((this->buffer.length + this->buffer.base) - this->buffer.writePtr));

        AuUInt read {};
        try
        {
            if (this->pSource->Read(AuMemoryViewStreamWrite(AuMemoryViewWrite(this->buffer.writePtr, canBuffer), read)) !=
                AuIO::EStreamError::eErrorNone)
            {
                return TryPump();
            }
        }
        catch (...)
        {
            SysPushErrorCatch();
        }

        this->buffer.writePtr += read;

        if (this->buffer.writePtr == this->buffer.base + this->buffer.length)
        {
            this->buffer.writePtr = this->buffer.base;
        }

        return TryPump();
    }

    AuUInt32 IOBufferedProcessor::TryPump()
    {
        AuUInt bytesProcessedTotal {};
        AuUInt bytesProcessed {};

        do
        {
            if (this->buffer.flagCircular && this->buffer.readPtr == this->buffer.base + this->buffer.length)
            {
                this->buffer.readPtr = this->buffer.base;
            }

            AuUInt canRead = this->buffer.RemainingBytes();
            canRead = AuMin<AuUInt>(canRead, (this->buffer.length + this->buffer.base) - this->buffer.readPtr);

            try
            {
                if (!this->pProcessor->OnDataAvailable(AuMemoryViewStreamRead(AuMemoryViewRead(this->buffer.readPtr, canRead), bytesProcessed), this->pDrain))
                {
                    break;
                }
            }
            catch (...)
            {
                SysPushErrorCatch();
            }

            this->buffer.readPtr += bytesProcessed;
            bytesProcessedTotal += bytesProcessed;

            if (this->buffer.flagCircular && this->buffer.readPtr == this->buffer.base + this->buffer.length)
            {
                this->buffer.readPtr = this->buffer.base;
            }
        }
        while (AuExchange(bytesProcessed, 0));

        return bytesProcessedTotal;
    }

    AuUInt32 IOBufferedProcessor::GetRawBytesBuffered()
    {
        return this->buffer.RemainingBytes();
    }

    AuUInt32 IOBufferedProcessor::GetRawBytesLimit()
    {
        return this->uBufferSize;
    }

    AUKN_SYM AuSPtr<IIOBufferedProcessor> NewBufferedProcessor(const AuSPtr<IStreamReader> &pSource,
                                                               const AuSPtr<IIOPipeInterceptor> &pProcessor,
                                                               const AuSPtr<IStreamWriter> &pDrain,
                                                               AuUInt32 uBufferSize)
    {
        return AuMakeShared<IOBufferedProcessor>(pSource, pDrain, pProcessor, uBufferSize);
    }
}