Squash commits for public release

libs/libipc/BUILD.gn

@@ -0,0 +1,6 @@
+import("//build/libs/TEMPLATE.gni")
+
+xOS_static_library("libipc") {
+  sources = []
+  configs = [ "//build/libs:libcxx_flags" ]
+}

libs/libipc/include/libipc/ClientConnection.h

@@ -0,0 +1,115 @@
+#pragma once
+#include <cstdlib>
+#include <libfoundation/Event.h>
+#include <libfoundation/EventLoop.h>
+#include <libfoundation/EventReceiver.h>
+#include <libfoundation/Logger.h>
+#include <libipc/DoubleSidedConnection.h>
+#include <libipc/Message.h>
+#include <libipc/MessageDecoder.h>
+#include <unistd.h>
+#include <vector>
+
+template <typename ServerDecoder, typename ClientDecoder>
+class ClientConnection : public LFoundation::EventReceiver {
+public:
+    ClientConnection(const LIPC::DoubleSidedConnection& connection, ServerDecoder& server_decoder, ClientDecoder& client_decoder)
+        : m_connection(connection)
+        , m_server_decoder(server_decoder)
+        , m_client_decoder(client_decoder)
+        , m_messages()
+    {
+    }
+
+    void set_accepted_key(int key) { m_accepted_key = key; }
+
+    bool send_message(const Message& msg) const
+    {
+        auto encoded_msg = msg.encode();
+        int wrote = write(m_connection.c2s_fd(), encoded_msg.data(), encoded_msg.size());
+        return wrote == encoded_msg.size();
+    }
+
+    std::unique_ptr<Message> send_sync(const Message& msg)
+    {
+        bool status = send_message(msg);
+        return wait_for_answer(msg);
+    }
+
+    std::unique_ptr<Message> wait_for_answer(const Message& msg)
+    {
+        for (;;) {
+            for (int i = 0; i < m_messages.size(); i++) {
+                if (m_messages[i] && m_messages[i]->key() == msg.key() && m_messages[i]->id() == msg.reply_id()) {
+                    return m_messages[i].release();
+                }
+            }
+            pump_messages();
+        }
+    }
+
+    void pump_messages()
+    {
+        std::vector<char> buf;
+
+        char tmpbuf[1024];
+
+        int read_cnt;
+        while ((read_cnt = read(m_connection.s2c_fd(), tmpbuf, sizeof(tmpbuf)))) {
+            if (read_cnt <= 0) {
+                Logger::debug << getpid() << " :: ClientConnection read error" << std::endl;
+                return;
+            }
+            size_t buf_size = buf.size();
+            buf.resize(buf_size + read_cnt);
+            memcpy((uint8_t*)&buf.data()[buf_size], (uint8_t*)tmpbuf, read_cnt);
+            if (read_cnt < sizeof(tmpbuf)) {
+                break;
+            }
+        }
+
+        size_t msg_len = 0;
+        size_t buf_size = buf.size();
+        for (size_t i = 0; i < buf_size; i += msg_len) {
+            msg_len = 0;
+            if (auto response = m_client_decoder.decode((buf.data() + i), read_cnt - i, msg_len)) {
+                m_messages.push_back(std::move(response));
+            } else if (auto response = m_server_decoder.decode((buf.data() + i), read_cnt - i, msg_len)) {
+                m_messages.push_back(std::move(response));
+            } else {
+                Logger::debug << getpid() << " :: ClientConnection read error" << std::endl;
+                std::abort();
+            }
+        }
+
+        if (m_messages.size() > 0) {
+            // Note: We send an event to ourselves and use CallEvent to recognize the
+            // event as sign to start processing of messages.
+            LFoundation::EventLoop::the().add(*this, new LFoundation::CallEvent(nullptr));
+        }
+    }
+
+    void receive_event(std::unique_ptr<LFoundation::Event> event) override
+    {
+        switch (event->type()) {
+        case LFoundation::Event::Type::DeferredInvoke: {
+            // Note: The event was sent from pump_messages() and callback of CallEvent is 0!
+            // Do NOT call callback here!
+            auto msg = std::move(m_messages);
+            for (int i = 0; i < msg.size(); i++) {
+                if (msg[i] && msg[i]->decoder_magic() == m_client_decoder.magic() && msg[i]->key() == m_accepted_key) {
+                    m_client_decoder.handle(*msg[i]);
+                }
+            }
+            break;
+        }
+        }
+    }
+
+private:
+    int m_accepted_key { -1 };
+    LIPC::DoubleSidedConnection m_connection;
+    std::vector<std::unique_ptr<Message>> m_messages;
+    ServerDecoder& m_server_decoder;
+    ClientDecoder& m_client_decoder;
+};

libs/libipc/include/libipc/Decodable.h

@@ -0,0 +1,8 @@
+#pragma once
+#include <cstddef>
+
+template <typename T>
+class Decodable {
+public:
+    virtual void decode(const char* buf, size_t& offset) { }
+};

libs/libipc/include/libipc/DoubleSidedConnection.h

@@ -0,0 +1,23 @@
+#pragma once
+
+namespace LIPC {
+
+class DoubleSidedConnection {
+public:
+    DoubleSidedConnection(int server_to_clients_fd, int clients_to_server_fd)
+        : m_clients_to_server_fd(clients_to_server_fd)
+        , m_server_to_clients_fd(server_to_clients_fd)
+    {
+    }
+
+    ~DoubleSidedConnection() = default;
+
+    inline int c2s_fd() const { return m_clients_to_server_fd; }
+    inline int s2c_fd() const { return m_server_to_clients_fd; }
+
+private:
+    int m_clients_to_server_fd;
+    int m_server_to_clients_fd;
+};
+
+}

libs/libipc/include/libipc/Encodable.h

@@ -0,0 +1,8 @@
+#pragma once
+#include <libipc/Encoder.h>
+
+template <typename T>
+class Encodable {
+public:
+    virtual void encode(EncodedMessage& buf) const { }
+};

libs/libipc/include/libipc/Encoder.h

@@ -0,0 +1,90 @@
+#pragma once
+#include <vector>
+
+typedef std::vector<uint8_t> EncodedMessage;
+
+class Encoder {
+public:
+    ~Encoder() = default;
+
+    static void append(EncodedMessage& buf, int val)
+    {
+        buf.push_back((uint8_t)val);
+        buf.push_back((uint8_t)(val >> 8));
+        buf.push_back((uint8_t)(val >> 16));
+        buf.push_back((uint8_t)(val >> 24));
+    }
+
+    static void append(EncodedMessage& buf, unsigned int val)
+    {
+        buf.push_back((uint8_t)val);
+        buf.push_back((uint8_t)(val >> 8));
+        buf.push_back((uint8_t)(val >> 16));
+        buf.push_back((uint8_t)(val >> 24));
+    }
+
+    static void append(EncodedMessage& buf, unsigned long val)
+    {
+        buf.push_back((uint8_t)val);
+        buf.push_back((uint8_t)(val >> 8));
+        buf.push_back((uint8_t)(val >> 16));
+        buf.push_back((uint8_t)(val >> 24));
+    }
+
+    static void decode(const char* buf, size_t& offset, unsigned long& val)
+    {
+        uint8_t b0 = buf[offset++];
+        uint8_t b1 = buf[offset++];
+        uint8_t b2 = buf[offset++];
+        uint8_t b3 = buf[offset++];
+
+        val = 0;
+        val |= (uint8_t(b3) << 24);
+        val |= (uint8_t(b2) << 16);
+        val |= (uint8_t(b1) << 8);
+        val |= (uint8_t(b0));
+    }
+
+    static void decode(const char* buf, size_t& offset, unsigned int& val)
+    {
+        uint8_t b0 = buf[offset++];
+        uint8_t b1 = buf[offset++];
+        uint8_t b2 = buf[offset++];
+        uint8_t b3 = buf[offset++];
+
+        val = 0;
+        val |= (uint8_t(b3) << 24);
+        val |= (uint8_t(b2) << 16);
+        val |= (uint8_t(b1) << 8);
+        val |= (uint8_t(b0));
+    }
+
+    static void decode(const char* buf, size_t& offset, int& val)
+    {
+        uint8_t b0 = buf[offset++];
+        uint8_t b1 = buf[offset++];
+        uint8_t b2 = buf[offset++];
+        uint8_t b3 = buf[offset++];
+
+        val = 0;
+        val |= (uint8_t(b3) << 24);
+        val |= (uint8_t(b2) << 16);
+        val |= (uint8_t(b1) << 8);
+        val |= (uint8_t(b0));
+    }
+
+    template <typename T>
+    static void append(EncodedMessage& buf, T& value)
+    {
+        value.encode(buf);
+    }
+
+    template <typename T>
+    static void decode(const char* buf, size_t& offset, T& value)
+    {
+        value.decode(buf, offset);
+    }
+
+private:
+    Encoder() = default;
+};

libs/libipc/include/libipc/Message.h

@@ -0,0 +1,18 @@
+#pragma once
+#include <sys/types.h>
+#include <vector>
+
+typedef std::vector<uint8_t> EncodedMessage;
+typedef int message_key_t;
+
+class Message {
+public:
+    Message() = default;
+    virtual ~Message() = default;
+
+    virtual int decoder_magic() const { return 0; }
+    virtual int id() const { return 0; }
+    virtual message_key_t key() const { return -1; }
+    virtual int reply_id() const { return -1; } // -1 means that there is no reply.
+    virtual EncodedMessage encode() const { return std::vector<uint8_t>(); }
+};

libs/libipc/include/libipc/MessageDecoder.h

@@ -0,0 +1,13 @@
+#pragma once
+#include <libipc/Message.h>
+#include <memory>
+
+class MessageDecoder {
+public:
+    MessageDecoder() = default;
+    virtual ~MessageDecoder() = default;
+
+    virtual int magic() { return 0; }
+    virtual std::unique_ptr<Message> decode(const char* buf, size_t size, size_t& decoded_msg_len) { return nullptr; }
+    virtual std::unique_ptr<Message> handle(Message&) { return nullptr; }
+};

libs/libipc/include/libipc/ServerConnection.h

@@ -0,0 +1,66 @@
+#pragma once
+#include <cstdlib>
+#include <libfoundation/Logger.h>
+#include <libipc/DoubleSidedConnection.h>
+#include <libipc/Message.h>
+#include <libipc/MessageDecoder.h>
+#include <vector>
+
+template <typename ServerDecoder, typename ClientDecoder>
+class ServerConnection {
+public:
+    ServerConnection(const LIPC::DoubleSidedConnection& connection, ServerDecoder& server_decoder, ClientDecoder& client_decoder)
+        : m_connection(connection)
+        , m_server_decoder(server_decoder)
+        , m_client_decoder(client_decoder)
+    {
+    }
+
+    bool send_message(const Message& msg) const
+    {
+        auto encoded_msg = msg.encode();
+        int wrote = write(m_connection.s2c_fd(), encoded_msg.data(), encoded_msg.size());
+        return wrote == encoded_msg.size();
+    }
+
+    void pump_messages()
+    {
+        std::vector<char> buf;
+
+        char tmpbuf[1024];
+
+        int read_cnt;
+        while ((read_cnt = read(m_connection.c2s_fd(), tmpbuf, sizeof(tmpbuf)))) {
+            if (read_cnt <= 0) {
+                Logger::debug << getpid() << " :: ServerConnection read error" << std::endl;
+                return;
+            }
+            size_t buf_size = buf.size();
+            buf.resize(buf_size + read_cnt);
+            memcpy(&buf.data()[buf_size], tmpbuf, read_cnt);
+            if (read_cnt < sizeof(tmpbuf)) {
+                break;
+            }
+        }
+
+        size_t msg_len = 0;
+        size_t buf_size = buf.size();
+        for (int i = 0; i < buf_size; i += msg_len) {
+            msg_len = 0;
+            if (auto response = m_server_decoder.decode((buf.data() + i), read_cnt - i, msg_len)) {
+                if (auto answer = m_server_decoder.handle(*response)) {
+                    send_message(*answer);
+                }
+            } else if (auto response = m_client_decoder.decode((buf.data() + i), read_cnt - i, msg_len)) {
+
+            } else {
+                std::abort();
+            }
+        }
+    }
+
+private:
+    LIPC::DoubleSidedConnection m_connection;
+    ServerDecoder& m_server_decoder;
+    ClientDecoder& m_client_decoder;
+};

libs/libipc/include/libipc/StringEncoder.h

@@ -0,0 +1,47 @@
+#pragma once
+
+#include <libipc/Decodable.h>
+#include <libipc/Encodable.h>
+#include <libipc/Encoder.h>
+#include <string>
+#include <sys/types.h>
+
+namespace LIPC {
+
+class StringEncoder : public Encodable<StringEncoder>, public Decodable<StringEncoder> {
+public:
+    StringEncoder() = default;
+    StringEncoder(const std::string& str)
+        : m_str(str)
+    {
+    }
+
+    StringEncoder(std::string&& str)
+        : m_str(std::move(str))
+    {
+    }
+
+    void encode(EncodedMessage& buf) const override
+    {
+        for (int i = 0; i < m_str.size(); i++) {
+            buf.push_back(m_str.at(i));
+        }
+        buf.push_back('\0');
+    }
+
+    void decode(const char* buf, size_t& offset) override
+    {
+        while (buf[offset] != '\0') {
+            m_str.push_back(buf[offset]);
+            offset++;
+        }
+        offset++;
+    }
+    std::string& string() { return m_str; }
+    const std::string& string() const { return m_str; }
+    std::string&& move_string() { return std::move(m_str); }
+
+private:
+    std::string m_str {};
+};
+} // namespace LIPC

libs/libipc/include/libipc/VectorEncoder.h

@@ -0,0 +1,55 @@
+#pragma once
+
+#include <iostream>
+#include <libipc/Decodable.h>
+#include <libipc/Encodable.h>
+#include <libipc/Encoder.h>
+#include <sys/types.h>
+#include <vector>
+
+namespace LIPC {
+
+template <class T>
+class VectorEncoder : public Encodable<VectorEncoder<T>>, public Decodable<VectorEncoder<T>> {
+public:
+    VectorEncoder() = default;
+    VectorEncoder(const std::vector<T>& vec)
+        : m_vec(vec)
+    {
+    }
+
+    VectorEncoder(std::vector<T>&& vec)
+        : m_vec(std::move(vec))
+    {
+    }
+
+    ~VectorEncoder() = default;
+
+    void encode(EncodedMessage& buf) const override
+    {
+        size_t sz = m_vec.size();
+        Encoder::append(buf, sz);
+        for (int i = 0; i < sz; i++) {
+            Encoder::append(buf, m_vec[i]);
+        }
+    }
+
+    void decode(const char* buf, size_t& offset) override
+    {
+        size_t sz = 0;
+        Encoder::decode(buf, offset, sz);
+        for (int i = 0; i < sz; i++) {
+            T el;
+            Encoder::decode(buf, offset, el);
+            m_vec.push_back(std::move(el));
+        }
+    }
+
+    std::vector<T>& vector() { return m_vec; }
+    const std::vector<T>& vector() const { return m_vec; }
+    std::vector<T>&& move_vector() { return std::move(m_vec); }
+
+private:
+    std::vector<T> m_vec {};
+};
+} // namespace LIPC