Client.h

/*+-------------------------------------------------------------------------+
  |                       MultiVehicle simulator (libmvsim)                 |
  |                                                                         |
  | Copyright (C) 2014-2026  Jose Luis Blanco Claraco                       |
  | Copyright (C) 2017  Borys Tymchenko (Odessa Polytechnic University)     |
  | Distributed under 3-clause BSD License                                  |
  |   See COPYING                                                           |
  +-------------------------------------------------------------------------+ */
 
#pragma once
 
#include <mrpt/system/COutputLogger.h>
#include <mrpt/system/CTimeLogger.h>
#include <mvsim/Comms/common.h>
#include <mvsim/Comms/zmq_fwrds.h>
 
#include <atomic>
#include <memory>
#include <thread>
 
namespace mvsim::internal
{
struct InfoPerService;
struct InfoPerSubscribedTopic;
}  // namespace mvsim::internal
 
namespace mvsim
{
class Client : public mrpt::system::COutputLogger
{
   public:
    Client();
    Client(const std::string& nodeName);
 
    ~Client();
 
    void setName(const std::string& nodeName);
 
    const std::string& serverHostAddress() const { return serverHostAddress_; }
    void serverHostAddress(const std::string& serverIpOrAddressName)
    {
        ASSERT_(!connected());
        serverHostAddress_ = serverIpOrAddressName;
    }
 
    void connect();
 
    bool connected() const;
 
    void shutdown() noexcept;
 
    template <typename T>
    void advertiseTopic(const std::string& topicName);
 
    void publishTopic(const std::string& topicName, const google::protobuf::Message& msg);
 
    template <typename MSG_T>
    void subscribeTopic(
        const std::string& topicName, const std::function<void(const MSG_T&)>& callback);
 
    template <typename INPUT_MSG_T, typename OUTPUT_MSG_T>
    void advertiseService(
        const std::string& serviceName,
        const std::function<OUTPUT_MSG_T(const INPUT_MSG_T&)>& callback);
 
    template <typename INPUT_MSG_T, typename OUTPUT_MSG_T>
    void callService(
        const std::string& serviceName, const INPUT_MSG_T& input, OUTPUT_MSG_T& output);
 
    std::string callService(const std::string& serviceName, const std::string& inputSerializedMsg);
 
    void subscribeTopic(
        const std::string& topicName,
        const std::function<
            void(const std::string& /*msgType*/, const std::vector<uint8_t>& /*serializedMsg*/)>&
            callback);
 
    struct InfoPerNode
    {
        std::string name;
    };
    std::vector<InfoPerNode> requestListOfNodes();
 
    struct InfoPerTopic
    {
        std::string name;
        std::string type;
        std::vector<std::string> endpoints, publishers;
    };
    std::vector<InfoPerTopic> requestListOfTopics();
 
    using topic_callback_t = std::function<void(const zmq::message_t& /*m*/)>;
 
    void subscribe_topic_raw(const std::string& topicName, const topic_callback_t& callback);
 
    const mrpt::system::CTimeLogger& profiler() const { return profiler_; }
    void enable_profiler(bool enable) { profiler_.enable(enable); }
 
   private:
    struct ZMQImpl;
    std::unique_ptr<ZMQImpl> zmq_;
 
    std::string serverHostAddress_ = "localhost";
    std::string nodeName_ = "anonymous";
 
    std::thread serviceInvokerThread_;
    std::thread topicUpdatesThread_;
 
    std::map<std::string, std::string> serviceToEndPointCache_;
    std::mutex serviceToEndPointCacheMtx_;
 
    mrpt::system::CTimeLogger profiler_{false, "mvsim::Client"};
 
    void doRegisterClient();
    void doUnregisterClient();
 
    void internalServiceServingThread();
    void internalTopicUpdatesThread();
    void internalTopicSubscribeThread(internal::InfoPerSubscribedTopic& ipt);
 
    using service_callback_t = std::function<std::shared_ptr<google::protobuf::Message>(
        const std::string& /*inAsString*/)>;
 
    void doAdvertiseTopic(
        const std::string& topicName, const google::protobuf::Descriptor* descriptor);
    void doAdvertiseService(
        const std::string& serviceName, const google::protobuf::Descriptor* descIn,
        const google::protobuf::Descriptor* descOut, service_callback_t callback);
 
    void doSubscribeTopic(
        const std::string& topicName, const google::protobuf::Descriptor* descriptor,
        const topic_callback_t& callback);
    void doCallService(
        const std::string& serviceName, const std::string& inputSerializedMsg,
        mrpt::optional_ref<google::protobuf::Message> outputMsg,
        mrpt::optional_ref<std::string> outputSerializedMsg = std::nullopt,
        mrpt::optional_ref<std::string> outputMsgTypeName = std::nullopt);
 
    friend struct internal::InfoPerService;
    friend struct internal::InfoPerSubscribedTopic;
};
 
template <typename T>
void Client::advertiseTopic(const std::string& topicName)
{
    doAdvertiseTopic(topicName, T::descriptor());
}
 
template <typename INPUT_MSG_T, typename OUTPUT_MSG_T>
void Client::advertiseService(
    const std::string& serviceName, const std::function<OUTPUT_MSG_T(const INPUT_MSG_T&)>& callback)
{
    doAdvertiseService(
        serviceName, INPUT_MSG_T::descriptor(), OUTPUT_MSG_T::descriptor(),
        service_callback_t(
            [callback](const std::string& inData)
            {
                INPUT_MSG_T in;
                in.ParseFromString(inData);
                return std::make_shared<OUTPUT_MSG_T>(callback(in));
            }));
}
 
template <typename MSG_T>
void Client::subscribeTopic(
    const std::string& topicName, const std::function<void(const MSG_T&)>& callback)
{
    doSubscribeTopic(
        topicName, MSG_T::descriptor(),
        topic_callback_t(
            [callback](const zmq::message_t& m)
            {
                MSG_T in;
                mvsim::parseMessage(m, in);
                callback(in);
            }));
}
 
template <typename INPUT_MSG_T, typename OUTPUT_MSG_T>
void Client::callService(
    const std::string& serviceName, const INPUT_MSG_T& input, OUTPUT_MSG_T& output)
{
    doCallService(serviceName, input.SerializeAsString(), output);
}
 
}  // namespace mvsim