AuroraRuntime/Source/IO/Protocol/AuProtocolPiece.cpp

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

    File: AuProtocolPiece.cpp
    Date: 2022-8-24
    Author: Reece
***/
#include "Protocol.hpp"
#include "AuProtocolPiece.hpp"
#include "AuProtocolStack.hpp"
#include "IProtocolNext.hpp"

// TODO: check correct thread or throw api
// restrict this to one thread

namespace Aurora::IO::Protocol
{
    AuSPtr<ProtocolStack> ProtocolPiece::GetParent()

    {
        if (this->pParent)
        {
            return this->pParent->SharedFromThis();
        }
        else
        {
            return {};
        }
    }

    AuSPtr<IStreamWriter> ProtocolPiece::GetNextWriter()
    {
        if (!this->pParent)
        {
            return {};
        }

        AU_LOCK_GUARD(this->pParent->mutex);

        if (this->pParent->pTopPiece.get() == this)
        {
            if (this->pParent->pDrainBuffer)
            {
                return AuMakeShared<Buffered::BlobWriter>(this->pParent->pDrainBuffer);
            }
        }

        return this->pNext ?
            this->pNext->pWriteInteface->GetStreamWriter() :
            this->pOuputWriter;
    }

    AuSPtr<IStreamWriter> ProtocolPiece::GetInputWriter()
    {
        return this->pWriteInteface ? this->pWriteInteface->GetStreamWriter() : AuSPtr<IStreamWriter> {};
    }

    AuSPtr<IProtocolPiece> ProtocolPiece::GetNextPiece()
    {
        return this->pNext;
    }

    AuSPtr<IProtocolPiece> ProtocolPiece::GetPreviousPiece()
    {
        return this->pBefore;
    }

    AuSPtr<IStreamReader> ProtocolPiece::GetOutputReader()
    {
        // TODO: cache if required
        return IO::Adapters::NewByteBufferReadAdapter(AuSPtr<AuByteBuffer>(this->SharedFromThis(), &this->outputBuffer));
    }

    bool ProtocolPiece::ReallocateDrainBuffer(AuUInt32 uOutputLength)
    {
        if (!this->pParent)
        {
            return false;
        }

        AU_LOCK_GUARD(this->pParent->mutex);

        if (!this->pNext)
        {
            if (this->pParent->pDrainBuffer)
            {
                return false;
            }
        }

        this->outputBuffer.flagNoRealloc = false;
        auto bRet = this->outputBuffer.Resize(uOutputLength);
        if (!this->uMax.has_value())
        {
            this->outputBuffer.flagNoRealloc = true;
        }
        return bRet;
    }

    bool ProtocolPiece::ReallocateSourceBuffer(AuUInt32 uInputLength)
    {
        if (!this->pParent)
        {
            return false;
        }

        AU_LOCK_GUARD(this->pParent->mutex);

        if (auto pBefore = this->pBefore)
        {
            if (pBefore->outputBuffer)
            {
                pBefore->outputBuffer.flagNoRealloc = false;
                auto bRet = pBefore->outputBuffer.Resize(uInputLength);
                if (!pBefore->uMax.has_value())
                {
                    pBefore->outputBuffer.flagNoRealloc = true;
                }
                return bRet;
            }
        }
        else
        {
            if (this->pParent->bOwnsSource &&
                this->pParent->pSourceBufer)
            {
                return this->pParent->pSourceBufer->Resize(uInputLength);
            }
        }
    
        return false;
    }

    void ProtocolPiece::Remove()
    {
        this->PrivateUserDataClear();
        if (!this->pParent)
        {
            return;
        }

        AU_LOCK_GUARD(this->pParent->mutex);

        auto &pBottomPiece = this->pParent->pBottomPiece;
        
        // fix chain
        AuSPtr<ProtocolPiece> pLast;
        auto pCurrent = pBottomPiece;
        while (true)
        {
            if (!pCurrent)
            {
                break;
            }

            if (pCurrent.get() == this)
            {
                if (auto pNext = pCurrent->pNext)
                {
                    pNext->pBefore = pLast;
                }
                pLast->pNext = pCurrent->pNext;
                break;
            }

            pLast = pCurrent;
            pCurrent = pCurrent->pNext;
        }

        // fix head pointers
        if (this->pParent->pTopPiece.get() == this)
        {
            this->pParent->pTopPiece = AuReinterpretCast<ProtocolPiece>(pLast);

            this->pParent->bWrittenEnd = false;
        }

        if (this->pParent->pBottomPiece.get() == this)
        {
            this->pParent->pBottomPiece = this->pNext;
        }

        this->PrivateUserDataClear();
        this->pParent->DiscardAllocCaches();

        this->pParent = nullptr;
    }

    AuSPtr<Memory::ByteBuffer> ProtocolPiece::GetNextPieceBuffer()
    {
        return AuSPtr<AuByteBuffer>(AuSharedFromThis(), &this->outputBuffer);
    }

    AuSPtr<IProtocolInterceptorEx> ProtocolPiece::GetExtendedInterceptor()
    {
        return this->pInterceptorEx;
    }

    AuSPtr<IProtocolInterceptor> ProtocolPiece::GetShortPipeInterceptor()
    {
        return this->pInterceptor;
    }

    AuOptional<AuUInt> ProtocolPiece::ExchangeMaximumBufferLength(AuOptional<AuUInt> optMax)
    {
        if (!optMax)
        {
            return {};
        }

        if (!this->uMax)
        {
            return {};
        }

        return AuExchange(this->uMax, optMax);
    }
}