Skip to content

Node - Mediator Interface Definitions #6

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
JosephUT wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Node - Mediator Interface Definitions #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
447df94
Added header comments
JKSully Jan 5, 2024
5d4c2ff
Added comments
JKSully Jan 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 36 additions & 36 deletions CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
)
202 changes: 175 additions & 27 deletions include/mediator/mediator.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,74 +11,214 @@
#include <nlohmann/json.hpp>
#include "logging/logging.hpp"

#include <socket/json_rpc_socket/connection_json_rpc_socket.hpp>
#include <socket/server_socket.hpp>

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<bool> status{false};
};

/**
* Mediator that keeps a ledger of pubs and subs
*/
class Mediator : public std::enable_shared_from_this<Mediator> {

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();

@JosephUT JosephUT Jan 5, 2024 •
edited
Loading

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is where the ServerSocket is accepting incoming connections made by Node's JsonRPCClientSocket, right?

We accept the connection, setup the appropriate callbacks (that you put below) for the resulting ConnectionSocket, and then start up the RPC for that socket and store the shared_ptr to the RPC socket so we can talk to that node in the future.


/*
* =======================================
* These are callbacks to be linked to RPC
* =======================================
* */
/**
* 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
*/
void addSubscriber(std::string const &topic_name, std::string const &host, int const port);
Comment thread
JKSully marked this conversation as resolved.

/**
* 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
*/
void addPublisher(std::string const &topic_name, std::string const &host, int const port);
Comment thread
JKSully marked this conversation as resolved.

/**
* 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);

@JosephUT JosephUT Jan 5, 2024 •
edited
Loading

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this maybe makes sense to call from the destructor of subscriber? But then subscriber has to internally make a call to a node to get node to call this callback on mediator over rpc since only node is connected to the mediator. There's probably two ways we could go:

  1. Since there isn't a delete_subscriber() function for node the subscriber will only be destroyed when the node is destroyed. We could demote this to a helper function that gets called by remove_node() callback you made below. As the node dies it calls the remove node callback on mediator and the mediator then looks up the publishers and subscribers associated with that node and calls remove publisher and remove subscriber on them.

The user can't delete the publisher's and subscriber's individually because it's a weak_ptr or something right? The publishers and subscribers are owned by node?

  1. We add a delete_subscriber() function to the Node interface. Then this makes sense and gets called internally by the delete_subscriber function over rpc.

Whichever seems best to you, or if you have another way.

@JKSully JKSully Jan 5, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Definitely should be called in the dtor of sub

@JosephUT JosephUT Jan 5, 2024

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

After reading the whole codebase again I really don't agree with this. I think our implementation will be much simpler if we just rely on Node getting destroyed to call the callbacks that remove the appropriate publishers and subscribers. The Publisher and Subscriber have life cycles only can end when Node is destroyed so it doesn't make sense to separate these into entirely different callbacks. Keeping them as helper functions is good for compartmentalization though.

@JosephUT JosephUT Jan 6, 2024

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Discussed


/** 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
*/
void removePublisher(std::string const &topic_name, std::string const &host, int const port);

@JosephUT JosephUT Jan 5, 2024

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Exactly the same comment as above about removePublisher().


/**
* 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
*/
void addNode(std::string const &node_name, std::string const &host, int const port);
Comment thread
JKSully marked this conversation as resolved.

/**
* 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);

@JosephUT JosephUT Jan 5, 2024

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Gets registered as the closing callback for the rpc connection to each Node?

If the Node gets killed locally, it closes the rpc connection and the mediator automatically calls this removeNode callback.

If the Mediator decides to kill the node, then the mediator closes the rpc connection and which automatically calls this removeNode callback and the Node's closing callback sequence.

@JosephUT JosephUT Jan 5, 2024

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Question that applies to all of the above callbacks:

Should we also have little wrapper functions for each of these callbacks that handle parsing the string and int arguments from a single JSON object using keywords?

To actually register these as callbacks for the JsonRPCSocket's they need to just take one JSON object as input. I think the way you have it right now is way more readable so maybe just little wrapper functions that take a JSON as a parameter then parse out std::string node_name, std::string host, int port, etc? Could call it like removeNodeJson, addSubscriberJson, or something


/*
* =============
* End callbacks
* =============
* */

/**
* Address of mediator
*/
std::string address_;

/**
* Mediator port
*/
int port_;

/**
* Logger
*/
Logger &logger_;

/**
* Signal handler
*/
MediatorSignalHandler handler;

int server_fd_;

/**
* owning pointer to ServerSocket for RPC calls between mediator and nodes
*/
std::shared_ptr<ServerSocket> server_rpc_socket_;

/**
* Data type that stores topic information. Allows comparison for all members except file descriptor for ease of use
*/
class Topic {
public:
Topic() = default;
Expand Down Expand Up @@ -137,40 +277,48 @@ class Mediator : public std::enable_shared_from_this<Mediator> {

};

std::unordered_map<std::string, std::vector<Topic>> subscriberTable_; //URI -> vector of topics
std::unordered_map<std::string, std::vector<Topic>> publisherTable_; //URI -> vector of topics
std::unordered_map<std::string, std::string> nodeTable_; //URI -> name
/**
* converts URI to topic for subscribers (uri example: http://127.0.0.1:8080/topic_name)
*/
std::unordered_map<std::string, std::vector<Topic>> subscriberTable_;

/**
* Converts URI to topic for publishers
*/
std::unordered_map<std::string, std::vector<Topic>> publisherTable_;

/**
* Converts URI to name of node
*/
std::unordered_map<std::string, std::string> nodeTable_;

@JosephUT JosephUT Jan 5, 2024

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we wanted to have the removeNode() callback also remove publishers and subscribers, we would need a node URI to publisher and subsciber URI map or maps.

Or we could make nodeTable map to a struct with the node's name, a vector of publisher URI's, and a vector of subscriber URI's.

Only necessary if we do option 1 mentioned above next to removeNode() function.

@JosephUT JosephUT Jan 5, 2024 •
edited
Loading

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We definitely also need a vector of Node URI's mapping shared_ptr to each other somewhere. We need to know which rpc socket talks to which node.

If we edit nodeTable to store a struct of stuff as the value for the unordered_map, we could add the pointer to the rpc socket in there as well. That would probably make the most sense since there's a lot of different information we might want to know about a Node.

  • name
  • publisher URI's
  • subscriber URI's
  • ptr to connection json rpc socket

/**
* 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);


};


Expand Down
15 changes: 15 additions & 0 deletions include/messages/exampleMessages.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,16 +9,31 @@
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
// i.e. Messages::String

@JosephUT JosephUT Jan 5, 2024

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ahhhhh the message interface is so tricky.

Maybe we want to have an abstract base message class?

  • Pure virtual function to initialize a the message from a json
  • Pure virtual function to generate a json from the content of the message (Both would use that cool macro on line 37)

Then to make different message types the users of the library just inherit from the base message class and define all the message member variables they want. They are forced to define to and from Json functions because of the pure virtual functions, so we can always rely on them being there in other pieces of code.

That's the simplest thing I can think of for making these a little more uniform. I'll open this as a maybe for later, low priority issue.

I think having a Messages:: namespace seems like a good idea for sure though

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)
};
}
Expand Down
Loading