Client.h

Go to the documentation of this file.

/*+-------------------------------------------------------------------------+
  |                       MultiVehicle simulator (libmvsim)                 |
  |                                                                         |
  | Copyright (C) 2014-2024  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