Participant_manager Class Reference

Collaboration diagram for Participant_manager:

Public Member Functions
	Participant_manager (module &mod, const char *compress)
	~Participant_manager ()
void	run_async (std::shared_ptr< socket_udp > server_sock)
void	tick (sockaddr_storage &sin, socklen_t addrlen)

Private Member Functions
void	worker ()

Private Attributes
std::string	compression
struct module &	mod
std::vector< Conf_participant >	participants
std::atomic< bool >	should_stop = false
std::shared_ptr< socket_udp >	sock
std::thread	worker_thread

Detailed Description

Definition at line 868 of file hd-rum-translator.cpp.

Constructor & Destructor Documentation

Participant_manager()

Participant_manager::Participant_manager ( module & mod, const char * compress )

inline

Definition at line 870 of file hd-rum-translator.cpp.

: mod(mod), compression(compress ? compress : "") {  }

~Participant_manager()

Participant_manager::~Participant_manager ( )

inline

Definition at line 871 of file hd-rum-translator.cpp.

                              {
                should_stop.store(true, std::memory_order_relaxed);
                if(worker_thread.joinable())
                    worker_thread.join();
        }

References should_stop, and worker_thread.

Member Function Documentation

run_async()

void Participant_manager::run_async ( std::shared_ptr< socket_udp > server_sock )

inline

Definition at line 940 of file hd-rum-translator.cpp.

                                                             {
                sock = std::move(server_sock);
                worker_thread = std::thread(&Participant_manager::worker, this);
        }

References sock, worker(), and worker_thread.

Referenced by main().

Here is the call graph for this function:

tick()

void Participant_manager::tick ( sockaddr_storage & sin, socklen_t addrlen )

inline

Definition at line 877 of file hd-rum-translator.cpp.

                                                           {
                struct timeval t;
                gettimeofday(&t, NULL);
                bool seen = false;
                for(auto it = participants.begin(); it != participants.end();){
                        if(addrlen > 0 && sockaddr_equal(&it->addr, &sin)){
                                it->last_recv = t;
                                seen = true;
                        }
                        const double participant_timeout = 10.0;
                        if(tv_diff(t, it->last_recv) > participant_timeout){
                                log_msg(LOG_LEVEL_NOTICE, "Removing participant\n");
                                std::string msg = "delete-port ";
                                auto addr = reinterpret_cast<struct sockaddr *>(&it->addr);
                                char addr_str[128];
                                get_sockaddr_addr_str(addr, addr_str, sizeof(addr_str));
                                char *replica_name = get_replica_mod_name(addr_str, get_sockaddr_addr_port(addr));
                                msg += replica_name;
                                free(replica_name);
 
                                std::swap(*it, participants.back());
                                participants.pop_back();
 
                                struct msg_universal *m = (struct msg_universal *) new_message(sizeof(struct msg_universal));
                                strncpy(m->text, msg.c_str(), sizeof(m->text) - 1);
                                log_msg(LOG_LEVEL_NOTICE, "Msg: %s\n", m->text);
                                struct response *r = send_message_to_receiver(&mod, (struct message *) m);
                                if (response_get_status(r) != RESPONSE_ACCEPTED) {
                                        log_msg(LOG_LEVEL_ERROR, "Couldn't remove participant (error %d)!\n", response_get_status(r));
                                }
                                free_response(r);
                        } else {
                                it++;
                        }
                }
 
                if(!seen && addrlen > 0){
                        Conf_participant p;
                        p.addr = sin;
                        p.addrlen = addrlen;
                        p.last_recv = t;
 
                        log_msg(LOG_LEVEL_NOTICE, "New participant\n");
                        std::string msg = "create-port ";
                        msg += get_sockaddr_str(reinterpret_cast<sockaddr *>(&sin));
                        if(!compression.empty()){
                                msg += " ";
                                msg += compression;
                        }
 
                        struct msg_universal *m = (struct msg_universal *) new_message(sizeof(struct msg_universal));
                        strncpy(m->text, msg.c_str(), sizeof(m->text) - 1);
                        log_msg(LOG_LEVEL_NOTICE, "Msg: %s\n", m->text);
                        struct response *r = send_message_to_receiver(&mod, (struct message *) m);
                        if (response_get_status(r) != RESPONSE_ACCEPTED) {
                                log_msg(LOG_LEVEL_ERROR, "Cannot add new participant (error %d)!\n", response_get_status(r));
                        } else{
                                participants.push_back(p);
                        }
                        free_response(r);
                }
        }

References Conf_participant::addr, Conf_participant::addrlen, compression, free_response(), get_replica_mod_name(), get_sockaddr_addr_port(), get_sockaddr_addr_str(), get_sockaddr_str(), Conf_participant::last_recv, LOG_LEVEL_ERROR, LOG_LEVEL_NOTICE, log_msg(), mod, new_message(), participants, RESPONSE_ACCEPTED, response_get_status(), send_message_to_receiver(), sockaddr_equal(), msg_universal::text, and tv_diff().

Referenced by main(), and worker().

Here is the call graph for this function:

worker()

void Participant_manager::worker ( )

inlineprivate

Definition at line 953 of file hd-rum-translator.cpp.

                     {
            struct timeval timeout = { 1, 0 };
            char dummy_buf[256];
 
            while(!should_stop.load(std::memory_order_relaxed)){
                    struct sockaddr_storage sin = {};
                    socklen_t addrlen = sizeof(sin);
                    int size = udp_recvfrom_timeout(sock.get(), dummy_buf, std::size(dummy_buf), &timeout, (sockaddr *) &sin, &addrlen);
 
                    if(size > 0)
                            tick(sin, addrlen);
            }
        }

References should_stop, sock, tick(), and udp_recvfrom_timeout().

Referenced by run_async().

Here is the call graph for this function:

Member Data Documentation

compression

std::string Participant_manager::compression

private

Definition at line 948 of file hd-rum-translator.cpp.

Referenced by tick().

mod

struct module& Participant_manager::mod

private

Definition at line 947 of file hd-rum-translator.cpp.

Referenced by tick().

participants

std::vector<Conf_participant> Participant_manager::participants

private

Definition at line 946 of file hd-rum-translator.cpp.

Referenced by tick().

should_stop

std::atomic<bool> Participant_manager::should_stop = false

private

Definition at line 951 of file hd-rum-translator.cpp.

Referenced by ~Participant_manager(), and worker().

sock

std::shared_ptr<socket_udp> Participant_manager::sock

private

Definition at line 949 of file hd-rum-translator.cpp.

Referenced by run_async(), and worker().

worker_thread

std::thread Participant_manager::worker_thread

private

Definition at line 950 of file hd-rum-translator.cpp.

Referenced by ~Participant_manager(), and run_async().

---

The documentation for this class was generated from the following file:

• src/hd-rum-translator/hd-rum-translator.cpp