aniani/icecast-server
1
0
Fork 0
mirror of https://gitlab.xiph.org/xiph/icecast-server.git synced 2024-09-29 04:25:55 -04:00
Code

Cleanup: Code style

Browse Source
This commit is contained in:
Philipp Schafft 2022-04-15 10:32:08 +00:00
parent 3583aefceb
commit d8b7e4f322
1 changed files with 135 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

133
src/slave.c
View File

@ -25,6 +25,7 @@
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <stdbool.h>
#include <string.h> #include <string.h>
#include <sys/types.h> #include <sys/types.h>
@ -261,8 +262,7 @@ static client_t *open_relay_connection (relay_t *relay, relay_config_upstream_t
config_release_config(); config_release_config();
/* build any authentication header before connecting */ /* build any authentication header before connecting */
if (_GET_UPSTREAM_SETTING(username) && _GET_UPSTREAM_SETTING(password)) if (_GET_UPSTREAM_SETTING(username) && _GET_UPSTREAM_SETTING(password)) {
{
char *esc_authorisation; char *esc_authorisation;
unsigned len = strlen(_GET_UPSTREAM_SETTING(username)) + strlen(_GET_UPSTREAM_SETTING(password)) + 2; unsigned len = strlen(_GET_UPSTREAM_SETTING(username)) + strlen(_GET_UPSTREAM_SETTING(password)) + 2;
@ -275,19 +275,17 @@ static client_t *open_relay_connection (relay_t *relay, relay_config_upstream_t
snprintf(auth_header, len, snprintf(auth_header, len,
"Authorization: Basic %s\r\n", esc_authorisation); "Authorization: Basic %s\r\n", esc_authorisation);
free(esc_authorisation); free(esc_authorisation);
} } else {
else
auth_header = strdup(""); auth_header = strdup("");
}
while (redirects < 10) while (redirects < 10) {
{
sock_t streamsock; sock_t streamsock;
ICECAST_LOG_INFO("connecting to %s:%d", server, port); ICECAST_LOG_INFO("connecting to %s:%d", server, port);
streamsock = sock_connect_wto_bind(server, port, _GET_UPSTREAM_SETTING(bind), 10); streamsock = sock_connect_wto_bind(server, port, _GET_UPSTREAM_SETTING(bind), 10);
if (streamsock == SOCK_ERROR) if (streamsock == SOCK_ERROR) {
{
ICECAST_LOG_WARN("Failed to connect to %s:%d", server, port); ICECAST_LOG_WARN("Failed to connect to %s:%d", server, port);
break; break;
} }
@ -310,22 +308,19 @@ static client_t *open_relay_connection (relay_t *relay, relay_config_upstream_t
_GET_UPSTREAM_SETTING(mp3metadata) ? "Icy-MetaData: 1\r\n" : "", _GET_UPSTREAM_SETTING(mp3metadata) ? "Icy-MetaData: 1\r\n" : "",
auth_header); auth_header);
memset(header, 0, sizeof(header)); memset(header, 0, sizeof(header));
if (util_read_header (con->sock, header, 4096, READ_ENTIRE_HEADER) == 0) if (util_read_header(con->sock, header, 4096, READ_ENTIRE_HEADER) == 0) {
{
ICECAST_LOG_ERROR("Header read failed for %s (%s:%d%s)", relay->config->localmount, server, port, mount); ICECAST_LOG_ERROR("Header read failed for %s (%s:%d%s)", relay->config->localmount, server, port, mount);
break; break;
} }
igloo_prng_write(igloo_instance, header, strlen(header), -1, igloo_PRNG_FLAG_NONE); igloo_prng_write(igloo_instance, header, strlen(header), -1, igloo_PRNG_FLAG_NONE);
parser = httpp_create_parser(); parser = httpp_create_parser();
httpp_initialize(parser, NULL); httpp_initialize(parser, NULL);
if (! httpp_parse_response (parser, header, strlen(header), relay->config->localmount)) if (! httpp_parse_response(parser, header, strlen(header), relay->config->localmount)) {
{
ICECAST_LOG_ERROR("Error parsing relay request for %s (%s:%d%s)", relay->config->localmount, ICECAST_LOG_ERROR("Error parsing relay request for %s (%s:%d%s)", relay->config->localmount,
server, port, mount); server, port, mount);
break; break;
} }
if (strcmp (httpp_getvar (parser, HTTPP_VAR_ERROR_CODE), "302") == 0) if (strcmp(httpp_getvar (parser, HTTPP_VAR_ERROR_CODE), "302") == 0) {
{
/* better retry the connection again but with different details */ /* better retry the connection again but with different details */
const char *uri, *mountpoint; const char *uri, *mountpoint;
int len; int len;
@ -353,20 +348,16 @@ static client_t *open_relay_connection (relay_t *relay, relay_config_upstream_t
httpp_destroy(parser); httpp_destroy(parser);
con = NULL; con = NULL;
parser = NULL; parser = NULL;
} } else {
else
{
client_t *client = NULL; client_t *client = NULL;
if (httpp_getvar (parser, HTTPP_VAR_ERROR_MESSAGE)) if (httpp_getvar(parser, HTTPP_VAR_ERROR_MESSAGE)) {
{
ICECAST_LOG_ERROR("Error from relay request: %s (%s)", relay->config->localmount, ICECAST_LOG_ERROR("Error from relay request: %s (%s)", relay->config->localmount,
httpp_getvar(parser, HTTPP_VAR_ERROR_MESSAGE)); httpp_getvar(parser, HTTPP_VAR_ERROR_MESSAGE));
break; break;
} }
global_lock(); global_lock();
if (client_create (&client, con, parser) < 0) if (client_create(&client, con, parser) < 0) {
{
global_unlock(); global_unlock();
/* make sure only the client_destroy frees these */ /* make sure only the client_destroy frees these */
con = NULL; con = NULL;
@ -410,8 +401,7 @@ static void *start_relay_stream (void *arg)
client_t *client; client_t *client;
ICECAST_LOG_INFO("Starting relayed source at mountpoint \"%s\"", relay->config->localmount); ICECAST_LOG_INFO("Starting relayed source at mountpoint \"%s\"", relay->config->localmount);
do do {
{
size_t i; size_t i;
for (i = 0; i < relay->config->upstreams; i++) { for (i = 0; i < relay->config->upstreams; i++) {
@ -434,8 +424,7 @@ static void *start_relay_stream (void *arg)
src->parser = client->parser; src->parser = client->parser;
src->con = client->con; src->con = client->con;
if (connection_complete_source (src, 0) < 0) if (connection_complete_source (src, 0) < 0) {
{
ICECAST_LOG_INFO("Failed to complete source initialisation"); ICECAST_LOG_INFO("Failed to complete source initialisation");
client_destroy (client); client_destroy (client);
src->client = NULL; src->client = NULL;
@ -446,8 +435,7 @@ static void *start_relay_stream (void *arg)
source_main(relay->source); source_main(relay->source);
if (relay->config->on_demand == 0) if (relay->config->on_demand == 0) {
{
/* only keep refreshing YP entries for inactive on-demand relays */ /* only keep refreshing YP entries for inactive on-demand relays */
yp_remove(relay->config->localmount); yp_remove(relay->config->localmount);
relay->source->yp_public = -1; relay->source->yp_public = -1;
@ -462,8 +450,7 @@ static void *start_relay_stream (void *arg)
return NULL; return NULL;
} while (0); /* TODO allow looping through multiple servers */ } while (0); /* TODO allow looping through multiple servers */
if (relay->source->fallback_mount) if (relay->source->fallback_mount) {
{
source_t *fallback_source; source_t *fallback_source;
ICECAST_LOG_DEBUG("failed relay, fallback to %s", relay->source->fallback_mount); ICECAST_LOG_DEBUG("failed relay, fallback to %s", relay->source->fallback_mount);
@ -494,21 +481,17 @@ static void *start_relay_stream (void *arg)
/* wrapper for starting the provided relay stream */ /* wrapper for starting the provided relay stream */
static void check_relay_stream (relay_t *relay) static void check_relay_stream (relay_t *relay)
{ {
if (relay->source == NULL) if (relay->source == NULL) {
{ if (relay->config->localmount[0] != '/') {
if (relay->config->localmount[0] != '/')
{
ICECAST_LOG_WARN("relay mountpoint \"%s\" does not start with /, skipping", ICECAST_LOG_WARN("relay mountpoint \"%s\" does not start with /, skipping",
relay->config->localmount); relay->config->localmount);
return; return;
} }
/* new relay, reserve the name */ /* new relay, reserve the name */
relay->source = source_reserve(relay->config->localmount); relay->source = source_reserve(relay->config->localmount);
if (relay->source) if (relay->source) {
{
ICECAST_LOG_DEBUG("Adding relay source at mountpoint \"%s\"", relay->config->localmount); ICECAST_LOG_DEBUG("Adding relay source at mountpoint \"%s\"", relay->config->localmount);
if (relay->config->on_demand) if (relay->config->on_demand) {
{
ice_config_t *config = config_get_config(); ice_config_t *config = config_get_config();
mount_proxy *mountinfo = config_find_mount(config, relay->config->localmount, MOUNT_TYPE_NORMAL); mount_proxy *mountinfo = config_find_mount(config, relay->config->localmount, MOUNT_TYPE_NORMAL);
relay->source->on_demand = relay->config->on_demand; relay->source->on_demand = relay->config->on_demand;
@ -518,11 +501,8 @@ static void check_relay_stream (relay_t *relay)
stats_event(relay->config->localmount, "listeners", "0"); stats_event(relay->config->localmount, "listeners", "0");
slave_update_all_mounts(); slave_update_all_mounts();
} }
} } else {
else if (relay->start == 0) {
{
if (relay->start == 0)
{
ICECAST_LOG_WARN("new relay but source \"%s\" already exists", relay->config->localmount); ICECAST_LOG_WARN("new relay but source \"%s\" already exists", relay->config->localmount);
relay->start = 1; relay->start = 1;
} }
@ -536,17 +516,14 @@ static void check_relay_stream (relay_t *relay)
if (relay->source == NULL || relay->running || relay->start > time(NULL)) if (relay->source == NULL || relay->running || relay->start > time(NULL))
break; break;
/* check if an inactive on-demand relay has a fallback that has listeners */ /* check if an inactive on-demand relay has a fallback that has listeners */
if (relay->config->on_demand && source->on_demand_req == 0) if (relay->config->on_demand && source->on_demand_req == 0) {
{
relay->source->on_demand = relay->config->on_demand; relay->source->on_demand = relay->config->on_demand;
if (source->fallback_mount && source->fallback_override != FALLBACK_OVERRIDE_NONE) if (source->fallback_mount && source->fallback_override != FALLBACK_OVERRIDE_NONE) {
{
source_t *fallback; source_t *fallback;
avl_tree_rlock(global.source_tree); avl_tree_rlock(global.source_tree);
fallback = source_find_mount(source->fallback_mount); fallback = source_find_mount(source->fallback_mount);
if (fallback && fallback->running && fallback->listeners) if (fallback && fallback->running && fallback->listeners) {
{
ICECAST_LOG_DEBUG("fallback running %d with %lu listeners", fallback->running, fallback->listeners); ICECAST_LOG_DEBUG("fallback running %d with %lu listeners", fallback->running, fallback->listeners);
source->on_demand_req = 1; source->on_demand_req = 1;
} }
@ -564,10 +541,8 @@ static void check_relay_stream (relay_t *relay)
} while (0); } while (0);
/* the relay thread may of shut down itself */ /* the relay thread may of shut down itself */
if (relay->cleanup) if (relay->cleanup) {
{ if (relay->thread) {
if (relay->thread)
{
ICECAST_LOG_DEBUG("waiting for relay thread for \"%s\"", relay->config->localmount); ICECAST_LOG_DEBUG("waiting for relay thread for \"%s\"", relay->config->localmount);
thread_join(relay->thread); thread_join(relay->thread);
relay->thread = NULL; relay->thread = NULL;
@ -575,8 +550,7 @@ static void check_relay_stream (relay_t *relay)
relay->cleanup = 0; relay->cleanup = 0;
relay->running = 0; relay->running = 0;
if (relay->config->on_demand && relay->source) if (relay->config->on_demand && relay->source) {
{
ice_config_t *config = config_get_config(); ice_config_t *config = config_get_config();
mount_proxy *mountinfo = config_find_mount(config, relay->config->localmount, MOUNT_TYPE_NORMAL); mount_proxy *mountinfo = config_find_mount(config, relay->config->localmount, MOUNT_TYPE_NORMAL);
source_update_settings(config, relay->source, mountinfo); source_update_settings(config, relay->source, mountinfo);
@ -657,8 +631,7 @@ update_relay_set(relay_t **current, relay_config_t **updated, size_t updated_len
existing_relay = *current; existing_relay = *current;
existing_p = current; existing_p = current;
while (existing_relay) while (existing_relay) {
{
/* break out if keeping relay */ /* break out if keeping relay */
if (strcmp(relay->localmount, existing_relay->config->localmount) == 0) if (strcmp(relay->localmount, existing_relay->config->localmount) == 0)
if (relay_has_changed(relay, existing_relay->config) == 0) if (relay_has_changed(relay, existing_relay->config) == 0)
@ -669,13 +642,10 @@ update_relay_set(relay_t **current, relay_config_t **updated, size_t updated_len
} }
if (existing_relay == NULL) if (existing_relay == NULL) {
{
/* new one, copy and insert */ /* new one, copy and insert */
existing_relay = relay_new(relay); existing_relay = relay_new(relay);
} } else {
else
{
*existing_p = existing_relay->next; *existing_p = existing_relay->next;
} }
existing_relay->next = new_list; existing_relay->next = new_list;
@ -710,27 +680,23 @@ static void relay_check_streams (relay_t *to_start,
{ {
relay_t *relay; relay_t *relay;
while (to_free) while (to_free) {
{ if (to_free->source) {
if (to_free->source) if (to_free->running) {
{
if (to_free->running)
{
/* relay has been removed from xml, shut down active relay */ /* relay has been removed from xml, shut down active relay */
ICECAST_LOG_DEBUG("source shutdown request on \"%s\"", to_free->config->localmount); ICECAST_LOG_DEBUG("source shutdown request on \"%s\"", to_free->config->localmount);
to_free->running = 0; to_free->running = 0;
to_free->source->running = 0; to_free->source->running = 0;
thread_join(to_free->thread); thread_join(to_free->thread);
} } else {
else
stats_event(to_free->config->localmount, NULL, NULL); stats_event(to_free->config->localmount, NULL, NULL);
} }
}
to_free = relay_free(to_free); to_free = relay_free(to_free);
} }
relay = to_start; relay = to_start;
while (relay) while (relay) {
{
if (skip_timer) if (skip_timer)
relay->start = 0; relay->start = 0;
check_relay_stream(relay); check_relay_stream(relay);
@ -746,8 +712,8 @@ static int update_from_master(ice_config_t *config)
sock_t mastersock; sock_t mastersock;
int ret = 0; int ret = 0;
char buf[256]; char buf[256];
do
{ do {
char *authheader, *data; char *authheader, *data;
relay_t *cleanup_relays; relay_t *cleanup_relays;
relay_config_t **new_relays = NULL; relay_config_t **new_relays = NULL;
@ -772,8 +738,7 @@ static int update_from_master(ice_config_t *config)
config_release_config(); config_release_config();
mastersock = sock_connect_wto(master, port, 10); mastersock = sock_connect_wto(master, port, 10);
if (mastersock == SOCK_ERROR) if (mastersock == SOCK_ERROR) {
{
ICECAST_LOG_WARN("Relay slave failed to contact master server to fetch stream list"); ICECAST_LOG_WARN("Relay slave failed to contact master server to fetch stream list");
break; break;
} }
@ -790,8 +755,7 @@ static int update_from_master(ice_config_t *config)
free(data); free(data);
if (sock_read_line(mastersock, buf, sizeof(buf)) == 0 || if (sock_read_line(mastersock, buf, sizeof(buf)) == 0 ||
((strncmp (buf, "HTTP/1.0 200", 12) != 0) && (strncmp (buf, "HTTP/1.1 200", 12) != 0))) ((strncmp (buf, "HTTP/1.0 200", 12) != 0) && (strncmp (buf, "HTTP/1.1 200", 12) != 0))) {
{
sock_close(mastersock); sock_close(mastersock);
ICECAST_LOG_WARN("Master rejected streamlist request"); ICECAST_LOG_WARN("Master rejected streamlist request");
break; break;
@ -904,8 +868,7 @@ static void *_slave_thread(void *arg)
config_release_config(); config_release_config();
source_recheck_mounts(1); source_recheck_mounts(1);
while (1) while (true) {
{
relay_t *cleanup_relays = NULL; relay_t *cleanup_relays = NULL;
int skip_timer = 0; int skip_timer = 0;
@ -925,12 +888,11 @@ static void *_slave_thread(void *arg)
} }
thread_mutex_unlock(&_slave_mutex); thread_mutex_unlock(&_slave_mutex);
++interval; interval++;
/* only update relays lists when required */ /* only update relays lists when required */
thread_mutex_lock(&_slave_mutex); thread_mutex_lock(&_slave_mutex);
if (max_interval <= interval) if (max_interval <= interval) {
{
ICECAST_LOG_DEBUG("checking master stream list"); ICECAST_LOG_DEBUG("checking master stream list");
config = config_get_config(); config = config_get_config();
@ -949,9 +911,7 @@ static void *_slave_thread(void *arg)
cleanup_relays = update_relays(&global.relays, config->relay, config->relay_length); cleanup_relays = update_relays(&global.relays, config->relay, config->relay_length);
config_release_config(); config_release_config();
} } else {
else
{
thread_mutex_unlock(&_slave_mutex); thread_mutex_unlock(&_slave_mutex);
thread_mutex_lock(&(config_locks()->relay_lock)); thread_mutex_lock(&(config_locks()->relay_lock));
} }
@ -961,8 +921,7 @@ static void *_slave_thread(void *arg)
thread_mutex_unlock(&(config_locks()->relay_lock)); thread_mutex_unlock(&(config_locks()->relay_lock));
thread_mutex_lock(&_slave_mutex); thread_mutex_lock(&_slave_mutex);
if (update_settings) if (update_settings) {
{
source_recheck_mounts(update_all_mounts); source_recheck_mounts(update_all_mounts);
update_settings = 0; update_settings = 0;
update_all_mounts = 0; update_all_mounts = 0;