MsgTopic.hpp

//
// Created by xiong on 2020/10/23.
//
 
#ifndef EMX_MSGTOPIC_HPP
#define EMX_MSGTOPIC_HPP
 
#include <vector>
#include "core/EmxTypeDef.hpp"
#include "core/EmxUV.hpp"
#include "core/utils/Socket.hpp"
 
namespace Emx {
    class MsgTopic {
    public:
 
        enum class ClientTypeE : uint8_t {
            Pub = 0, 
            Sub 
        };
 
        enum class MsgTypeE : uint8_t {
            Transfer = 0, 
            Register, 
        };
 
        static const int32_t MaxTopicNum = 64;
 
        struct Header {
            uint32_t sync;
            MsgTypeE msgType;
            uint32_t size;
        };
 
        struct SubRegister {
            Header header; 
            bool sendByPack; 
            int32_t maxSend; 
            uint32_t topicArray[MaxTopicNum]; 
        };
 
        struct Transfer {
            Header header; 
            uint32_t topic; 
        };
    };
 
    class MsgTopicPublisherAsync : public EuvPipeClientBase {
    public:
        using Callback = std::function<void()>;
 
        struct Config {
            EuvLoop *loop; 
            Callback OnConnected; 
            Callback OnDisConnected; 
        };
 
        MsgTopicPublisherAsync() : EuvPipeClientBase() {
            m_connected = false;
        }
 
        ErrCodeE CreateMsgTopicPublisherAsync(const Config &cfg);
 
        void DestroyMsgTopicPublisherAsync() { Destroy(); }
 
        ErrCodeE Publish(uint32_t topic, const char *data, int32_t size);
 
    protected:
        void OnConnect() override {
            m_connected = true;
            if (m_cbConnected)m_cbConnected();
        }
 
        void OnDisConnect() override {
            m_connected = false;
            if (m_cbDisConnected)m_cbDisConnected();
        }
 
    private:
        bool m_connected;
        Callback m_cbConnected;
        Callback m_cbDisConnected;
    };
 
    class MsgTopicPublisherSync {
    public:
        static ErrCodeE Publish(uint32_t topic, const char *data, int32_t size, int32_t timeoutMs = 1000);
    };
 
    class MsgTopicPublisherSync2 {
    public:
        ErrCodeE Create(int32_t timeoutMs = 3000);
 
        void Destroy();
 
        ErrCodeE Publish(uint32_t topic, const char *data, int32_t size, int32_t timeoutMs = 3000);
    private:
        SocketUnStream m_sock;
        char m_name[EMX_MAX_PATH_SIZE];
    };
 
    class MsgTopicSubscriberAsync : public EuvPipeClientBase {
    public:
        using Callback = std::function<void()>;
 
        using RecvMsgCallback = std::function<void(uint32_t topic, const char *data, int32_t size)>;
 
        struct Config {
            EuvLoop *loop; 
            bool sendByPack; 
            int32_t maxSend; 
            std::vector<uint32_t> topicArray; 
            char *buffer; 
            int32_t size; 
            Callback OnConnected; 
            Callback OnDisConnected; 
            RecvMsgCallback OnRecvMsg; 
        };
 
        MsgTopicSubscriberAsync() : EuvPipeClientBase() {
            m_waitSize = sizeof(MsgTopic::Header);
            m_waitHeader = true;
            m_reg = nullptr;
        }
 
        ErrCodeE CreateMsgTopicSubscriberAsync(const Config &cfg);
 
        void DestroyMsgTopicSubscriberAsync() { Destroy(); }
 
    protected:
        void OnConnect() override;
 
        void OnDisConnect() override {
            if (m_cbDisConnected)m_cbDisConnected();
        }
 
        void OnRecv(const char *data, int32_t size) override;
 
    private:
        bool m_waitHeader;
        uint32_t m_waitSize;
        MsgTopic::Header m_header;
        std::string m_data;
        std::shared_ptr<std::string> m_reg;
        Callback m_cbConnected;
        Callback m_cbDisConnected;
        RecvMsgCallback m_cbRecvMsg;
    };
 
    class MsgTopicSubscriberSync {
    public:
 
        struct Config {
            bool sendByPack; 
            int32_t maxSend; 
            std::vector<uint32_t> topicArray; 
        };
 
        MsgTopicSubscriberSync() {}
 
        ErrCodeE Connect(const Config &cfg, int32_t timeoutMs);
 
        ErrCodeE GetMsg(uint32_t &topic, std::string &data, int32_t timeoutMs);
 
        void DisConnect();
 
        const char *GetSockName() { return m_path; }
 
    private:
        SocketUnStream m_sock;
        char m_path[EMX_MAX_PATH_SIZE];
    };
}
 
#endif //EMX_MSGTOPIC_HPP