From 447df941b4ace128c490675bc25775cd51fca228 Mon Sep 17 00:00:00 2001 From: jksully Date: Fri, 5 Jan 2024 13:45:45 -0500 Subject: [PATCH 1/2] Added header comments --- CMakeLists.txt | 72 +++++------ include/mediator/mediator.hpp | 186 +++++++++++++++++++++++---- include/messages/exampleMessages.hpp | 15 +++ include/mros/mros.hpp | 70 +++++++++- include/mros/node.hpp | 60 +++++++++ include/mros/publisher.hpp | 39 +++++- include/mros/subscriber.hpp | 88 ++++++++++++- test/mros/test_publisher.cpp | 2 +- 8 files changed, 461 insertions(+), 71 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index ded372b..60b8072 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -281,40 +281,40 @@ target_link_libraries(test_logger ) -#add_executable(mros_pub_test -# src/logging/logging.cpp -# src/mros/node.cpp -# src/mros/mros.cpp -# src/mros/publisher.cpp -# test/mros/test_publisher.cpp -# ${SOCKET_SOURCES} -#) -#target_include_directories(mros_pub_test -# PRIVATE -# include -#) -#target_link_libraries(mros_pub_test -# PRIVATE -# nlohmann_json::nlohmann_json -# log4cxx -# EXPAT::EXPAT -#) +add_executable(mros_pub_test + src/logging/logging.cpp + src/mros/node.cpp + src/mros/mros.cpp + src/mros/publisher.cpp + test/mros/test_publisher.cpp + ${SOCKET_SOURCES} +) +target_include_directories(mros_pub_test + PRIVATE + include +) +target_link_libraries(mros_pub_test + PRIVATE + nlohmann_json::nlohmann_json + log4cxx + EXPAT::EXPAT +) -#add_executable(mros_sub_test -# src/logging/logging.cpp -# src/mros/node.cpp -# src/mros/mros.cpp -# src/mros/subscriber.cpp -# test/mros/test_subscriber.cpp -# ${SOCKET_SOURCES} -#) -#target_include_directories(mros_sub_test -# PRIVATE -# include -#) -#target_link_libraries(mros_sub_test -# PRIVATE -# nlohmann_json::nlohmann_json -# log4cxx -# EXPAT::EXPAT -#) +add_executable(mros_sub_test + src/logging/logging.cpp + src/mros/node.cpp + src/mros/mros.cpp + src/mros/subscriber.cpp + test/mros/test_subscriber.cpp + ${SOCKET_SOURCES} +) +target_include_directories(mros_sub_test + PRIVATE + include +) +target_link_libraries(mros_sub_test + PRIVATE + nlohmann_json::nlohmann_json + log4cxx + EXPAT::EXPAT +) diff --git a/include/mediator/mediator.hpp b/include/mediator/mediator.hpp index 19d9eeb..a2a41fc 100644 --- a/include/mediator/mediator.hpp +++ b/include/mediator/mediator.hpp @@ -14,71 +14,197 @@ using namespace std::chrono_literals; using Json = nlohmann::json; + +/** + * Signal handler for mediator. Similar to MROS + */ class MediatorSignalHandler { public: friend class Mediator; + /** + * Constructor for MediatorSignalHandler + * @param argc number of command line arguments + * @param argv command line arguments. Number denoted by argc + */ MediatorSignalHandler(int argc, char** argv); + /** + * Default dtor for MediatorSignalHandler + */ ~MediatorSignalHandler() = default; + /** + * Register signal handler for Mediator + */ void registerHandler(); + /** + * Status of MediatorSignalHandler + * @return boolean status controlled by SIGINT + */ bool getStatus(); private: + /** + * Base ctor hidden from public interface + */ MediatorSignalHandler(); + /** + * Signal handler that controls status boolean + * @param signal + */ void handleSignal(int signal); + /** + * Signal handler that is invoked in signal + * @param signal type of signal received i.e. SIGINT + */ static void staticHandleSignal(int signal); + /** + * Sets status_ to false, allowing destruction of mediator + */ void deactivateSignalHandler(); + /** + * Reference to logger + */ Logger &logger_; + + /** + * Static pointer for MROS. + */ static MediatorSignalHandler *instancePtr; + + /** + * Status controlled by signal handler + */ std::atomic status{false}; }; +/** + * Mediator that keeps a ledger of pubs and subs + */ class Mediator : public std::enable_shared_from_this { - public: + /** + * Constructor of mediator that defines non-default address and port + * @param argc number of command line arguments + * @param argv command line arguments + * @param address string address that is used + * @param port port used + */ Mediator(int argc, char** argv, std::string address, int port); + /** + * Mediator ctor with loopback address and 13330 port + * @param argc number of command line arguments + * @param argv command line arguments + */ Mediator(int argc, char** argv); + /** + * Deleted base destructor of mediator + */ Mediator() = delete; + /** + * Mediator destructor + * Cleans up threads and sends closing messages to pubs and subs + */ ~Mediator(); + /** + * Handles incoming RPC calls initializes listener + */ void spin(); private: + /** + * Created listener thread for RPC calls + */ void RPCListenerThread(); + /** + * Callback to add subscriber + * @param topic_name name of topic to be subscribed + * @param host socket host + * @param port socket port + */ void addSubscriber(std::string const &topic_name, std::string const &host, int const port); + /** + * Callback to add publisher + * @param topic_name name of topic to be published + * @param host socket host + * @param port socket port + */ void addPublisher(std::string const &topic_name, std::string const &host, int const port); + /** + * Callback to remove subscriber + * @param topic_name name of topic to be unsubscribed + * @param host socket host + * @param port socket port + */ void removeSubscriber(std::string const &topic_name, std::string const &host, int const port); + /** Callback to remove publisher + * @param topic_name name of topic to be unpublished + * @param host socket host + * @param port socket port + */ void removePublisher(std::string const &topic_name, std::string const &host, int const port); + /** + * Callback to add node + * @param node_name name of node + * @param host socket host + * @param port socket port + */ void addNode(std::string const &node_name, std::string const &host, int const port); + /** + * Callback to remove node + * @param node_name name of node + * @param host socket host + * @param port socket port + */ void removeNode(std::string const &node_name, std::string const &host, int const port); + /** + * Address of mediator + */ std::string address_; + + /** + * Mediator port + */ int port_; + + /** + * Logger + */ Logger &logger_; + /** + * Signal handler + */ MediatorSignalHandler handler; - + + /** + * file descriptor for server + */ int server_fd_; + /** + * Data type that stores topic information. Allows comparison for all members except file descriptor for ease of use + */ class Topic { public: Topic() = default; @@ -137,40 +263,48 @@ class Mediator : public std::enable_shared_from_this { }; - std::unordered_map> subscriberTable_; //URI -> vector of topics - std::unordered_map> publisherTable_; //URI -> vector of topics - std::unordered_map nodeTable_; //URI -> name + /** + * converts URI to topic for subscribers (uri example: http://127.0.0.1:8080/topic_name) + */ + std::unordered_map> subscriberTable_; + + /** + * Converts URI to topic for publishers + */ + std::unordered_map> publisherTable_; + + /** + * Converts URI to name of node + */ + std::unordered_map nodeTable_; + /** + * Mutex to control subscriberTable_ + */ std::mutex subMutex_; + + /** + * Mutex to control publisherTable_ + */ std::mutex pubMutex_; + + /** + * Mutex to control nodeTable_ + */ std::mutex nodeMutex_; + /** + * Thread that allows for incoming connections and messages + */ std::thread RPCListenerThread_; - + /** + * Accessed the signal handler's state + * @return boolean controlled by SIGINT + */ bool status() { return handler.getStatus(); } - - // void createMainSocket(); - - // void startListening(); - - // void jsonCallback(Json const &json, int const new_sock_fd); - - // bool publishCallback(Topic const &topic); - - // bool subscribeCallback(Topic const &topic); - - // bool unpublishCallback(Topic const &topic); - - // bool unsubscribeCallback(Topic const &topic); - - // bool createNodeCallback(std::string const &node_name); - - // bool bsonSender(const Json &json, int new_sock_fd); - - }; diff --git a/include/messages/exampleMessages.hpp b/include/messages/exampleMessages.hpp index e8982bf..e4d75f5 100644 --- a/include/messages/exampleMessages.hpp +++ b/include/messages/exampleMessages.hpp @@ -9,9 +9,19 @@ using Json = nlohmann::json; namespace Messages { +/** + * Sample string message + */ struct String { + /** + * String message to be sent + */ std::string base; + /** + * Name of message. Must be known at compile time (static) and must be in each message definition + * @return + */ static std::string messageName() { // return the exact name of the struct // idk how I want to handle namespaces @@ -19,6 +29,11 @@ namespace Messages { return "String"; } + /** + * Conversion to and from Json MACRO + * @param nlohmann_json_j name of struct + * @param nlohmann_json_t list of type unsafe names in order + */ NLOHMANN_DEFINE_TYPE_INTRUSIVE(String, base) }; } diff --git a/include/mros/mros.hpp b/include/mros/mros.hpp index 6b1c5e4..e3e597b 100644 --- a/include/mros/mros.hpp +++ b/include/mros/mros.hpp @@ -6,37 +6,105 @@ #include #include "logging/logging.hpp" +/** + * MROS class that handles signals and logger + */ class MROS { public: + /** + * Destructor for MROS singleton + */ ~MROS(); + /** + * Static initializer for MROS. Determines whether logger is in debugging mode or etc + * @param argc number of arguments from command line + * @param argv command line arguments. Number determined by argc + */ static void init(int argc, char** argv); + /** + * Gives access to MROS as a signal handler and the logger + * @return Reference to static MROS object if and only if MROS::init has been called prior + */ static MROS& getMROS(); + /** + * Access logger through MROS + * @return reference to logger, as defined in logging + */ Logger& getLogger(); + /** + * State variable for overriden signal handler + * @return boolean denoting state of MROS + */ bool status(); + /** + * Public interface for signal handler + */ void registerHandler(); - MROS(MROS const& mr) = delete; + /** + * Removes copy constructor for MROS. Move only + * @param mros + */ + MROS(MROS const& mros) = delete; + /** + * Removes assignment operator for MROS. Move only + * @param mros + */ void operator=(MROS const& mros) = delete; protected: + /** + * MROS object pointer. Static for use in signal handler + */ static MROS* mros_ptr_; + + /** + * Reference to Logger + */ Logger& logger_; + + /** + * Boolean for whether MROS::init has been called prior to fetching logger reference + */ bool check_logger_; + + /** + * Thread-safe (atomic) boolean controlled by SIGINT. Used for determining when to kill node + */ std::atomic_bool status_; private: + /** + * Base constructor removed from public interface. + */ MROS(); + /** + * Constructor to create static mros pointer. Initializes MROS privately + * @param argc number of command line arguments + * @param argv command line arguments, where number is given by argc + */ MROS(int argc, char** argv); + /** + * Signal handler invoked by staticHandleSignal that changes status_ variable + * @param signal required parameter that denotes signal type i.e. SIGINT + */ void handleSignal(int signal); + /** + * Static signal handler for use in registerHandler + * @param signal required parameter that denotes signal type i.e. SIGINT + */ static void staticHandleSignal(int signal); + /** + * invoked by static signal handler if SIGINT is received. Changes status_ to false + */ void deactivateSignal(); }; diff --git a/include/mros/node.hpp b/include/mros/node.hpp index 247ca9e..de7727a 100644 --- a/include/mros/node.hpp +++ b/include/mros/node.hpp @@ -16,34 +16,94 @@ using namespace std::chrono_literals; +/** + * Factory class for Publishers and Subscribers + */ class Node : public std::enable_shared_from_this { public: + /** + * Base contructor for Node class + * @param node_name String that denoted the name of Node. Must be unique to other nodes + */ explicit Node(std::string const &node_name); + /** + * Destructor for Node. Must + * Must close threads and do closing handshake via JSON RPC + */ ~Node(); + /** + * Creator of subscribers. + * @tparam MessageT Type of message to be received. Requires conversion from bson to json to MessageT + * @tparam CallbackT Callback to be called with a type MessageT + * @tparam SubscriberT The type of subscriber created by by create_subscriber. We only have one subscriber type + * @param topic_name String name of topic. Shared between publishers and subscribers + * @param queue_size The number of messages to be allowed on the message queue before they are dismissed + * @param callback The callback to be called + * @return shared pointer to subscriber + */ template > std::shared_ptr create_subscriber(std::string topic_name, std::uint32_t queue_size, CallbackT &&callback); + /** + * Creator of subscribers + * @tparam MessageT Type of message to be received. Requires conversion from bson to json to MessageT + * @tparam PublisherT The type of publisher created by by create_publisher. We only have one publisher type + * @param topic_name String name of topic. Shared between publishers and subscribers + * @param queue_size The number of messages to be allowed on the message queue before they are dismissed + * @return shared pointer to publisher + */ template > std::shared_ptr create_publisher(std::string topic_name, uint32_t queue_size); + /** + * Continuously handles incoming information such as new publishers and subscribed messages + */ void spin(); + /** + * handles incoming information such as new publishers and subscribed messages once + */ void spinOnce(); + /** + * MROS status accessor from node + * @return status of MROS + */ bool status() const; private: + /** + * Name of node + */ std::string node_name_; + /** + * Reference to logger, invoked by Logger::getLogger() + */ Logger &logger_; + + /** + * Reference to MROS + */ MROS &core_; + /** + * List of subscribers, owned in main + */ std::vector> subs_; + + /** + * List of publishers, owned in main + */ std::vector> pubs_; + + /** + * Thread pool as needed + */ // std::vector threadPool_; }; diff --git a/include/mros/publisher.hpp b/include/mros/publisher.hpp index a9e1339..6a34a1c 100644 --- a/include/mros/publisher.hpp +++ b/include/mros/publisher.hpp @@ -18,8 +18,9 @@ using namespace std::chrono_literals; +// Lets PublisherBase know Node exists without inclusion to know it exists class Node; - +// Non-templated base class for Publishers to allow for ownership by Pointer class PublisherBase { public: friend class Node; @@ -27,23 +28,55 @@ class PublisherBase { virtual ~PublisherBase() = default; }; +/** + * Publisher class + * @tparam MessageT Message Type. Requires conversion to and from json + */ template class Publisher : public std::enable_shared_from_this>, public PublisherBase { public: - + /** + * Deleted base constructor + */ Publisher() = delete; + /** + * Publisher constructor. Please use constructor from node + * @param node Non-owning pointer of node + * @param topic_name String name of topic shared between publishers and subscribers + * @param queue_size Number of messages allowed on the message queue before refusal + */ + Publisher(std::weak_ptr node, std::string topic_name, std::uint32_t queue_size); + + /** + * Overridden publisher destructor + * Requirements currently unknown TODO + */ ~Publisher() override; + /** + * Pubishes message on topic defined by ctor + * @param msg Message to be sent + */ void publish(MessageT const &msg); + /** + * Trivial accessor for Topic Name + * @return topic name + */ std::string getTopicName() const; private: - Publisher(std::weak_ptr node, std::string topic_name, std::uint32_t queue_size); + /** + * MROS status accessor for Publisher + * @return + */ bool status(); + /** + * TODO + */ void socketListener(); std::string topic_name_; diff --git a/include/mros/subscriber.hpp b/include/mros/subscriber.hpp index c5af86c..46f7b48 100644 --- a/include/mros/subscriber.hpp +++ b/include/mros/subscriber.hpp @@ -20,6 +20,11 @@ using namespace std::chrono_literals; +class Node; + +/* + * Non-templated base class for ownership + */ class SubscriberBase { public: friend class Node; @@ -33,44 +38,119 @@ class SubscriberBase { }; class Node; - +/** + * Subscriber class to send messages + * @tparam MessageT Type of message to be sent. Requires conversion to and from json + */ template class Subscriber : public std::enable_shared_from_this>, public SubscriberBase { public: friend class Node; + /** + * Deleted base ctor for Subscriber + */ Subscriber() = delete; + /** + * Ctor for subscriber. Please use Node::create_subscriber + * @param node Non-owning node pointer + * @param topic_name String name of topic + * @param queue_size The number of messages allowed on the queue before refusal + * @param callback The callback function to be called + */ + Subscriber(std::weak_ptr node, std::string topic_name, std::uint32_t queue_size, std::function callback); + + /** + * Overriden destructor for subscriber + * TODO: outline functionality + */ ~Subscriber() override; + /** + * Trivial accessor for topic name + * @return topic name + */ std::string getTopicName() const; private: - Subscriber(std::weak_ptr node, std::string topic_name, std::uint32_t queue_size, std::function callback); + /** + * Handles continous incoming messages to be called + */ void spin() override; + /** + * Handles message once + */ void spinOnce() override; + /** + * Access status of MROS + * @return MROS status + */ bool status(); + /** + * Creates thread to receive incoming messages on one file descriptor + * @param socket_fd File descriptor + */ void receiverThread(int socket_fd); + /** + * string name of topic + */ std::string topic_name_; + + /** + * number of messages to be allowed on the queue + */ std::uint32_t queue_size_; + /** + * Callback to be called with. Takes in Message of type MessageT + */ std::function callbackFunc; - + + /** + * Non-owning pointer to node used to construct object + */ std::weak_ptr node_; + /** + * Conidition variable that locks receiving messages until new publisher is added to list and connection is created + */ std::condition_variable spin_cv; //Broadcasted when spin() or spinOnce() is called. Blocks all receivers. - std::unordered_map publisherListenerThreads_; // socket fd -> thread. Makes sure we have one thread per socket exactly + /** + * Lookup socket file descriptor -> thread. One to one + */ + std::unordered_map publisherListenerThreads_; + + /** + * Mutex for publisherListenerThreads_ to prevent race conditions + */ std::mutex publisherListenerThreadsMutex_; + + /** + * Number of threads waiting + * TODO: More information + */ int numThreadsWaiting_ = 0; + + /** + * TODO: Unsure + */ bool listenerThreadsCondition = false; + /** + * Reference to logger + */ Logger &logger_; + + /** + * Reference to MROS + */ MROS &core_; }; diff --git a/test/mros/test_publisher.cpp b/test/mros/test_publisher.cpp index 0ae9f32..a7b20c3 100644 --- a/test/mros/test_publisher.cpp +++ b/test/mros/test_publisher.cpp @@ -1,7 +1,7 @@ #include #include "messages/exampleMessages.hpp" -#include "mros/node.hpp" +#include int main(int argc, char **argv) { MROS::init(argc, argv); From 5d4c2ffeaa3afe4d5ed87013d43fac7d218165e4 Mon Sep 17 00:00:00 2001 From: jksully Date: Fri, 5 Jan 2024 16:00:25 -0500 Subject: [PATCH 2/2] Added comments --- include/mediator/mediator.hpp | 30 ++++++++++++++++++++++-------- include/mros/publisher.hpp | 30 +++++++++++++++++++++++++++++- 2 files changed, 51 insertions(+), 9 deletions(-) diff --git a/include/mediator/mediator.hpp b/include/mediator/mediator.hpp index a2a41fc..51f4b05 100644 --- a/include/mediator/mediator.hpp +++ b/include/mediator/mediator.hpp @@ -11,6 +11,9 @@ #include #include "logging/logging.hpp" +#include +#include + using namespace std::chrono_literals; using Json = nlohmann::json; @@ -130,8 +133,13 @@ class Mediator : public std::enable_shared_from_this { */ void RPCListenerThread(); + /* + * ======================================= + * These are callbacks to be linked to RPC + * ======================================= + * */ /** - * Callback to add subscriber + * When we call Node::create_subscriber(), the node calls this callback over RPC internally * @param topic_name name of topic to be subscribed * @param host socket host * @param port socket port @@ -139,7 +147,7 @@ class Mediator : public std::enable_shared_from_this { void addSubscriber(std::string const &topic_name, std::string const &host, int const port); /** - * Callback to add publisher + * When we call Node::create_publisher(), the node calls this callback over RPC internally * @param topic_name name of topic to be published * @param host socket host * @param port socket port @@ -147,14 +155,14 @@ class Mediator : public std::enable_shared_from_this { void addPublisher(std::string const &topic_name, std::string const &host, int const port); /** - * Callback to remove subscriber + * when we destruct a subscriber, the node calls this callback over RPC internally * @param topic_name name of topic to be unsubscribed * @param host socket host * @param port socket port */ void removeSubscriber(std::string const &topic_name, std::string const &host, int const port); - /** Callback to remove publisher + /** When we destruct a publisher, the node calls this callback over RPC internally * @param topic_name name of topic to be unpublished * @param host socket host * @param port socket port @@ -162,7 +170,7 @@ class Mediator : public std::enable_shared_from_this { void removePublisher(std::string const &topic_name, std::string const &host, int const port); /** - * Callback to add node + * When a node is constructed, the node calls this callback over RPC internally * @param node_name name of node * @param host socket host * @param port socket port @@ -170,13 +178,19 @@ class Mediator : public std::enable_shared_from_this { void addNode(std::string const &node_name, std::string const &host, int const port); /** - * Callback to remove node + * When a node is destructed, the node calls this callback over RPC internally * @param node_name name of node * @param host socket host * @param port socket port */ void removeNode(std::string const &node_name, std::string const &host, int const port); + /* + * ============= + * End callbacks + * ============= + * */ + /** * Address of mediator */ @@ -198,9 +212,9 @@ class Mediator : public std::enable_shared_from_this { MediatorSignalHandler handler; /** - * file descriptor for server + * owning pointer to ServerSocket for RPC calls between mediator and nodes */ - int server_fd_; + std::shared_ptr server_rpc_socket_; /** * Data type that stores topic information. Allows comparison for all members except file descriptor for ease of use diff --git a/include/mros/publisher.hpp b/include/mros/publisher.hpp index 6a34a1c..6872bd4 100644 --- a/include/mros/publisher.hpp +++ b/include/mros/publisher.hpp @@ -16,6 +16,8 @@ #include "logging/logging.hpp" #include "mros/mros.hpp" +#include + using namespace std::chrono_literals; // Lets PublisherBase know Node exists without inclusion to know it exists @@ -79,17 +81,43 @@ class Publisher : public std::enable_shared_from_this>, publ */ void socketListener(); + /** + * name of topic + */ std::string topic_name_; + + /** + * Number of messages that can be published before refusal + */ std::uint32_t queue_size_; + /** + * Non-owning pointer of node + */ std::weak_ptr node_; + /** + * Thread for listening for incoming subscriber connections + */ std::thread socketListenerThread_; - std::vector subscribers_; + /** + * Vector to contain information about subscribers + */ + std::vector> subscribers_; + + /** + * Mutex for controlling subscribers_ + */ std::mutex subscribersMutex_; + /** + * Logger for debugging + */ Logger &logger_; + /** + * MROS for signal handling and logger initialization + */ MROS &core_; };