AuroraRuntime/Source/IPC/IPCPipe.Unix.cpp

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

    File: IPCPipe.Linux.cpp
    Date: 2022-4-14
    Author: Reece
***/
#include <Source/RuntimeInternal.hpp>
#include "IPC.hpp"
#include "IPCHandle.hpp"
#include "IPCPipe.Unix.hpp"

#include <Source/IO/UNIX/FDIpcServer.hpp>

#include <Source/Loop/ILoopSourceEx.hpp>
#include <Source/Loop/LSHandle.hpp>
#include <fcntl.h>

namespace Aurora::IPC
{
    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // Pipes
    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    
    struct IPCPipeImpl : IPCPipe, Loop::LSHandle
    {
        IPCPipeImpl(int (fds)[2], IPCHandle readEnd, IPCHandle writeEnd);
        ~IPCPipeImpl();

        PROXY_INTERNAL_INTERFACE_(LSHandle::)

        virtual bool Read(const Memory::MemoryViewStreamWrite &write, bool nonblock) override;
        virtual bool Write(const Memory::MemoryViewStreamRead &read, bool nonblock) override;
        virtual AuString ExportToString() override;

        bool IsSignaled() override;
        bool WaitOn(AuUInt32 timeout) override;
        Loop::ELoopSource GetType() override;

    private:
        int fds[2] {-1, -1};
        //Loop::LSHandle lsHandle_;
        IPCHandle readEnd_;
        IPCHandle writeEnd_;
    };

    IPCPipeImpl::IPCPipeImpl(int (fds2)[2], IPCHandle readEnd, IPCHandle writeEnd) : 
        fds {fds2[0], fds2[1]}, readEnd_(readEnd), writeEnd_(writeEnd)
    {
        this->handle = fds[0];
    }
    
    IPCPipeImpl::~IPCPipeImpl()
    {
        int fd {-1};

        if ((fd = AuExchange(fds[0], -1)) != -1)
        {
            IO::UNIX::FDServeEnd(readEnd_);
            ::close(fd);
        }

        if ((fd = AuExchange(fds[1], -1)) != -1)
        {
            IO::UNIX::FDServeEnd(writeEnd_);
            ::close(fd);
        }
    }
    
    bool IPCPipeImpl::Read(const Memory::MemoryViewStreamWrite &write, bool nonblock)
    {
        auto handle = fds[0];
        auto control = ::fcntl(handle, F_GETFL);
        auto ref = control;

        if (nonblock)
        {
            control |= O_NONBLOCK;
        }
        else
        {
            control &= ~O_NONBLOCK;
        }

        if (ref != control)
        {
            ::fcntl(handle, F_SETFL, control);
        }

        int tmp;
        do
        {
            tmp = ::read(handle, write.ptr, write.length);
        } while ((tmp == -1 && errno == EINTR));

        if (tmp <= 0)
        {
            if (tmp == 0)
            {
                return nonblock;
            }
        
            SysPushErrorMem();
            return false;
        }

        write.outVariable = tmp;
        return true;
    }

    bool IPCPipeImpl::Write(const Memory::MemoryViewStreamRead &read, bool nonblock)
    {
        auto handle = this->fds[1];
        auto control = ::fcntl(handle, F_GETFL);
        auto ref = control;

        if (nonblock)
        {
            control |= O_NONBLOCK;
        }
        else
        {
            control &= ~O_NONBLOCK;
        }

        if (ref != control)
        {
            ::fcntl(handle, F_SETFL, control);
        }

        int tmp;
        do
        {
            tmp = ::write(handle, read.ptr, read.length);
        } while ((tmp == -1 && errno == EINTR));

        if (tmp <= 0)
        {
            if (tmp == 0)
            {
                return nonblock;
            }
        
            SysPushErrorMem();
            return false;
        }

        read.outVariable = tmp;
        return true;
    }

    bool IPCPipeImpl::IsSignaled()
    {
        return LSHandle::IsSignaled();
    }

    bool IPCPipeImpl::WaitOn(AuUInt32 timeout)
    {
        return LSHandle::WaitOn(timeout);
    }

    Loop::ELoopSource IPCPipeImpl::GetType()
    {
        return Loop::ELoopSource::eSourceIPCReadPipe;
    }

    AuString IPCPipeImpl::ExportToString()
    {
        return this->readEnd_.ToString() + "." + this->writeEnd_.ToString();
    }

    AUKN_SYM AuSPtr<IPCPipe>      NewPipe()
    {
        IPCHandle readEnd, writeEnd;
        int fds[2];

        if (::pipe(fds) == -1)
        {
            SysPushErrorIO();
            return {};
        }

        if (!IO::UNIX::FDServe(true, true, true, true, fds[0], readEnd))
        {
            SysPushErrorIO();
            ::close(fds[0]);
            ::close(fds[1]);
            return {};
        }

        if (!IO::UNIX::FDServe(true, true, true, true, fds[1], writeEnd))
        {
            SysPushErrorIO();
            IO::UNIX::FDServeEnd(readEnd);
            ::close(fds[0]);
            ::close(fds[1]);
            return {};
        }

        auto handle = AuMakeShared<IPCPipeImpl>(fds, readEnd, writeEnd);
        if (!handle)
        {
            SysPushErrorMem();
            IO::UNIX::FDServeEnd(readEnd);
            IO::UNIX::FDServeEnd(writeEnd);
            ::close(fds[0]);
            ::close(fds[1]);
            return {};
        }

        return handle;
    }

    AUKN_SYM AuSPtr<IPCPipe>      ImportPipe(const AuString &handle)
    {
        IPCHandle readEnd, writeEnd;
        int fds[2] {-1, -1};

        auto itr = handle.find('.');
        if (itr == AuString::npos)
        {
            return {};
        }

        auto readString = handle.substr(0, itr);
        auto writeString = handle.substr(itr + 1);

        if (!readEnd.FromString(readString))
        {
            SysPushErrorParseError();
            return {};
        }

        if (!writeEnd.FromString(writeString))
        {
            SysPushErrorParseError();
            return {};
        }

        if (!IO::UNIX::FDAccept(readEnd, fds[0]))
        {
            SysPushErrorNested();
            return {};
        }

        if (!IO::UNIX::FDAccept(writeEnd, fds[1]))
        {
            ::close(fds[0]);
            SysPushErrorNested();
            return {};
        }

        if (!IO::UNIX::FDServe(true, true, true, true, fds[0], readEnd))
        {
            SysPushErrorIO();
            ::close(fds[0]);
            ::close(fds[1]);
            return {};
        }

        if (!IO::UNIX::FDServe(true, true, true, true, fds[1], writeEnd))
        {
            SysPushErrorIO();
            IO::UNIX::FDServeEnd(readEnd);
            ::close(fds[0]);
            ::close(fds[1]);
            return {};
        }

        auto object = AuMakeShared<IPCPipeImpl>(fds, readEnd, writeEnd);
        if (!object)
        {
            SysPushErrorMem();
            IO::UNIX::FDServeEnd(readEnd);
            IO::UNIX::FDServeEnd(writeEnd);
            ::close(fds[0]);
            ::close(fds[1]);
            return {};
        }

        return object;
    }
}