capnp router implementation: rpc client server comm

.gitignore

@@ -1,3 +1,6 @@
 /cmake-build-debug/
 /cmake-build-release/
 .idea
+.vscode/
+build/
+test/bizlogic/python/__pycache__/

CMakeLists.txt

@@ -5,6 +5,9 @@ project(UniversalActorFramework CXX)
 set(THREADS_PREFER_PTHREAD_FLAG ON)
 find_package(Threads REQUIRED)
 
+set(CAPNP_LINKER_FLAGS "-lcapnp -lkj -lcapnp-rpc -lkj-async")
+
+
 message("Detected cmake version ${CMAKE_VERSION}")
 
 if(${CMAKE_VERSION} VERSION_GREATER_EQUAL 3.27.0)
@@ -19,16 +22,16 @@ set(CMAKE_EXPERIMENTAL_CXX_MODULE_DYNDEP 1)
 if (CMAKE_CXX_COMPILER_ID STREQUAL "Clang")
     # using Clang
     set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -stdlib=libc++")
-    set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -stdlib=libc++ -lc++abi")
+    set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -stdlib=libc++ -lc++abi ${CAPNP_LINKER_FLAGS}")
 elseif (CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
     # using GCC
     set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -stdlib=stdlibc++")
-    set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -stdlib=stdlibc++")
+    set(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} -stdlib=stdlibc++ ${CAPNP_LINKER_FLAGS}")
 
     include(gcc_modules.cmake)
 endif()
 
-#link_libraries("-fsanitize=undefined")
+link_libraries("-fsanitize=undefined")
 
 # Default to C++ extensions being off. Clang's modules support have trouble
 # with extensions right now.

src/node/CMakeLists.txt

@@ -1,10 +1,10 @@
 
-
 add_subdirectory(rule_engine/abstract)
 add_subdirectory(rule_engine/python)
 add_subdirectory(cluster)
+add_subdirectory(network)
 
 add_executable(node main.cpp)
 
 
-target_link_libraries(node PRIVATE cluster python_rule_engine)
+target_link_libraries(node PRIVATE cluster python_rule_engine network)

src/node/main.cpp

@@ -4,6 +4,7 @@
 
 import rule_engine.python;
 import cluster;
+import network.router;
 
 std::string actor_ids[4] = {"ActorA", "ActorB", "ActorC", "ActorD"};
 
@@ -23,5 +24,5 @@ int main(int argc, char *argv[]) {
         auto& actor_id = actor_ids[i%4];
         runtime.message_to(actor_id, {});
     }
-
+    test_network();
 }

src/node/network/CMakeLists.txt

@@ -0,0 +1,22 @@
+#FetchContent_Declare(
+#        gRPC
+#        GIT_REPOSITORY https://github.com/grpc/grpc
+#        GIT_TAG        v1.54.0
+#)
+
+add_library(network)
+
+target_sources(
+        network
+
+        PUBLIC
+        FILE_SET cxx_modules TYPE CXX_MODULES FILES
+        router.cppm
+)
+
+add_subdirectory(rpc)
+
+target_link_libraries(
+        network
+        rpc
+)

src/node/network/router.cppm

@@ -0,0 +1,33 @@
+module;
+#include <iostream>
+#include <thread>
+#include <cstring>
+#include <cstdlib>
+#include <unistd.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+import rpc.client;
+import rpc.server;
+
+export module network.router;
+
+export void test_network()
+{
+    pid_t serverProcess = fork();
+
+    if (serverProcess < 0) {
+        perror("Error in forking server process");
+        exit(1);
+    } else if (serverProcess == 0) { 
+        start_server("localhost:8081");
+    } else { 
+        std::thread clientThread(start_client, "localhost:8081");
+
+        int status;
+        waitpid(serverProcess, &status, 0);   //kill the server process manually on the terminal. Currently the server will run forever
+        clientThread.join();
+    }
+}
+

src/node/network/rpc/CMakeLists.txt

@@ -0,0 +1,47 @@
+set(RPC_SRCS
+    src/rpc_client.cppm
+    src/rpc_server.cppm
+)
+
+set(CAPNPC_SRCS
+    schema/message_intf.capnp
+)
+
+set(GLOBAL_INCLUDES
+    include/rpc_defs.cppm
+)
+
+foreach(schema ${CAPNPC_SRCS})
+    add_custom_command(
+        OUTPUT "${schema}.c++" "${schema}.h"
+        COMMAND capnp compile -oc++ "${schema}"
+        DEPENDS ${schema}
+        WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}
+        COMMENT "Compiling Cap'n Proto schema: ${schema}"
+    )
+    list(APPEND CAPNPC_CXX_SOURCES "${CMAKE_CURRENT_SOURCE_DIR}/${schema}.c++")
+endforeach()
+
+add_library(rpc)
+include_directories(schema)
+target_sources(
+    rpc
+    PUBLIC
+    ${CAPNPC_CXX_SOURCES}
+)
+target_sources(
+    rpc
+    PUBLIC
+    FILE_SET cxx_modules
+    TYPE CXX_MODULES
+    FILES
+    ${GLOBAL_INCLUDES}
+    ${RPC_SRCS}
+
+)
+target_link_libraries(
+    rpc
+    PRIVATE
+    capnp 
+    kj
+)

src/node/network/rpc/include/rpc_defs.cppm

@@ -0,0 +1,43 @@
+module;
+#include <iostream>
+#include <kj/common.h>
+#include <vector>
+#include <cstdint>
+#include <memory>
+export module rpc_defs;
+
+export namespace uaf
+{
+    typedef unsigned char byte;
+    enum class StatusCode : std::uint16_t
+    {
+        STATUS_NONE,
+        STATUS_ERROR,
+        STATUS_ERROR_WITH_DATA,
+        STATUS_SUCCESS,
+        STATUS_SUCCESS_WITH_DATA
+    };
+
+    struct ResultDefStruct
+    {
+        uaf::StatusCode status;
+        std::vector<kj::byte> data;
+
+        ResultDefStruct(StatusCode status, std::vector<kj::byte> &&data) : status(status), data(std::move(data))
+        {
+        }
+
+        ResultDefStruct(ResultDefStruct &&other) noexcept
+            : status(other.status), data(std::move(other.data)) {}
+
+        ResultDefStruct &operator=(ResultDefStruct &&other) noexcept
+        {
+            if (this != &other)
+            {
+                status = other.status;
+                data = std::move(other.data);
+            }
+            return *this;
+        }
+    };
+}

src/node/network/rpc/schema/message_intf.capnp

@@ -0,0 +1,15 @@
+@0x9c4f255fd44cfc76;
+interface MessageDefInterface 
+{
+  struct MessageDef {
+    id @0 : Text;
+    message @1 : Data;
+  }
+
+  interface ResultDef {
+    getStatus @0 () -> (status : UInt16);
+    getData @1 () -> (data : Data);
+  }
+
+  schedule @0 (messageDefObj : MessageDef) -> (resultDefObj : ResultDef);
+}