vertx_bridge.cpp

#ifdef _WIN32
#include <winsock2.h>
#else
#include <arpa/inet.h> // Linux/Unix: htonl, ntohl
#endif

#include "vertx_bridge.h"
#include "socket_client.h"
#include <chrono>
#include <random>
#include <sstream>
#include <mutex>

using namespace chrono;

class VertxEventBusClient::Impl {
public:
    Impl(const string& host, int port) : host(host), port(port), sock(1000, 3000) {
        sock.connectToServer(host, port);
    }

    bool isConnected() const {
        return sock.isConnected();
    }

    bool publish(const string& address, json body, json headers) {
        return sendMessage({
            {"type", "publish"},
            {"address", address},
            {"headers", move(headers)},
            {"body", move(body)}
        });
    }

    bool send(const string& address, json body, json headers) {
        return sendMessage({
            {"type", "send"},
            {"address", address},
            {"headers", move(headers)},
            {"body", move(body)}
        });
    }

    bool fail(const string& address, int failureCode, string message) {
        return sendMessage({
            {"type", "err"},
            {"address", address},
            {"failureCode", failureCode},
            {"failureType", "RECIPIENT_FAILURE"},
            {"message", message}
        });
    }

    bool request(const string& address,
                 const function<void(json)>& handler,
                 json body, json headers, int timeout
    ) {
        string replyAddress = generateReplyAddress();
        {
            lock_guard lock(handlers_mtx);
            replyHandlers[replyAddress] = ReplyHandler(handler, timeout);
        }
        return sendMessage({
            {"type", "send"},
            {"address", address},
            {"replyAddress", replyAddress},
            {"headers", move(headers)},
            {"body", move(body)}
        });
    }

    bool consume(const string& address, const std::function<void(json message)>& handler) {
        bool ok = sendMessage({
            {"type", "register"},
            {"address", address},
        });
        if (ok) {
            lock_guard lock(handlers_mtx);
            consumeHandlers[address] = handler;
        }
        return ok;
    }

    bool unregister(const string& address) {
        {
            lock_guard lock(handlers_mtx);
            consumeHandlers.erase(address);
        }
        return sendMessage({
            {"type", "unregister"},
            {"address", address},
        });
    }

    struct ReplyHandler {
        function<void(json)> handler;
        time_point<system_clock> deadline;

        ReplyHandler() = default;

        explicit ReplyHandler(
            const function<void(json)>& handler, int timeout
        ): handler(handler), deadline(system_clock::now()) {
            deadline += milliseconds(timeout);
        }
    };

    const string host;
    const int port;
    SocketClient sock;
    map<string, ReplyHandler> replyHandlers;
    map<string, function<void(json)>> consumeHandlers;
    mutex handlers_mtx;

    volatile bool stopping;

    bool sendMessage(const json& message) {
        string msgStr = message.dump();
        uint32_t len = htonl(msgStr.length());
        return sock.sendData(&len, sizeof(len)) && sock.sendData(msgStr.data(), msgStr.size());
    }

    void stop() {
        stopping = true;
    }

    void run() {
        vector<char> buffer;
        stopping = false;
        while (!stopping) {
            if (!sock.isConnected()) {
                if (!sock.connectToServer(host, port)) {
                    for (int i = 0; i < 30 && !stopping; i++) {
                        this_thread::sleep_for(milliseconds(100));
                    }
                    continue;
                }
            }
            {
                lock_guard lock(handlers_mtx);
                //reregister
                for (const auto& it : consumeHandlers) {
                    sendMessage({
                        {"type", "register"},
                        {"address", it.first}
                    });
                }
            }
            auto last_ping_time = system_clock::now();
            while (!stopping) {
                if (!sock.isDataAvailable()) {
                    removeTimeoutReplyHandlers();
                    auto time = duration_cast<milliseconds>(system_clock::now() - last_ping_time).count();
                    if (time > 5000) {
                        sendMessage({
                            {"type", "ping"}
                        });
                    }
                    if (time > 10000) {
                        fprintf(stderr, "socket connection broken\n");
                        sock.disconnect();
                        break;
                    }
                    continue;
                }
                // 读取消息长度
                uint32_t len;
                if (!sock.receiveData(&len, sizeof(len))) break;
                len = ntohl(len);

                // 读取消息内容
                if (buffer.size() < len) buffer.resize(len);
                if (!sock.receiveData(buffer.data(), len)) break;

                json message = json::parse(buffer.data(), buffer.data() + len, nullptr, false);
                if (message.is_discarded() || !message.is_object()) {
                    fprintf(stderr, "message format error: %s\n", string(buffer.data(), buffer.data() + len).c_str());
                } else {
                    handleMessage(move(message));
                }
                last_ping_time = system_clock::now();
            }
            {
                lock_guard lock(handlers_mtx);
                for (auto& [address, rh] : replyHandlers) {
                    rh.handler({
                        {"type", "err"},
                        {"address", address},
                        {"failureCode", -1},
                        {"failureType", "ERROR"},
                        {"message", "TCP Bridge is broken"}
                    });
                }
                replyHandlers.clear();
            }
        }
    }

    void removeTimeoutReplyHandlers() {
        lock_guard lock(handlers_mtx);
        auto now = system_clock::now();
        for (auto it = replyHandlers.begin(); it != replyHandlers.end();) {
            if (now > it->second.deadline) {
                it->second.handler({
                    {"type", "err"},
                    {"address", it->first},
                    {"failureCode", -1},
                    {"failureType", "TIMEOUT"},
                    {"message", "Reply message is timeout"}
                });
                it = replyHandlers.erase(it);
            } else {
                ++it;
            }
        }
    }

    void handleMessage(const json message) {
        string type = message.value("type", "");
        if (type == "pong") return;
        string address = message.value("address", "");
        function<void(json)> handler;
        {
            lock_guard lock(handlers_mtx);
            auto it = consumeHandlers.find(address);
            if (it != consumeHandlers.end()) {
                handler = it->second;
            } else {
                auto it2 = replyHandlers.find(address);
                if (it2 != replyHandlers.end()) {
                    handler = it2->second.handler;
                    replyHandlers.erase(it2);
                }
            }
        }
        if (handler) handler(move(message));
    }

    static string generateReplyAddress() {
        static random_device rd;
        static mt19937 gen(rd());
        static uniform_int_distribution<> dis(0, 0xFFFF);
        static atomic_int id(0);

        stringstream ss;
        ss << "_vb.reply." << ++id << "." << hex << dis(gen);
        return ss.str();
    }
};

VertxEventBusClient::VertxEventBusClient(const string& host, int port) : impl_(new Impl(host, port)) {
}

VertxEventBusClient::~VertxEventBusClient() {
    delete impl_;
}

void VertxEventBusClient::stop() {
    impl_->stop();
}

void VertxEventBusClient::run() {
    impl_->run();
}

bool VertxEventBusClient::isConnected() const {
    return impl_->isConnected();
}

bool VertxEventBusClient::publish(const string& address, json body, json headers) {
    return impl_->publish(address, move(body), move(headers));
}

bool VertxEventBusClient::send(const string& address, json body, json headers) {
    return impl_->send(address, move(body), move(headers));
}

bool VertxEventBusClient::fail(const string& address, int failureCode, string message) {
    return impl_->fail(address, failureCode, move(message));
}

bool VertxEventBusClient::request(
    const string& address, const std::function<void(json message)>& handler,
    json body, json headers, int timeout
) {
    return impl_->request(address, handler, move(body), move(headers), timeout);
}

bool VertxEventBusClient::consume(
    const string& address, const std::function<void(json message)>& handler
) {
    return impl_->consume(address, handler);
}

bool VertxEventBusClient::unregister(const std::string& address) {
    return impl_->unregister(address);
}