HelloAurora/Tests/Public/21. Hello IPC/Main.cpp

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

    File: Main.cpp
    Date: 2022-2-18
    Author: Reece
***/
#include <AuroraRuntime.hpp>
#include <gtest/gtest.h>

#if defined(AURORA_IS_POSIX_DERIVED)

TEST(IPC, ShareUnixFd)
{
    int fds[2];

    ASSERT_EQ(pipe(fds), 0);

    ASSERT_EQ(write(fds[1], "hello", 5), 5);

    auto handleString = AuIO::UNIX::ShareFileDescriptor(fds[0]);
    AuLogDbg("UNIX IPC Handle: {}", handleString);
    ASSERT_TRUE(bool(handleString.size()));

    auto acceptFd = AuIO::UNIX::ShareFileDescriptorAccept(handleString);
    AuLogDbg("Got FD: {}, started with: {}", acceptFd, fds[0]);

    char hello[5];
    ASSERT_EQ(read(acceptFd, hello, 5), 5);

    AuLogDbg("Write side said: {}", AuString(hello, hello + 5));
    ASSERT_EQ(AuString(hello, hello + 5), "hello");

    close(fds[0]);
    close(fds[1]);
    close(acceptFd);

    AuIO::UNIX::ShareFileDescriptorStop(handleString);
}

#endif

TEST(IPC, Event)
{
    auto event = AuIPC::NewEvent(false, true);
    ASSERT_TRUE(bool(event));

    auto handleString = event->ExportToString();
    ASSERT_TRUE(bool(handleString.size()));
    AuLogDbg("Exported event handle: {}", handleString);

    auto eventImported = AuIPC::ImportEvent(handleString);
    ASSERT_TRUE(bool(eventImported));

    ASSERT_FALSE(eventImported->IsSignaled());
    ASSERT_FALSE(event->IsSignaled());

    event->Set();
    ASSERT_TRUE(eventImported->IsSignaled());
    
    ASSERT_FALSE(eventImported->IsSignaled());
    ASSERT_FALSE(event->IsSignaled());
}

TEST(IPC, Semaphore)
{
    auto semaphore = AuIPC::NewSemaphore(2);
    ASSERT_TRUE(bool(semaphore));

    auto handleString = semaphore->ExportToString();
    ASSERT_TRUE(bool(handleString.size()));
    AuLogDbg("Exported semaphore handle: {}", handleString);

    auto semaphoreImported = AuIPC::ImportSemaphore(handleString);
    ASSERT_TRUE(bool(semaphoreImported));

    ASSERT_TRUE(semaphoreImported->IsSignaled());
    ASSERT_TRUE(semaphore->IsSignaled());

    ASSERT_FALSE(semaphoreImported->IsSignaled());
    ASSERT_FALSE(semaphore->IsSignaled());

    semaphore->AddOne();

    ASSERT_TRUE(semaphore->IsSignaled());

    ASSERT_FALSE(semaphoreImported->IsSignaled());
    ASSERT_FALSE(semaphore->IsSignaled());
}

TEST(IPC, Memory)
{
    auto memory = AuIPC::NewSharedMemory(4096);
    ASSERT_TRUE(bool(memory));

    auto handleString = memory->ExportToString();
    ASSERT_TRUE(bool(handleString.size()));
    AuLogDbg("Exported shared view handle: {}", handleString);

    auto memoryImported = AuIPC::ImportSharedMemory(handleString);
    ASSERT_TRUE(bool(memoryImported));

    static const AuString kHelloWorld = "Hello IPC";
    AuMemcpy(memory->GetMemory().ptr, kHelloWorld.c_str(), kHelloWorld.size() + 1);

    AuLogDbg("Shared Memory String: {}", memoryImported->GetMemory().Begin<char>());

    ASSERT_EQ(memoryImported->GetMemory().Begin<char>(), kHelloWorld);
}

TEST(IPC, Pipe)
{
    auto pipe = AuIPC::NewPipe();
    ASSERT_TRUE(bool(pipe));
    
    ASSERT_FALSE(pipe->AsReadChannelIsOpen()->IsSignaled());

    auto handleString = pipe->ExportToString();
    ASSERT_TRUE(bool(handleString.size()));
    AuLogDbg("Exported pipe handle: {}", handleString);

    auto pipeImported = AuIPC::ImportPipe(handleString);
    ASSERT_TRUE(bool(pipeImported));

    ASSERT_TRUE(pipe->AsReadChannelIsOpen()->IsSignaled());

    static const AuString kHelloWorldClient = "Hello Client";
    static const AuString kHelloWorldServer = "Hello Server";

    //ASSERT_FALSE(pipe->AsReadChannelHasData()->IsSignaled());
    ASSERT_FALSE(pipeImported->AsReadChannelHasData()->IsSignaled());

    AuUInt bytesWritten;
    ASSERT_TRUE(pipe->Write(AuMemoryViewStreamRead(kHelloWorldClient, bytesWritten)));
    ASSERT_EQ(bytesWritten, kHelloWorldClient.size());

    ASSERT_TRUE(pipeImported->AsReadChannelHasData()->IsSignaled());
    ASSERT_FALSE(pipe->AsReadChannelHasData()->IsSignaled());

    AuString temp(kHelloWorldClient.size(), '\00');
    AuUInt bytesRead;

    ASSERT_TRUE(pipeImported->Read(AuMemoryViewStreamWrite(temp.data(), temp.data() + temp.size(), bytesRead), false));

    AuLogDbg("Pipe Message: {}", temp);

    ASSERT_EQ(bytesRead, kHelloWorldClient.size());
    ASSERT_EQ(temp, kHelloWorldClient);

    ASSERT_FALSE(pipe->AsReadChannelHasData()->IsSignaled());
    ASSERT_FALSE(pipeImported->AsReadChannelHasData()->IsSignaled());

    ASSERT_TRUE(pipeImported->Write(AuMemoryViewStreamRead(kHelloWorldServer, bytesWritten)));
    ASSERT_EQ(bytesWritten, kHelloWorldServer.size());

    ASSERT_TRUE(pipe->AsReadChannelHasData()->IsSignaled());
    ASSERT_FALSE(pipeImported->AsReadChannelHasData()->IsSignaled());

    temp.resize(kHelloWorldServer.size());

    ASSERT_TRUE(pipe->Read(AuMemoryViewStreamWrite(temp.data(), temp.data() + temp.size(), bytesRead), false));

    AuLogDbg("Pipe Message: {}", temp);

    ASSERT_EQ(bytesRead, kHelloWorldServer.size());
    ASSERT_EQ(temp, kHelloWorldServer);

    ASSERT_FALSE(pipeImported->AsReadChannelHasData()->IsSignaled());
    ASSERT_FALSE(pipe->AsReadChannelHasData()->IsSignaled());

    pipeImported.reset();
    ASSERT_FALSE(pipe->AsReadChannelIsOpen()->IsSignaled());
}

#pragma optimize("", off)

TEST(IPC, AsyncPipe)
{
    auto pipe = AuIPC::NewPipe();
    ASSERT_TRUE(bool(pipe));
    
    ASSERT_FALSE(pipe->AsReadChannelIsOpen()->IsSignaled());

    auto handleString = pipe->ExportToString();
    ASSERT_TRUE(bool(handleString.size()));
    AuLogDbg("Exported pipe handle: {}", handleString);

    auto pipeImported = AuIPC::ImportPipe(handleString);
    ASSERT_TRUE(bool(pipeImported));

    ASSERT_TRUE(pipe->AsReadChannelIsOpen()->IsSignaled());

    static const AuString kHelloWorldClient = "Hello Client";
    static const AuString kHelloWorldServer = "Hello Server";

    auto transactionA = pipe->NewAsyncTransaction();
    auto transactionB = pipeImported->NewAsyncTransaction();


    // Set callback
    transactionA->SetCallback(AuMakeShared<AuIO::IAsyncFinishedSubscriberFunctional>([](AuUInt64 offset, AuUInt32 length)
    {
        AuLogDbg("IPC server callback: {} {}", offset, length);
    }));

    transactionB->SetCallback(AuMakeShared<AuIO::IAsyncFinishedSubscriberFunctional>([](AuUInt64 offset, AuUInt32 length)
    {
        AuLogDbg("IPC client callback: {} {}", offset, length);
    }));


    // 
    AuByteBuffer writeBuffer(512);
    
    AuRng::RngFillRange(writeBuffer);
    AuMemoryViewRead writeView(writeBuffer);
    ASSERT_TRUE(transactionA->StartWrite(0, AuUnsafeRaiiToShared(&writeView)));

    AuByteBuffer readBuffer(512);
    AuMemoryViewWrite readView(readBuffer);
    ASSERT_TRUE(transactionB->StartRead(0, AuUnsafeRaiiToShared(&readView)));

    // Create loop to sync against the two outstanding IO requests
    auto loop = AuLoop::NewLoopQueue();
    
    // Add initial loop sources
    ASSERT_TRUE(loop->SourceAdd(transactionA->NewLoopSource()));
    ASSERT_TRUE(loop->SourceAdd(transactionB->NewLoopSource()));
   
    ASSERT_TRUE(loop->Commit());

    // Wait for 100 MS
    ASSERT_TRUE(loop->WaitAll(100));

    ASSERT_EQ(writeBuffer, readBuffer);

    // Reset client pipe
    pipeImported.reset();
    transactionB.reset();
    transactionA.reset();

    // Assert dead
    ASSERT_FALSE(pipe->AsReadChannelIsOpen()->IsSignaled());
}

void RunTests()
{
    Aurora::RuntimeStartInfo info;
    info.console.fio.bEnableLogging = false;
    info.console.asyncVSLog = false;
    Aurora::RuntimeStart(info);
}