135 lines
4.3 KiB
C++
135 lines
4.3 KiB
C++
/***
|
|
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);
|
|
}
|
|
} |