TT#52651 Squashed commit of the following:

commit 2029144368
Author: Richard Fuchs <rfuchs@sipwise.com>
Date:   Thu Feb 21 13:12:50 2019 -0500

    TT#52651 switch TCP to TLS

    Change-Id: Iab6b05d3b5c88553cbd6f531f3189084d9e71995

commit b28e718ee4
Author: Richard Fuchs <rfuchs@sipwise.com>
Date:   Fri Feb 22 09:20:54 2019 -0500

    TT#52651 generalise streambuf interface

    Change-Id: I7d5ab8ffe13e52d4dbb1901531cc13fcc173d60d

commit cb2dbd2a92
Author: Richard Fuchs <rfuchs@sipwise.com>
Date:   Tue Feb 19 09:32:56 2019 -0500

    TT#52651 add start/stop forwarding commands and party selection logic

    Change-Id: I8ef7e288d3a3e485bd2fa14e1a2407a0c8d94bac

commit 442c48f627
Author: Richard Fuchs <rfuchs@sipwise.com>
Date:   Thu Feb 14 15:43:23 2019 -0500

    TT#52651 produce output for TCP forwarding feature

    Change-Id: I18543921577faf655679829684f5af46c0af5054

commit 2ef8028eb2
Author: Richard Fuchs <rfuchs@sipwise.com>
Date:   Thu Feb 14 10:18:21 2019 -0500

    TT#52651 make recording to output files optional

    Change-Id: I12c288b965641352658ce3b499c2ee90593e1322

commit 10a58cd7a0
Author: Richard Fuchs <rfuchs@sipwise.com>
Date:   Wed Feb 13 16:02:16 2019 -0500

    TT#52651 strip streambuf into lib and include in recording daemon

    Change-Id: I1f6638961e9e767063e0b4e6b5d55d88799366d3

commit 9d3bb5bffc
Author: Richard Fuchs <rfuchs@sipwise.com>
Date:   Wed Feb 13 15:40:12 2019 -0500

    TT#52651 extract/move unrelated old legacy decoder struct members

    Change-Id: Iffd79b43180c30a9e128a460f7ba85ba49dedeaf

commit 1bc38e4201
Author: Richard Fuchs <rfuchs@sipwise.com>
Date:   Tue Feb 12 16:43:42 2019 -0500

    TT#52651 config options for forwarding option

    Change-Id: Ieaa2ee0e55a0c531158174bc6a534738a64dbee6

commit 06d61cd3dd
Author: Richard Fuchs <rfuchs@sipwise.com>
Date:   Tue Feb 12 16:29:52 2019 -0500

    TT#52651 move socket.[ch] into lib/

    includes necessary re-shuffling of additional code pieces

    Change-Id: I74b314ab5936ac8a0eeaff94e084617b59b28d79

Change-Id: I025e8ec86b90ede79565542dff57ec1559d04200

daemon/.gitignore

@@ -11,3 +11,5 @@ codeclib.c
 resample.c
 str.c
 fix_frame_channel_layout.h
+socket.c
+streambuf.c

daemon/Makefile

@@ -119,12 +119,12 @@ ifeq ($(have_bcg729),yes)
 LDLIBS+=	$(bcg729_lib)
 endif
 
-SRCS=		main.c kernel.c poller.c aux.c control_tcp.c streambuf.c call.c control_udp.c redis.c \
+SRCS=		main.c kernel.c poller.c aux.c control_tcp.c call.c control_udp.c redis.c \
 		bencode.c cookie_cache.c udp_listener.c control_ng.strhash.c sdp.strhash.c stun.c rtcp.c \
-		crypto.c rtp.c call_interfaces.strhash.c dtls.c log.c cli.c graphite.c ice.c socket.c \
+		crypto.c rtp.c call_interfaces.strhash.c dtls.c log.c cli.c graphite.c ice.c \
 		media_socket.c homer.c recording.c statistics.c cdr.c ssrc.c iptables.c tcp_listener.c \
 		codec.c load.c dtmf.c
-LIBSRCS=	loglib.c auxlib.c rtplib.c str.c
+LIBSRCS=	loglib.c auxlib.c rtplib.c str.c socket.c streambuf.c ssllib.c
 ifeq ($(with_transcoding),yes)
 LIBSRCS+=	codeclib.c resample.c
 endif

daemon/aux.c

@@ -27,9 +27,6 @@ struct detach_thread {
 	const char	*scheduler;
 	int		priority;
 };
-struct thread_buf {
-	char buf[THREAD_BUF_SIZE];
-};
 struct scheduler {
 	const char *name;
 	int num;
@@ -42,10 +39,6 @@ static GList *threads_to_join;
 static GList *threads_running;
 static cond_t threads_cond = COND_STATIC_INIT;
 
-static struct thread_buf __thread t_bufs[NUM_THREAD_BUFS];
-static int __thread t_buf_idx;
-
-__thread struct timeval rtpe_now;
 volatile int rtpe_shutdown;
 
 #ifdef NEED_ATOMIC64_MUTEX
@@ -249,25 +242,6 @@ void thread_create_detach_prio(void (*f)(void *), void *d, const char *scheduler
 		abort();
 }
 
-unsigned int in6_addr_hash(const void *p) {
-	const struct in6_addr *a = p;
-	return a->s6_addr32[0] ^ a->s6_addr32[3];
-}
-
-int in6_addr_eq(const void *a, const void *b) {
-	const struct in6_addr *A = a, *B = b;
-	return !memcmp(A, B, sizeof(*A));
-}
-
-char *get_thread_buf(void) {
-	char *ret;
-	ret = t_bufs[t_buf_idx].buf;
-	t_buf_idx++;
-	if (t_buf_idx >= G_N_ELEMENTS(t_bufs))
-		t_buf_idx = 0;
-	return ret;
-}
-
 int g_tree_find_first_cmp(void *k, void *v, void *d) {
 	void **p = d;
 	GEqualFunc f = p[1];
@@ -285,14 +259,6 @@ int g_tree_find_all_cmp(void *k, void *v, void *d) {
 		g_queue_push_tail(q, v);
 	return FALSE;
 }
-unsigned int uint32_hash(const void *p) {
-	const u_int32_t *a = p;
-	return *a;
-}
-int uint32_eq(const void *a, const void *b) {
-	const u_int32_t *A = a, *B = b;
-	return (*A == *B) ? TRUE : FALSE;
-}
 
 void free_buf(char **p) {
 	if (*p)

daemon/call_interfaces.c

@@ -1004,8 +1004,10 @@ static const char *call_offer_answer_ng(bencode_item_t *input,
 	bencode_buffer_destroy_add(output->buffer, (free_func_t) sdp_chopper_destroy, chopper);
 
 	detect_setup_recording(call, &flags.record_call_str, &flags.metadata);
-	if (flags.record_call)
+	if (flags.record_call) {
+		call->recording_on = 1;
 		recording_start(call, NULL, &flags.metadata);
+	}
 
 	ret = monologue_offer_answer(monologue, &streams, &flags);
 	if (!ret) {
@@ -1438,6 +1440,7 @@ const char *call_start_recording_ng(bencode_item_t *input, bencode_item_t *outpu
 	if (!call)
 		return "Unknown call-id";
 
+	call->recording_on = 1;
 	recording_start(call, NULL, &metadata);
 
 	rwlock_unlock_w(&call->master_lock);
@@ -1456,6 +1459,7 @@ const char *call_stop_recording_ng(bencode_item_t *input, bencode_item_t *output
 	if (!call)
 		return "Unknown call-id";
 
+	call->recording_on = 0;
 	recording_stop(call);
 
 	rwlock_unlock_w(&call->master_lock);
@@ -1512,6 +1516,80 @@ found:
 	return NULL;
 }
 
+// XXX these are all identical - unify and use a flags int and/or callback
+const char *call_start_forwarding_ng(bencode_item_t *input, bencode_item_t *output) {
+	struct call *call;
+	struct call_monologue *monologue;
+	const char *errstr = NULL;
+	str metadata;
+
+	errstr = media_block_match(&call, &monologue, input);
+	if (errstr)
+		goto out;
+
+	bencode_dictionary_get_str(input, "metadata", &metadata);
+
+	if (monologue) {
+		ilog(LOG_INFO, "Start forwarding for single party (tag '" STR_FORMAT ")",
+				STR_FMT(&monologue->tag));
+		monologue->rec_forwarding = 1;
+	}
+	else {
+		ilog(LOG_INFO, "Start forwarding (entire call)");
+		call->rec_forwarding = 1;
+	}
+
+	recording_start(call, NULL, &metadata);
+	errstr = NULL;
+out:
+	if (call) {
+		rwlock_unlock_w(&call->master_lock);
+		obj_put(call);
+	}
+
+	return errstr;
+}
+
+const char *call_stop_forwarding_ng(bencode_item_t *input, bencode_item_t *output) {
+	struct call *call;
+	struct call_monologue *monologue;
+	const char *errstr = NULL;
+	struct sdp_ng_flags flags;
+
+	errstr = media_block_match(&call, &monologue, input);
+	if (errstr)
+		goto out;
+
+	call_ng_process_flags(&flags, input, OP_OTHER);
+
+	if (monologue) {
+		ilog(LOG_INFO, "Stop forwarding for single party (tag '" STR_FORMAT ")",
+				STR_FMT(&monologue->tag));
+		monologue->rec_forwarding = 0;
+	}
+	else {
+		ilog(LOG_INFO, "Stop forwarding (entire call)");
+		call->rec_forwarding = 0;
+		if (flags.all) {
+			for (GList *l = call->monologues.head; l; l = l->next) {
+				monologue = l->data;
+				monologue->rec_forwarding = 0;
+			}
+		}
+	}
+
+	recording_stop(call);
+
+	errstr = NULL;
+out:
+	if (call) {
+		rwlock_unlock_w(&call->master_lock);
+		obj_put(call);
+	}
+
+	return NULL;
+}
+
 const char *call_block_dtmf_ng(bencode_item_t *input, bencode_item_t *output) {
 	struct call *call;
 	struct call_monologue *monologue;

daemon/control_ng.c

@@ -219,6 +219,14 @@ static void control_ng_incoming(struct obj *obj, str *buf, const endpoint_t *sin
 			errstr = call_stop_recording_ng(dict, resp);
 			g_atomic_int_inc(&cur->stop_recording);
 			break;
+		case CSH_LOOKUP("start forwarding"):
+			errstr = call_start_forwarding_ng(dict, resp);
+			g_atomic_int_inc(&cur->start_forwarding);
+			break;
+		case CSH_LOOKUP("stop forwarding"):
+			errstr = call_stop_forwarding_ng(dict, resp);
+			g_atomic_int_inc(&cur->stop_forwarding);
+			break;
 		case CSH_LOOKUP("block DTMF"):
 			errstr = call_block_dtmf_ng(dict, resp);
 			g_atomic_int_inc(&cur->block_dtmf);

daemon/log.h

@@ -48,5 +48,12 @@ void dtmflog(GString *s);
 void log_format(enum log_format);
 void __ilog(int prio, const char *fmt, ...) __attribute__ ((format (printf, 2, 3)));
 
+// call debug
+#ifdef __DEBUG
+#define __C_DBG(x...) ilog(LOG_DEBUG, x)
+#else
+#define __C_DBG(x...) ((void)0)
+#endif
+
 
 #endif

daemon/main.c

@@ -12,7 +12,6 @@
 #include <errno.h>
 #include <stdlib.h>
 #include <time.h>
-#include <openssl/ssl.h>
 #include <ifaddrs.h>
 #include <net/if.h>
 
@@ -42,6 +41,7 @@
 #include "graphite.h"
 #include "codeclib.h"
 #include "load.h"
+#include "ssllib.h"
 
 
 
@@ -586,59 +586,15 @@ void fill_initial_rtpe_cfg(struct rtpengine_config* ini_rtpe_cfg) {
 
 }
 
-#if OPENSSL_VERSION_NUMBER < 0x10100000L
-static mutex_t *openssl_locks;
-
-static void cb_openssl_threadid(CRYPTO_THREADID *tid) {
-	pthread_t me;
-
-	me = pthread_self();
-
-	if (sizeof(me) == sizeof(void *))
-		CRYPTO_THREADID_set_pointer(tid, (void *) me);
-	else
-		CRYPTO_THREADID_set_numeric(tid, (unsigned long) me);
-}
-
-static void cb_openssl_lock(int mode, int type, const char *file, int line) {
-	if ((mode & CRYPTO_LOCK))
-		mutex_lock(&openssl_locks[type]);
-	else
-		mutex_unlock(&openssl_locks[type]);
-}
-
-static void make_OpenSSL_thread_safe(void) {
-	int i;
-
-	openssl_locks = malloc(sizeof(*openssl_locks) * CRYPTO_num_locks());
-	for (i = 0; i < CRYPTO_num_locks(); i++)
-		mutex_init(&openssl_locks[i]);
-
-	CRYPTO_THREADID_set_callback(cb_openssl_threadid);
-	CRYPTO_set_locking_callback(cb_openssl_lock);
-}
-#else
-static void make_OpenSSL_thread_safe(void) {
-	;
-}
-#endif
-
-
 static void early_init() {
 	socket_init(); // needed for socktype_udp
 }
 
 static void init_everything() {
-	struct timespec ts;
-
 	log_init("rtpengine");
 	log_format(rtpe_config.log_format);
 	recording_fs_init(rtpe_config.spooldir, rtpe_config.rec_method, rtpe_config.rec_format);
-	clock_gettime(CLOCK_REALTIME, &ts);
-	srandom(ts.tv_sec ^ ts.tv_nsec);
-	SSL_library_init();
-	SSL_load_error_strings();
-	make_OpenSSL_thread_safe();
+	rtpe_ssl_init();
 
 #if !GLIB_CHECK_VERSION(2,32,0)
 	g_thread_init(NULL);

daemon/poller.c

@@ -366,7 +366,8 @@ out:
 }
 
 
-void poller_blocked(struct poller *p, int fd) {
+void poller_blocked(struct poller *p, void *fdp) {
+	int fd = GPOINTER_TO_INT(fdp);
 	struct epoll_event e;
 
 	if (!p || fd < 0)
@@ -392,7 +393,8 @@ fail:
 	mutex_unlock(&p->lock);
 }
 
-void poller_error(struct poller *p, int fd) {
+void poller_error(struct poller *p, void *fdp) {
+	int fd = GPOINTER_TO_INT(fdp);
 	if (!p || fd < 0)
 		return;
 
@@ -412,7 +414,8 @@ fail:
 	mutex_unlock(&p->lock);
 }
 
-int poller_isblocked(struct poller *p, int fd) {
+int poller_isblocked(struct poller *p, void *fdp) {
+	int fd = GPOINTER_TO_INT(fdp);
 	int ret;
 
 	if (!p || fd < 0)

daemon/recording.c

@@ -35,6 +35,9 @@ struct pcap_format {
 static int check_main_spool_dir(const char *spoolpath);
 static char *recording_setup_file(struct recording *recording);
 static char *meta_setup_file(struct recording *recording);
+static int append_meta_chunk(struct recording *recording, const char *buf, unsigned int buflen,
+		const char *label_fmt, ...)
+	__attribute__((format(printf,4,5)));
 
 // pcap methods
 static int pcap_create_spool_dir(const char *dirpath);
@@ -58,6 +61,9 @@ static void kernel_info_proc(struct packet_stream *, struct rtpengine_target_inf
 
 static void pcap_eth_header(unsigned char *, struct packet_stream *);
 
+#define append_meta_chunk_str(r, str, f...) append_meta_chunk(r, (str)->s, (str)->len, f)
+#define append_meta_chunk_s(r, str, f...) append_meta_chunk(r, (str), strlen(str), f)
+#define append_meta_chunk_null(r,f...) append_meta_chunk(r, "", 0, f)
 
 
 static const struct recording_method methods[] = {
@@ -224,12 +230,26 @@ static void update_metadata(struct call *call, str *metadata) {
 	}
 }
 
+// lock must be held
+static void recording_update_flags(struct call *call) {
+	append_meta_chunk_null(call->recording, "RECORDING %u", call->recording_on ? 1 : 0);
+	append_meta_chunk_null(call->recording, "FORWARDING %u", call->rec_forwarding ? 1 : 0);
+	for (GList *l = call->streams.head; l; l = l->next) {
+		struct packet_stream *ps = l->data;
+		append_meta_chunk_null(call->recording, "STREAM %u FORWARDING %u",
+				ps->unique_id, ps->media->monologue->rec_forwarding ? 1 : 0);
+	}
+}
+
 // lock must be held
 void recording_start(struct call *call, const char *prefix, str *metadata) {
 	update_metadata(call, metadata);
 
-	if (call->recording) // already active
+	if (call->recording) {
+		// already active
+		recording_update_flags(call);
 		return;
+	}
 
 	if (!spooldir) {
 		ilog(LOG_ERR, "Call recording requested, but no spool directory configured");
@@ -267,11 +287,27 @@ void recording_start(struct call *call, const char *prefix, str *metadata) {
 		__unkernelize(ps);
 		ps->handler = NULL;
 	}
+
+	recording_update_flags(call);
 }
 void recording_stop(struct call *call) {
 	if (!call->recording)
 		return;
 
+	// check if all recording options are disabled
+	if (call->recording_on || call->rec_forwarding) {
+		recording_update_flags(call);
+		return;
+	}
+
+	for (GList *l = call->monologues.head; l; l = l->next) {
+		struct call_monologue *ml = l->data;
+		if (ml->rec_forwarding) {
+			recording_update_flags(call);
+			return;
+		}
+	}
+
 	ilog(LOG_NOTICE, "Turning off call recording.");
 	recording_finish(call);
 }
@@ -628,10 +664,6 @@ static int vappend_meta_chunk_iov(struct recording *recording, struct iovec *in_
 	return 0;
 }
 
-static int append_meta_chunk(struct recording *recording, const char *buf, unsigned int buflen,
-		const char *label_fmt, ...)
-	__attribute__((format(printf,4,5)));
-
 static int append_meta_chunk(struct recording *recording, const char *buf, unsigned int buflen,
 		const char *label_fmt, ...)
 {
@@ -646,8 +678,6 @@ static int append_meta_chunk(struct recording *recording, const char *buf, unsig
 
 	return ret;
 }
-#define append_meta_chunk_str(r, str, f...) append_meta_chunk(r, (str)->s, (str)->len, f)
-#define append_meta_chunk_s(r, str, f...) append_meta_chunk(r, (str), strlen(str), f)
 
 static void proc_init(struct call *call) {
 	struct recording *recording = call->recording;
@@ -694,8 +724,14 @@ static void finish_proc(struct call *call) {
 	struct recording *recording = call->recording;
 	if (!kernel.is_open)
 		return;
-	if (recording->u.proc.call_idx != UNINIT_IDX)
+	if (recording->u.proc.call_idx != UNINIT_IDX) {
 		kernel_del_call(recording->u.proc.call_idx);
+		recording->u.proc.call_idx = UNINIT_IDX;
+	}
+	for (GList *l = call->streams.head; l; l = l->next) {
+		struct packet_stream *ps = l->data;
+		ps->recording.u.proc.stream_idx = UNINIT_IDX;
+	}
 	unlink(recording->meta_filepath);
 }
 

include/aux.h

@@ -49,24 +49,8 @@ G_STATIC_ASSERT (sizeof *(atomic) == sizeof (gint));                     \
 
 
 
-/*** HELPER MACROS ***/
-
-#define ZERO(x)			memset(&(x), 0, sizeof(x))
-
-#define UINT64F			"%" G_GUINT64_FORMAT
-
-#define THREAD_BUF_SIZE		64
-#define NUM_THREAD_BUFS		8
-
-#define AUTO_CLEANUP(decl, func)		decl __attribute__ ((__cleanup__(func)))
-#define AUTO_CLEANUP_INIT(decl, func, val)	AUTO_CLEANUP(decl, func) = val
-#define AUTO_CLEANUP_NULL(decl, func)		AUTO_CLEANUP_INIT(decl, func, 0)
-#define AUTO_CLEANUP_BUF(var)			AUTO_CLEANUP_NULL(char *var, free_buf)
-
-
 /*** GLOBALS ***/
 
-extern __thread struct timeval rtpe_now;
 extern volatile int rtpe_shutdown;
 
 
@@ -79,11 +63,6 @@ typedef int (*parse_func)(char **, void **, void *);
 int pcre_multi_match(pcre *, pcre_extra *, const char *, unsigned int, parse_func, void *, GQueue *);
 INLINE void strmove(char **, char **);
 INLINE void strdupfree(char **, const char *);
-char *get_thread_buf(void);
-unsigned int in6_addr_hash(const void *p);
-int in6_addr_eq(const void *a, const void *b);
-unsigned int uint32_hash(const void *p);
-int uint32_eq(const void *a, const void *b);
 
 
 
@@ -274,135 +253,6 @@ void free_buf(char **);
 
 
 
-/*** MUTEX ABSTRACTION ***/
-
-typedef pthread_mutex_t mutex_t;
-typedef pthread_rwlock_t rwlock_t;
-typedef pthread_cond_t cond_t;
-
-#define mutex_init(m) __debug_mutex_init(m, __FILE__, __LINE__)
-#define mutex_destroy(m) __debug_mutex_destroy(m, __FILE__, __LINE__)
-#define mutex_lock(m) __debug_mutex_lock(m, __FILE__, __LINE__)
-#define mutex_trylock(m) __debug_mutex_trylock(m, __FILE__, __LINE__)
-#define mutex_unlock(m) __debug_mutex_unlock(m, __FILE__, __LINE__)
-#define MUTEX_STATIC_INIT PTHREAD_MUTEX_INITIALIZER
-
-#define rwlock_init(l) __debug_rwlock_init(l, __FILE__, __LINE__)
-#define rwlock_destroy(l) __debug_rwlock_destroy(l, __FILE__, __LINE__)
-#define rwlock_lock_r(l) __debug_rwlock_lock_r(l, __FILE__, __LINE__)
-#define rwlock_unlock_r(l) __debug_rwlock_unlock_r(l, __FILE__, __LINE__)
-#define rwlock_lock_w(l) __debug_rwlock_lock_w(l, __FILE__, __LINE__)
-#define rwlock_unlock_w(l) __debug_rwlock_unlock_w(l, __FILE__, __LINE__)
-
-#define cond_init(c) __debug_cond_init(c, __FILE__, __LINE__)
-#define cond_wait(c,m) __debug_cond_wait(c,m, __FILE__, __LINE__)
-#define cond_timedwait(c,m,t) __debug_cond_timedwait(c,m,t, __FILE__, __LINE__)
-#define cond_signal(c) __debug_cond_signal(c, __FILE__, __LINE__)
-#define cond_broadcast(c) __debug_cond_broadcast(c, __FILE__, __LINE__)
-#define COND_STATIC_INIT PTHREAD_COND_INITIALIZER
-
-INLINE int __cond_timedwait_tv(cond_t *c, mutex_t *m, const struct timeval *tv) {
-	struct timespec ts;
-	ts.tv_sec = tv->tv_sec;
-	ts.tv_nsec = tv->tv_usec * 1000;
-	return pthread_cond_timedwait(c, m, &ts);
-}
-
-#ifndef __THREAD_DEBUG
-
-#define __debug_mutex_init(m, F, L) pthread_mutex_init(m, NULL)
-#define __debug_mutex_destroy(m, F, L) pthread_mutex_destroy(m)
-#define __debug_mutex_lock(m, F, L) pthread_mutex_lock(m)
-#define __debug_mutex_trylock(m, F, L) pthread_mutex_trylock(m)
-#define __debug_mutex_unlock(m, F, L) pthread_mutex_unlock(m)
-
-#define __debug_rwlock_init(l, F, L) pthread_rwlock_init(l, NULL)
-#define __debug_rwlock_destroy(l, F, L) pthread_rwlock_destroy(l)
-#define __debug_rwlock_lock_r(l, F, L) pthread_rwlock_rdlock(l)
-#define __debug_rwlock_unlock_r(l, F, L) pthread_rwlock_unlock(l)
-#define __debug_rwlock_lock_w(l, F, L) pthread_rwlock_wrlock(l)
-#define __debug_rwlock_unlock_w(l, F, L) pthread_rwlock_unlock(l)
-
-#define __debug_cond_init(c, F, L) pthread_cond_init(c, NULL)
-#define __debug_cond_wait(c, m, F, L) pthread_cond_wait(c,m)
-#define __debug_cond_timedwait(c, m, t, F, L) __cond_timedwait_tv(c,m,t)
-#define __debug_cond_signal(c, F, L) pthread_cond_signal(c)
-#define __debug_cond_broadcast(c, F, L) pthread_cond_broadcast(c)
-
-#else
-
-
-#include "log.h"
-
-
-
-INLINE int __debug_mutex_init(mutex_t *m, const char *file, unsigned int line) {
-	write_log(LOG_DEBUG, "mutex_init(%p) at %s:%u", m, file, line);
-	return pthread_mutex_init(m, NULL);
-}
-INLINE int __debug_mutex_destroy(mutex_t *m, const char *file, unsigned int line) {
-	write_log(LOG_DEBUG, "mutex_destroy(%p) at %s:%u", m, file, line);
-	return pthread_mutex_destroy(m);
-}
-INLINE int __debug_mutex_lock(mutex_t *m, const char *file, unsigned int line) {
-	int ret;
-	write_log(LOG_DEBUG, "mutex_lock(%p) at %s:%u ...", m, file, line);
-	ret = pthread_mutex_lock(m);
-	write_log(LOG_DEBUG, "mutex_lock(%p) at %s:%u returning %i", m, file, line, ret);
-	return ret;
-}
-INLINE int __debug_mutex_trylock(mutex_t *m, const char *file, unsigned int line) {
-	int ret;
-	write_log(LOG_DEBUG, "mutex_trylock(%p) at %s:%u ...", m, file, line);
-	ret = pthread_mutex_trylock(m);
-	write_log(LOG_DEBUG, "mutex_trylock(%p) at %s:%u returning %i", m, file, line, ret);
-	return ret;
-}
-INLINE int __debug_mutex_unlock(mutex_t *m, const char *file, unsigned int line) {
-	write_log(LOG_DEBUG, "mutex_unlock(%p) at %s:%u", m, file, line);
-	return pthread_mutex_unlock(m);
-}
-
-INLINE int __debug_rwlock_init(rwlock_t *m, const char *file, unsigned int line) {
-	write_log(LOG_DEBUG, "rwlock_init(%p) at %s:%u", m, file, line);
-	return pthread_rwlock_init(m, NULL);
-}
-INLINE int __debug_rwlock_destroy(rwlock_t *m, const char *file, unsigned int line) {
-	write_log(LOG_DEBUG, "rwlock_destroy(%p) at %s:%u", m, file, line);
-	return pthread_rwlock_destroy(m);
-}
-INLINE int __debug_rwlock_lock_r(rwlock_t *m, const char *file, unsigned int line) {
-	int ret;
-	write_log(LOG_DEBUG, "rwlock_lock_r(%p) at %s:%u ...", m, file, line);
-	ret = pthread_rwlock_rdlock(m);
-	write_log(LOG_DEBUG, "rwlock_lock_r(%p) at %s:%u returning %i", m, file, line, ret);
-	return ret;
-}
-INLINE int __debug_rwlock_lock_w(rwlock_t *m, const char *file, unsigned int line) {
-	int ret;
-	write_log(LOG_DEBUG, "rwlock_lock_w(%p) at %s:%u ...", m, file, line);
-	ret = pthread_rwlock_wrlock(m);
-	write_log(LOG_DEBUG, "rwlock_lock_w(%p) at %s:%u returning %i", m, file, line, ret);
-	return ret;
-}
-INLINE int __debug_rwlock_unlock_r(rwlock_t *m, const char *file, unsigned int line) {
-	write_log(LOG_DEBUG, "rwlock_unlock_r(%p) at %s:%u", m, file, line);
-	return pthread_rwlock_unlock(m);
-}
-INLINE int __debug_rwlock_unlock_w(rwlock_t *m, const char *file, unsigned int line) {
-	write_log(LOG_DEBUG, "rwlock_unlock_w(%p) at %s:%u", m, file, line);
-	return pthread_rwlock_unlock(m);
-}
-
-#define __debug_cond_init(c, F, L) pthread_cond_init(c, NULL)
-#define __debug_cond_wait(c, m, F, L) pthread_cond_wait(c,m)
-#define __debug_cond_timedwait(c, m, t, F, L) __cond_timedwait_tv(c,m,t)
-#define __debug_cond_signal(c, F, L) pthread_cond_signal(c)
-#define __debug_cond_broadcast(c, F, L) pthread_cond_broadcast(c)
-
-#endif
-
-
 
 
 /*** THREAD HELPERS ***/

include/call.h

@@ -90,12 +90,6 @@ enum call_type {
 #define RTP_LOOP_MAX_COUNT	30 /* number of consecutively detected dupes to trigger protection */
 #endif
 
-#ifdef __DEBUG
-#define __C_DBG(x...) ilog(LOG_DEBUG, x)
-#else
-#define __C_DBG(x...) ((void)0)
-#endif
-
 #define IS_FOREIGN_CALL(c) (c->foreign_call)
 #define IS_OWN_CALL(c) !IS_FOREIGN_CALL(c)
 
@@ -370,6 +364,7 @@ struct call_monologue {
 
 	int			block_dtmf:1;
 	int			block_media:1;
+	int			rec_forwarding:1;
 };
 
 struct call {
@@ -408,6 +403,8 @@ struct call {
 
 	int			block_dtmf:1;
 	int			block_media:1;
+	int			recording_on:1;
+	int			rec_forwarding:1;
 };
 
 

include/call_interfaces.h

@@ -102,6 +102,8 @@ const char *call_query_ng(bencode_item_t *, bencode_item_t *);
 const char *call_list_ng(bencode_item_t *, bencode_item_t *);
 const char *call_start_recording_ng(bencode_item_t *, bencode_item_t *);
 const char *call_stop_recording_ng(bencode_item_t *, bencode_item_t *);
+const char *call_start_forwarding_ng(bencode_item_t *, bencode_item_t *);
+const char *call_stop_forwarding_ng(bencode_item_t *, bencode_item_t *);
 const char *call_block_dtmf_ng(bencode_item_t *, bencode_item_t *);
 const char *call_unblock_dtmf_ng(bencode_item_t *, bencode_item_t *);
 const char *call_block_media_ng(bencode_item_t *, bencode_item_t *);

include/control_ng.h

@@ -19,6 +19,8 @@ struct control_ng_stats {
 	int list;
 	int start_recording;
 	int stop_recording;
+	int start_forwarding;
+	int stop_forwarding;
 	int block_dtmf;
 	int unblock_dtmf;
 	int block_media;

include/poller.h

@@ -34,9 +34,10 @@ struct poller *poller_new(void);
 int poller_add_item(struct poller *, struct poller_item *);
 int poller_update_item(struct poller *, struct poller_item *);
 int poller_del_item(struct poller *, int);
-void poller_blocked(struct poller *, int);
-int poller_isblocked(struct poller *, int);
-void poller_error(struct poller *, int);
+
+void poller_blocked(struct poller *, void *);
+int poller_isblocked(struct poller *, void *);
+void poller_error(struct poller *, void *);
 
 int poller_poll(struct poller *, int);
 void poller_timer_loop(void *);

include/tcp_listener.h

@@ -3,6 +3,7 @@
 
 #include "socket.h"
 #include "obj.h"
+#include "aux.h"
 
 
 struct poller;

lib/auxlib.c

@@ -10,12 +10,21 @@
 #ifdef HAVE_LIBSYSTEMD
 #include <systemd/sd-daemon.h>
 #endif
+#include <sys/socket.h>
+#include <netinet/in.h>
 #include "log.h"
 #include "loglib.h"
 
+struct thread_buf {
+	char buf[THREAD_BUF_SIZE];
+};
 
 static int version;
 struct rtpengine_common_config *rtpe_common_config_ptr;
+__thread struct timeval rtpe_now;
+
+static struct thread_buf __thread t_bufs[NUM_THREAD_BUFS];
+static int __thread t_buf_idx;
 
 
 void daemonize(void) {
@@ -200,3 +209,31 @@ out:
 err:
 	die("Bad command line: %s", er->message);
 }
+
+char *get_thread_buf(void) {
+	char *ret;
+	ret = t_bufs[t_buf_idx].buf;
+	t_buf_idx++;
+	if (t_buf_idx >= G_N_ELEMENTS(t_bufs))
+		t_buf_idx = 0;
+	return ret;
+}
+
+unsigned int in6_addr_hash(const void *p) {
+	const struct in6_addr *a = p;
+	return a->s6_addr32[0] ^ a->s6_addr32[3];
+}
+
+int in6_addr_eq(const void *a, const void *b) {
+	const struct in6_addr *A = a, *B = b;
+	return !memcmp(A, B, sizeof(*A));
+}
+
+unsigned int uint32_hash(const void *p) {
+	const u_int32_t *a = p;
+	return *a;
+}
+int uint32_eq(const void *a, const void *b) {
+	const u_int32_t *A = a, *B = b;
+	return (*A == *B) ? TRUE : FALSE;
+}

lib/auxlib.h

@@ -6,6 +6,11 @@
 #include "compat.h"
 #include <openssl/rand.h>
 
+
+#define THREAD_BUF_SIZE		64
+#define NUM_THREAD_BUFS		8
+
+
 struct rtpengine_common_config {
 	char *config_file;
 	char *config_section;
@@ -18,6 +23,18 @@ struct rtpengine_common_config {
 
 extern struct rtpengine_common_config *rtpe_common_config_ptr;
 
+
+
+/*** GLOBALS ***/
+
+extern __thread struct timeval rtpe_now;
+extern volatile int rtpe_shutdown;
+
+
+
+
+/*** PROTOTYPES ***/
+
 void daemonize(void);
 void wpidfile(void);
 void service_notify(const char *message);
@@ -25,9 +42,163 @@ void config_load(int *argc, char ***argv, GOptionEntry *entries, const char *des
 		char *default_config, char *default_section,
 		struct rtpengine_common_config *);
 
+char *get_thread_buf(void);
+
+unsigned int in6_addr_hash(const void *p);
+int in6_addr_eq(const void *a, const void *b);
+unsigned int uint32_hash(const void *p);
+int uint32_eq(const void *a, const void *b);
+
+
+/*** HELPER MACROS ***/
+
+#define ZERO(x)			memset(&(x), 0, sizeof(x))
+
+#define UINT64F			"%" G_GUINT64_FORMAT
+
+#define AUTO_CLEANUP(decl, func)		decl __attribute__ ((__cleanup__(func)))
+#define AUTO_CLEANUP_INIT(decl, func, val)	AUTO_CLEANUP(decl, func) = val
+#define AUTO_CLEANUP_NULL(decl, func)		AUTO_CLEANUP_INIT(decl, func, 0)
+#define AUTO_CLEANUP_BUF(var)			AUTO_CLEANUP_NULL(char *var, free_buf)
+
+
+/*** STRING HELPERS ***/
+
 INLINE void random_string(unsigned char *buf, int len) {
 	int ret = RAND_bytes(buf, len);
 	assert(ret == 1);
 }
 
+
+/*** MUTEX ABSTRACTION ***/
+
+typedef pthread_mutex_t mutex_t;
+typedef pthread_rwlock_t rwlock_t;
+typedef pthread_cond_t cond_t;
+
+#define mutex_init(m) __debug_mutex_init(m, __FILE__, __LINE__)
+#define mutex_destroy(m) __debug_mutex_destroy(m, __FILE__, __LINE__)
+#define mutex_lock(m) __debug_mutex_lock(m, __FILE__, __LINE__)
+#define mutex_trylock(m) __debug_mutex_trylock(m, __FILE__, __LINE__)
+#define mutex_unlock(m) __debug_mutex_unlock(m, __FILE__, __LINE__)
+#define MUTEX_STATIC_INIT PTHREAD_MUTEX_INITIALIZER
+
+#define rwlock_init(l) __debug_rwlock_init(l, __FILE__, __LINE__)
+#define rwlock_destroy(l) __debug_rwlock_destroy(l, __FILE__, __LINE__)
+#define rwlock_lock_r(l) __debug_rwlock_lock_r(l, __FILE__, __LINE__)
+#define rwlock_unlock_r(l) __debug_rwlock_unlock_r(l, __FILE__, __LINE__)
+#define rwlock_lock_w(l) __debug_rwlock_lock_w(l, __FILE__, __LINE__)
+#define rwlock_unlock_w(l) __debug_rwlock_unlock_w(l, __FILE__, __LINE__)
+
+#define cond_init(c) __debug_cond_init(c, __FILE__, __LINE__)
+#define cond_wait(c,m) __debug_cond_wait(c,m, __FILE__, __LINE__)
+#define cond_timedwait(c,m,t) __debug_cond_timedwait(c,m,t, __FILE__, __LINE__)
+#define cond_signal(c) __debug_cond_signal(c, __FILE__, __LINE__)
+#define cond_broadcast(c) __debug_cond_broadcast(c, __FILE__, __LINE__)
+#define COND_STATIC_INIT PTHREAD_COND_INITIALIZER
+
+INLINE int __cond_timedwait_tv(cond_t *c, mutex_t *m, const struct timeval *tv) {
+	struct timespec ts;
+	ts.tv_sec = tv->tv_sec;
+	ts.tv_nsec = tv->tv_usec * 1000;
+	return pthread_cond_timedwait(c, m, &ts);
+}
+
+#ifndef __THREAD_DEBUG
+
+#define __debug_mutex_init(m, F, L) pthread_mutex_init(m, NULL)
+#define __debug_mutex_destroy(m, F, L) pthread_mutex_destroy(m)
+#define __debug_mutex_lock(m, F, L) pthread_mutex_lock(m)
+#define __debug_mutex_trylock(m, F, L) pthread_mutex_trylock(m)
+#define __debug_mutex_unlock(m, F, L) pthread_mutex_unlock(m)
+
+#define __debug_rwlock_init(l, F, L) pthread_rwlock_init(l, NULL)
+#define __debug_rwlock_destroy(l, F, L) pthread_rwlock_destroy(l)
+#define __debug_rwlock_lock_r(l, F, L) pthread_rwlock_rdlock(l)
+#define __debug_rwlock_unlock_r(l, F, L) pthread_rwlock_unlock(l)
+#define __debug_rwlock_lock_w(l, F, L) pthread_rwlock_wrlock(l)
+#define __debug_rwlock_unlock_w(l, F, L) pthread_rwlock_unlock(l)
+
+#define __debug_cond_init(c, F, L) pthread_cond_init(c, NULL)
+#define __debug_cond_wait(c, m, F, L) pthread_cond_wait(c,m)
+#define __debug_cond_timedwait(c, m, t, F, L) __cond_timedwait_tv(c,m,t)
+#define __debug_cond_signal(c, F, L) pthread_cond_signal(c)
+#define __debug_cond_broadcast(c, F, L) pthread_cond_broadcast(c)
+
+#else
+
+
+#include "log.h"
+
+
+
+INLINE int __debug_mutex_init(mutex_t *m, const char *file, unsigned int line) {
+	write_log(LOG_DEBUG, "mutex_init(%p) at %s:%u", m, file, line);
+	return pthread_mutex_init(m, NULL);
+}
+INLINE int __debug_mutex_destroy(mutex_t *m, const char *file, unsigned int line) {
+	write_log(LOG_DEBUG, "mutex_destroy(%p) at %s:%u", m, file, line);
+	return pthread_mutex_destroy(m);
+}
+INLINE int __debug_mutex_lock(mutex_t *m, const char *file, unsigned int line) {
+	int ret;
+	write_log(LOG_DEBUG, "mutex_lock(%p) at %s:%u ...", m, file, line);
+	ret = pthread_mutex_lock(m);
+	write_log(LOG_DEBUG, "mutex_lock(%p) at %s:%u returning %i", m, file, line, ret);
+	return ret;
+}
+INLINE int __debug_mutex_trylock(mutex_t *m, const char *file, unsigned int line) {
+	int ret;
+	write_log(LOG_DEBUG, "mutex_trylock(%p) at %s:%u ...", m, file, line);
+	ret = pthread_mutex_trylock(m);
+	write_log(LOG_DEBUG, "mutex_trylock(%p) at %s:%u returning %i", m, file, line, ret);
+	return ret;
+}
+INLINE int __debug_mutex_unlock(mutex_t *m, const char *file, unsigned int line) {
+	write_log(LOG_DEBUG, "mutex_unlock(%p) at %s:%u", m, file, line);
+	return pthread_mutex_unlock(m);
+}
+
+INLINE int __debug_rwlock_init(rwlock_t *m, const char *file, unsigned int line) {
+	write_log(LOG_DEBUG, "rwlock_init(%p) at %s:%u", m, file, line);
+	return pthread_rwlock_init(m, NULL);
+}
+INLINE int __debug_rwlock_destroy(rwlock_t *m, const char *file, unsigned int line) {
+	write_log(LOG_DEBUG, "rwlock_destroy(%p) at %s:%u", m, file, line);
+	return pthread_rwlock_destroy(m);
+}
+INLINE int __debug_rwlock_lock_r(rwlock_t *m, const char *file, unsigned int line) {
+	int ret;
+	write_log(LOG_DEBUG, "rwlock_lock_r(%p) at %s:%u ...", m, file, line);
+	ret = pthread_rwlock_rdlock(m);
+	write_log(LOG_DEBUG, "rwlock_lock_r(%p) at %s:%u returning %i", m, file, line, ret);
+	return ret;
+}
+INLINE int __debug_rwlock_lock_w(rwlock_t *m, const char *file, unsigned int line) {
+	int ret;
+	write_log(LOG_DEBUG, "rwlock_lock_w(%p) at %s:%u ...", m, file, line);
+	ret = pthread_rwlock_wrlock(m);
+	write_log(LOG_DEBUG, "rwlock_lock_w(%p) at %s:%u returning %i", m, file, line, ret);
+	return ret;
+}
+INLINE int __debug_rwlock_unlock_r(rwlock_t *m, const char *file, unsigned int line) {
+	write_log(LOG_DEBUG, "rwlock_unlock_r(%p) at %s:%u", m, file, line);
+	return pthread_rwlock_unlock(m);
+}
+INLINE int __debug_rwlock_unlock_w(rwlock_t *m, const char *file, unsigned int line) {
+	write_log(LOG_DEBUG, "rwlock_unlock_w(%p) at %s:%u", m, file, line);
+	return pthread_rwlock_unlock(m);
+}
+
+#define __debug_cond_init(c, F, L) pthread_cond_init(c, NULL)
+#define __debug_cond_wait(c, m, F, L) pthread_cond_wait(c,m)
+#define __debug_cond_timedwait(c, m, t, F, L) __cond_timedwait_tv(c,m,t)
+#define __debug_cond_signal(c, F, L) pthread_cond_signal(c)
+#define __debug_cond_broadcast(c, F, L) pthread_cond_broadcast(c)
+
+#endif
+
+
+
+
 #endif

lib/codeclib.c

@@ -456,7 +456,6 @@ decoder_t *decoder_new_fmtp(const codec_def_t *def, int clockrate, int channels,
 
 	ret->pts = (uint64_t) -1LL;
 	ret->rtp_ts = (unsigned long) -1L;
-	ret->mixer_idx = (unsigned int) -1;
 
 	return ret;
 
@@ -488,7 +487,6 @@ void decoder_close(decoder_t *dec) {
 		dec->def->codec_type->decoder_close(dec);
 
 	resample_shutdown(&dec->resampler);
-	resample_shutdown(&dec->mix_resampler);
 	g_slice_free1(sizeof(*dec), dec);
 }
 

lib/codeclib.h

@@ -131,8 +131,7 @@ struct decoder_s {
 	format_t in_format,
 		 out_format;
 
-	resample_t resampler,
-		   mix_resampler; // XXX move this out of here - specific to recording-daemon
+	resample_t resampler;
 
 	union {
 		struct {
@@ -146,8 +145,6 @@ struct decoder_s {
 
 	unsigned long rtp_ts;
 	uint64_t pts;
-
-	unsigned int mixer_idx;
 };
 
 struct encoder_s {

lib/ssllib.c

@@ -0,0 +1,51 @@
+#include "ssllib.h"
+#include <openssl/ssl.h>
+#include <time.h>
+
+
+#if OPENSSL_VERSION_NUMBER < 0x10100000L
+static mutex_t *openssl_locks;
+
+static void cb_openssl_threadid(CRYPTO_THREADID *tid) {
+	pthread_t me;
+
+	me = pthread_self();
+
+	if (sizeof(me) == sizeof(void *))
+		CRYPTO_THREADID_set_pointer(tid, (void *) me);
+	else
+		CRYPTO_THREADID_set_numeric(tid, (unsigned long) me);
+}
+
+static void cb_openssl_lock(int mode, int type, const char *file, int line) {
+	if ((mode & CRYPTO_LOCK))
+		mutex_lock(&openssl_locks[type]);
+	else
+		mutex_unlock(&openssl_locks[type]);
+}
+
+static void make_OpenSSL_thread_safe(void) {
+	int i;
+
+	openssl_locks = malloc(sizeof(*openssl_locks) * CRYPTO_num_locks());
+	for (i = 0; i < CRYPTO_num_locks(); i++)
+		mutex_init(&openssl_locks[i]);
+
+	CRYPTO_THREADID_set_callback(cb_openssl_threadid);
+	CRYPTO_set_locking_callback(cb_openssl_lock);
+}
+#else
+static void make_OpenSSL_thread_safe(void) {
+	;
+}
+#endif
+
+
+void rtpe_ssl_init(void) {
+	struct timespec ts;
+	clock_gettime(CLOCK_REALTIME, &ts);
+	srandom(ts.tv_sec ^ ts.tv_nsec);
+	SSL_library_init();
+	SSL_load_error_strings();
+	make_OpenSSL_thread_safe();
+}

lib/ssllib.h

@@ -0,0 +1,8 @@
+#ifndef __SSLLIB_H__
+#define __SSLLIB_H__
+
+
+void rtpe_ssl_init(void);
+
+
+#endif

recording-daemon/.gitignore

@@ -11,3 +11,5 @@ codeclib.c
 resample.c
 str.c
 fix_frame_channel_layout.h
+socket.c
+streambuf.c

recording-daemon/Makefile

@@ -1,6 +1,6 @@
 TARGET=		rtpengine-recording
 
-CFLAGS=		-g -Wall -pthread -I. -I../lib/
+CFLAGS=		-g -Wall -pthread -I. -I../lib/ -I../kernel-module/
 CFLAGS+=	-std=c99
 CFLAGS+=	-D_GNU_SOURCE -D_POSIX_SOURCE -D_POSIX_C_SOURCE
 CFLAGS+=	$(shell pkg-config --cflags glib-2.0)
@@ -25,8 +25,8 @@ LDLIBS+=	$(shell mysql_config --libs)
 LDLIBS+=	$(shell pkg-config --libs openssl)
 
 SRCS=		epoll.c garbage.c inotify.c main.c metafile.c stream.c recaux.c packet.c \
-		decoder.c output.c mix.c db.c log.c forward.c tag.c
-LIBSRCS=	loglib.c auxlib.c rtplib.c codeclib.c resample.c str.c
+		decoder.c output.c mix.c db.c log.c forward.c tag.c poller.c
+LIBSRCS=	loglib.c auxlib.c rtplib.c codeclib.c resample.c str.c socket.c streambuf.c ssllib.c
 OBJS=		$(SRCS:.c=.o) $(LIBSRCS:.c=.o)
 
 include ../lib/common.Makefile

recording-daemon/db.c

@@ -214,7 +214,7 @@ static void db_do_call_id(metafile_t *mf) {
 	execute_wrap(&stm_insert_call, b, &mf->db_id);
 }
 static void db_do_call_metadata(metafile_t *mf) {
-	if (!mf->metadata)
+	if (!mf->metadata_db)
 		return;
 	if (mf->db_id == 0)
 		return;
@@ -224,7 +224,7 @@ static void db_do_call_metadata(metafile_t *mf) {
 
 	// XXX offload this parsing to proxy module -> bencode list/dictionary
 	str all_meta;
-	str_init(&all_meta, mf->metadata);
+	str_init(&all_meta, mf->metadata_db);
 	while (all_meta.len > 1) {
 		str token;
 		if (str_token_sep(&token, &all_meta, '|'))
@@ -242,7 +242,7 @@ static void db_do_call_metadata(metafile_t *mf) {
 		execute_wrap(&stm_insert_metadata, b, NULL);
 	}
 
-	mf->metadata = NULL;
+	mf->metadata_db = NULL;
 }
 
 void db_do_call(metafile_t *mf) {

recording-daemon/decoder.c

@@ -15,13 +15,16 @@
 #include "mix.h"
 #include "resample.h"
 #include "codeclib.h"
+#include "streambuf.h"
+#include "main.h"
+#include "packet.h"
 
 
 int resample_audio;
 
 
 
-decoder_t *decoder_new(const char *payload_str, output_t *outp) {
+decode_t *decoder_new(const char *payload_str, output_t *outp) {
 	str name;
 	char *slash = strchr(payload_str, '/');
 	if (!slash) {
@@ -69,13 +72,22 @@ decoder_t *decoder_new(const char *payload_str, output_t *outp) {
 		outp->encoder->requested_format.format = out_format.format;
 	}
 
-	return decoder_new_fmt(def, clockrate, channels, &out_format);
+	decoder_t *dec = decoder_new_fmt(def, clockrate, channels, &out_format);
+	if (!dec)
+		return NULL;
+	decode_t *deco = g_slice_alloc0(sizeof(decode_t));
+	deco->dec = dec;
+	deco->mixer_idx = (unsigned int) -1;
+	return deco;
 }
 
 
-static int decoder_got_frame(decoder_t *dec, AVFrame *frame, void *op, void *mp) {
-	metafile_t *metafile = mp;
-	output_t *output = op;
+static int decoder_got_frame(decoder_t *dec, AVFrame *frame, void *sp, void *dp) {
+	ssrc_t *ssrc = sp;
+	metafile_t *metafile = ssrc->metafile;
+	output_t *output = ssrc->output;
+	stream_t *stream = ssrc->stream;
+	decode_t *deco = dp;
 
 	dbg("got frame pts %llu samples %u contents %02x%02x%02x%02x...", (unsigned long long) frame->pts, frame->nb_samples,
 			(unsigned int) frame->extended_data[0][0],
@@ -83,34 +95,66 @@ static int decoder_got_frame(decoder_t *dec, AVFrame *frame, void *op, void *mp)
 			(unsigned int) frame->extended_data[0][2],
 			(unsigned int) frame->extended_data[0][3]);
 
+	if (!metafile->recording_on)
+		goto no_recording;
+
 	// handle mix output
 	pthread_mutex_lock(&metafile->mix_lock);
 	if (metafile->mix_out) {
-		if (G_UNLIKELY(dec->mixer_idx == (unsigned int) -1))
-			dec->mixer_idx = mix_get_index(metafile->mix);
+		dbg("adding packet from stream #%lu to mix output", stream->id);
+		if (G_UNLIKELY(deco->mixer_idx == (unsigned int) -1))
+			deco->mixer_idx = mix_get_index(metafile->mix);
 		format_t actual_format;
 		if (output_config(metafile->mix_out, &dec->out_format, &actual_format))
 			goto no_mix_out;
 		mix_config(metafile->mix, &actual_format);
 		// XXX might be a second resampling to same format
-		AVFrame *dec_frame = resample_frame(&dec->mix_resampler, frame, &actual_format);
+		AVFrame *dec_frame = resample_frame(&deco->mix_resampler, frame, &actual_format);
 		if (!dec_frame) {
 			pthread_mutex_unlock(&metafile->mix_lock);
 			goto err;
 		}
-		if (mix_add(metafile->mix, dec_frame, dec->mixer_idx, metafile->mix_out))
+		if (mix_add(metafile->mix, dec_frame, deco->mixer_idx, metafile->mix_out))
 			ilog(LOG_ERR, "Failed to add decoded packet to mixed output");
 	}
 no_mix_out:
 	pthread_mutex_unlock(&metafile->mix_lock);
 
 	if (output) {
+		dbg("SSRC %lx of stream #%lu has single output", ssrc->ssrc, stream->id);
 		if (output_config(output, &dec->out_format, NULL))
 			goto err;
 		if (output_add(output, frame))
 			ilog(LOG_ERR, "Failed to add decoded packet to individual output");
 	}
 
+no_recording:
+	if (ssrc->tls_fwd_stream) {
+		// XXX might be a second resampling to same format
+		dbg("SSRC %lx of stream #%lu has TLS forwarding stream", ssrc->ssrc, stream->id);
+		AVFrame *dec_frame = resample_frame(&ssrc->tls_fwd_resampler, frame, &ssrc->tls_fwd_format);
+
+		ssrc_tls_state(ssrc);
+
+		if (!ssrc->sent_intro) {
+			if (metafile->metadata) {
+				dbg("Writing metadata header to TLS");
+				streambuf_write(ssrc->tls_fwd_stream, metafile->metadata, strlen(metafile->metadata) + 1);
+			}
+			else {
+				ilog(LOG_WARN, "No metadata present for forwarding connection");
+				streambuf_write(ssrc->tls_fwd_stream, "\0", 1);
+			}
+			ssrc->sent_intro = 1;
+		}
+
+		dbg("Writing %u bytes PCM to TLS", dec_frame->linesize[0]);
+		streambuf_write(ssrc->tls_fwd_stream, (char *) dec_frame->extended_data[0],
+				dec_frame->linesize[0]);
+		av_frame_free(&dec_frame);
+
+	}
+
 	av_frame_free(&frame);
 	return 0;
 
@@ -120,6 +164,14 @@ err:
 }
 
 
-int decoder_input(decoder_t *dec, const str *data, unsigned long ts, output_t *output, metafile_t *metafile) {
-	return decoder_input_data(dec, data, ts, decoder_got_frame, output, metafile);
+int decoder_input(decode_t *deco, const str *data, unsigned long ts, ssrc_t *ssrc) {
+	return decoder_input_data(deco->dec, data, ts, decoder_got_frame, ssrc, deco);
+}
+
+void decoder_free(decode_t *deco) {
+	if (!deco)
+		return;
+	decoder_close(deco->dec);
+	resample_shutdown(&deco->mix_resampler);
+	g_slice_free1(sizeof(*deco), deco);
 }

recording-daemon/decoder.h

@@ -8,8 +8,9 @@
 extern int resample_audio;
 
 
-decoder_t *decoder_new(const char *payload_str, output_t *);
-int decoder_input(decoder_t *, const str *, unsigned long ts, output_t *, metafile_t *);
+decode_t *decoder_new(const char *payload_str, output_t *);
+int decoder_input(decode_t *, const str *, unsigned long ts, ssrc_t *);
+void decoder_free(decode_t *);
 
 
 #endif

recording-daemon/log.h

@@ -9,6 +9,7 @@
 #include <stdlib.h>
 
 #define dbg(fmt, ...) ilog(LOG_DEBUG, "[%s:%i] " fmt, __FILE__, __LINE__, ##__VA_ARGS__)
+#define __C_DBG(x...) ilog(LOG_DEBUG, x)
 
 void __ilog(int prio, const char *fmt, ...) __attribute__ ((format (printf, 2, 3)));
 

recording-daemon/main.c

@@ -24,6 +24,8 @@
 #include "output.h"
 #include "forward.h"
 #include "codeclib.h"
+#include "socket.h"
+#include "ssllib.h"
 
 
 
@@ -36,12 +38,16 @@ static const char *output_format = "wav";
 int output_mixed;
 int output_single;
 int output_enabled = 1;
+int decoding_enabled;
 const char *c_mysql_host,
       *c_mysql_user,
       *c_mysql_pass,
       *c_mysql_db;
 int c_mysql_port;
 const char *forward_to = NULL;
+static const char *tls_send_to = NULL;
+endpoint_t tls_send_to_ep;
+int tls_resample = 8000;
 
 static GQueue threads = G_QUEUE_INIT; // only accessed from main thread
 
@@ -65,8 +71,11 @@ static void signals(void) {
 
 static void setup(void) {
 	log_init("rtpengine-recording");
-	if (output_enabled) {
+	rtpe_ssl_init();
+	socket_init();
+	if (decoding_enabled)
 		codeclib_init(0);
+	if (output_enabled) {
 		output_init(output_format);
 		if (!g_file_test(output_dir, G_FILE_TEST_IS_DIR)) {
 			ilog(LOG_INFO, "Creating output dir '%s'", output_dir);
@@ -153,24 +162,35 @@ static void options(int *argc, char ***argv) {
 		{ "mysql-user",		0,   0,	G_OPTION_ARG_STRING,	&c_mysql_user,	"MySQL connection credentials",		"USERNAME"	},
 		{ "mysql-pass",		0,   0,	G_OPTION_ARG_STRING,	&c_mysql_pass,	"MySQL connection credentials",		"PASSWORD"	},
 		{ "mysql-db",		0,   0,	G_OPTION_ARG_STRING,	&c_mysql_db,	"MySQL database name",			"STRING"	},
-		{ "forward-to", 	0,   0, G_OPTION_ARG_STRING,    &forward_to,	"Where to forward to (unix socket)",	"PATH"		},
+		{ "forward-to", 	0,   0, G_OPTION_ARG_STRING,	&forward_to,	"Where to forward to (unix socket)",	"PATH"		},
+		{ "tls-send-to", 	0,   0, G_OPTION_ARG_STRING,	&tls_send_to,	"Where to send to (TLS destination)",	"IP:PORT"	},
+		{ "tls-resample", 	0,   0, G_OPTION_ARG_INT,	&tls_resample,	"Sampling rate for TLS PCM output",	"INT"		},
 		{ NULL, }
 	};
 
 	config_load(argc, argv, e, " - rtpengine recording daemon",
 			"/etc/rtpengine/rtpengine-recording.conf", "rtpengine-recording", &rtpe_common_config);
 
+	if (tls_send_to) {
+		if (endpoint_parse_any_getaddrinfo_full(&tls_send_to_ep, tls_send_to))
+			die("Failed to parse 'tls-send-to' option");
+	}
+
 	if (!strcmp(output_format, "none")) {
 		output_enabled = 0;
 		if (output_mixed || output_single)
 			die("Output is disabled, but output-mixed or output-single is set");
-		if (!forward_to) {
+		if (!forward_to && !tls_send_to_ep.port) {
 			//the daemon has no function
-			die("Both output and packet forwarding are disabled");
+			die("Both output and forwarding are disabled");
 		}
+		output_format = NULL;
 	} else if (!output_mixed && !output_single)
 		output_mixed = output_single = 1;
 
+	if (output_enabled || tls_send_to_ep.port)
+		decoding_enabled = 1;
+
 	if (!os_str || !strcmp(os_str, "file"))
 		output_storage = OUTPUT_STORAGE_FILE;
 	else if (!strcmp(os_str, "db"))

recording-daemon/main.h

@@ -3,6 +3,7 @@
 
 
 #include "auxlib.h"
+#include "socket.h"
 
 
 enum output_storage_enum {
@@ -19,12 +20,15 @@ extern const char *output_dir;
 extern int output_mixed;
 extern int output_single;
 extern int output_enabled;
+extern int decoding_enabled;
 extern const char *c_mysql_host,
       *c_mysql_user,
       *c_mysql_pass,
       *c_mysql_db;
 extern int c_mysql_port;
 extern const char *forward_to;
+extern endpoint_t tls_send_to_ep;
+extern int tls_resample;
 
 extern volatile int shutdown_flag;
 

recording-daemon/metafile.c

@@ -70,9 +70,9 @@ static void meta_destroy(metafile_t *mf) {
 // mf is locked
 static void meta_stream_interface(metafile_t *mf, unsigned long snum, char *content) {
 	db_do_call(mf);
-	if (output_enabled) {
+	if (output_enabled && output_mixed) {
 		pthread_mutex_lock(&mf->mix_lock);
-		if (!mf->mix && output_mixed) {
+		if (!mf->mix) {
 			char buf[256];
 			snprintf(buf, sizeof(buf), "%s-mix", mf->parent);
 			mf->mix_out = output_new(output_dir, buf);
@@ -107,7 +107,7 @@ static void meta_rtp_payload_type(metafile_t *mf, unsigned long mnum, unsigned i
 		ilog(LOG_ERR, "Payload type number %u is invalid", payload_num);
 		return;
 	}
-	if (output_enabled) {
+	if (decoding_enabled) {
 		pthread_mutex_lock(&mf->payloads_lock);
 		mf->payload_types[payload_num] = g_string_chunk_insert(mf->gsc,
 				payload_type);
@@ -119,6 +119,7 @@ static void meta_rtp_payload_type(metafile_t *mf, unsigned long mnum, unsigned i
 // mf is locked
 static void meta_metadata(metafile_t *mf, char *content) {
 	mf->metadata = g_string_chunk_insert(mf->gsc, content);
+	mf->metadata_db = mf->metadata;
 	db_do_call(mf);
 	if (forward_to)
 		start_forwarding_capture(mf, content);
@@ -146,6 +147,12 @@ static void meta_section(metafile_t *mf, char *section, char *content, unsigned
 		tag_name(mf, lu, content);
 	else if (sscanf_match(section, "LABEL %lu", &lu) == 1)
 		tag_label(mf, lu, content);
+	else if (sscanf_match(section, "RECORDING %u", &u) == 1)
+		mf->recording_on = u ? 1 : 0;
+	else if (sscanf_match(section, "FORWARDING %u", &u) == 1)
+		mf->forwarding_on = u ? 1 : 0;
+	else if (sscanf_match(section, "STREAM %lu FORWARDING %u", &lu, &u) == 2)
+		stream_forwarding_on(mf, lu, u);
 }
 
 
@@ -167,8 +174,9 @@ static metafile_t *metafile_get(char *name) {
 	mf->forward_fd = -1;
 	mf->forward_count = 0;
 	mf->forward_failed = 0;
+	mf->recording_on = 1;
 
-	if (output_enabled) {
+	if (decoding_enabled) {
 		pthread_mutex_init(&mf->payloads_lock, NULL);
 		pthread_mutex_init(&mf->mix_lock, NULL);
 		mf->ssrc_hash = g_hash_table_new_full(g_direct_hash, g_direct_equal, NULL, ssrc_free);

recording-daemon/packet.c

@@ -4,6 +4,7 @@
 #include <netinet/udp.h>
 #include <glib.h>
 #include <unistd.h>
+#include <openssl/err.h>
 #include "types.h"
 #include "log.h"
 #include "rtplib.h"
@@ -13,8 +14,65 @@
 #include "main.h"
 #include "output.h"
 #include "db.h"
+#include "streambuf.h"
+#include "resample.h"
 
 
+static ssize_t ssrc_tls_write(void *, const void *, size_t);
+static ssize_t ssrc_tls_read(void *, void *, size_t);
+
+static struct streambuf_funcs ssrc_tls_funcs = {
+	.write = ssrc_tls_write,
+	.read = ssrc_tls_read,
+};
+
+static void ssrc_tls_log_errors(void) {
+	int i;
+	char err[160];
+	while ((i = ERR_get_error())) {
+		ERR_error_string(i, err);
+		dbg("TLS error: %s", err);
+	}
+}
+
+static int ssrc_tls_check_blocked(SSL *ssl, int ret) {
+	if (!ssl)
+		return 0;
+	int err = SSL_get_error(ssl, ret);
+	dbg("TLS error code: %i -> %i", ret, err);
+	switch (err) {
+		case SSL_ERROR_ZERO_RETURN:
+			return 0; // eof
+		case SSL_ERROR_WANT_READ:
+		case SSL_ERROR_WANT_WRITE:
+		case SSL_ERROR_WANT_CONNECT:
+		case SSL_ERROR_WANT_ACCEPT:
+			errno = EAGAIN;
+			return -1;
+		case SSL_ERROR_SYSCALL:
+			return -1;
+	}
+	errno = EFAULT;
+	return -1;
+}
+
+static ssize_t ssrc_tls_write(void *fd, const void *b, size_t s) {
+	SSL *ssl = fd;
+	ssrc_tls_log_errors();
+	int ret = SSL_write(ssl, b, s);
+	if (ret > 0)
+		return ret;
+	return ssrc_tls_check_blocked(ssl, ret);
+}
+static ssize_t ssrc_tls_read(void *fd, void *b, size_t s) {
+	SSL *ssl = fd;
+	ssrc_tls_log_errors();
+	int ret = SSL_read(ssl, b, s);
+	if (ret > 0)
+		return ret;
+	return ssrc_tls_check_blocked(ssl, ret);
+}
+
 static void packet_free(void *p) {
 	packet_t *packet = p;
 	if (!packet)
@@ -24,12 +82,73 @@ static void packet_free(void *p) {
 }
 
 
+static void ssrc_tls_shutdown(ssrc_t *ssrc) {
+	streambuf_destroy(ssrc->tls_fwd_stream);
+	ssrc->tls_fwd_stream = NULL;
+	resample_shutdown(&ssrc->tls_fwd_resampler);
+	if (ssrc->ssl)
+		SSL_free(ssrc->ssl);
+	ssrc->ssl = NULL;
+	if (ssrc->ssl_ctx)
+		SSL_CTX_free(ssrc->ssl_ctx);
+	ssrc->ssl_ctx = NULL;
+	close_socket(&ssrc->tls_fwd_sock);
+	ssrc->sent_intro = 0;
+}
+
+
+void ssrc_tls_state(ssrc_t *ssrc) {
+	int ret;
+
+	ssrc_tls_log_errors();
+	if (ssrc->tls_fwd_poller.state == PS_CONNECTING) {
+		int status = connect_socket_retry(&ssrc->tls_fwd_sock);
+		if (status == 0) {
+			dbg("TLS connection to %s doing handshake",
+				endpoint_print_buf(&tls_send_to_ep));
+			ssrc->tls_fwd_poller.state = PS_HANDSHAKE;
+			if ((ret = SSL_connect(ssrc->ssl)) == 1) {
+				dbg("TLS connection to %s established",
+						endpoint_print_buf(&tls_send_to_ep));
+				ssrc->tls_fwd_poller.state = PS_OPEN;
+				streambuf_writeable(ssrc->tls_fwd_stream);
+			}
+			else
+				ssrc_tls_check_blocked(ssrc->ssl, ret);
+		}
+		else if (status < 0) {
+			ilog(LOG_ERR, "Failed to connect TLS socket: %s", strerror(errno));
+			ssrc_tls_shutdown(ssrc);
+		}
+	}
+	else if (ssrc->tls_fwd_poller.state == PS_HANDSHAKE) {
+		if ((ret = SSL_connect(ssrc->ssl)) == 1) {
+			dbg("TLS connection to %s established",
+					endpoint_print_buf(&tls_send_to_ep));
+			ssrc->tls_fwd_poller.state = PS_OPEN;
+			streambuf_writeable(ssrc->tls_fwd_stream);
+		}
+		else
+			ssrc_tls_check_blocked(ssrc->ssl, ret);
+	}
+	else if (ssrc->tls_fwd_poller.state == PS_WRITE_BLOCKED) {
+		ssrc->tls_fwd_poller.state = PS_OPEN;
+		streambuf_writeable(ssrc->tls_fwd_stream);
+	}
+	else if (ssrc->tls_fwd_poller.state == PS_ERROR)
+		ssrc_tls_shutdown(ssrc);
+	ssrc_tls_log_errors();
+}
+
+
 void ssrc_free(void *p) {
 	ssrc_t *s = p;
 	packet_sequencer_destroy(&s->sequencer);
 	output_close(s->output);
 	for (int i = 0; i < G_N_ELEMENTS(s->decoders); i++)
-		decoder_close(s->decoders[i]);
+		decoder_free(s->decoders[i]);
+	if (s->tls_fwd_stream)
+		ssrc_tls_shutdown(s);
 	g_slice_free1(sizeof(*s), s);
 }
 
@@ -49,18 +168,66 @@ static ssrc_t *ssrc_get(stream_t *stream, unsigned long ssrc) {
 	ret->ssrc = ssrc;
 	packet_sequencer_init(&ret->sequencer, packet_free);
 
-	char buf[256];
-	snprintf(buf, sizeof(buf), "%s-%08lx", mf->parent, ssrc);
-	if (output_single) {
-		ret->output = output_new(output_dir, buf);
-		db_do_stream(mf, ret->output, "single", stream, ssrc);
-	}
-
 	g_hash_table_insert(mf->ssrc_hash, GUINT_TO_POINTER(ssrc), ret);
 
 out:
 	pthread_mutex_lock(&ret->lock);
 	pthread_mutex_unlock(&mf->lock);
+
+	dbg("Init for SSRC %lx of stream #%lu", ret->ssrc, stream->id);
+
+	if (mf->recording_on && !ret->output && output_single) {
+		char buf[256];
+		snprintf(buf, sizeof(buf), "%s-%08lx", mf->parent, ssrc);
+		ret->output = output_new(output_dir, buf);
+		db_do_stream(mf, ret->output, "single", stream, ssrc);
+	}
+	if ((stream->forwarding_on || mf->forwarding_on) && !ret->tls_fwd_stream) {
+		// initialise the connection
+		ZERO(ret->tls_fwd_poller);
+		dbg("Starting TLS connection to %s", endpoint_print_buf(&tls_send_to_ep));
+		ret->ssl_ctx = SSL_CTX_new(TLS_client_method());
+		if (!ret->ssl_ctx) {
+			ilog(LOG_ERR, "Failed to create TLS context");
+			ssrc_tls_shutdown(ret);
+			goto tls_out;
+		}
+		ret->ssl = SSL_new(ret->ssl_ctx);
+		if (!ret->ssl) {
+			ilog(LOG_ERR, "Failed to create TLS connection");
+			ssrc_tls_shutdown(ret);
+			goto tls_out;
+		}
+		int status = connect_socket_nb(&ret->tls_fwd_sock, SOCK_STREAM, &tls_send_to_ep);
+		if (status < 0) {
+			ilog(LOG_ERR, "Failed to open/connect TLS socket to %s: %s",
+				endpoint_print_buf(&tls_send_to_ep),
+				strerror(errno));
+			ssrc_tls_shutdown(ret);
+			goto tls_out;
+		}
+
+		ret->tls_fwd_poller.state = PS_CONNECTING;
+		if (SSL_set_fd(ret->ssl, ret->tls_fwd_sock.fd) != 1) {
+			ilog(LOG_ERR, "Failed to set TLS fd");
+			ssrc_tls_shutdown(ret);
+			goto tls_out;
+		}
+		ret->tls_fwd_stream = streambuf_new_ptr(&ret->tls_fwd_poller, ret->ssl, &ssrc_tls_funcs);
+
+		ssrc_tls_state(ret);
+
+		ret->tls_fwd_format = (format_t) {
+			.clockrate = tls_resample,
+			.channels = 1,
+			.format = AV_SAMPLE_FMT_S16,
+		};
+tls_out:
+		;
+	}
+	else if (!(stream->forwarding_on || mf->forwarding_on) && ret->tls_fwd_stream)
+		ssrc_tls_shutdown(ret);
+
 	return ret;
 }
 
@@ -103,7 +270,7 @@ static void packet_decode(ssrc_t *ssrc, packet_t *packet) {
 	}
 
 	if (decoder_input(ssrc->decoders[payload_type], &packet->payload, ntohl(packet->rtp->timestamp),
-			ssrc->output, ssrc->metafile))
+			ssrc))
 		ilog(LOG_ERR, "Failed to decode media packet");
 }
 

recording-daemon/packet.h

@@ -7,4 +7,6 @@ void ssrc_free(void *p);
 
 void packet_process(stream_t *, unsigned char *, unsigned len);
 
+void ssrc_tls_state(ssrc_t *ssrc);
+
 #endif

recording-daemon/poller.c

@@ -0,0 +1,11 @@
+#include "poller.h"
+
+void poller_blocked(struct poller *p, void *fdp) {
+	p->state = PS_WRITE_BLOCKED;
+}
+int poller_isblocked(struct poller *p, void *fdp) {
+	return p->state != PS_OPEN;
+}
+void poller_error(struct poller *p, void *fdp) {
+	p->state = PS_ERROR;
+}

recording-daemon/poller.h

@@ -0,0 +1,22 @@
+#ifndef __POLLER_H__
+#define __POLLER_H__
+
+
+// dummy poller
+struct poller {
+	enum {
+		PS_CLOSED = 0,
+		PS_CONNECTING,
+		PS_HANDSHAKE,
+		PS_OPEN,
+		PS_WRITE_BLOCKED,
+		PS_ERROR,
+	} state;
+};
+
+void poller_blocked(struct poller *, void *);
+int poller_isblocked(struct poller *, void *);
+void poller_error(struct poller *, void *);
+
+
+#endif

recording-daemon/stream.c

@@ -75,7 +75,7 @@ static void stream_handler(handler_t *handler) {
 		else
 			g_atomic_int_inc(&stream->metafile->forward_count);
 	}
-	if (output_enabled)
+	if (decoding_enabled)
 		packet_process(stream, buf, ret); // consumes buf
 	else
 		free(buf);
@@ -140,3 +140,9 @@ void stream_details(metafile_t *mf, unsigned long id, unsigned int tag) {
 	stream_t *stream = stream_get(mf, id);
 	stream->tag = tag;
 }
+
+void stream_forwarding_on(metafile_t *mf, unsigned long id, unsigned int on) {
+	stream_t *stream = stream_get(mf, id);
+	dbg("Setting forwarding flag to %u for stream #%lu", on, stream->id);
+	stream->forwarding_on = on ? 1 : 0;
+}

recording-daemon/stream.h

@@ -5,6 +5,7 @@
 
 void stream_open(metafile_t *mf, unsigned long id, char *name);
 void stream_details(metafile_t *mf, unsigned long id, unsigned int tag);
+void stream_forwarding_on(metafile_t *mf, unsigned long id, unsigned int on);
 void stream_close(stream_t *stream);
 void stream_free(stream_t *stream);
 

recording-daemon/types.h

@@ -10,14 +10,19 @@
 #include <libavutil/channel_layout.h>
 #include <libavutil/samplefmt.h>
 #include <libavutil/audio_fifo.h>
+#include <openssl/ssl.h>
+#include <openssl/bio.h>
 #include "str.h"
 #include "codeclib.h"
+#include "poller.h"
+#include "socket.h"
 
 
 struct iphdr;
 struct ip6_hdr;
 struct udphdr;
 struct rtp_header;
+struct streambuf;
 
 
 struct handler_s;
@@ -28,6 +33,8 @@ struct output_s;
 typedef struct output_s output_t;
 struct mix_s;
 typedef struct mix_s mix_t;
+struct decode_s;
+typedef struct decode_s decode_t;
 
 
 typedef void handler_func(handler_t *);
@@ -47,6 +54,7 @@ struct stream_s {
 	unsigned long tag;
 	int fd;
 	handler_t handler;
+	int forwarding_on:1;
 };
 typedef struct stream_s stream_t;
 
@@ -71,8 +79,19 @@ struct ssrc_s {
 	metafile_t *metafile;
 	unsigned long ssrc;
 	packet_sequencer_t sequencer;
-	decoder_t *decoders[128];
+	decode_t *decoders[128];
 	output_t *output;
+
+	// TLS output
+	format_t tls_fwd_format;
+	resample_t tls_fwd_resampler;
+	socket_t tls_fwd_sock;
+	//BIO *bio;
+	SSL_CTX *ssl_ctx;
+	SSL *ssl;
+	struct streambuf *tls_fwd_stream;
+	struct poller tls_fwd_poller;
+	int sent_intro:1;
 };
 typedef struct ssrc_s ssrc_t;
 
@@ -91,6 +110,7 @@ struct metafile_s {
 	char *parent;
 	char *call_id;
 	char *metadata;
+	char *metadata_db;
 	off_t pos;
 	unsigned long long db_id;
 
@@ -110,6 +130,9 @@ struct metafile_s {
 
 	pthread_mutex_t payloads_lock;
 	char *payload_types[128];
+
+	int recording_on:1;
+	int forwarding_on:1;
 };
 
 
@@ -135,5 +158,12 @@ struct output_s {
 };
 
 
+struct decode_s {
+	decoder_t *dec;
+	resample_t mix_resampler;
+	unsigned int mixer_idx;
+};
+
+
 
 #endif

t/Makefile

@@ -62,10 +62,10 @@ SRCS+=		transcode-test.c
 ifeq ($(with_amr_tests),yes)
 SRCS+=		amr-decode-test.c amr-encode-test.c
 endif
-LIBSRCS+=	codeclib.c resample.c
-DAEMONSRCS+=	codec.c call.c ice.c kernel.c media_socket.c stun.c bencode.c socket.c poller.c \
+LIBSRCS+=	codeclib.c resample.c socket.c streambuf.c
+DAEMONSRCS+=	codec.c call.c ice.c kernel.c media_socket.c stun.c bencode.c poller.c \
 		dtls.c recording.c statistics.c rtcp.c redis.c iptables.c graphite.c \
-		streambuf.c cookie_cache.c udp_listener.c homer.c load.c cdr.c dtmf.c
+		cookie_cache.c udp_listener.c homer.c load.c cdr.c dtmf.c
 HASHSRCS+=	call_interfaces.c control_ng.c sdp.c
 endif
 

t/log.h

@@ -3,6 +3,7 @@
 
 #include "loglib.h"
 #define __ilog(prio, fmt, ...) fprintf(stderr, fmt "\n", ##__VA_ARGS__)
+#define __C_DBG(x...) ilog(LOG_DEBUG, x)
 
 INLINE void rtcplog(const char *x) {
 }