/* * rharchive * * rharchive is a simple tcp connection proxy which combines the * features of rinetd and 6tunnel. rharchive supports IPv4 and * IPv6 and also supports connections from IPv6 to IPv4 * endpoints and vice versa. * * * Copyright (C) 2010-2011 Christian Pointner * * This file is part of rharchive. * * rharchive 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. * * rharchive 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 rharchive. If not, see . */ #include #include #include #include #include #include #include #include "writer.h" #include "datatypes.h" #include "log.h" static int init_time_boundaries(writer_t* writer) { if(!writer) return; struct timespec now; clock_gettime(CLOCK_REALTIME, &now); struct tm bd_time; localtime_r(&(now.tv_sec), &bd_time); log_printf(INFO, "it's now: %02d:%02d:%02d on %d.%d.%d", bd_time.tm_hour, bd_time.tm_min, bd_time.tm_sec, bd_time.tm_mday, bd_time.tm_mon+1, bd_time.tm_year+1900); char name[256]; strftime(name, sizeof(name), writer->name_format_, &bd_time); asprintf(&(writer->current_.path_), "%s/%s", writer->output_dir_, name); if(!writer->current_.path_) { return -2; } log_printf(INFO, "current filename is: %s(.?)", writer->current_.path_); writer->current_.fd_ = -1; writer->current_.mode_ = writer->mode_; bd_time.tm_sec = 0; bd_time.tm_min = 0; time_t T = mktime(&bd_time); T+=3600; localtime_r(&T, &bd_time); log_printf(INFO, "next boundary is at: %02d:%02d:%02d on %d.%d.%d", bd_time.tm_hour, bd_time.tm_min, bd_time.tm_sec, bd_time.tm_mday, bd_time.tm_mon+1, bd_time.tm_year+1900); strftime(name, sizeof(name), writer->name_format_, &bd_time); asprintf(&(writer->next_.path_), "%s/%s", writer->output_dir_, name); if(!writer->next_.path_) { free(writer->current_.path_); return -2; } log_printf(INFO, "next filename will be: %s(.?)", writer->next_.path_); writer->next_.fd_ = -1; writer->next_.mode_ = writer->mode_; struct timespec b = { T, 0 }; writer->next_boundary_ = b; return 0; } int writer_init(writer_t* writer, GMainLoop *loop, const char* name_format, mode_t mode, const char* output_dir, int interval, int offset) { if(!writer) return -1; writer->loop_ = loop; writer->sink_ = gst_element_factory_make("fakesink", "writer"); if(!writer->sink_) { log_printf(ERROR, "the writer object could not be created. Exiting."); return -1; } 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->output_dir_ = output_dir; writer->interval_ = interval * GST_MSECOND; writer->offset_ = offset * GST_MSECOND; writer->current_.path_ = NULL; writer->next_.path_ = NULL; writer->clock_id_ = NULL; writer->thread_ = NULL; return init_time_boundaries(writer); } static int open_file(file_t* file) { char* orig_path = file->path_; int cnt = 0; do { file->fd_ = open(file->path_, O_WRONLY | O_CREAT | O_EXCL, file->mode_); if(file->fd_ < 0) { if(errno != EEXIST) { // TODO: thread safe strerror log_printf(ERROR, "can't open file '%s': %s", file->path_, strerror(errno)); if(orig_path != file->path_) free(orig_path); return -1; } cnt++; char* tmp; asprintf(&tmp, "%s.%d", orig_path, cnt); if(!tmp) { if(orig_path != file->path_) free(orig_path); return -2; } if(file->path_ != orig_path) free(file->path_); file->path_ = tmp; } fchmod(file->fd_, file->mode_); } while(file->fd_ < 0); if(orig_path != file->path_) free(orig_path); } static gpointer writer_thread_func(gpointer data) { writer_t *writer = (writer_t*)data; log_printf(NOTICE, "writer thread started"); GstBuffer* buf = NULL; for(;;) { GstClockReturn ret = gst_clock_id_wait(writer->clock_id_, NULL); if(GST_CLOCK_UNSCHEDULED == ret) break; if(GST_CLOCK_EARLY == ret) continue; struct timespec now; clock_gettime(CLOCK_REALTIME, &now); struct tm now_bd; localtime_r(&(now.tv_sec), &now_bd); log_printf(DEBUG, "it's now: %02d:%02d:%02d on %d.%d.%d", now_bd.tm_hour, now_bd.tm_min, now_bd.tm_sec, now_bd.tm_mday, now_bd.tm_mon+1, now_bd.tm_year+1900); } 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; 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_create(writer_thread_func, writer, TRUE, NULL); 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; if(writer->current_.path_) free(writer->current_.path_); if(writer->current_.fd_ >= 0) close(writer->current_.fd_); if(writer->next_.path_) free(writer->next_.path_); if(writer->next_.fd_ >= 0) close(writer->next_.fd_); 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_)); }