/* * sydra * * sydra is a toolbox which allows you to set up multiple bidirectional * Video/Audio streams from external locations. * sydra has been written to be used for the Elevate Festival in Graz * Austria in order to involve external locations to present themselves * at the festival. * Sydra is based on GStreamer and is written in C. * * * Copyright (C) 2014 Christian Pointner * * This file is part of sydra. * * sydra is free software: you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation, either version 3 of the License, or * any later version. * * sydra is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with sydra. If not, see . * * In addition, as a special exception, the copyright holders hereby * grant permission for non-GPL-compatible GStreamer plugins to be used * and distributed together with GStreamer and sydra. * This permission goes above and beyond the permissions granted by the * GPL license sydra is covered by. */ #define _GNU_SOURCE #include #include #include #include #include #include #include #include #include #include #include "datatypes.h" #include "options.h" #include "string_list.h" #include "log.h" #include "daemon.h" #include "utils.h" #include "pipelines.h" static gboolean sig_handler_terminate(gpointer user_data) { GMainLoop *loop = (GMainLoop *)user_data; log_printf(NOTICE, "signal received, closing application"); g_main_loop_quit(loop); return TRUE; } static gboolean bus_call(GstBus *bus, GstMessage *msg, gpointer data) { GMainLoop *loop = (GMainLoop *)data; switch (GST_MESSAGE_TYPE(msg)) { case GST_MESSAGE_EOS: { log_printf(NOTICE, "End of stream"); g_main_loop_quit(loop); break; } case GST_MESSAGE_INFO: { GError *info; gst_message_parse_info(msg, &info, NULL); log_printf(INFO, "%s", info->message); g_error_free(info); break; } case GST_MESSAGE_WARNING: { GError *warning; gst_message_parse_warning(msg, &warning, NULL); log_printf(WARNING, "%s", warning->message); g_error_free(warning); break; } case GST_MESSAGE_ERROR: { GError *error; gst_message_parse_error(msg, &error, NULL); log_printf(ERROR, "%s", error->message); g_error_free(error); g_main_loop_quit(loop); break; } case GST_MESSAGE_STATE_CHANGED: { GstState old_state, new_state; gst_message_parse_state_changed(msg, &old_state, &new_state, NULL); log_printf(DEBUG, "Element '%s' changed state from %s to %s", (msg->src ? GST_OBJECT_NAME(msg->src) : "NULL"), gst_element_state_get_name(old_state), gst_element_state_get_name(new_state)); break; } case GST_MESSAGE_NEW_CLOCK: { GstClock *clock; gst_message_parse_new_clock(msg, &clock); log_printf(INFO, "New clock: %s", (clock ? GST_OBJECT_NAME (clock) : "NULL")); break; } case GST_MESSAGE_QOS: { guint64 running_time, stream_time, timestamp, duration; gst_message_parse_qos(msg, NULL, &running_time, &stream_time, ×tamp, &duration); log_printf(WARNING, "Element '%s' dropped frames running_time=%lu, stream_time=%lu, timestamp=%lu, duration=%lu", (msg->src ? GST_OBJECT_NAME(msg->src) : "NULL"), running_time, stream_time, timestamp, duration); break; } /* case GST_MESSAGE_STREAM_STATUS: */ /* { */ /* GstStreamStatusType type; */ /* GstElement *owner; */ /* const GValue *val; */ /* gchar *path, *ownerstr; */ /* GstTask *task = NULL; */ /* gst_message_parse_stream_status (msg, &type, &owner); */ /* val = gst_message_get_stream_status_object (msg); */ /* path = gst_object_get_path_string (GST_MESSAGE_SRC (msg)); */ /* ownerstr = gst_object_get_path_string (GST_OBJECT (owner)); */ /* log_printf(DEBUG,"Recevied Stream-Status message type: %d, source: %s, owner: %s, object: type %s, value %p", */ /* type, path, ownerstr, G_VALUE_TYPE_NAME (val), g_value_get_object (val)); */ /* g_free (path); */ /* g_free (ownerstr); */ /* /\* see if we know how to deal with this object *\/ */ /* if (G_VALUE_TYPE (val) == GST_TYPE_TASK) { */ /* task = g_value_get_object (val); */ /* } */ /* switch (type) { */ /* case GST_STREAM_STATUS_TYPE_CREATE: */ /* log_printf(DEBUG," created task %p", task); */ /* break; */ /* case GST_STREAM_STATUS_TYPE_ENTER: */ /* /\* log_printf(DEBUG," raising task priority"); *\/ */ /* /\* setpriority (PRIO_PROCESS, 0, -10); *\/ */ /* break; */ /* case GST_STREAM_STATUS_TYPE_LEAVE: */ /* break; */ /* default: */ /* break; */ /* } */ /* break; */ /* } */ default: log_printf(DEBUG, "unkonwn message %s from %s", GST_MESSAGE_TYPE_NAME(msg), GST_MESSAGE_SRC_NAME(msg)); return TRUE; } return TRUE; } static void udp_add_client(struct sockaddr_storage *addr, socklen_t addrlen, struct udp_sink* sink) { struct addr_port* client = g_new(struct addr_port, 1); if(!client) return; switch(addr->ss_family) { case AF_INET: client->type_ = IPv4; client->port_ = ntohs(((struct sockaddr_in*)addr)->sin_port); break; case AF_INET6: client->type_ = IPv6; client->port_ = ntohs(((struct sockaddr_in6*)addr)->sin6_port); break; default: return; } int errcode = getnameinfo((struct sockaddr *)addr, addrlen, client->addr_, sizeof(client->addr_), NULL, 0, NI_NUMERICHOST | NI_NUMERICSERV); if (errcode != 0) { log_printf(WARNING, "getnameinfo() error: %s", gai_strerror(errcode)); return; } gchar* name = gst_element_get_name(sink->udp_); if(!g_list_find_custom(sink->clients_, client, cmp_addr_port)) { log_printf(INFO, "adding host %s%c%d to list of receivers for element %s", client->addr_, client->type_ == IPv4 ? ':' : '.', client->port_, name); g_signal_emit_by_name(G_OBJECT(sink->udp_), "add", client->addr_, client->port_, NULL); sink->clients_ = g_list_append(sink->clients_, client); } else { log_printf(DEBUG, "not adding host %s%c%d to list of receivers for element %s - already added", client->addr_, client->type_ == IPv4 ? ':' : '.', client->port_, name); g_free(client); } g_free(name); } static void udp_remove_client(struct sockaddr_storage *addr, socklen_t addrlen, struct udp_sink* sink) { // TODO: implement this! gchar* name = gst_element_get_name(sink->udp_); log_printf(ERROR, "removing client from %s failed! - removing not yet implemented!", name); g_free(name); } #define UDP_PROTO_MAGIC "SYDRA:" #define UDP_PROTO_MAGIC_LEN (sizeof(UDP_PROTO_MAGIC)-1) #define UDP_PROTO_CMD_ADD_CLIENT "add_client\n" #define UDP_PROTO_CMD_ADD_CLIENT_LEN (sizeof(UDP_PROTO_CMD_ADD_CLIENT)-1) #define UDP_PROTO_CMD_RM_CLIENT "remove_client\n" #define UDP_PROTO_CMD_RM_CLIENT_LEN (sizeof(UDP_PROTO_CMD_RM_CLIENT)-1) static gboolean on_udp_desc_ready(gint fd, GIOCondition cond, gpointer user_data) { struct udp_sink* sink = (struct udp_sink*)user_data; if(!sink || !(sink->udp_)) { log_printf(WARNING, "File descriptor %d is ready for reading - but supplied element is NULL removing callback", fd); return FALSE; } struct sockaddr_storage addr; socklen_t addrlen = sizeof(addr); u_int8_t buf[2048]; ssize_t bytes = recvfrom(fd, buf, sizeof(buf), 0, (struct sockaddr *)&addr, &addrlen); if(bytes < 1) { if(errno == EINTR) return TRUE; log_printf(WARNING, "Error while receiving UDP data on fd %d, will remove callback", fd); return FALSE; } if(bytes < UDP_PROTO_MAGIC_LEN || memcmp(UDP_PROTO_MAGIC, buf, UDP_PROTO_MAGIC_LEN)) { log_printf(DEBUG, "client discovery: ignoring invalid incoming packet"); return TRUE; } u_int8_t* cmd = &(buf[UDP_PROTO_MAGIC_LEN]); if(bytes == (UDP_PROTO_MAGIC_LEN+UDP_PROTO_CMD_ADD_CLIENT_LEN) && !memcmp(UDP_PROTO_CMD_ADD_CLIENT, cmd, UDP_PROTO_CMD_ADD_CLIENT_LEN)) { udp_add_client(&addr, addrlen, sink); } else if(bytes == (UDP_PROTO_MAGIC_LEN+UDP_PROTO_CMD_RM_CLIENT_LEN) && !memcmp(UDP_PROTO_CMD_RM_CLIENT, cmd, UDP_PROTO_CMD_RM_CLIENT_LEN)) { udp_remove_client(&addr, addrlen, sink); } else { log_printf(DEBUG, "client discovery: ignoring invalid command"); } return TRUE; } static gboolean attach_udp_descriptor(struct udp_sink* sink, const char* name, const char* prop) { GSocket *sock; g_object_get(G_OBJECT(sink->udp_), prop, &sock, NULL); int fd = g_socket_get_fd(sock); guint id = g_unix_fd_add(fd, G_IO_IN, on_udp_desc_ready, sink); if(id <= 0) { log_printf(ERROR, "Error while adding %s file descriptor (%d) to main loop", name, fd); return FALSE; } log_printf(DEBUG, "successfully installed callback for %s (fd: %d) for reading (id: %d)", name, fd, id); return TRUE; } static gboolean attach_udp_descriptors(struct udp_sinks *sinks) { if(!attach_udp_descriptor(&(sinks->rtp_video_), "RTP(video) IPv4", "used-socket") || !attach_udp_descriptor(&(sinks->rtp_video_), "RTP(video) IPv6", "used-socket-v6") || !attach_udp_descriptor(&(sinks->rtcp_video_), "RTCP(video) IPv4", "used-socket") || !attach_udp_descriptor(&(sinks->rtcp_video_), "RTCP(video) IPv6", "used-socket-v6") || !attach_udp_descriptor(&(sinks->rtp_audio_), "RTP(audio) IPv4", "used-socket") || !attach_udp_descriptor(&(sinks->rtp_audio_), "RTP(audio) IPv6", "used-socket-v6") || !attach_udp_descriptor(&(sinks->rtcp_audio_), "RTCP(audio) IPv4", "used-socket") || !attach_udp_descriptor(&(sinks->rtcp_audio_), "RTCP(audio) IPv6", "used-socket-v6")) { return FALSE; } return TRUE; } int main_loop(options_t* opt) { log_printf(INFO, "entering main loop"); struct udp_sinks udp = { { NULL, NULL }, { NULL, NULL }, { NULL, NULL }, { NULL, NULL } }; GstElement *pipeline = create_pipeline(opt, &udp); if(!pipeline) { log_printf(ERROR, "creating pipeline failed"); return -1; } GMainLoop *loop = g_main_loop_new(NULL, FALSE); GstBus *bus = gst_pipeline_get_bus(GST_PIPELINE(pipeline)); gst_bus_add_watch(bus, bus_call, loop); gst_object_unref(GST_OBJECT(bus)); gulong deep_notify_id = 0; if(opt->debug_) { deep_notify_id = g_signal_connect (pipeline, "deep-notify", G_CALLBACK (gst_object_default_deep_notify), NULL); } log_printf(INFO, "Set State: Paused"); gst_element_set_state(pipeline, GST_STATE_PAUSED); if(!attach_udp_descriptors(&udp)) return -1; log_printf(INFO, "Set State: Playing"); gst_element_set_state(pipeline, GST_STATE_PLAYING); g_unix_signal_add(SIGHUP, sig_handler_terminate, loop); g_unix_signal_add(SIGINT, sig_handler_terminate, loop); g_unix_signal_add(SIGTERM, sig_handler_terminate, loop); g_main_loop_run(loop); if (deep_notify_id != 0) g_signal_handler_disconnect (pipeline, deep_notify_id); log_printf(NOTICE, "Stopping pipeline"); gst_element_set_state(pipeline, GST_STATE_NULL); gst_object_unref(GST_OBJECT(pipeline)); return 0; } int main(int argc, char* argv[]) { log_init(); options_t opt; int ret = options_parse(&opt, argc, argv); if(ret) { if(ret > 0) fprintf(stderr, "syntax error near: %s\n\n", argv[ret]); if(ret == -2) fprintf(stderr, "memory error on options_parse, exitting\n"); if(ret == -3) options_print_version(); if(ret == -4) fprintf(stderr, "the port number is invalid\n"); if(ret != -2 && ret != -3 && ret != -4) options_print_usage(); if(ret == -1 || ret == -3) ret = 0; options_clear(&opt); log_close(); exit(ret); } slist_element_t* tmp = opt.log_targets_.first_; while(tmp) { ret = log_add_target(tmp->data_); if(ret) { switch(ret) { case -2: fprintf(stderr, "memory error on log_add_target, exitting\n"); break; case -3: fprintf(stderr, "unknown log target: '%s', exitting\n", (char*)(tmp->data_)); break; case -4: fprintf(stderr, "this log target is only allowed once: '%s', exitting\n", (char*)(tmp->data_)); break; default: fprintf(stderr, "syntax error near: '%s', exitting\n", (char*)(tmp->data_)); break; } options_clear(&opt); log_close(); exit(ret); } tmp = tmp->next_; } log_printf(NOTICE, "just started..."); options_parse_post(&opt); if(opt.debug_) options_print(&opt); if(opt.appname_) g_set_prgname (opt.appname_); else g_set_prgname (opt.progname_); priv_info_t priv; if(opt.username_) if(priv_init(&priv, opt.username_, opt.groupname_)) { options_clear(&opt); log_close(); exit(-1); } FILE* pid_file = NULL; if(opt.pid_file_) { pid_file = fopen(opt.pid_file_, "w"); if(!pid_file) { log_printf(WARNING, "unable to open pid file: %s", strerror(errno)); } } if(opt.chroot_dir_) if(do_chroot(opt.chroot_dir_)) { options_clear(&opt); log_close(); exit(-1); } if(opt.username_) if(priv_drop(&priv)) { options_clear(&opt); log_close(); exit(-1); } if(opt.daemonize_) { pid_t oldpid = getpid(); daemonize(); log_printf(INFO, "running in background now (old pid: %d)", oldpid); } if(pid_file) { pid_t pid = getpid(); fprintf(pid_file, "%d", pid); fclose(pid_file); } gst_init(NULL, NULL); const gchar *nano_str; guint major, minor, micro, nano; gst_version(&major, &minor, µ, &nano); if (nano == 1) nano_str = " (CVS)"; else if (nano == 2) nano_str = " (Prerelease)"; else nano_str = ""; log_printf(NOTICE, "sydra linked against GStreamer %d.%d.%d%s", major, minor, micro, nano_str); ret = main_loop(&opt); options_clear(&opt); log_printf(NOTICE, "sydra shutdown"); gst_deinit(); log_close(); return ret; }