qiurui
/
uvgrtp-base
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Allow pkt_dispatcher thread access to pkt_dispatcher's state 

This makes the code and interface cleaner
This commit is contained in:
Aaro Altonen 2021-02-26 00:55:30 +02:00
parent 6ceec24473
commit 367a8ad465
2 changed files with 12 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
include/pkt_dispatch.hh
View File

@ -87,8 +87,9 @@ namespace uvgrtp {
uvgrtp::frame::rtp_frame *pull_frame(); uvgrtp::frame::rtp_frame *pull_frame();
uvgrtp::frame::rtp_frame *pull_frame(size_t ms); uvgrtp::frame::rtp_frame *pull_frame(size_t ms);
/* Return reference to the map that holds all installed handlers */ private:
std::unordered_map<uint32_t, uvgrtp::packet_handlers>& get_handlers(); /* RTP packet dispatcher thread */
void runner(uvgrtp::socket *socket, int flags);
/* Return a processed RTP frame to user either through frame queue or receive hook */ /* Return a processed RTP frame to user either through frame queue or receive hook */
void return_frame(uvgrtp::frame::rtp_frame *frame); void return_frame(uvgrtp::frame::rtp_frame *frame);
@ -96,15 +97,7 @@ namespace uvgrtp {
/* Call auxiliary handlers of a primary handler */ /* Call auxiliary handlers of a primary handler */
void call_aux_handlers(uint32_t key, int flags, uvgrtp::frame::rtp_frame **frame); void call_aux_handlers(uint32_t key, int flags, uvgrtp::frame::rtp_frame **frame);
/* RTP packet dispatcher thread */ /* Primary handlers for the socket */
static void runner(
uvgrtp::pkt_dispatcher *dispatcher,
uvgrtp::socket *socket,
int flags,
std::mutex *exit_mtx
);
private:
std::unordered_map<uint32_t, packet_handlers> packet_handlers_; std::unordered_map<uint32_t, packet_handlers> packet_handlers_;
/* If receive hook has not been installed, frames are pushed to "frames_" /* If receive hook has not been installed, frames are pushed to "frames_"

27
src/pkt_dispatch.cc
View File

@ -24,7 +24,7 @@ uvgrtp::pkt_dispatcher::~pkt_dispatcher()
rtp_error_t uvgrtp::pkt_dispatcher::start(uvgrtp::socket *socket, int flags) rtp_error_t uvgrtp::pkt_dispatcher::start(uvgrtp::socket *socket, int flags)
{ {
if (!(runner_ = new std::thread(runner, this, socket, flags, &exit_mtx_))) if (!(runner_ = new std::thread(&uvgrtp::pkt_dispatcher::runner, this, socket, flags)))
return RTP_MEMORY_ERROR; return RTP_MEMORY_ERROR;
runner_->detach(); runner_->detach();
@ -122,11 +122,6 @@ rtp_error_t uvgrtp::pkt_dispatcher::install_aux_handler(
return RTP_OK; return RTP_OK;
} }
std::unordered_map<uint32_t, uvgrtp::packet_handlers>& uvgrtp::pkt_dispatcher::get_handlers()
{
return packet_handlers_;
}
void uvgrtp::pkt_dispatcher::return_frame(uvgrtp::frame::rtp_frame *frame) void uvgrtp::pkt_dispatcher::return_frame(uvgrtp::frame::rtp_frame *frame)
{ {
if (recv_hook_) { if (recv_hook_) {
@ -212,12 +207,7 @@ void uvgrtp::pkt_dispatcher::call_aux_handlers(uint32_t key, int flags, uvgrtp::
* *
* If a handler receives a non-null "out", it can safely ignore "packet" and operate just on * If a handler receives a non-null "out", it can safely ignore "packet" and operate just on
* the "out" parameter because at that point it already contains all needed information. */ * the "out" parameter because at that point it already contains all needed information. */
void uvgrtp::pkt_dispatcher::runner( void uvgrtp::pkt_dispatcher::runner(uvgrtp::socket *socket, int flags)
uvgrtp::pkt_dispatcher *dispatcher,
uvgrtp::socket *socket,
int flags,
std::mutex *exit_mtx
)
{ {
int nread; int nread;
fd_set read_fds; fd_set read_fds;
@ -232,14 +222,13 @@ void uvgrtp::pkt_dispatcher::runner(
const size_t recv_buffer_len = 8192; const size_t recv_buffer_len = 8192;
uint8_t recv_buffer[recv_buffer_len] = { 0 }; uint8_t recv_buffer[recv_buffer_len] = { 0 };
auto handlers = dispatcher->get_handlers();
while (!dispatcher->active()) while (!this->active())
; ;
exit_mtx->lock(); exit_mtx_.lock();
while (dispatcher->active()) { while (this->active()) {
FD_SET(socket->get_raw_socket(), &read_fds); FD_SET(socket->get_raw_socket(), &read_fds);
int sret = ::select(socket->get_raw_socket() + 1, &read_fds, nullptr, nullptr, &t_val); int sret = ::select(socket->get_raw_socket() + 1, &read_fds, nullptr, nullptr, &t_val);
@ -257,7 +246,7 @@ void uvgrtp::pkt_dispatcher::runner(
break; break;
} }
for (auto& handler : handlers) { for (auto& handler : packet_handlers_) {
switch ((ret = (*handler.second.primary)(nread, recv_buffer, flags, &frame))) { switch ((ret = (*handler.second.primary)(nread, recv_buffer, flags, &frame))) {
/* packet was handled successfully */ /* packet was handled successfully */
case RTP_OK: case RTP_OK:
@ -270,7 +259,7 @@ void uvgrtp::pkt_dispatcher::runner(
/* packet was handled by the primary handler /* packet was handled by the primary handler
* and should be dispatched to the auxiliary handler(s) */ * and should be dispatched to the auxiliary handler(s) */
case RTP_PKT_MODIFIED: case RTP_PKT_MODIFIED:
dispatcher->call_aux_handlers(handler.first, flags, &frame); this->call_aux_handlers(handler.first, flags, &frame);
break; break;
case RTP_GENERIC_ERROR: case RTP_GENERIC_ERROR:
@ -285,5 +274,5 @@ void uvgrtp::pkt_dispatcher::runner(
} while (ret == RTP_OK); } while (ret == RTP_OK);
} }
exit_mtx->unlock(); exit_mtx_.unlock();
} }