/* * 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 compute_filname(struct tm* time, const char* type, file_t* file, const char* format, const char* dir, mode_t mode) { log_printf(INFO, "%s time: %02d:%02d:%02d on %d.%d.%d", type, time->tm_hour, time->tm_min, time->tm_sec, time->tm_mday, time->tm_mon+1, time->tm_year+1900); char name[256]; strftime(name, sizeof(name), format, time); asprintf(&(file->path_), "%s/%s", dir, name); if(!file->path_) return -2; log_printf(INFO, "%s filename is: %s(.?)", type, file->path_); file->fd_ = -1; file->mode_ = mode; return 0; } 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); int ret = compute_filname(&bd_time, "current", &(writer->current_), writer->name_format_, writer->output_dir_, writer->mode_); if(ret) return ret; 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; return compute_filname(&bd_time, "next", &(writer->next_), writer->name_format_, writer->output_dir_, writer->mode_); } 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) { if(file->fd_ > 0) // file already open! return -1; 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); return 0; } 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(DEBUG, "boundary reached! it's now: %02d:%02d:%02d.%06d on %d.%d.%d", 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); int ret = open_file(&(writer->next_)); // if(ret) return ret; // TODO: stop writer on open_file error ??? // - add new file // - remove old file // - calculate next filename and boundary // - call post processing script from fd_removed callback } return 0; } static gpointer writer_thread_func(gpointer data) { writer_t *writer = (writer_t*)data; log_printf(NOTICE, "writer thread started"); GstBuffer* buf = NULL; 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; // TODO: children handling (waitpid) } 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; // TODO: add fd to fdsink 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_)); }