control_socket.cpp File Reference

#include "control_socket.h"
#include "compat/platform_pipe.h"
#include <condition_variable>
#include <mutex>
#include <queue>
#include <string>
#include <thread>
#include "debug.h"
#include "host.h"
#include "messaging.h"
#include "module.h"
#include "rtp/net_udp.h"
#include "tv.h"
#include "utils/net.h"
#include "utils/thread.h"

Include dependency graph for control_socket.cpp:

Go to the source code of this file.

Classes
struct	client
struct	control_state

Macros
#define	CONTROL_CLOSE_HANDLE   -2
#define	CONTROL_EXIT   -1
#define	is_internal_port(x)
#define	MAX_CLIENTS   16
#define	MAX_STAT_EVENT_QUEUE   100
#define	MSG_NOSIGNAL   0
#define	prefix_matches(x, y)
#define	suffix(x, y)

Typedefs
typedef void *	sso_val_type

Enumerations
enum	connection_type { SERVER , CLIENT }

Functions
static struct client *	add_client (struct client *clients, fd_t fd)
	ADD_TO_PARAM ("control-accept-global", "* control-accept-global\n" " Open control socket to public network.\n")
	ADD_TO_PARAM ("control-report-audio-ch-count", "* control-report-audio-ch-count=<count>\n" " The number of channels reported over control port.\n")
int	control_audio_ch_report_count (struct control_state *state)
void	control_done (struct control_state *s)
int	control_init (int port, int connection_type, struct control_state **state, struct module *root_module, int force_ip_version)
void	control_report_event (struct control_state *s, const std::string &report_line)
void	control_report_stats (struct control_state *s, const std::string &report_line)
static void	control_report_stats_event (struct control_state *s, const std::string &report_line)
void	control_start (struct control_state *s)
bool	control_stats_enabled (struct control_state *s)
static void *	control_thread (void *args)
static void	new_message (struct module *m)
static bool	parse_msg (char *buffer, int buffer_len, char *message, int *new_buffer_len)
static void	print_control_help ()
static struct response *	process_audio_message (struct module *root_module, const char *cmd)
static void	process_messages (struct control_state *s)
static int	process_msg (struct control_state *s, fd_t client_fd, char *message, struct client *clients)
static void	send_response (fd_t fd, struct response *resp)
static void	set_socket_nonblock (fd_t fd)
static void *	stat_event_thread (void *args)
static ssize_t	write_all (fd_t fd, const void *buf, size_t count)

Detailed Description

Author

Martin Pulec pulec.nosp@m.@ces.nosp@m.net.c.nosp@m.z

Definition in file control_socket.cpp.

Macro Definition Documentation

CONTROL_CLOSE_HANDLE

#define CONTROL_CLOSE_HANDLE   -2

Definition at line 113 of file control_socket.cpp.

Referenced by control_thread(), and process_msg().

CONTROL_EXIT

#define CONTROL_EXIT   -1

Definition at line 112 of file control_socket.cpp.

Referenced by control_thread(), and process_msg().

is_internal_port

#define is_internal_port

(

x

)

Value:

(x == s->internal_fd[0])

Definition at line 314 of file control_socket.cpp.

Referenced by control_thread(), and process_msg().

MAX_CLIENTS

#define MAX_CLIENTS   16

Definition at line 62 of file control_socket.cpp.

Referenced by control_init().

MAX_STAT_EVENT_QUEUE

#define MAX_STAT_EVENT_QUEUE   100

Definition at line 86 of file control_socket.cpp.

Referenced by control_report_stats_event().

MSG_NOSIGNAL

#define MSG_NOSIGNAL   0

Definition at line 124 of file control_socket.cpp.

Referenced by write_all().

prefix_matches

#define prefix_matches	(		x,
			y )

Value:

strncasecmp(x, y, strlen(y)) == 0

Definition at line 312 of file control_socket.cpp.

Referenced by process_audio_message(), and process_msg().

suffix

#define suffix	(		x,
			y )

Value:

x + strlen(y)

Definition at line 313 of file control_socket.cpp.

Referenced by get_configured_capture_aliases(), get_interlacing_from_suffix(), get_nth_word(), initialize_import(), process_audio_message(), and process_msg().

Typedef Documentation

sso_val_type

typedef void* sso_val_type

Definition at line 67 of file control_socket.cpp.

Enumeration Type Documentation

connection_type

enum connection_type

Enumerator
SERVER
CLIENT

Definition at line 81 of file control_socket.cpp.

                     {
        SERVER,
        CLIENT
};

Function Documentation

add_client()

static struct client * add_client ( struct client * clients, fd_t fd )

static

prepends itself at the head of the list

Definition at line 719 of file control_socket.cpp.

                                                                  {
        struct client *new_client = (struct client *)
                malloc(sizeof(struct client));
        new_client->fd = fd;
        new_client->prev = NULL;
        new_client->next = clients;
        if (new_client->next) {
                new_client->next->prev = new_client;
        }
        new_client->buff_len = 0;
 
        return new_client;
}

References client::buff_len, client::fd, client::next, and client::prev.

Referenced by control_thread().

ADD_TO_PARAM() [1/2]

ADD_TO_PARAM	(	"control-accept-global"	,
		"* control-accept-global\n" " Open control socket to public network.\n"	)

ADD_TO_PARAM() [2/2]

ADD_TO_PARAM

(

"control-report-audio-ch-count"

)

control_audio_ch_report_count()

int control_audio_ch_report_count

(

struct control_state *

state

)

Definition at line 1021 of file control_socket.cpp.

                                                              {
        return state ? state->audio_channel_report_count : 0;
}

References state.

Referenced by decode_audio_frame().

control_done()

void control_done

(

struct control_state *

s

)

Definition at line 953 of file control_socket.cpp.

{
        if(!s) {
                return;
        }
 
        module_done(&s->mod);
 
        if(s->started) {
                s->stats_lock.lock();
                s->stat_event_queue.push({});
                s->stats_lock.unlock();
                s->stat_event_cv.notify_one();
                s->stat_event_thread_id.join();
 
                int ret = write_all(s->internal_fd[1], "quit\r\n", 6);
                if (ret > 0) {
                        s->control_thread_id.join();
                        platform_pipe_close(s->internal_fd[1]);
                } else {
                        fprintf(stderr, "Cannot exit control thread!\n");
                }
        }
        if(s->connection_type == SERVER && s->socket_fd != INVALID_SOCKET) {
                // for client, the socket has already been closed
                // by the time of control_thread exit
                CLOSESOCKET(s->socket_fd);
        }
 
        delete s;
}

References CLOSESOCKET, control_state::connection_type, control_state::control_thread_id, control_state::internal_fd, INVALID_SOCKET, control_state::mod, module_done(), platform_pipe_close(), SERVER, control_state::socket_fd, control_state::started, control_state::stat_event_cv, control_state::stat_event_queue, control_state::stat_event_thread_id, control_state::stats_lock, and write_all().

Referenced by hd_rum_translator_deinit(), and main().

Here is the call graph for this function:

control_init()

int control_init	(	int	port,
		int	connection_type,
		struct control_state **	state,
		struct module *	root_module,
		int	force_ip_version )

Parameters

[in]

force_ip_version

IP version to force (4 or 6). Use 0 to use default (both 4 and 6 if available)

Return values

0	if success

Definition at line 159 of file control_socket.cpp.

{
        control_state *s = new control_state();
 
        s->root_module = root_module;
        s->started = false;
 
        module_init_default(&s->mod);
        s->mod.cls = MODULE_CLASS_CONTROL;
        s->mod.new_message = new_message;
        s->mod.priv_data = s;
 
        if(connection_type == 0) {
                s->network_port = port;
                s->connection_type = SERVER;
        } else {
                s->network_port = port;
                s->connection_type = CLIENT;
        }
 
        if(s->connection_type == SERVER) {
                int ip_version = 6;
                if (force_ip_version != 4) {
                        s->socket_fd = socket(AF_INET6, SOCK_STREAM, 0);
                        if (s->socket_fd == INVALID_SOCKET) {
                                socket_error("Control socket - IPv6 not available");
                        }
                }
                if (force_ip_version == 4 || (force_ip_version == 0 && s->socket_fd == INVALID_SOCKET && errno == EAFNOSUPPORT)) { // try IPv4
                        ip_version = 4;
                        s->socket_fd = socket(AF_INET, SOCK_STREAM, 0);
                }
                if (s->socket_fd == INVALID_SOCKET) {
                        socket_error("Control socket - socket");
                        goto error;
                }
                int val = 1;
                int rc;
                rc = setsockopt(s->socket_fd, SOL_SOCKET, SO_REUSEADDR,
                                (sso_val_type) &val, sizeof(val));
                if (rc != 0) {
                        socket_error("Control socket - setsockopt");
                }
 
                int ipv6only = 0;
                if (ip_version == 6 && setsockopt(s->socket_fd, IPPROTO_IPV6, IPV6_V6ONLY, (char *)&ipv6only,
                                        sizeof(ipv6only)) != 0) {
                        socket_error("setsockopt IPV6_V6ONLY");
                }
 
                /* setting address to in6addr_any allows connections to be established
                 * from both IPv4 and IPv6 hosts. This behavior can be modified
                 * using the IPPROTO_IPV6 level socket option IPV6_V6ONLY if required.*/
                struct sockaddr_storage ss{};
                socklen_t s_len = 0;
                if (ip_version == 4) {
                        auto *s_in = reinterpret_cast<struct sockaddr_in *>(&ss);
                        s_in->sin_family = AF_INET;
                        s_in->sin_addr.s_addr = htonl(INADDR_ANY);
                        s_in->sin_port = htons(s->network_port);
                        s_len = sizeof *s_in;
                } else {
                        auto *s_in6 = reinterpret_cast<struct sockaddr_in6 *>(&ss);
                        s_in6->sin6_family = AF_INET6;
                        s_in6->sin6_addr = in6addr_any;
                        s_in6->sin6_port = htons(s->network_port);
                        s_len = sizeof *s_in6;
                }
 
                rc = ::bind(s->socket_fd, reinterpret_cast<const struct sockaddr *>(&ss), s_len);
                if (rc != 0) {
                        socket_error("Control socket - bind");
                        goto error;
                } else {
                        rc = listen(s->socket_fd, MAX_CLIENTS);
                        if (rc != 0) {
                                socket_error("Control socket - listen");
                                goto error;
                        }
                }
        } else {
                if (force_ip_version == 6) {
                        log_msg(LOG_LEVEL_ERROR, "Control socket: IPv6 unimplemented in client mode!\n");
                        goto error;
                }
                s->socket_fd = socket(AF_INET, SOCK_STREAM, 0);
                assert(s->socket_fd != INVALID_SOCKET);
                struct addrinfo hints, *res, *res0;
                int err;
 
                memset(&hints, 0, sizeof(hints));
                hints.ai_family = AF_INET;
                hints.ai_socktype = SOCK_STREAM;
                char port_str[10];
                snprintf(port_str, 10, "%d", s->network_port);
                err = getaddrinfo("127.0.0.1", port_str, &hints, &res0);
 
                if(err) {
                        fprintf(stderr, "Unable to get address: %s\n", gai_strerror(err));
                        goto error;
                }
                bool connected = false;
 
                for (res = res0; res; res = res->ai_next) {
                    if(connect(s->socket_fd, res->ai_addr, res->ai_addrlen) == -1) {
                        continue;
                    }
                    connected = true;
                    break; /* okay we got one */
                }
 
                freeaddrinfo(res0);
 
                if(!connected) {
                        fprintf(stderr, "Unable to connect to localhost:%d\n", s->network_port);
                        goto error;
                }
        }
 
 
        module_register(&s->mod, root_module);
 
        if (const char *val = get_commandline_param("control-report-audio-ch-count")) {
                s->audio_channel_report_count = strtoll(val, NULL, 0);
        }
 
        *state = s;
        return 0;
 
error:
        if (s->socket_fd != INVALID_SOCKET) {
                CLOSESOCKET(s->socket_fd);
        }
        delete s;
        return -1;
}

References control_state::audio_channel_report_count, CLIENT, CLOSESOCKET, module::cls, control_state::connection_type, get_commandline_param(), INVALID_SOCKET, LOG_LEVEL_ERROR, log_msg(), MAX_CLIENTS, control_state::mod, MODULE_CLASS_CONTROL, module_init_default(), module_register(), control_state::network_port, module::new_message, new_message(), module::priv_data, control_state::root_module, SERVER, socket_error, control_state::socket_fd, control_state::started, state, and val.

Referenced by main(), and main().

Here is the call graph for this function:

control_report_event()

void control_report_event	(	struct control_state *	s,
		const std::string &	report_line )

Definition at line 1007 of file control_socket.cpp.

{
        if (!s) {
                return;
        }
 
        control_report_stats_event(s, "event " + report_line + "\r\n");
}

References control_report_stats_event().

Here is the call graph for this function:

control_report_stats()

void control_report_stats	(	struct control_state *	s,
		const std::string &	report_line )

Definition at line 998 of file control_socket.cpp.

{
        if (!s || !s->stats_on) {
                return;
        }
 
        control_report_stats_event(s, "stats " + report_line + "\r\n");
}

References control_report_stats_event(), and control_state::stats_on.

Referenced by decode_audio_frame(), filter(), and report_stats().

Here is the call graph for this function:

control_report_stats_event()

static void control_report_stats_event ( struct control_state * s, const std::string & report_line )

static

Definition at line 985 of file control_socket.cpp.

{
        std::unique_lock<std::mutex> lk(s->stats_lock);
 
        if (s->stat_event_queue.size() < MAX_STAT_EVENT_QUEUE) {
                s->stat_event_queue.push(report_line);
        } else {
                log_msg(LOG_LEVEL_WARNING, "Cannot write stats/event - queue full!!!");
        }
        lk.unlock();
        s->stat_event_cv.notify_one();
}

References LOG_LEVEL_WARNING, log_msg(), MAX_STAT_EVENT_QUEUE, control_state::stat_event_cv, control_state::stat_event_queue, and control_state::stats_lock.

Referenced by control_report_event(), and control_report_stats().

Here is the call graph for this function:

control_start()

void control_start

(

struct control_state *

s

)

Definition at line 296 of file control_socket.cpp.

{
        if (s == NULL) {
                return;
        }
 
        platform_pipe_init(s->internal_fd);
 
        s->control_thread_id = thread(control_thread, s);
        s->stat_event_thread_id = thread(stat_event_thread, s);
 
        log_msg(LOG_LEVEL_NOTICE, "Control socket listening on port %d\n",
                        socket_get_recv_port(s->socket_fd));
        s->started = true;
}

References control_thread(), control_state::control_thread_id, control_state::internal_fd, LOG_LEVEL_NOTICE, log_msg(), platform_pipe_init(), control_state::socket_fd, socket_get_recv_port(), control_state::started, stat_event_thread(), and control_state::stat_event_thread_id.

Referenced by main(), and main().

Here is the call graph for this function:

control_stats_enabled()

bool control_stats_enabled

(

struct control_state *

s

)

Definition at line 1016 of file control_socket.cpp.

{
        return s && s->stats_on;
}

References control_state::stats_on.

Referenced by decode_audio_frame(), filter(), and report_stats().

control_thread()

static void * control_thread ( void * args )

static

Definition at line 773 of file control_socket.cpp.

{
        set_thread_name(__func__);
        struct control_state *s = (struct control_state *) args;
        struct client *clients = NULL;
 
        assert(s->socket_fd != INVALID_SOCKET);
        assert(s->internal_fd[0] != INVALID_SOCKET);
 
        if(s->connection_type == CLIENT) {
                clients = add_client(clients, s->socket_fd);
        }
        struct sockaddr_storage client_addr;
        socklen_t len = sizeof client_addr;
 
        errno = 0;
 
        clients = add_client(clients, s->internal_fd[0]);
 
        bool should_exit = false;
 
        struct timeval last_report_sent;
        const int report_interval_sec = 5;
        gettimeofday(&last_report_sent, NULL);
 
        while(!should_exit) {
                process_messages(s);
 
                fd_t max_fd = 0;
                fd_set fds;
                FD_ZERO(&fds);
                if (s->connection_type == SERVER) {
                        FD_SET(s->socket_fd, &fds);
                        max_fd = s->socket_fd + 1;
                }
 
                struct client *cur = clients;
 
                while(cur) {
                        FD_SET(cur->fd, &fds);
                        if(cur->fd + 1 > max_fd) {
                                max_fd = cur->fd + 1;
                        }
                        cur = cur->next;
                }
 
                struct timeval timeout;
                timeout.tv_sec = report_interval_sec;
                timeout.tv_usec = 0;
 
                struct timeval *timeout_ptr = NULL;
                if(clients->next != NULL) { // some remote client
                        timeout_ptr = &timeout;
                }
 
                int rc;
 
                if ((rc = select(max_fd, &fds, NULL, NULL, timeout_ptr)) >= 1) {
                        if(s->connection_type == SERVER && FD_ISSET(s->socket_fd, &fds)) {
                                fd_t fd = accept(s->socket_fd, (struct sockaddr *) &client_addr, &len);
                                if (fd == INVALID_SOCKET) {
                                        socket_error("[control socket] accept");
                                        continue;
                                }
 
                                // all remote sockets are written sequentially so
                                // we don't want to block if one gets stuck
                                set_socket_nonblock(fd);
 
                                // refuse remote connections by default
                                if (!get_commandline_param("control-accept-global") &&
                                                !is_addr_loopback((struct sockaddr *) &client_addr)) {
                                        log_msg(LOG_LEVEL_WARNING, "[control socket] Refusing remote connection. Use \"--param control-accept-global\" to allow UG control from a remote host.\n");
                                        const char *msg = "unauthorized\r\n";
                                        write_all(fd, msg, strlen(msg));
                                        CLOSESOCKET(fd);
                                        continue;
                                }
 
                                // add to list
                                clients = add_client(clients, fd);
                        }
 
                        struct client *cur = clients;
 
                        while(cur) {
                                if(FD_ISSET(cur->fd, &fds)) {
                                        ssize_t ret = PLATFORM_PIPE_READ(cur->fd, cur->buff + cur->buff_len,
                                                        sizeof(cur->buff) - cur->buff_len);
                                        if(ret == -1) {
                                                fprintf(stderr, "Error reading socket, closing!!!\n");
                                        }
                                        if(ret <= 0) {
                                                struct client *next;
                                                CLOSESOCKET(cur->fd);
                                                if (cur->prev) {
                                                        cur->prev->next = cur->next;
                                                } else {
                                                        clients = cur->next;
                                                }
                                                if (cur->next) {
                                                        cur->next->prev = cur->prev;
                                                }
                                                next = cur->next;
                                                free(cur);
                                                cur = next;
                                                continue;
                                        }
                                        cur->buff_len += ret;
                                }
                                cur = cur->next;
                        }
                } else if (rc < 0) {
                        socket_error("[control socket] select");
                }
 
                cur = clients;
                while(cur) {
                        char msg[sizeof(cur->buff) + 1];
                        int cur_buffer_len;
                        while(parse_msg(cur->buff, cur->buff_len, msg, &cur_buffer_len)) {
                                cur->buff_len = cur_buffer_len;
                                int ret = process_msg(s, cur->fd, msg, clients);
                                if(ret == CONTROL_EXIT && is_internal_port(cur->fd)) {
                                        should_exit = true;
                                } else if(ret == CONTROL_CLOSE_HANDLE) {
                                        shutdown(cur->fd, SHUT_RDWR);
                                }
                        }
                        if(cur->buff_len == sizeof(cur->buff)) {
                                fprintf(stderr, "Socket buffer full and no delimited message. Discarding.\n");
                                cur->buff_len = 0;
                        }
 
                        cur = cur->next;
                }
        }
 
        // notify clients about exit
        struct client *cur = clients;
        while(cur) {
                struct client *tmp = cur;
                if (!is_internal_port(cur->fd)) {
                        const char *msg = "event exit\r\n";
                        write_all(cur->fd, msg, strlen(msg));
                }
                CLOSESOCKET(cur->fd);
                cur = cur->next;
                free(tmp);
        }
 
        platform_pipe_close(s->internal_fd[0]);
 
        return NULL;
}

References add_client(), client::buff, client::buff_len, CLIENT, CLOSESOCKET, control_state::connection_type, CONTROL_CLOSE_HANDLE, CONTROL_EXIT, client::fd, get_commandline_param(), control_state::internal_fd, INVALID_SOCKET, is_addr_loopback(), is_internal_port, len, LOG_LEVEL_WARNING, log_msg(), max_fd, client::next, parse_msg(), platform_pipe_close(), PLATFORM_PIPE_READ, client::prev, process_messages(), process_msg(), SERVER, set_socket_nonblock(), set_thread_name(), socket_error, control_state::socket_fd, and write_all().

Referenced by control_start().

Here is the call graph for this function:

new_message()

static void new_message ( struct module * m )

static

Definition at line 145 of file control_socket.cpp.

                                          {
        control_state *s = (control_state *) m->priv_data;
        if(s->started) {
                // just wake up from select
                int ret = write_all(s->internal_fd[1], "noop\r\n", 6);
                if (ret <= 0) {
                        log_msg(LOG_LEVEL_ERROR, "[control] Cannot write to pipe!\n");
                }
        }
}

References control_state::internal_fd, LOG_LEVEL_ERROR, log_msg(), module::priv_data, control_state::started, and write_all().

Referenced by compress_process_message(), control_init(), BasicRTSPOnlySubsession::deleteStream(), keyboard_control::impl::exec_external_commands(), keyboard_control::impl::info(), keycontrol_register_key(), keycontrol_send_key(), process_audio_message(), process_msg(), reconfigure_decoder(), keyboard_control::impl::run(), sdp_send_change_address_message(), h264_sdp_video_rxtx::send_frame(), set_fec(), BasicRTSPOnlySubsession::startStream(), Participant_manager::tick(), tx_update(), and vidcap_switcher_register_keyboard_ctl().

Here is the call graph for this function:

parse_msg()

static bool parse_msg ( char * buffer, int buffer_len, char * message, int * new_buffer_len )

static

Definition at line 684 of file control_socket.cpp.

{
        bool ret = false;
        int i = 0;
 
        while (i < buffer_len) {
                if(buffer[i] == '\0' || buffer[i] == '\n' || buffer[i] == '\r') {
                        ++i;
                } else {
                        break;
                }
        }
 
        int start = i;
 
        for( ; i < buffer_len; ++i) {
                if(buffer[i] == '\0' || buffer[i] == '\n' || buffer[i] == '\r') {
                        memcpy(message, buffer + start, i - start);
                        message[i - start] = '\0';
                        ret = true;
                        break;
                }
        }
 
        if(ret) {
                memmove(buffer, buffer + i, buffer_len - i);
                *new_buffer_len = buffer_len - i;
        }
 
        return ret;
}

References start().

Referenced by control_thread().

Here is the call graph for this function:

print_control_help()

static void print_control_help ( )

static

Definition at line 1025 of file control_socket.cpp.

                                 {
        color_printf("Control internal commands:\n"
                        TBOLD("\texit") "\n"
                        TBOLD("\tpause") "\n"
                        TBOLD("\tplay") "\n"
                        TBOLD("\treciever {pause|play}") "\n"
                        TBOLD("\treset-ssrc") "\n"
                        TBOLD("\t{receiver|sender}-port <XY>") "\n"
                        TBOLD("\tfec {audio|video} <fec-string>") "\n"
                        TBOLD("\tcompress <new-compress>") "\n"
                        TBOLD("\tcompress param <new-compress-param>") "\n"
                        TBOLD("\tvolume {up|down}") u8"¹\n"
                        TBOLD("\tav-delay <ms>") u8"¹\n"
                        TBOLD("\tmute") " - toggles receiver mute\n"
                        TBOLD("\t[un]mute-{receiver,sender}")
                                " - (un)mutes audio sender or receiver\n"
                        TBOLD("\tpostprocess <new_postprocess> | flush") "\n"
                        TBOLD("\tdump-tree")"\n");
        color_printf("\nOther commands can be issued directly to individual "
                        "modules (see \"" TBOLD("dump-tree") "\"), eg.:\n"
                        "\t" TBOLD("capture.filter mirror") "\n"
                        "\nSometimes those modules support help (eg. \"" TBOLD("capture.filter help") "\")\n\n");
        color_printf(TBOLD(u8"¹") " audio commands applying to receiver\n\n");
}

References color_printf(), and TBOLD.

Referenced by process_msg().

Here is the call graph for this function:

process_audio_message()

static struct response * process_audio_message ( struct module * root_module, const char * cmd )

static

Definition at line 317 of file control_socket.cpp.

{
        char path[STR_LEN] = "";
        if (strcmp(cmd, "mute") == 0 || strstr(cmd, "-receiver") != nullptr) {
                strncpy(path, "audio.receiver", sizeof path);
                auto *msg = (struct msg_receiver *) new_message(
                    sizeof(struct msg_receiver));
                if (strcmp(cmd, "mute") == 0) {
                        msg->type = RECEIVER_MSG_MUTE_TOGGLE;
                } else if (prefix_matches(cmd, "mute-")) {
                        msg->type = RECEIVER_MSG_MUTE;
                } else if (prefix_matches(cmd, "unmute-")) {
                        msg->type = RECEIVER_MSG_UNMUTE;
                } else {
                        free_message((struct message *) msg, nullptr);
                        return new_response(RESPONSE_BAD_REQUEST,
                                            "malformed audio recv mute msg");
                }
                return send_message(root_module, path, (struct message *) msg);
        }
        if (strstr(cmd, "-sender") != nullptr) {
                strncpy(path, "audio.sender", sizeof path);
                auto *msg = (struct msg_sender *) new_message(
                    sizeof(struct msg_sender));
                if (prefix_matches(cmd, "mute-")) {
                        msg->type = SENDER_MSG_MUTE;
                } else if (prefix_matches(cmd, "unmute-")) {
                        msg->type = SENDER_MSG_UNMUTE;
                } else {
                        free_message((struct message *) msg, nullptr);
                        return new_response(RESPONSE_BAD_REQUEST,
                                            "malformed audio send mute msg");
                }
                return send_message(root_module, path, (struct message *) msg);
        }
        if (prefix_matches(cmd, "volume ")) {
                strncpy(path, "audio.receiver", sizeof path);
                auto *msg = (struct msg_receiver *) new_message(
                    sizeof(struct msg_receiver));
                const char *dir = suffix(cmd, "volume ");
                if (strcmp(dir, "up") == 0) {
                        msg->type = RECEIVER_MSG_INCREASE_VOLUME;
                } else if (strcmp(dir, "down") == 0) {
                        msg->type = RECEIVER_MSG_DECREASE_VOLUME;
                } else {
                        free_message((struct message *) msg, nullptr);
                        return new_response(RESPONSE_BAD_REQUEST, nullptr);
                }
                return send_message(root_module, path, (struct message *) msg);
        }
        return new_response(RESPONSE_BAD_REQUEST, "unexpected audio msg");
}

References free_message(), new_message(), new_response(), prefix_matches, RECEIVER_MSG_DECREASE_VOLUME, RECEIVER_MSG_INCREASE_VOLUME, RECEIVER_MSG_MUTE, RECEIVER_MSG_MUTE_TOGGLE, RECEIVER_MSG_UNMUTE, RESPONSE_BAD_REQUEST, send_message(), SENDER_MSG_MUTE, SENDER_MSG_UNMUTE, STR_LEN, and suffix.

Referenced by process_msg().

Here is the call graph for this function:

process_messages()

static void process_messages ( struct control_state * s )

static

Definition at line 733 of file control_socket.cpp.

{
        struct message *msg;
        while((msg = check_message(&s->mod))) {
                struct msg_universal *m = (struct msg_universal *) msg;
                if (strcmp(m->text, "get_port") == 0) {
                        struct response *r;
                        uint16_t port = socket_get_recv_port(s->socket_fd);
                        if (port) {
                                char port_str[6];
                                snprintf(port_str, sizeof port_str, "%hu", port);
                                r = new_response(RESPONSE_OK, port_str);
                        } else {
                                r = new_response(RESPONSE_INT_SERV_ERR, "get_recv_port");
                        }
                        free_message(msg, r);
                } else if (strstr(m->text, "execute ") == m->text) {
                        process_msg(s, 1, m->text + strlen("execute "), nullptr);
                        free_message(msg, new_response(RESPONSE_OK, nullptr));
                } else {
                        log_msg(LOG_LEVEL_WARNING, "[control] Unrecognized command: %s\n", m->text);
                        free_message(msg, new_response(RESPONSE_NOT_IMPL, NULL));
                        continue;
                }
        }
}

References check_message(), free_message(), LOG_LEVEL_WARNING, log_msg(), msg_universal::m, control_state::mod, msg, new_response(), process_msg(), RESPONSE_INT_SERV_ERR, RESPONSE_NOT_IMPL, RESPONSE_OK, control_state::socket_fd, socket_get_recv_port(), and msg_universal::text.

Referenced by control_thread().

Here is the call graph for this function:

process_msg()

static int process_msg ( struct control_state * s, fd_t client_fd, char * message, struct client * clients )

static

Return values

-1	exit thread
-2	close handle

Definition at line 374 of file control_socket.cpp.

{
        int ret = 0;
        struct response *resp = NULL;
        char path[1024] = ""; // path for msg receiver (usually video)
        char path_audio[1024] = ""; // auxiliary buffer used when we need to signalize both audio
                                    // and video
        char buf[1048];
 
        if(prefix_matches(message, "port ")) {
                message = suffix(message, "port ");
                if (isdigit(message[0])) { // index is a number
                        snprintf(path, 1024, "%s[%d]", module_class_name(MODULE_CLASS_PORT), atoi(message));
                        while(isdigit(*message))
                                message++;
                        while(isspace(*message))
                                message++;
                } else { // index is a name
                        char *port_id_name = message;
                        while (!isspace(message[0]) && message[0] != '\0')
                                message++;
                        if (message[0] != '\0') {
                                message[0] = '\0';
                                message++;
                        }
                        snprintf(path, 1024, "%s[%s]", module_class_name(MODULE_CLASS_PORT), port_id_name);
                        while (isspace(message[0]) && message[0] != '\0')
                                message++;
                }
 
                /* "port n compress" messages get forwarded to
                 * port[n].sender.compress, but that is wrong, because frames
                 * are now compressed before they are passed to the sender
                 * (sender compression is now always set to "none").
                 */
                if (prefix_matches(message, "compress ")){
                    log_msg(LOG_LEVEL_ERROR, "\"port n compress\" was deprecated. Use \"port[n] compress\" instead\n");
                    return ret;
                }
        }
 
        if(strcasecmp(message, "quit") == 0) {
                return CONTROL_EXIT;
        } else if (strcasecmp(message, "exit") == 0) {
                exit_uv(0);
                resp = new_response(RESPONSE_OK, NULL);
        } else if (strcasecmp(message, "help") == 0) {
                print_control_help();
                return ret;
        } else if (strcasecmp(message, "noop") == 0) {
                return ret;
        } else if (prefix_matches(message, "stats ") || prefix_matches(message, "event ")) {
                if (is_internal_port(client_fd)) {
                        struct client *cur = clients;
                        char *new_msg = NULL;
 
                        while (cur) {
                                if(is_internal_port(cur->fd)) { // skip local FD
                                        cur = cur->next;
                                        continue;
                                }
                                if (!new_msg) {
                                        // append <CR><LF> again
                                        new_msg = (char *) malloc(strlen(message) + 1 + 2);
                                        strcpy(new_msg, message);
                                        new_msg[strlen(message)] = '\r';
                                        new_msg[strlen(message) + 1] = '\n';
                                        new_msg[strlen(message) + 2] = '\0';
                                }
 
                                int ret = write_all(cur->fd, new_msg, strlen(new_msg));
                                if (ret != (int) strlen(new_msg)) {
                                        log_msg(LOG_LEVEL_WARNING, "Cannot write stats/event!\n");
                                }
                                cur = cur->next;
                        }
                        free(new_msg);
                        return ret;
                } else if (prefix_matches(message, "stats ")) {
                        const char *toggle = suffix(message, "stats ");
                        if (strcasecmp(toggle, "on") == 0) {
                                s->stats_on = true;
                                resp = new_response(RESPONSE_OK, NULL);
                        } else if (strcasecmp(toggle, "off") == 0) {
                                s->stats_on = false;
                                resp = new_response(RESPONSE_OK, NULL);
                        } else {
                                resp = new_response(RESPONSE_BAD_REQUEST, NULL);
                        }
                }
        } else if (prefix_matches(message, "sender-port ")) {
                struct msg_sender *msg = (struct msg_sender *)
                        new_message(sizeof(struct msg_sender));
                struct msg_sender *msg_audio = (struct msg_sender *)
                        new_message(sizeof(struct msg_sender));
                char *port_str = suffix(message, "sender-port ");
 
                msg->type = SENDER_MSG_CHANGE_PORT;
                msg_audio->type = SENDER_MSG_CHANGE_PORT;
 
                if (strchr(port_str, ':')) {
                        char *save_ptr, *item;
                        item = strtok_r(port_str, ":", &save_ptr);
                        msg->rx_port = atoi(item);
                        item = strtok_r(NULL, ":", &save_ptr);
                        msg->tx_port = atoi(item);
 
                        item = strtok_r(NULL, ":", &save_ptr);
                        if (item) {
                                char *item2 = strtok_r(NULL, ":", &save_ptr);
                                if (item2) { // change only if both RX and TX given
                                        msg_audio->rx_port = atoi(item);
                                        msg_audio->tx_port = atoi(item2);
                                } else {
                                        log_msg(LOG_LEVEL_ERROR, "Not changing audio port!\n");
                                }
                        }
                } else {
                        msg->tx_port = atoi(port_str);
                }
 
                if (msg_audio->rx_port == 0) {
                        msg_audio->rx_port = msg->rx_port ? msg->rx_port + 2 : 0;
                }
 
                if (msg_audio->tx_port == 0) {
                        msg_audio->tx_port = msg->tx_port + 2;
                }
 
                enum module_class path_sender[] = { MODULE_CLASS_SENDER, MODULE_CLASS_NONE };
                enum module_class path_sender_audio[] = { MODULE_CLASS_AUDIO, MODULE_CLASS_SENDER, MODULE_CLASS_NONE };
                memcpy(path_audio, path, sizeof(path_audio));
                append_message_path(path, sizeof(path), path_sender);
                append_message_path(path_audio, sizeof(path_audio), path_sender_audio);
 
                resp =
                        send_message(s->root_module, path, (struct message *) msg);
                struct response *resp_audio =
                        send_message(s->root_module, path_audio, (struct message *) msg_audio);
                free_response(resp_audio);
        } else if(prefix_matches(message, "receiver ") || prefix_matches(message, "play") ||
                        prefix_matches(message, "pause") || prefix_matches(message, "reset-ssrc")) {
                struct msg_sender *msg =
                        (struct msg_sender *)
                        new_message(sizeof(struct msg_sender));
                if(prefix_matches(message, "receiver ")) {
                        strncpy(msg->receiver, suffix(message, "receiver "), sizeof(msg->receiver) - 1);
                        msg->type = SENDER_MSG_CHANGE_RECEIVER;
                } else if(prefix_matches(message, "reset-ssrc")) {
                        msg->type = SENDER_MSG_RESET_SSRC;
                } else {
                        abort();
                }
 
                struct msg_sender *msg_audio = (struct msg_sender *) malloc(sizeof(struct msg_sender));
                memcpy(msg_audio, msg, sizeof(struct msg_sender));
 
                enum module_class path_sender[] = { MODULE_CLASS_SENDER, MODULE_CLASS_NONE };
                enum module_class path_sender_audio[] = { MODULE_CLASS_AUDIO, MODULE_CLASS_SENDER, MODULE_CLASS_NONE };
                memcpy(path_audio, path, sizeof(path_audio));
                append_message_path(path, sizeof(path), path_sender);
                append_message_path(path_audio, sizeof(path_audio), path_sender_audio);
 
                resp =
                        send_message(s->root_module, path, (struct message *) msg);
                struct response *resp_audio =
                        send_message(s->root_module, path_audio, (struct message *) msg_audio);
                free_response(resp_audio);
        } else if (prefix_matches(message, "receiver-port ")) {
                struct msg_receiver *msg =
                        (struct msg_receiver *)
                        new_message(sizeof(struct msg_receiver));
                struct msg_receiver *msg_audio =
                        (struct msg_receiver *)
                        new_message(sizeof(struct msg_receiver));
                msg->type = RECEIVER_MSG_CHANGE_RX_PORT;
                msg_audio->type = RECEIVER_MSG_CHANGE_RX_PORT;
 
                char *port_str = suffix(message, "receiver-port ");
 
                msg->new_rx_port = atoi(port_str);
                if (strchr(port_str, ':')) {
                        msg_audio->new_rx_port = atoi(strchr(port_str, ':') + 1);
                } else {
                        msg_audio->new_rx_port = msg->new_rx_port + 2;
                }
 
                enum module_class path_receiver[] = { MODULE_CLASS_RECEIVER, MODULE_CLASS_NONE };
                enum module_class path_audio_receiver[] = { MODULE_CLASS_AUDIO, MODULE_CLASS_RECEIVER, MODULE_CLASS_NONE };
                append_message_path(path, sizeof(path), path_receiver);
                append_message_path(path_audio, sizeof(path_audio), path_audio_receiver);
                resp =
                        send_message(s->root_module, path, (struct message *) msg);
                struct response *resp_audio =
                        send_message(s->root_module, path_audio, (struct message *) msg_audio);
                free_response(resp_audio);
        } else if(prefix_matches(message, "fec ")) {
                auto *msg = reinterpret_cast<struct msg_universal *>(new_message(sizeof(struct msg_universal)));
                char *fec = suffix(message, "fec ");
                enum tx_media_type media_type{};
                strncpy(msg->text, MSG_UNIVERSAL_TAG_TX "fec ", sizeof(msg->text) - 1);
 
                if(strncasecmp(fec, "audio ", 6) == 0) {
                        media_type = TX_MEDIA_AUDIO;
                        strncat(msg->text, fec + 6, sizeof(msg->text) - strlen(msg->text) - 1);
                } else if(strncasecmp(fec, "video ", 6) == 0) {
                        media_type = TX_MEDIA_VIDEO;
                        strncat(msg->text, fec + 6, sizeof(msg->text) - strlen(msg->text) - 1);
                } else {
                        resp = new_response(RESPONSE_NOT_FOUND, "unknown media type");
                        free(msg);
                        msg = nullptr;
                }
 
                if (msg) {
                        if (media_type == TX_MEDIA_VIDEO) {
                                enum module_class path_tx[] = { MODULE_CLASS_SENDER, MODULE_CLASS_TX, MODULE_CLASS_NONE };
                                append_message_path(path, sizeof(path),
                                                path_tx);
                        } else {
                                enum module_class path_audio_tx[] = { MODULE_CLASS_AUDIO, MODULE_CLASS_SENDER, MODULE_CLASS_TX, MODULE_CLASS_NONE };
                                append_message_path(path, sizeof(path),
                                                path_audio_tx);
                        }
                        resp = send_message(s->root_module, path, (struct message *) msg);
                }
        } else if(prefix_matches(message, "compress ")) {
                struct msg_change_compress_data *msg =
                        (struct msg_change_compress_data *)
                        new_message(sizeof(struct msg_change_compress_data));
                char *compress = suffix(message, "compress ");
 
                if(prefix_matches(compress, "param ")) {
                        compress = suffix(compress, "param ");
                        msg->what = CHANGE_PARAMS;
                } else {
                        msg->what = CHANGE_COMPRESS;
                }
                strncpy(msg->config_string, compress, sizeof(msg->config_string) - 1);
 
                if(!resp) {
                        enum module_class path_compress[] = { MODULE_CLASS_SENDER, MODULE_CLASS_COMPRESS, MODULE_CLASS_NONE };
                        append_message_path(path, sizeof(path), path_compress);
                        resp = send_message(s->root_module, path, (struct message *) msg);
                }
        } else if (prefix_matches(message, "volume ") ||
                   prefix_matches(message, "mute") ||
                   prefix_matches(message, "unmute")) {
                resp = process_audio_message(s->root_module, message);
        } else if (prefix_matches(message, "av-delay ")) {
                int val = atoi(suffix(message, "av-delay "));
                set_audio_delay(val);
                resp = new_response(RESPONSE_OK, NULL);
        } else if (prefix_matches(message, "postprocess ")) {
                strncpy(path, "display", sizeof path);
                struct msg_universal *msg = (struct msg_universal *) new_message(sizeof(struct msg_universal));
                strncpy(msg->text, message, sizeof msg->text - 1);
                msg->text[sizeof msg->text - 1] = '\0';
                resp = send_message(s->root_module, path, (struct message *) msg);
        } else if(strcasecmp(message, "bye") == 0) {
                ret = CONTROL_CLOSE_HANDLE;
                resp = new_response(RESPONSE_OK, NULL);
        } else if(strcmp(message, "dump-tree") == 0) {
                dump_tree(s->root_module, 0);
                resp = new_response(RESPONSE_OK, NULL);
        } else { // assume message in format "path message"
                struct msg_universal *msg = (struct msg_universal *)
                        new_message(sizeof(struct msg_universal));
 
                if (strchr(message, ' ')) {
                        memcpy(path, message, strchr(message, ' ') - message);
                        strncpy(msg->text, strchr(message, ' ') + 1, sizeof(path) - 1);
                } else {
                        strncpy(path, message, sizeof(path) - 1); // empty message ??
                }
 
                resp = send_message(s->root_module, path, (struct message *) msg);
        }
 
        if(!resp) {
                fprintf(stderr, "No response received!\n");
                snprintf(buf, sizeof(buf), "(unknown path: %s)", path);
                resp = new_response(RESPONSE_INT_SERV_ERR, buf);
        }
        send_response(client_fd, resp);
 
        return ret;
}

References append_message_path(), CHANGE_COMPRESS, CHANGE_PARAMS, CONTROL_CLOSE_HANDLE, CONTROL_EXIT, dump_tree(), exit_uv(), client::fd, free_response(), is_internal_port, LOG_LEVEL_ERROR, LOG_LEVEL_WARNING, log_msg(), MODULE_CLASS_AUDIO, MODULE_CLASS_COMPRESS, module_class_name(), MODULE_CLASS_NONE, MODULE_CLASS_PORT, MODULE_CLASS_RECEIVER, MODULE_CLASS_SENDER, MODULE_CLASS_TX, MSG_UNIVERSAL_TAG_TX, new_message(), new_msg(), new_response(), msg_receiver::new_rx_port, client::next, prefix_matches, print_control_help(), process_audio_message(), RECEIVER_MSG_CHANGE_RX_PORT, RESPONSE_BAD_REQUEST, RESPONSE_INT_SERV_ERR, RESPONSE_NOT_FOUND, RESPONSE_OK, control_state::root_module, msg_sender::rx_port, send_message(), send_response(), SENDER_MSG_CHANGE_PORT, SENDER_MSG_CHANGE_RECEIVER, SENDER_MSG_RESET_SSRC, set_audio_delay(), control_state::stats_on, suffix, TX_MEDIA_AUDIO, TX_MEDIA_VIDEO, msg_sender::tx_port, msg_receiver::type, msg_sender::type, val, and write_all().

Referenced by control_thread(), and process_messages().

Here is the call graph for this function:

send_response()

static void send_response ( fd_t fd, struct response * resp )

static

Definition at line 663 of file control_socket.cpp.

{
        char buffer[1024];
 
        snprintf(buffer, sizeof(buffer) - 2, "%d %s", response_get_status(resp),
                        response_status_to_text(response_get_status(resp)));
 
        if (response_get_text(resp)) {
                strncat(buffer, " ", sizeof(buffer) - strlen(buffer) - 1 - 2);
                strncat(buffer, response_get_text(resp), sizeof(buffer) - strlen(buffer) - 1 - 2);
        }
        strcat(buffer, "\r\n");
 
        ssize_t ret = write_all(fd, buffer, strlen(buffer));
        if (ret < 0) {
                socket_error("Unable to write response");
        }
 
        free_response(resp);
}

References free_response(), response_get_status(), response_get_text(), response_status_to_text(), socket_error, and write_all().

Referenced by process_msg().

Here is the call graph for this function:

set_socket_nonblock()

static void set_socket_nonblock ( fd_t fd )

static

Definition at line 760 of file control_socket.cpp.

                                         {
#ifdef _WIN32
    unsigned long ul;
    ioctlsocket(fd, FIONBIO, &ul);
#else
    if (fcntl(fd, F_SETFL, O_NONBLOCK) == -1) {
            perror("fcntl");
    }
#endif
}

Referenced by control_thread().

stat_event_thread()

static void * stat_event_thread ( void * args )

static

Definition at line 929 of file control_socket.cpp.

{
        set_thread_name(__func__);
        struct control_state *s = (struct control_state *) args;
 
        while (1) {
                std::unique_lock<std::mutex> lk(s->stats_lock);
                s->stat_event_cv.wait(lk, [s] { return s->stat_event_queue.size() > 0; });
                string &line = s->stat_event_queue.front();
 
                if (line.empty()) {
                        break;
                }
 
                int ret = write_all(s->internal_fd[1], line.c_str(), line.length());
                s->stat_event_queue.pop();
                if (ret <= 0) {
                        fprintf(stderr, "Cannot write stat line!\n");
                }
        }
 
        return NULL;
}

References control_state::internal_fd, set_thread_name(), control_state::stat_event_cv, control_state::stat_event_queue, control_state::stats_lock, and write_all().

Referenced by control_start().

Here is the call graph for this function:

write_all()

static ssize_t write_all ( fd_t fd, const void * buf, size_t count )

static

Definition at line 127 of file control_socket.cpp.

{
    const char *p = (const char *) buf;
    size_t rest = count;
    ssize_t w = 0;
 
    while (rest > 0 && ((fd > 2 && (w = send(fd, p, rest, MSG_NOSIGNAL)) > 0) ||
                            (w = write(fd, p, rest)) > 0)) {
        p += w;
        rest -= w;
    }
 
    if (rest > 0)
        return w;
    else
        return count;
}

References count, and MSG_NOSIGNAL.

Referenced by control_done(), control_thread(), crash_signal_handler(), hang_signal_handler(), new_message(), print_backtrace(), process_msg(), send_response(), signal_handler(), and stat_event_thread().