From a7c3a5c479812f1f69acac276402fdffc60371aa Mon Sep 17 00:00:00 2001 From: Christian Pointner Date: Wed, 3 Sep 2014 23:09:49 +0200 Subject: added initial code - based on rharchive --- src/writer.c | 243 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 243 insertions(+) create mode 100644 src/writer.c (limited to 'src/writer.c') diff --git a/src/writer.c b/src/writer.c new file mode 100644 index 0000000..f282f8d --- /dev/null +++ b/src/writer.c @@ -0,0 +1,243 @@ +/* + * 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 . + */ + +#include + +#include + +#include +#include +#include + +#include "writer.h" + +#include "datatypes.h" +#include "log.h" +#include "file_list.h" + +static int init_time_boundaries(writer_t* writer) +{ + if(!writer) + return -1; + + struct timespec now; + clock_gettime(CLOCK_REALTIME, &now); + + struct tm bd_time; + localtime_r(&(now.tv_sec), &bd_time); + + writer->current_ = file_list_add(&(writer->files_), &bd_time, "current", writer->name_format_, writer->output_dir_, writer->mode_, writer->nocache_); + if(writer->current_ == NULL) return -2; + + bd_time.tm_sec = 0; + bd_time.tm_min = 0; + time_t T = mktime(&bd_time); + T+=3600; + localtime_r(&T, &bd_time); + + struct timespec b = { T, 0 }; + writer->next_boundary_ = b; + + writer->next_ = file_list_add(&(writer->files_), &bd_time, "next", writer->name_format_, writer->output_dir_, writer->mode_, writer->nocache_); + if(writer->next_ == NULL) return -2; + + return 0; +} + +static void added_cb(GstElement* sink, gint fd, gpointer data) +{ + gint num_fds; + g_object_get(G_OBJECT(sink), "num_fds", &num_fds, NULL); + log_printf(INFO, "fdsink: successfully added client %d (sink has now %d fds)", fd, num_fds); +} + +static void removed_cb(GstElement* sink, gint fd, gpointer data) +{ + gint num_fds; + g_object_get(G_OBJECT(sink), "num_fds", &num_fds, NULL); + log_printf(INFO, "fdsink: successfully removed client %d (sink has now %d fds)", fd, num_fds); +} + +static void fdremoved_cb(GstElement* sink, gint fd, gpointer data) +{ + gint num_fds; + g_object_get(G_OBJECT(sink), "num_fds", &num_fds, NULL); + log_printf(INFO, "fdsink: successfully removed fd %d (sink has now %d fds)", fd, num_fds); + + writer_t *writer = (writer_t*)data; + if(writer->post_process_) + file_list_call_post_process(&(writer->files_), fd, writer->post_process_); + else + file_list_remove(&(writer->files_), fd); +} + +int writer_init(writer_t* writer, GMainLoop *loop, const char* name_format, mode_t mode, int nocache, const char* output_dir, int interval, int offset, char* post_process) +{ + if(!writer) + return -1; + + writer->loop_ = loop; + writer->sink_ = gst_element_factory_make("multifdsink", "writer"); + if(!writer->sink_) { + log_printf(ERROR, "the writer object could not be created. Exiting."); + return -1; + } + // TODO: how the heck should we get the right value? 3 means keyframe... + g_object_set(G_OBJECT(writer->sink_), "recover-policy", 3, NULL); + g_signal_connect(G_OBJECT(writer->sink_), "client-added", G_CALLBACK(added_cb), writer); + g_signal_connect(G_OBJECT(writer->sink_), "client-removed", G_CALLBACK(removed_cb), writer); + g_signal_connect(G_OBJECT(writer->sink_), "client-fd-removed", G_CALLBACK(fdremoved_cb), writer); + + writer->clock_ = gst_system_clock_obtain(); + if(!writer->clock_) { + log_printf(ERROR, "unable to obtain the system clock"); + return -1; + } + writer->name_format_ = name_format; + writer->mode_ = mode; + writer->nocache_ = nocache; + writer->output_dir_ = output_dir; + writer->interval_ = interval * GST_MSECOND; + writer->offset_ = offset * GST_MSECOND; + writer->post_process_ = post_process; + writer->clock_id_ = NULL; + writer->thread_ = NULL; + int ret = file_list_init(&(writer->files_)); + if(ret) return ret; + + return init_time_boundaries(writer); +} + +static void add_fd(writer_t* writer, int fd) +{ + log_printf(INFO, "adding fd %d to fdsink", fd); + g_signal_emit_by_name(G_OBJECT(writer->sink_), "add", fd, NULL); +} + +static void remove_fd(writer_t* writer, int fd) +{ + log_printf(INFO, "removing fd %d from fdsink", fd); + g_signal_emit_by_name(G_OBJECT(writer->sink_), "remove-flush", fd, NULL); +} + +static int check_boundaries(writer_t* writer) +{ + struct timespec now; + clock_gettime(CLOCK_REALTIME, &now); + + GstClockTime tmp = GST_TIMESPEC_TO_TIME(now); + GstClockTime boundary = GST_TIMESPEC_TO_TIME(writer->next_boundary_); + tmp -= writer->offset_; + if(tmp >= boundary) { + struct tm now_bd; + localtime_r(&(now.tv_sec), &now_bd); + log_printf(INFO, "boundary reached! it's now: %02d:%02d:%02d.%06d on %d.%d.%d%s (%d ms offset)", now_bd.tm_hour, now_bd.tm_min, now_bd.tm_sec, now.tv_nsec/1000, now_bd.tm_mday, now_bd.tm_mon+1, now_bd.tm_year+1900, now_bd.tm_isdst > 0 ? " (DST)": "", GST_TIME_AS_MSECONDS(writer->offset_)); + + int ret = open_file(writer->next_); + if(ret) return ret; // TODO: stop writer on open_file error ??? + + add_fd(writer, writer->next_->fd_); + remove_fd(writer, writer->current_->fd_); + writer->current_ = writer->next_; + + writer->next_boundary_.tv_sec += 3600; + struct tm bd_time; + localtime_r(&(writer->next_boundary_.tv_sec), &bd_time); + + writer->next_ = file_list_add(&(writer->files_), &bd_time, "next", writer->name_format_, writer->output_dir_, writer->mode_, writer->nocache_); + if(writer->next_ == NULL) return -2; + } + + return 0; +} + +static gpointer writer_thread_func(gpointer data) +{ + writer_t *writer = (writer_t*)data; + log_printf(NOTICE, "writer thread started"); + + for(;;) { + GstClockReturn wret = gst_clock_id_wait(writer->clock_id_, NULL); + if(GST_CLOCK_UNSCHEDULED == wret) + break; + if(GST_CLOCK_EARLY == wret) + continue; + + int ret = check_boundaries(writer); + if(ret) break; + + ret = file_list_waitpid(&(writer->files_)); + if(ret) break; + } + + log_printf(NOTICE, "writer thread stopped"); + g_main_loop_quit(writer->loop_); + return NULL; +} + +int writer_start(writer_t* writer) +{ + if(!writer) + return -1; + + int ret = open_file(writer->current_); + if(ret) + return ret; + + add_fd(writer, writer->current_->fd_); + + writer->clock_id_ = gst_clock_new_periodic_id(writer->clock_, 0, writer->interval_); + if(!writer->clock_id_) { + log_printf(ERROR, "clock id could not be created"); + return -1; + } + writer->thread_ = g_thread_new("writer", writer_thread_func, writer); + if(!writer->thread_) { + log_printf(ERROR, "writer thread could not be started"); + return -1; + } + + return 0; +} + +void writer_stop(writer_t* writer) +{ + if(!writer) + return; + + file_list_clear(&(writer->files_)); + if(writer->clock_id_) { + gst_clock_id_unschedule(writer->clock_id_); + } + if(writer->thread_) { + log_printf(NOTICE, "waiting for writer thread to stop"); + g_thread_join(writer->thread_); + } + gst_object_unref(GST_OBJECT(writer->clock_)); +} -- cgit v1.2.3