AuroraRuntime/Source/IO/Net/AuNetStream.Linux.cpp

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

    File: AuNetStream.Linux.cpp
    Date: 2022-8-26
    Author: Reece
***/
#include "Networking.hpp"
#include "AuNetStream.Linux.hpp"
#include "AuNetSocket.hpp"
#include "AuNetWorker.hpp"
#include "AuNetEndpoint.hpp"
#include <Source/IO/Loop/LSEvent.hpp>

namespace Aurora::IO::Net
{
    LinuxAsyncNetworkTransaction::LinuxAsyncNetworkTransaction(SocketBase *pSocket) :
        pSocket(pSocket)
    {
        if (this->pSocket->GetRemoteEndpoint().transportProtocol == ETransportProtocol::eProtocolTCP)
        {
            //this->dwRecvFlags = MSG_PARTIAL;
        }
    }

    LinuxAsyncNetworkTransaction::~LinuxAsyncNetworkTransaction()
    {
        Reset();
    }

    bool     LinuxAsyncNetworkTransaction::StartRead(AuUInt64 offset, const AuSPtr<AuMemoryViewWrite> &memoryView)
    {
        if (this->bDisallowRecv)
        {
            SysPushErrorIO("Recv isn't allowed");
            return false;
        }

        if (this->bIsIrredeemable)
        {
            SysPushErrorIO("Transaction was signaled to be destroyed to reset mid synchronizable operation. You can no longer use this stream object");
            return false;
        }

        if (!IDontWannaUsePorts())
        {
            return false;
        }

        if (!memoryView)
        {
            SysPushErrorArg();
            return {};
        }

        //if (this->pMemoryHold)
        //{
        //    SysPushErrorIO("IO Operation in progress");
        //    return {};
        //}

        this->bLatch = false;

        //this->pMemoryHold = memoryView;
        this->bHasFailed = false;

        this->dwLastAbstractStat = memoryView->length;
        this->dwLastAbstractOffset = offset;
        this->dwLastBytes = 0;

        this->bIsWriting = false;

        LIOS_Init(AuSharedFromThis());
        SetMemory(memoryView);

        if (!this->bDatagramMode)
        {
            if (!UNIX::LinuxOverlappedSubmitRead(this->GetSocket(), 0, this, this->pWaitable ? this->pWaitable.get() : nullptr, true))
            {
                LIOS_SendProcess(0, true, errno);
                return true;
            }
            else
            {
                if (gRuntimeConfig.bFIODisableBatching)
                {
                    UNIX::SendIOBuffers(); 
                }

                return true;
            }
        }
        else
        {
            return false;
        }
    }

    bool     LinuxAsyncNetworkTransaction::StartWrite(AuUInt64 offset, const AuSPtr<AuMemoryViewRead> &memoryView)
    {
        if (this->bDisallowSend)
        {
            SysPushErrorIO("Send isn't allowed");
            return false;
        }

        if (this->bIsIrredeemable)
        {
            SysPushErrorIO("Transaction was signaled to be destroyed to reset mid synchronizable operation. You can no longer use this stream object");
            return false;
        }

        if (!IDontWannaUsePorts())
        {
            return false;
        }

        if (!memoryView)
        {
            SysPushErrorArg();
            return {};
        }

        //if (this->pMemoryHold)
        //{
        //    SysPushErrorIO("IO Operation in progress");
        //    return {};
        //}

        this->bLatch = false;

        //this->pMemoryHold = memoryView;
        this->bHasFailed = false;

        this->dwLastAbstractStat = memoryView->length;
        this->dwLastAbstractOffset = offset;
        this->dwLastBytes = 0;

        this->bIsWriting = true;

        LIOS_Init(AuSharedFromThis());
        SetMemory(memoryView);

        if (!this->bDatagramMode)
        {
            if (!UNIX::LinuxOverlappedSubmitWrite(this->GetSocket(), 0, this, this->pWaitable ? this->pWaitable.get() : nullptr))
            {
                LIOS_SendProcess(0, true, errno);
                return false;
            }
            else
            {
                if (gRuntimeConfig.bFIODisableBatching)
                {
                    UNIX::SendIOBuffers(); 
                }

                return true;
            }
        }
        else
        {
            this->iSocketLength = this->pSocket->endpointSize_;

        #if 0
            AuLogDbg("{} -> {} {}, {}, {} {}", this->GetSocket(),
                         memoryView->ptr,
                         memoryView->length,
                         0,
                         (void *)netEndpoint.hint,
                         this->iSocketLength);
        #endif

            if (::sendto(this->GetSocket(),
                         memoryView->ptr,
                         memoryView->length,
                         0,
                         (struct sockaddr *)netEndpoint.hint,
                         EndpointToLength(netEndpoint)) != memoryView->length)
            {
                LIOS_SendProcess(0, false, errno);
                return true;
            }

            LIOS_SendProcess(memoryView->length, true, errno);
            return true;
        }
    }

    void LinuxAsyncNetworkTransaction::LIOS_Process(AuUInt32 read, bool failure, int err, bool mark)
    {
        this->dwLastBytes = failure ? 0 : read;
        this->bHasFailed |= failure;
        this->dwOsErrorCode = err;

        if (mark)
        {
            return;
        }

        this->DispatchCb(read);
    }

    bool     LinuxAsyncNetworkTransaction::TranslateLastError(bool bReturnValue)
    {
        return false;
    }

    bool     LinuxAsyncNetworkTransaction::Complete()
    {
        return this->bLatch;
    }

    bool     LinuxAsyncNetworkTransaction::CompleteEx(AuUInt completeRoutine)
    {
        return false;
    }

    bool     LinuxAsyncNetworkTransaction::Failed()
    {
        return this->bHasFailed;
    }
    
    AuUInt   LinuxAsyncNetworkTransaction::GetOSErrorCode()
    {
        return this->dwOsErrorCode;
    }

    AuUInt32 LinuxAsyncNetworkTransaction::GetLastPacketLength()
    {
        return this->dwLastBytes;
    }

    void     LinuxAsyncNetworkTransaction::SetCallback(const AuSPtr<IAsyncFinishedSubscriber> &sub)
    {
        this->pSub = sub;
    }

    bool LinuxAsyncNetworkTransaction::Wait(AuUInt32 timeout)
    {
        return this->bLatch;
    }

    AuSPtr<AuLoop::ILoopSource> LinuxAsyncNetworkTransaction::NewLoopSource()
    {
        return this->pWaitable;
    }

    void LinuxAsyncNetworkTransaction::Reset()
    {
        if (this->dwLastAbstractStat)
        {
            this->bIsIrredeemable = true;
            this->bHasFailed = true;
            this->dwOsErrorCode = 0;
        }
        else
        {
            this->bHasFailed = false;
        }

        this->dwLastBytes = 0;
        this->dwLastAbstractStat = 0;
    }

    void LinuxAsyncNetworkTransaction::MakeSyncable()
    {
        this->pWaitable = AuLoop::NewLSEvent(false, true);
        SysAssert(this->pWaitable);
        //this->overlap.hEvent = (HANDLE)AuStaticCast<Loop::LSEvent>(this->pWaitable)->GetHandle();
    }

    void LinuxAsyncNetworkTransaction::ForceNextWriteWait()
    {
        this->bForceNextWait = true;
    }

    bool LinuxAsyncNetworkTransaction::IDontWannaUsePorts()
    {

        return true;
    }

    void LinuxAsyncNetworkTransaction::DispatchCb(AuUInt32 read)
    {
        if (this->bIsWriting)
        {
            if (read != this->dwLastAbstractStat)
            {
                this->dwOsErrorCode = 69;
                this->bHasFailed = true;
            }
        }

        this->dwLastAbstractStat = 0;
        this->dwLastBytes = read;

        if (AuExchange(this->bLatch, true))
        {
            return;
        }

        if (this->pSub)
        {
            this->pSub->OnAsyncFileOpFinished(this->dwLastAbstractOffset, read);
        }
    }

    int LinuxAsyncNetworkTransaction::GetSocket()
    {
        return this->pSocket->ToPlatformHandle();
    }

    int LinuxAsyncNetworkTransaction::GetAlertable()
    {
        return (int)AuStaticCast<AuLoop::LSEvent>(this->pSocket->ToWorkerEx()->ToEvent())->GetHandle();
    }
}