Initial import

.gitignore

@@ -0,0 +1,3 @@
+*.o
+*.d
+tsdumper2

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "libfuncs"]
+	path = libfuncs
+	url = git://github.com/gfto/libfuncs.git

COPYING

@@ -0,0 +1,340 @@
+                    GNU GENERAL PUBLIC LICENSE
+                       Version 2, June 1991
+
+ Copyright (C) 1989, 1991 Free Software Foundation, Inc.
+ 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA
+ Everyone is permitted to copy and distribute verbatim copies
+ of this license document, but changing it is not allowed.
+
+                            Preamble
+
+  The licenses for most software are designed to take away your
+freedom to share and change it.  By contrast, the GNU General Public
+License is intended to guarantee your freedom to share and change free
+software--to make sure the software is free for all its users.  This
+General Public License applies to most of the Free Software
+Foundation's software and to any other program whose authors commit to
+using it.  (Some other Free Software Foundation software is covered by
+the GNU Library General Public License instead.)  You can apply it to
+your programs, too.
+
+  When we speak of free software, we are referring to freedom, not
+price.  Our General Public Licenses are designed to make sure that you
+have the freedom to distribute copies of free software (and charge for
+this service if you wish), that you receive source code or can get it
+if you want it, that you can change the software or use pieces of it
+in new free programs; and that you know you can do these things.
+
+  To protect your rights, we need to make restrictions that forbid
+anyone to deny you these rights or to ask you to surrender the rights.
+These restrictions translate to certain responsibilities for you if you
+distribute copies of the software, or if you modify it.
+
+  For example, if you distribute copies of such a program, whether
+gratis or for a fee, you must give the recipients all the rights that
+you have.  You must make sure that they, too, receive or can get the
+source code.  And you must show them these terms so they know their
+rights.
+
+  We protect your rights with two steps: (1) copyright the software, and
+(2) offer you this license which gives you legal permission to copy,
+distribute and/or modify the software.
+
+  Also, for each author's protection and ours, we want to make certain
+that everyone understands that there is no warranty for this free
+software.  If the software is modified by someone else and passed on, we
+want its recipients to know that what they have is not the original, so
+that any problems introduced by others will not reflect on the original
+authors' reputations.
+
+  Finally, any free program is threatened constantly by software
+patents.  We wish to avoid the danger that redistributors of a free
+program will individually obtain patent licenses, in effect making the
+program proprietary.  To prevent this, we have made it clear that any
+patent must be licensed for everyone's free use or not licensed at all.
+
+  The precise terms and conditions for copying, distribution and
+modification follow.
+
+                    GNU GENERAL PUBLIC LICENSE
+   TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION
+
+  0. This License applies to any program or other work which contains
+a notice placed by the copyright holder saying it may be distributed
+under the terms of this General Public License.  The "Program", below,
+refers to any such program or work, and a "work based on the Program"
+means either the Program or any derivative work under copyright law:
+that is to say, a work containing the Program or a portion of it,
+either verbatim or with modifications and/or translated into another
+language.  (Hereinafter, translation is included without limitation in
+the term "modification".)  Each licensee is addressed as "you".
+
+Activities other than copying, distribution and modification are not
+covered by this License; they are outside its scope.  The act of
+running the Program is not restricted, and the output from the Program
+is covered only if its contents constitute a work based on the
+Program (independent of having been made by running the Program).
+Whether that is true depends on what the Program does.
+
+  1. You may copy and distribute verbatim copies of the Program's
+source code as you receive it, in any medium, provided that you
+conspicuously and appropriately publish on each copy an appropriate
+copyright notice and disclaimer of warranty; keep intact all the
+notices that refer to this License and to the absence of any warranty;
+and give any other recipients of the Program a copy of this License
+along with the Program.
+
+You may charge a fee for the physical act of transferring a copy, and
+you may at your option offer warranty protection in exchange for a fee.
+
+  2. You may modify your copy or copies of the Program or any portion
+of it, thus forming a work based on the Program, and copy and
+distribute such modifications or work under the terms of Section 1
+above, provided that you also meet all of these conditions:
+
+    a) You must cause the modified files to carry prominent notices
+    stating that you changed the files and the date of any change.
+
+    b) You must cause any work that you distribute or publish, that in
+    whole or in part contains or is derived from the Program or any
+    part thereof, to be licensed as a whole at no charge to all third
+    parties under the terms of this License.
+
+    c) If the modified program normally reads commands interactively
+    when run, you must cause it, when started running for such
+    interactive use in the most ordinary way, to print or display an
+    announcement including an appropriate copyright notice and a
+    notice that there is no warranty (or else, saying that you provide
+    a warranty) and that users may redistribute the program under
+    these conditions, and telling the user how to view a copy of this
+    License.  (Exception: if the Program itself is interactive but
+    does not normally print such an announcement, your work based on
+    the Program is not required to print an announcement.)
+
+These requirements apply to the modified work as a whole.  If
+identifiable sections of that work are not derived from the Program,
+and can be reasonably considered independent and separate works in
+themselves, then this License, and its terms, do not apply to those
+sections when you distribute them as separate works.  But when you
+distribute the same sections as part of a whole which is a work based
+on the Program, the distribution of the whole must be on the terms of
+this License, whose permissions for other licensees extend to the
+entire whole, and thus to each and every part regardless of who wrote it.
+
+Thus, it is not the intent of this section to claim rights or contest
+your rights to work written entirely by you; rather, the intent is to
+exercise the right to control the distribution of derivative or
+collective works based on the Program.
+
+In addition, mere aggregation of another work not based on the Program
+with the Program (or with a work based on the Program) on a volume of
+a storage or distribution medium does not bring the other work under
+the scope of this License.
+
+  3. You may copy and distribute the Program (or a work based on it,
+under Section 2) in object code or executable form under the terms of
+Sections 1 and 2 above provided that you also do one of the following:
+
+    a) Accompany it with the complete corresponding machine-readable
+    source code, which must be distributed under the terms of Sections
+    1 and 2 above on a medium customarily used for software interchange; or,
+
+    b) Accompany it with a written offer, valid for at least three
+    years, to give any third party, for a charge no more than your
+    cost of physically performing source distribution, a complete
+    machine-readable copy of the corresponding source code, to be
+    distributed under the terms of Sections 1 and 2 above on a medium
+    customarily used for software interchange; or,
+
+    c) Accompany it with the information you received as to the offer
+    to distribute corresponding source code.  (This alternative is
+    allowed only for noncommercial distribution and only if you
+    received the program in object code or executable form with such
+    an offer, in accord with Subsection b above.)
+
+The source code for a work means the preferred form of the work for
+making modifications to it.  For an executable work, complete source
+code means all the source code for all modules it contains, plus any
+associated interface definition files, plus the scripts used to
+control compilation and installation of the executable.  However, as a
+special exception, the source code distributed need not include
+anything that is normally distributed (in either source or binary
+form) with the major components (compiler, kernel, and so on) of the
+operating system on which the executable runs, unless that component
+itself accompanies the executable.
+
+If distribution of executable or object code is made by offering
+access to copy from a designated place, then offering equivalent
+access to copy the source code from the same place counts as
+distribution of the source code, even though third parties are not
+compelled to copy the source along with the object code.
+
+  4. You may not copy, modify, sublicense, or distribute the Program
+except as expressly provided under this License.  Any attempt
+otherwise to copy, modify, sublicense or distribute the Program is
+void, and will automatically terminate your rights under this License.
+However, parties who have received copies, or rights, from you under
+this License will not have their licenses terminated so long as such
+parties remain in full compliance.
+
+  5. You are not required to accept this License, since you have not
+signed it.  However, nothing else grants you permission to modify or
+distribute the Program or its derivative works.  These actions are
+prohibited by law if you do not accept this License.  Therefore, by
+modifying or distributing the Program (or any work based on the
+Program), you indicate your acceptance of this License to do so, and
+all its terms and conditions for copying, distributing or modifying
+the Program or works based on it.
+
+  6. Each time you redistribute the Program (or any work based on the
+Program), the recipient automatically receives a license from the
+original licensor to copy, distribute or modify the Program subject to
+these terms and conditions.  You may not impose any further
+restrictions on the recipients' exercise of the rights granted herein.
+You are not responsible for enforcing compliance by third parties to
+this License.
+
+  7. If, as a consequence of a court judgment or allegation of patent
+infringement or for any other reason (not limited to patent issues),
+conditions are imposed on you (whether by court order, agreement or
+otherwise) that contradict the conditions of this License, they do not
+excuse you from the conditions of this License.  If you cannot
+distribute so as to satisfy simultaneously your obligations under this
+License and any other pertinent obligations, then as a consequence you
+may not distribute the Program at all.  For example, if a patent
+license would not permit royalty-free redistribution of the Program by
+all those who receive copies directly or indirectly through you, then
+the only way you could satisfy both it and this License would be to
+refrain entirely from distribution of the Program.
+
+If any portion of this section is held invalid or unenforceable under
+any particular circumstance, the balance of the section is intended to
+apply and the section as a whole is intended to apply in other
+circumstances.
+
+It is not the purpose of this section to induce you to infringe any
+patents or other property right claims or to contest validity of any
+such claims; this section has the sole purpose of protecting the
+integrity of the free software distribution system, which is
+implemented by public license practices.  Many people have made
+generous contributions to the wide range of software distributed
+through that system in reliance on consistent application of that
+system; it is up to the author/donor to decide if he or she is willing
+to distribute software through any other system and a licensee cannot
+impose that choice.
+
+This section is intended to make thoroughly clear what is believed to
+be a consequence of the rest of this License.
+
+  8. If the distribution and/or use of the Program is restricted in
+certain countries either by patents or by copyrighted interfaces, the
+original copyright holder who places the Program under this License
+may add an explicit geographical distribution limitation excluding
+those countries, so that distribution is permitted only in or among
+countries not thus excluded.  In such case, this License incorporates
+the limitation as if written in the body of this License.
+
+  9. The Free Software Foundation may publish revised and/or new versions
+of the General Public License from time to time.  Such new versions will
+be similar in spirit to the present version, but may differ in detail to
+address new problems or concerns.
+
+Each version is given a distinguishing version number.  If the Program
+specifies a version number of this License which applies to it and "any
+later version", you have the option of following the terms and conditions
+either of that version or of any later version published by the Free
+Software Foundation.  If the Program does not specify a version number of
+this License, you may choose any version ever published by the Free Software
+Foundation.
+
+  10. If you wish to incorporate parts of the Program into other free
+programs whose distribution conditions are different, write to the author
+to ask for permission.  For software which is copyrighted by the Free
+Software Foundation, write to the Free Software Foundation; we sometimes
+make exceptions for this.  Our decision will be guided by the two goals
+of preserving the free status of all derivatives of our free software and
+of promoting the sharing and reuse of software generally.
+
+                            NO WARRANTY
+
+  11. BECAUSE THE PROGRAM IS LICENSED FREE OF CHARGE, THERE IS NO WARRANTY
+FOR THE PROGRAM, TO THE EXTENT PERMITTED BY APPLICABLE LAW.  EXCEPT WHEN
+OTHERWISE STATED IN WRITING THE COPYRIGHT HOLDERS AND/OR OTHER PARTIES
+PROVIDE THE PROGRAM "AS IS" WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESSED
+OR IMPLIED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.  THE ENTIRE RISK AS
+TO THE QUALITY AND PERFORMANCE OF THE PROGRAM IS WITH YOU.  SHOULD THE
+PROGRAM PROVE DEFECTIVE, YOU ASSUME THE COST OF ALL NECESSARY SERVICING,
+REPAIR OR CORRECTION.
+
+  12. IN NO EVENT UNLESS REQUIRED BY APPLICABLE LAW OR AGREED TO IN WRITING
+WILL ANY COPYRIGHT HOLDER, OR ANY OTHER PARTY WHO MAY MODIFY AND/OR
+REDISTRIBUTE THE PROGRAM AS PERMITTED ABOVE, BE LIABLE TO YOU FOR DAMAGES,
+INCLUDING ANY GENERAL, SPECIAL, INCIDENTAL OR CONSEQUENTIAL DAMAGES ARISING
+OUT OF THE USE OR INABILITY TO USE THE PROGRAM (INCLUDING BUT NOT LIMITED
+TO LOSS OF DATA OR DATA BEING RENDERED INACCURATE OR LOSSES SUSTAINED BY
+YOU OR THIRD PARTIES OR A FAILURE OF THE PROGRAM TO OPERATE WITH ANY OTHER
+PROGRAMS), EVEN IF SUCH HOLDER OR OTHER PARTY HAS BEEN ADVISED OF THE
+POSSIBILITY OF SUCH DAMAGES.
+
+                     END OF TERMS AND CONDITIONS
+
+            How to Apply These Terms to Your New Programs
+
+  If you develop a new program, and you want it to be of the greatest
+possible use to the public, the best way to achieve this is to make it
+free software which everyone can redistribute and change under these terms.
+
+  To do so, attach the following notices to the program.  It is safest
+to attach them to the start of each source file to most effectively
+convey the exclusion of warranty; and each file should have at least
+the "copyright" line and a pointer to where the full notice is found.
+
+    <one line to give the program's name and a brief idea of what it does.>
+    Copyright (C) <year>  <name of author>
+
+    This program 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 2 of the License, or
+    (at your option) any later version.
+
+    This program 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 this program; if not, write to the Free Software
+    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA
+
+
+Also add information on how to contact you by electronic and paper mail.
+
+If the program is interactive, make it output a short notice like this
+when it starts in an interactive mode:
+
+    Gnomovision version 69, Copyright (C) year name of author
+    Gnomovision comes with ABSOLUTELY NO WARRANTY; for details type `show w'.
+    This is free software, and you are welcome to redistribute it
+    under certain conditions; type `show c' for details.
+
+The hypothetical commands `show w' and `show c' should show the appropriate
+parts of the General Public License.  Of course, the commands you use may
+be called something other than `show w' and `show c'; they could even be
+mouse-clicks or menu items--whatever suits your program.
+
+You should also get your employer (if you work as a programmer) or your
+school, if any, to sign a "copyright disclaimer" for the program, if
+necessary.  Here is a sample; alter the names:
+
+  Yoyodyne, Inc., hereby disclaims all copyright interest in the program
+  `Gnomovision' (which makes passes at compilers) written by James Hacker.
+
+  <signature of Ty Coon>, 1 April 1989
+  Ty Coon, President of Vice
+
+This General Public License does not permit incorporating your program into
+proprietary programs.  If your program is a subroutine library, you may
+consider it more useful to permit linking proprietary applications with the
+library.  If this is what you want to do, use the GNU Library General
+Public License instead of this License.

ChangeLog

@@ -0,0 +1,2 @@
+2013-07-22 : Version 0.9
+ * Initial public release.

Makefile

@@ -0,0 +1,112 @@
+CC = $(CROSS)$(TARGET)cc
+STRIP = $(CROSS)$(TARGET)strip
+MKDEP = $(CC) -MP -MM -o $*.d $<
+RM = rm -f
+MV = mv -f
+
+BUILD_ID = $(shell date +%Y%m%d_%H%M%S)
+VERSION = $(shell cat RELEASE)
+GIT_VER = $(shell git describe --tags --dirty --always 2>/dev/null)
+ifeq "$(GIT_VER)" ""
+GIT_VER = "release"
+endif
+
+ifndef V
+Q = @
+endif
+
+CFLAGS ?= -O2 -ggdb \
+ -W -Wall -Wextra \
+ -Wshadow -Wformat-security -Wstrict-prototypes
+
+DEFS = -DBUILD_ID=\"$(BUILD_ID)\" \
+ -DVERSION=\"$(VERSION)\" -DGIT_VER=\"$(GIT_VER)\"
+DEFS += -D_FILE_OFFSET_BITS=64
+
+PREFIX ?= /usr/local
+
+INSTALL_PRG = tsdumper2
+INSTALL_PRG_DIR = $(subst //,/,$(DESTDIR)/$(PREFIX)/bin)
+
+INSTALL_DOC = tsdumper2.1
+INSTALL_DOC_DIR = $(subst //,/,$(DESTDIR)/$(PREFIX)/share/man/man1)
+
+FUNCS_DIR = libfuncs
+FUNCS_LIB = $(FUNCS_DIR)/libfuncs.a
+
+tsdumper_SRC = \
+ udp.c \
+ util.c \
+ process.c \
+ tsdumper2.c
+tsdumper_LIBS = -lpthread
+
+tsdumper_OBJS = $(FUNCS_LIB) $(tsdumper_SRC:.c=.o)
+
+CLEAN_OBJS = tsdumper2 $(tsdumper_SRC:.c=.o) $(tsdumper_SRC:.c=.d)
+
+PROGS = tsdumper2
+
+.PHONY: help distclean clean install uninstall
+
+all: $(PROGS)
+
+$(FUNCS_LIB): $(FUNCS_DIR)/libfuncs.h
+	$(Q)echo "  MAKE	$(FUNCS_LIB)"
+	$(Q)$(MAKE) -s -C $(FUNCS_DIR)
+
+tsdumper2: $(tsdumper_OBJS)
+	$(Q)echo "  LINK	tsdumper2"
+	$(Q)$(CC) $(CFLAGS) $(DEFS) $(tsdumper_OBJS) $(tsdumper_LIBS) -o tsdumper2
+
+%.o: %.c Makefile RELEASE
+	@$(MKDEP)
+	$(Q)echo "  CC	tsdumper2	$<"
+	$(Q)$(CC) $(CFLAGS) $(DEFS) -c $<
+
+-include $(tsdumper_SRC:.c=.d)
+
+strip:
+	$(Q)echo "  STRIP	$(PROGS)"
+	$(Q)$(STRIP) $(PROGS)
+
+clean:
+	$(Q)echo "  RM	$(CLEAN_OBJS)"
+	$(Q)$(RM) $(CLEAN_OBJS)
+
+distclean: clean
+	$(Q)$(MAKE) -s -C $(FUNCS_DIR) clean
+
+install: all
+	@install -d "$(INSTALL_PRG_DIR)"
+	@echo "INSTALL $(INSTALL_PRG) -> $(INSTALL_PRG_DIR)"
+	$(Q)-install $(INSTALL_PRG) "$(INSTALL_PRG_DIR)"
+	@echo "INSTALL $(INSTALL_DOC) -> $(INSTALL_DOC_DIR)"
+	$(Q)-install --mode 0644 $(INSTALL_DOC) "$(INSTALL_DOC_DIR)"
+
+uninstall:
+	@-for FILE in $(INSTALL_PRG); do \
+		echo "RM       $(INSTALL_PRG_DIR)/$$FILE"; \
+		rm "$(INSTALL_PRG_DIR)/$$FILE"; \
+	done
+	@-for FILE in $(INSTALL_DOC); do \
+		echo "RM       $(INSTALL_DOC_DIR)/$$FILE"; \
+		rm "$(INSTALL_DOC_DIR)/$$FILE"; \
+	done
+
+help:
+	$(Q)echo -e "\
+tsdumper2 $(VERSION) ($(GIT_VER)) build\n\n\
+Build targets:\n\
+  tsdumper2|all   - Build tsdumper2.\n\
+\n\
+  install         - Install tsdumper2 in PREFIX ($(PREFIX))\n\
+  uninstall       - Uninstall tsdumper2 from PREFIX\n\
+\n\
+Cleaning targets:\n\
+  clean           - Remove tsdumper2 build files\n\
+  distclean       - Remove all build files\n\
+\n\
+  make V=1          Enable verbose build\n\
+  make PREFIX=dir   Set install prefix\n\
+  make CROSS=prefix Cross compile\n"

RELEASE

@@ -0,0 +1 @@
+0.9

TODO

@@ -0,0 +1,2 @@
+- Add README and man page.
+- Add http input.

libfuncs

@@ -0,0 +1 @@
+Subproject commit 2d8b0cb1051ce0196ee387ab6ae731c2f841c67b

process.c

@@ -0,0 +1,208 @@
+/*
+ * Process incoming data and save it into files.
+ * Copyright (C) 2013 Unix Solutions Ltd.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2
+ * as published by the Free Software Foundation.
+ *
+ * This program 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 (COPYING file) for more details.
+ *
+ */
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <errno.h>
+
+#include "tsdumper2.h"
+
+#define NO_UNLINK  0
+#define UNLINK_OLD 1
+
+#define ALIGN_DOWN(__src, __value) (__src - (__src % __value))
+
+static mode_t dir_perm;
+
+static int file_exists(char *filename) {
+	return access(filename, W_OK) == 0;
+}
+
+static void format_output_filename(struct ts *ts, time_t file_time) {
+	struct tm file_tm;
+	localtime_r(&file_time, &file_tm);
+
+	ts->output_startts = file_time;
+
+	ts->output_filename[0] = '\0';
+	strcat(ts->output_filename, ts->prefix);
+	strcat(ts->output_filename, "-");
+	strftime(ts->output_filename + strlen(ts->output_filename), OUTFILE_NAME_MAX, "%Y%m%d_%H%M%S-%s.ts", &file_tm);
+
+	ts->output_dirname[0] = '\0';
+	strftime(ts->output_dirname, OUTFILE_NAME_MAX, "%Y/%m/%d/%H", &file_tm);
+
+	ts->output_full_filename[0] = '\0';
+	snprintf(ts->output_full_filename, sizeof(ts->output_full_filename), "%s/%s",
+		ts->output_dirname, ts->output_filename);
+}
+
+static void report_file_creation(struct ts *ts, char *text_prefix, char *filename) {
+	char qdepth[32];
+	qdepth[0] = '\0';
+	if (ts->packet_queue->items)
+		snprintf(qdepth, sizeof(qdepth), " (depth:%d)", ts->packet_queue->items);
+	p_info("%s%s%s\n", text_prefix, filename, qdepth);
+}
+
+static void create_output_directory(struct ts *ts) {
+	if (!ts->create_dirs)
+		return;
+	if (!file_exists(ts->output_dirname)) {
+		p_info(" = Create directory %s", ts->output_dirname);
+		create_dir(ts->output_dirname, dir_perm);
+	}
+}
+
+static int create_output_file(struct ts *ts) {
+	char *filename = ts->output_filename;
+	if (ts->create_dirs)
+		filename = ts->output_full_filename;
+	create_output_directory(ts);
+	report_file_creation(ts, " = Create new file ", filename);
+	int fd = open(ts->output_filename, O_CREAT | O_WRONLY | O_TRUNC, 0644);
+	if (fd < 0) {
+		p_err("Can't create output file %s", ts->output_filename);
+		return -1;
+	}
+	if (ts->create_dirs) {
+		link(ts->output_filename, ts->output_full_filename);
+	}
+	return fd;
+}
+
+static int append_output_file(struct ts *ts) {
+	char *filename = ts->output_filename;
+	if (ts->create_dirs)
+		filename = ts->output_full_filename;
+	create_output_directory(ts);
+	report_file_creation(ts, " + Append to file ", filename);
+	int fd = open(ts->output_filename, O_APPEND | O_WRONLY);
+	if (fd < 0) {
+		p_err("Can't append to output file %s", ts->output_filename);
+		return -1;
+	}
+	if (ts->create_dirs) {
+		link(ts->output_filename, ts->output_full_filename);
+	}
+	return fd;
+}
+
+static void close_output_file(struct ts *ts, int unlink_file) {
+	if (ts->output_fd > -1) {
+		close(ts->output_fd);
+		if (unlink_file && ts->create_dirs) {
+			// The file is hard linked into the subdirectory. There is no need
+			// to keep it in the main directory.
+			unlink(ts->output_filename);
+		}
+	}
+}
+
+static void handle_files(struct ts *ts, struct packet *packet) {
+	int file_time = ALIGN_DOWN(packet->ts.tv_sec, ts->rotate_secs);
+
+	// Is this file already created?
+	if (file_time <= ts->output_startts)
+		return;
+
+	close_output_file(ts, UNLINK_OLD);
+	format_output_filename(ts, file_time);
+
+	/*
+	 * When tsdumper2 is started, try to continue writing into "current" file.
+	 * This allows the program to be killed/restarted.
+	 *
+	 * If current file does not exist, create new file with the time of the start
+	 * (not aligned to rotate_secs).
+	 */
+	int append = 0;
+	if (ts->output_fd < 0) { // First file (or error).
+		append = file_exists(ts->output_filename);
+		if (!append) { // Create first file *NOT ALIGNED*
+			format_output_filename(ts, packet->ts.tv_sec);
+		}
+	}
+
+	ts->output_fd = append ? append_output_file(ts) : create_output_file(ts);
+}
+
+void *write_thread(void *_ts) {
+	struct ts *ts = _ts;
+	struct packet *packet;
+
+	mode_t umask_val = umask(0);
+	dir_perm = (0777 & ~umask_val) | (S_IWUSR | S_IXUSR);
+
+	set_thread_name("tsdump-write");
+	while ((packet = queue_get(ts->packet_queue))) {
+		if (!packet->data_len)
+			continue;
+
+		time_t packet_time = packet->ts.tv_sec;
+		time_t file_time   = ALIGN_DOWN(packet_time, ts->rotate_secs);
+		p_dbg1(" - Got packet %d, size: %u, file_time:%lu packet_time:%lu depth:%d\n",
+			packet->num, packet->data_len, file_time, packet_time, ts->packet_queue->items);
+
+		handle_files(ts, packet);
+
+		if (ts->output_fd > -1) {
+			p_dbg2(" - Writing into fd:%d size:%d file:%s\n", ts->output_fd, packet->data_len, ts->output_filename);
+			ssize_t written = write(ts->output_fd, packet->data, packet->data_len);
+			if (written != packet->data_len) {
+				p_err("Can not write data (fd:%d written %zd of %d file:%s)",
+					ts->output_fd, written, packet->data_len, ts->output_filename);
+			}
+		}
+		free_packet(packet);
+	}
+	close_output_file(ts, NO_UNLINK);
+	return NULL;
+}
+
+static struct packet *add_to_queue(struct ts *ts) {
+	queue_add(ts->packet_queue, ts->current_packet);
+	ts->current_packet = alloc_packet(ts);
+	return ts->current_packet;
+}
+
+void process_packets(struct ts *ts, uint8_t *ts_packet, ssize_t readen) {
+	struct timeval now;
+	struct packet *packet = ts->current_packet;
+
+	if (packet->data_len + readen < PACKET_MAX_LENGTH) {
+		// Add data to buffer
+		memcpy(packet->data + packet->data_len, ts_packet, readen);
+		packet->data_len += readen;
+	} else {
+		// Too much data, add to queue
+		p_dbg1("*** Reached buffer end (%zd + %zd > %d)\n", packet->data_len + readen, readen, PACKET_MAX_LENGTH);
+		packet = add_to_queue(ts);
+	}
+
+	if (!packet->ts.tv_sec)
+		gettimeofday(&packet->ts, NULL);
+
+	gettimeofday(&now, NULL);
+	unsigned long long diff = timeval_diff_msec(&packet->ts, &now);
+	if (diff > PACKET_MAX_TIME) {
+		// Too much time have passed, add to queue
+		p_dbg1("+++ Reached time limit (%llu > %d)\n", diff, PACKET_MAX_TIME);
+		add_to_queue(ts);
+	}
+}

tsdumper2.c

@@ -0,0 +1,313 @@
+/*
+ * tsdumper2
+ * Copyright (C) 2013 Unix Solutions Ltd.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2
+ * as published by the Free Software Foundation.
+ *
+ * This program 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 (COPYING file) for more details.
+ *
+ */
+#include <stdlib.h>
+#include <stdarg.h>
+#include <unistd.h>
+#include <getopt.h>
+#include <string.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <signal.h>
+#include <fcntl.h>
+#include <errno.h>
+#include <sys/resource.h>
+
+#include "tsdumper2.h"
+
+extern int ai_family;
+
+#define PROGRAM_NAME "tsdumper2"
+static const char *program_id = PROGRAM_NAME " v" VERSION " (git-" GIT_VER ", build date " BUILD_ID ")";
+
+static int keep_running = 1;
+static unsigned long long total_read;
+
+static const char short_options[] = "n:s:d:i:z46DhV";
+
+static const struct option long_options[] = {
+	{ "prefix",				required_argument, NULL, 'n' },
+	{ "seconds",			required_argument, NULL, 's' },
+	{ "output-dir",			required_argument, NULL, 'd' },
+	{ "create-dirs",		no_argument,       NULL, 'D' },
+
+	{ "input",				required_argument, NULL, 'i' },
+	{ "input-ignore-disc",	no_argument,       NULL, 'z' },
+	{ "ipv4",				no_argument,       NULL, '4' },
+	{ "ipv6",				no_argument,       NULL, '6' },
+
+	{ "help",				no_argument,       NULL, 'h' },
+	{ "version",			no_argument,       NULL, 'V' },
+
+	{ 0, 0, 0, 0 }
+};
+
+static void show_help(struct ts *ts) {
+	printf("%s\n", program_id);
+	printf("Copyright (C) 2013 Unix Solutions Ltd.\n");
+	printf("\n");
+	printf("	Usage: " PROGRAM_NAME " -n <name> -i <input>\n");
+	printf("\n");
+	printf("Settings:\n");
+	printf(" -n --prefix <name>         | Filename prefix.\n");
+	printf(" -s --seconds <seconds>     | How much to save (default: %u sec).\n", ts->rotate_secs);
+	printf(" -d --output-dir <dir>      | Startup directory (default: %s).\n", ts->output_dir);
+	printf(" -D --create-dirs           | Save files in subdirs YYYY/MM/DD/HH/file.\n");
+	printf("\n");
+	printf("Input options:\n");
+	printf(" -i --input <source>        | Where to read from. Multicast address.\n");
+	printf("                            .  -i udp://224.0.0.1:5000    (v4 multicast)\n");
+	printf("                            .  -i udp://[ff01::1111]:5000 (v6 multicast)\n");
+	printf(" -z --input-ignore-disc     | Do not report discontinuty errors in input.\n");
+	printf(" -4 --ipv4                  | Use only IPv4 addresses.\n");
+	printf(" -6 --ipv6                  | Use only IPv6 addresses.\n");
+	printf("\n");
+	printf("Misc options:\n");
+	printf(" -h --help                  | Show help screen.\n");
+	printf(" -V --version               | Show program version.\n");
+	printf("\n");
+}
+
+static int parse_io_param(struct io *io, char *opt) {
+	int port_set = 0;
+	io->type = NET_IO;
+	int host_set = parse_host_and_port(opt, &io->hostname, &io->service, &port_set);
+	return !(!port_set || !host_set);
+}
+
+extern char *optarg;
+extern int optind, opterr, optopt;
+
+static void parse_options(struct ts *ts, int argc, char **argv) {
+	int j, input_addr_err = 1;
+	opterr = 0; // Prevent printing of error messages for unknown options in getopt()
+	while ((j = getopt_long(argc, argv, short_options, long_options, NULL)) != -1) {
+		if (j == '?')
+			die("Unknown parameter '%s'.\n", argv[optind - 1]);
+		switch (j) {
+			case 'n': // --prefix
+				ts->prefix = optarg;
+				if (strlen(optarg) >= PREFIX_MAX_LENGTH)
+					die("Prefix is longer than %d characters!", PREFIX_MAX_LENGTH);
+				break;
+			case 's': // --seconds
+				ts->rotate_secs = atoi(optarg);
+				break;
+			case 'd': // --output-dir
+				ts->output_dir = optarg;
+				break;
+			case 'D': // --create-dirs
+				ts->create_dirs = !ts->create_dirs;
+				break;
+			case 'i': // --input
+				input_addr_err = !parse_io_param(&ts->input, optarg);
+				break;
+			case 'z': // --input-ignore-disc
+				ts->ts_discont = !ts->ts_discont;
+				break;
+			case '4': // --ipv4
+				ai_family = AF_INET;
+				break;
+			case '6': // --ipv6
+				ai_family = AF_INET6;
+				break;
+			case 'h': // --help
+				show_help(ts);
+				exit(EXIT_SUCCESS);
+			case 'V': // --version
+				printf("%s\n", program_id);
+				exit(EXIT_SUCCESS);
+		}
+	}
+	if (input_addr_err || !ts->prefix) {
+		show_help(ts);
+		if (!ts->prefix)
+			fprintf(stderr, "ERROR: File name prefix is not set (--prefix XXX | -n XXX).\n");
+		if (input_addr_err)
+			fprintf(stderr, "ERROR: Input address is invalid (--input XXX | -i XXX).\n");
+		exit(EXIT_FAILURE);
+	}
+
+	p_info("Prefix     : %s\n", ts->prefix);
+	p_info("Input addr : udp://%s:%s/\n", ts->input.hostname, ts->input.service);
+	p_info("Seconds    : %u\n", ts->rotate_secs);
+	p_info("Output dir : %s (create directories: %s)\n", ts->output_dir,
+		ts->create_dirs ? "YES" : "no");
+	if (chdir(ts->output_dir) < 0)
+		die("Can not change directory to %s: %s\n", ts->output_dir, strerror(errno));
+}
+
+void signal_quit(int sig) {
+	if (!keep_running)
+		raise(sig);
+	keep_running = 0;
+	p_info("Killed %s with signal %d\n", program_id, sig);
+	signal(sig, SIG_DFL);
+}
+
+static void clear_packet(struct packet *p) {
+	p->ts.tv_sec  = 0;
+	p->ts.tv_usec = 0;
+	p->data_len   = 0;
+	p->allocated  = 0;
+	p->in_use     = 0;
+}
+
+struct packet *alloc_packet(struct ts *ts) {
+	// check for free static allocations
+	struct packet *p;
+	int i;
+	for (i = 0; i < NUM_PACKETS; i++) {
+		p = &ts->packets[i];
+		if (!p->in_use) {
+			p->in_use = 1;
+			p_dbg2("STATIC packet, num %d\n", p->num);
+			goto OUT;
+		}
+	}
+	// Dynamically allocate packet
+	p = malloc(sizeof(struct packet));
+	if (!p)
+		die("Can't alloc %lu bytes.\n", (unsigned long)sizeof(struct packet));
+	clear_packet(p);
+	p->num       = time(NULL);
+	p->allocated = 1;
+	p_dbg2("ALLOC  packet, num %d\n", p->num);
+OUT:
+	return p;
+}
+
+void free_packet(struct packet *packet) {
+	if (!packet->allocated) {
+		clear_packet(packet);
+		return;
+	}
+	p_dbg2("FREE   packet, num %d\n", packet->num);
+	free(packet);
+}
+
+#define RTP_HDR_SZ  12
+
+static uint8_t ts_packet[FRAME_SIZE + RTP_HDR_SZ];
+static uint8_t rtp_hdr[2][RTP_HDR_SZ];
+static struct ts ts;
+
+int main(int argc, char **argv) {
+	ssize_t readen;
+	int i;
+	int have_data = 1;
+	int ntimeouts = 0;
+	int rtp_hdr_pos = 0, num_packets = 0;
+	struct rlimit rl;
+
+	if (getrlimit(RLIMIT_STACK, &rl) == 0) {
+		if (rl.rlim_cur > THREAD_STACK_SIZE) {
+			rl.rlim_cur = THREAD_STACK_SIZE;
+			setrlimit(RLIMIT_STACK, &rl);
+		}
+	}
+
+	memset(rtp_hdr[0], 0, RTP_HDR_SZ);
+	memset(rtp_hdr[1], 0, RTP_HDR_SZ);
+
+	for (i = 0; i < NUM_PACKETS; i++) {
+		struct packet *p = &ts.packets[i];
+		p->num = i + 1;
+	}
+
+	ts.ts_discont     = 1;
+	ts.output_dir     = ".";
+	ts.rotate_secs    = 60;
+	ts.current_packet = alloc_packet(&ts);
+	ts.output_fd      = -1;
+
+	pthread_attr_init(&ts.thread_attr);
+	size_t stack_size;
+	pthread_attr_getstacksize(&ts.thread_attr, &stack_size);
+	if (stack_size > THREAD_STACK_SIZE)
+		pthread_attr_setstacksize(&ts.thread_attr, THREAD_STACK_SIZE);
+
+	parse_options(&ts, argc, argv);
+
+	ts.packet_queue   = queue_new();
+
+	p_info("Start %s\n", program_id);
+
+	if (ts.input.type == NET_IO && udp_connect_input(&ts.input) < 1)
+		goto EXIT;
+
+	signal(SIGCHLD, SIG_IGN);
+	signal(SIGPIPE, SIG_IGN);
+
+	signal(SIGINT , signal_quit);
+	signal(SIGTERM, signal_quit);
+
+	pthread_create(&ts.write_thread , &ts.thread_attr, &write_thread , &ts);
+
+	int data_received = 0;
+	do {
+		set_log_io_errors(0);
+		if (!ts.rtp_input) {
+			readen = fdread_ex(ts.input.fd, (char *)ts_packet, FRAME_SIZE, 250, 4, 1);
+		} else {
+			readen = fdread_ex(ts.input.fd, (char *)ts_packet, FRAME_SIZE + RTP_HDR_SZ, 250, 4, 1);
+			if (readen > RTP_HDR_SZ) {
+				memcpy(rtp_hdr[rtp_hdr_pos], ts_packet, RTP_HDR_SZ);
+				memmove(ts_packet, ts_packet + RTP_HDR_SZ, FRAME_SIZE);
+				readen -= RTP_HDR_SZ;
+				uint16_t ssrc  = (rtp_hdr[rtp_hdr_pos][2] << 8) | rtp_hdr[rtp_hdr_pos][3];
+				uint16_t pssrc = (rtp_hdr[!rtp_hdr_pos][2] << 8) | rtp_hdr[!rtp_hdr_pos][3];
+				rtp_hdr_pos = !rtp_hdr_pos;
+				if (pssrc + 1 != ssrc && (ssrc != 0 && pssrc != 0xffff) && num_packets > 2)
+					if (ts.ts_discont)
+						p_info(" *** RTP discontinuity last_ssrc %5d, curr_ssrc %5d, lost %d packet ***\n",
+							pssrc, ssrc, ((ssrc - pssrc)-1) & 0xffff);
+				num_packets++;
+			}
+		}
+		set_log_io_errors(1);
+		if (readen < 0) {
+			p_info(" *** Input read timeout ***\n");
+			data_received = 0;
+			ntimeouts++;
+		} else {
+			if (ntimeouts && readen > 0) {
+				ntimeouts = 0;
+			}
+		}
+		if (readen > 0) {
+			if (!data_received) {
+				p_info("Data received.\n");
+				data_received = 1;
+			}
+			total_read += readen;
+			process_packets(&ts, ts_packet, readen);
+		}
+		if (!keep_running)
+			break;
+	} while (have_data);
+EXIT:
+
+	queue_add(ts.packet_queue, ts.current_packet);
+	queue_add(ts.packet_queue, NULL); // Exit write_thread
+	pthread_join(ts.write_thread, NULL);
+
+	p_info("Stop %s (bytes_processed:%llu).\n", program_id, total_read);
+
+	queue_free(&ts.packet_queue);
+
+	pthread_attr_destroy(&ts.thread_attr);
+
+	exit(EXIT_SUCCESS);
+}

tsdumper2.h

@@ -0,0 +1,105 @@
+/*
+ * Data definitions
+ * Copyright (C) 2013 Unix Solutions Ltd.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2
+ * as published by the Free Software Foundation.
+ *
+ * This program 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 (COPYING file) for more details.
+ *
+ */
+#ifndef DATA_H
+#define DATA_H
+
+#include <pthread.h>
+#include <inttypes.h>
+#include <stdbool.h>
+
+#include "libfuncs/libfuncs.h"
+#include "util.h"
+
+// Supported values 0, 1 and 2. Higher value equals more spam in the log.
+#define DEBUG 0
+
+// 7 * 188
+#define FRAME_SIZE 1316
+
+// 64k should be enough for everybody
+#define THREAD_STACK_SIZE (64 * 1024)
+
+enum io_type {
+	FILE_IO,
+	NET_IO,
+	WTF_IO
+};
+
+// 1.2MB = ~13Mbit/s
+#define PACKET_MAX_LENGTH (1316 * 1024)
+
+// Maximum packet fill time in ms
+#define PACKET_MAX_TIME 1000
+
+#define PREFIX_MAX_LENGTH 64
+
+// PREFIX-20130717_000900-1374008940.ts (PREFIX-YYYYMMDD_HHMMSS-0123456789.ts)
+#define OUTFILE_NAME_MAX  (PREFIX_MAX_LENGTH + 128)
+
+#define NUM_PACKETS 16
+
+struct packet {
+	int					num;
+	struct timeval		ts;							// packet start time
+	int					allocated;					// set to true if the struct is dynamically allocated
+	int					in_use;						// this packet is currently being used
+	int					data_len;					// data length
+	uint8_t				data[PACKET_MAX_LENGTH];	// the data
+};
+
+struct io {
+	int					fd;
+	enum io_type		type;
+	char				*hostname;
+	char				*service;
+};
+
+struct ts {
+	char				*prefix;
+	char				*output_dir;
+	int					create_dirs;
+	int					rotate_secs;
+	int					ts_discont;
+	int					rtp_input;
+	struct io			input;
+
+	pthread_attr_t		thread_attr;
+	pthread_t			write_thread;
+
+	struct packet		packets[NUM_PACKETS];
+	int					packet_ptr;
+
+	struct packet		*current_packet;
+	QUEUE				*packet_queue;
+
+	int					output_fd;
+	time_t				output_startts;
+	char				output_dirname[OUTFILE_NAME_MAX];
+	char				output_filename[OUTFILE_NAME_MAX];
+	char				output_full_filename[OUTFILE_NAME_MAX];
+};
+
+// From tsdumper2.c
+struct packet *alloc_packet(struct ts *ts);
+void free_packet(struct packet *packet);
+
+// From process.c
+void *write_thread(void *_ts);
+void process_packets(struct ts *ts, uint8_t *ts_packet, ssize_t readen);
+
+// From udp.c
+int udp_connect_input(struct io *io);
+
+#endif

udp.c

@@ -0,0 +1,150 @@
+/*
+ * UDP functions
+ * Copyright (C) 2013 Unix Solutions Ltd.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2
+ * as published by the Free Software Foundation.
+ *
+ * This program 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 (COPYING file) for more details.
+ *
+ */
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <errno.h>
+
+#include "tsdumper2.h"
+
+#ifndef IPV6_ADD_MEMBERSHIP
+#define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
+#define IPV6_DROP_MEMBERSHIP IPV6_LEAVE_GROUP
+#endif
+
+int ai_family = AF_UNSPEC;
+
+static int is_multicast(struct sockaddr_storage *addr) {
+	int ret = 0;
+	switch (addr->ss_family) {
+	case AF_INET: {
+		struct sockaddr_in *addr4 = (struct sockaddr_in *)addr;
+		ret = IN_MULTICAST(ntohl(addr4->sin_addr.s_addr));
+		break;
+	}
+	case AF_INET6: {
+		struct sockaddr_in6 *addr6 = (struct sockaddr_in6 *)addr;
+		ret = IN6_IS_ADDR_MULTICAST(&addr6->sin6_addr);
+		break;
+	} }
+	return ret;
+}
+
+static int bind_addr(const char *hostname, const char *service, int socktype, struct sockaddr_storage *addr, int *addrlen, int *sock) {
+	struct addrinfo hints, *res, *ressave;
+	int n, ret = -1;
+
+	memset(&hints, 0, sizeof(struct addrinfo));
+	hints.ai_family = ai_family;
+	hints.ai_socktype = socktype;
+
+	n = getaddrinfo(hostname, service, &hints, &res);
+	if (n < 0) {
+		p_info("ERROR: getaddrinfo(%s): %s\n", hostname, gai_strerror(n));
+		return ret;
+	}
+
+	ressave = res;
+	while (res) {
+		*sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
+		if (*sock > -1) {
+			int on = 1;
+			setsockopt(*sock, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
+			set_sock_nonblock(*sock);
+			if (bind(*sock, res->ai_addr, res->ai_addrlen) == 0) {
+				memcpy(addr, res->ai_addr, res->ai_addrlen);
+				*addrlen = res->ai_addrlen;
+				ret = 0;
+				goto OUT;
+			} else {
+				char str_addr[INET6_ADDRSTRLEN];
+				my_inet_ntop(res->ai_family, res->ai_addr, str_addr, sizeof(str_addr));
+				p_err("bind: %s:%s (%s)", hostname, service, str_addr);
+			}
+			close(*sock);
+			*sock = -1;
+		}
+		res = res->ai_next;
+	}
+OUT:
+	freeaddrinfo(ressave);
+
+	return ret;
+}
+
+static int join_multicast_group(int sock, struct sockaddr_storage *addr) {
+	switch (addr->ss_family) {
+	case AF_INET: {
+		struct ip_mreq mreq;
+		mreq.imr_multiaddr.s_addr = ((struct sockaddr_in *)addr)->sin_addr.s_addr;
+		mreq.imr_interface.s_addr = INADDR_ANY;
+
+		if (setsockopt(sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, (const void *)&mreq, sizeof(mreq)) < 0) {
+			p_err("setsockopt(IP_ADD_MEMBERSHIP)");
+			return -1;
+		}
+		break;
+	}
+
+	case AF_INET6: {
+		struct ipv6_mreq mreq6;
+		memcpy(&mreq6.ipv6mr_multiaddr, &(((struct sockaddr_in6 *)addr)->sin6_addr), sizeof(struct in6_addr));
+		mreq6.ipv6mr_interface = 0; // interface index, will be set later
+
+		if (setsockopt(sock, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &mreq6, sizeof(mreq6)) < 0) {
+			p_err("setsockopt(IPV6_ADD_MEMBERSHIP)");
+			return -1;
+		}
+		break;
+	}
+	}
+
+	return 0;
+}
+
+int udp_connect_input(struct io *io) {
+	struct sockaddr_storage addr;
+	int addrlen = sizeof(addr);
+	int sock = -1;
+
+	memset(&addr, 0, sizeof(addr));
+
+	p_info("Connecting input to %s port %s\n", io->hostname, io->service);
+	if (bind_addr(io->hostname, io->service, SOCK_DGRAM, &addr, &addrlen, &sock) < 0)
+		return -1;
+
+	/* Set receive buffer size to ~4.0MB */
+	int bufsize = (4000000 / 1316) * 1316;
+	setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (void *)&bufsize, sizeof(bufsize));
+
+	if (is_multicast(&addr)) {
+		if (join_multicast_group(sock, &addr) < 0) {
+			close(sock);
+			return -1;
+		}
+	}
+
+	io->fd = sock;
+	p_info("Input connected to fd:%d\n", io->fd);
+
+	return 1;
+}

util.c

@@ -0,0 +1,167 @@
+/*
+ * Utility functions
+ * Copyright (C) 2013 Unix Solutions Ltd.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2
+ * as published by the Free Software Foundation.
+ *
+ * This program 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 (COPYING file) for more details.
+ *
+ */
+
+#include <stdio.h>
+#include <stdarg.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <ctype.h>
+#include <time.h>
+#include <string.h>
+#include <errno.h>
+#include <arpa/inet.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+
+#include "tsdumper2.h"
+
+#ifdef __linux__
+#include <sys/prctl.h>
+
+void set_thread_name(char *thread_name) {
+	prctl(PR_SET_NAME, thread_name, NULL, NULL, NULL);
+}
+
+#else
+void set_thread_name(char *thread_name) {
+    (void)thread_name;
+}
+
+#endif
+
+int parse_host_and_port(char *input, char **hostname, char **service, int *port_set) {
+	char *p, *proto;
+	if (strlen(input) == 0)
+		return 0;
+	proto = strstr(input, "http://");
+	if (proto == input)
+		die("HTTP input is unsupported (patch welcome): %s", input);
+	proto = strstr(input, "://");
+	if (proto)
+		input = proto + 3;
+	*hostname = input;
+	if (input[0] == '[') { // Detect IPv6 static address
+		p = strrchr(input, ']');
+		if (!p)
+			die("Invalid IPv6 address format: %s\n", input);
+		*hostname = input + 1; // Remove first [
+		*p = 0x00; // Remove last ]
+		char *p2 = strchr(p + 1, ':');
+		if (p2) {
+			*p2 = 0x00;
+			*service = p2 + 1;
+			*port_set = 1;
+		}
+	} else {
+		p = strrchr(input, ':');
+		if (p) {
+			*p = 0x00;
+			*service = p + 1;
+			*port_set = 1;
+		}
+	}
+	if (*service) {
+		char *path = strstr(*service, "/");
+		if (path)
+			path[0] = 0;
+	}
+	return 1;
+}
+
+char *my_inet_ntop(int family, struct sockaddr *addr, char *dest, int dest_len) {
+	struct sockaddr_in  *addr_v4 = (struct sockaddr_in  *)addr;
+	struct sockaddr_in6 *addr_v6 = (struct sockaddr_in6 *)addr;
+	switch (family) {
+		case AF_INET:
+			return (char *)inet_ntop(AF_INET, &addr_v4->sin_addr, dest, dest_len);
+			break;
+		case AF_INET6:
+			return (char *)inet_ntop(AF_INET6, &addr_v6->sin6_addr, dest, dest_len);
+			break;
+		default:
+			memset(dest, 0, dest_len);
+			strcpy(dest, "unknown");
+			return dest;
+	}
+}
+
+int create_dir(const char *dir, mode_t mode) {
+	int ret = 0;
+	unsigned int i;
+
+	// Shortcut
+	if (strchr(dir, '/') == NULL)
+		return mkdir(dir, mode);
+
+	char *d = strdup(dir);
+	unsigned int dlen = strlen(dir);
+
+	// Skip first char (it can be /)
+	for (i = 1; i < dlen; i++) {
+		if (d[i] != '/')
+			continue;
+		d[i] = '\0';
+		ret = mkdir(d, mode);
+		d[i] = '/';
+		if (ret < 0 && errno != EEXIST)
+			goto OUT;
+	}
+	ret = mkdir(d, mode);
+OUT:
+	free(d);
+	return ret;
+}
+
+void die(const char *fmt, ...) {
+	va_list args;
+	va_start(args, fmt);
+	fprintf(stderr, "ERROR: ");
+	vfprintf(stderr, fmt, args);
+	if (fmt[strlen(fmt) - 1] != '\n')
+		fprintf(stderr, "\n");
+	va_end(args);
+	exit(EXIT_FAILURE);
+}
+
+void p_err(const char *fmt, ...) {
+	va_list args;
+	va_start(args, fmt);
+	fprintf(stderr, "ERROR: ");
+	vfprintf(stderr, fmt, args);
+	if (fmt[strlen(fmt) - 1] != '\n')
+		fprintf(stderr, "\n");
+	va_end(args);
+}
+
+void p_info(const char *fmt, ...) {
+	if (DEBUG > 0) {
+		struct timeval tv;
+		gettimeofday(&tv, NULL);
+		fprintf(stdout, "%08ld.%08ld ", tv.tv_sec, tv.tv_usec);
+
+		char date[64];
+		struct tm tm;
+		localtime_r(&tv.tv_sec, &tm);
+		strftime(date, sizeof(date), "%F %H:%M:%S", localtime_r(&tv.tv_sec, &tm));
+		fprintf(stdout, "%s | ", date);
+	}
+	va_list args;
+	va_start(args, fmt);
+	vfprintf(stdout, fmt, args);
+	if (fmt[strlen(fmt) - 1] != '\n')
+		fprintf(stdout, "\n");
+	va_end(args);
+	fflush(stdout);
+}

util.h

@@ -0,0 +1,39 @@
+/*
+ * Utility functions header
+ * Copyright (C) 2013 Unix Solutions Ltd.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2
+ * as published by the Free Software Foundation.
+ *
+ * This program 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 (COPYING file) for more details.
+ *
+ */
+#ifndef UTIL_H
+#define UTIL_H
+
+#include <inttypes.h>
+#include <arpa/inet.h>
+
+void set_thread_name(char *thread_name);
+
+int parse_host_and_port(char *input, char **hostname, char **service, int *port_set);
+char *my_inet_ntop(int family, struct sockaddr *addr, char *dest, int dest_len);
+
+int create_dir(const char *dir, mode_t mode);
+
+#define p_dbg1(fmt, ...) \
+	do { if (DEBUG > 0) p_info(fmt, __VA_ARGS__); } while(0)
+
+#define p_dbg2(fmt, ...) \
+	do { if (DEBUG > 1) p_info(fmt, __VA_ARGS__); } while(0)
+
+__attribute__ ((format(printf, 1, 2))) void die(const char *fmt, ...);
+
+__attribute__ ((format(printf, 1, 2))) void p_err(const char *fmt, ...);
+__attribute__ ((format(printf, 1, 2))) void p_info(const char *fmt, ...);
+
+#endif