MT#62571 turn tls_fwd into sink

Change-Id: Ic859d0dd65156761007641ccd3f4e710d7e7fa9f

recording-daemon/decoder.c

@@ -105,71 +105,25 @@ static int decoder_got_frame(decoder_t *dec, AVFrame *frame, void *sp, void *dp)
 			(unsigned int) frame->extended_data[0][2],
 			(unsigned int) frame->extended_data[0][3]);
 
-	if (!metafile->recording_on)
-		goto no_recording;
+	if (metafile->recording_on) {
+		sink_add(&deco->mix_sink, frame, &dec->dest_format);
 
-	sink_add(&deco->mix_sink, frame, &dec->dest_format);
-
-	if (output) {
-		dbg("SSRC %lx of stream #%lu has single output", ssrc->ssrc, stream->id);
-		if (!sink_add(&output->sink, frame, &dec->dest_format))
-			ilog(LOG_ERR, "Failed to add decoded packet to individual output");
+		if (output) {
+			dbg("SSRC %lx of stream #%lu has single output", ssrc->ssrc, stream->id);
+			if (!sink_add(&output->sink, frame, &dec->dest_format))
+				ilog(LOG_ERR, "Failed to add decoded packet to individual output");
+		}
 	}
 
-no_recording:
 	if (ssrc->tls_fwd) {
-		// XXX might be a second resampling to same format
 		dbg("SSRC %lx of stream #%lu has TLS forwarding stream", ssrc->ssrc, stream->id);
-
-		tls_fwd_state(&ssrc->tls_fwd);
-		// if we're in the middle of a disconnect then ssrc_tls_state may have destroyed the streambuf
-		// so we need to skip the below to ensure we only send metadata for the new connection
-		// once we've got a new streambuf
-		if (!ssrc->tls_fwd)
-			goto err;
-
-		AVFrame *dec_frame = resample_frame(&ssrc->tls_fwd->resampler, frame, &ssrc->tls_fwd->format);
-
-		if (!ssrc->tls_fwd->sent_intro) {
-			tag_t *tag = NULL;
-
-			if (ssrc->stream)
-				tag = tag_get(metafile, ssrc->stream->tag);
-
-			if (tag && tag->metadata) {
-				dbg("Writing tag metadata header to TLS");
-				streambuf_write(ssrc->tls_fwd->stream, tag->metadata, strlen(tag->metadata) + 1);
-			}
-			else if (metafile->metadata) {
-				dbg("Writing call 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->tls_fwd->sent_intro = 1;
-		}
-
-		tls_fwd_silence_frames_upto(ssrc->tls_fwd, dec_frame, dec_frame->pts);
-		uint64_t next_pts = dec_frame->pts + dec_frame->nb_samples;
-		if (next_pts > ssrc->tls_fwd->in_pts)
-			ssrc->tls_fwd->in_pts = next_pts;
-
-		int linesize = av_get_bytes_per_sample(dec_frame->format) * dec_frame->nb_samples;
-		dbg("Writing %u bytes PCM to TLS", linesize);
-		streambuf_write(ssrc->tls_fwd->stream, (char *) dec_frame->extended_data[0], linesize);
-		if (dec_frame != frame)
-			av_frame_free(&dec_frame);
+		if (!sink_add(&ssrc->tls_fwd->sink, frame, &ssrc->tls_fwd->format))
+			ilog(LOG_ERR, "Failed to add decoded packet to TLS/TCP forward output");
 
 	}
 
 	av_frame_free(&frame);
 	return 0;
-
-err:
-	av_frame_free(&frame);
-	return -1;
 }
 
 

recording-daemon/tls_send.c

@@ -9,6 +9,8 @@
 #include "main.h"
 #include "streambuf.h"
 #include "fix_frame_channel_layout.h"
+#include "output.h"
+#include "tag.h"
 
 
 static ssize_t ssrc_tls_write(void *, const void *, size_t);
@@ -73,7 +75,6 @@ void tls_fwd_shutdown(tls_fwd_t **p) {
 		return;
 	streambuf_destroy(tls_fwd->stream);
 	tls_fwd->stream = NULL;
-	resample_shutdown(&tls_fwd->resampler);
 	if (tls_fwd->ssl) {
 		SSL_free(tls_fwd->ssl);
 		tls_fwd->ssl = NULL;
@@ -84,11 +85,12 @@ void tls_fwd_shutdown(tls_fwd_t **p) {
 	}
 	close_socket(&tls_fwd->sock);
 	av_frame_free(&tls_fwd->silence_frame);
+	sink_close(&tls_fwd->sink);
 	g_clear_pointer(p, g_free);
 }
 
 
-void tls_fwd_state(tls_fwd_t **p) {
+static void tls_fwd_state(tls_fwd_t **p) {
 	tls_fwd_t *tls_fwd = *p;
 	int ret;
 
@@ -140,7 +142,7 @@ void tls_fwd_state(tls_fwd_t **p) {
 }
 
 
-void tls_fwd_silence_frames_upto(tls_fwd_t *tls_fwd, AVFrame *frame, int64_t upto) {
+static void tls_fwd_silence_frames_upto(tls_fwd_t *tls_fwd, AVFrame *frame, int64_t upto) {
 	unsigned int silence_samples = tls_fwd->format.clockrate / 100;
 
 	while (tls_fwd->in_pts < upto) {
@@ -183,6 +185,69 @@ void tls_fwd_silence_frames_upto(tls_fwd_t *tls_fwd, AVFrame *frame, int64_t upt
 }
 
 
+static bool tls_fwd_add(sink_t *sink, AVFrame *frame) {
+	tls_fwd_t **p = sink->tls_fwd;
+
+	tls_fwd_state(p);
+
+	// if we're in the middle of a disconnect then ssrc_tls_state may have destroyed the streambuf
+	// so we need to skip the below to ensure we only send metadata for the new connection
+	// once we've got a new streambuf
+	tls_fwd_t *tls_fwd = *p;
+	if (!tls_fwd) {
+		av_frame_free(&frame);
+		return false;
+	}
+
+	if (!tls_fwd->sent_intro) {
+		ssrc_t *ssrc = tls_fwd->ssrc;
+		metafile_t *metafile = tls_fwd->metafile;
+		tag_t *tag = NULL;
+
+		if (ssrc->stream)
+			tag = tag_get(metafile, ssrc->stream->tag);
+
+		if (tag && tag->metadata) {
+			dbg("Writing tag metadata header to TLS");
+			streambuf_write(tls_fwd->stream, tag->metadata, strlen(tag->metadata) + 1);
+		}
+		else if (metafile->metadata) {
+			dbg("Writing call metadata header to TLS");
+			streambuf_write(tls_fwd->stream, metafile->metadata, strlen(metafile->metadata) + 1);
+		}
+		else {
+			ilog(LOG_WARN, "No metadata present for forwarding connection");
+			streambuf_write(tls_fwd->stream, "\0", 1);
+		}
+		tls_fwd->sent_intro = 1;
+	}
+
+	tls_fwd_silence_frames_upto(tls_fwd, frame, frame->pts);
+	uint64_t next_pts = frame->pts + frame->nb_samples;
+	if (next_pts > tls_fwd->in_pts)
+		tls_fwd->in_pts = next_pts;
+
+	int linesize = av_get_bytes_per_sample(frame->format) * frame->nb_samples;
+	dbg("Writing %u bytes PCM to TLS", linesize);
+	streambuf_write(tls_fwd->stream, (char *) frame->extended_data[0], linesize);
+
+	av_frame_free(&frame);
+
+	return true;
+}
+
+
+static bool tls_fwd_config(sink_t *sink, const format_t *requested_format, format_t *actual_format)
+{
+	*actual_format = (format_t) {
+		.clockrate = tls_resample,
+		.format = AV_SAMPLE_FMT_S16,
+		.channels = 1,
+	};
+	return true;
+}
+
+
 void tls_fwd_init(stream_t *stream, metafile_t *mf, ssrc_t *ssrc) {
 	if ((!stream->forwarding_on && !mf->forwarding_on) || !tls_send_to_ep.port) {
 		tls_fwd_shutdown(&ssrc->tls_fwd);
@@ -238,6 +303,9 @@ void tls_fwd_init(stream_t *stream, metafile_t *mf, ssrc_t *ssrc) {
 		tls_fwd->stream = streambuf_new(&tls_fwd->poller, tls_fwd->sock.fd);
 	}
 
+	tls_fwd->ssrc = ssrc;
+	tls_fwd->metafile = mf;
+
 	tls_fwd_state(&ssrc->tls_fwd);
 	if (!ssrc->tls_fwd) // may have been closed
 		return;
@@ -247,4 +315,9 @@ void tls_fwd_init(stream_t *stream, metafile_t *mf, ssrc_t *ssrc) {
 		.channels = 1,
 		.format = AV_SAMPLE_FMT_S16,
 	};
+
+	sink_init(&tls_fwd->sink);
+	tls_fwd->sink.tls_fwd = &ssrc->tls_fwd;
+	tls_fwd->sink.add = tls_fwd_add;
+	tls_fwd->sink.config = tls_fwd_config;
 }

recording-daemon/tls_send.h

@@ -6,7 +6,4 @@
 void tls_fwd_init(stream_t *stream, metafile_t *mf, ssrc_t *ssrc);
 void tls_fwd_shutdown(tls_fwd_t **);
 
-void tls_fwd_state(tls_fwd_t **);
-void tls_fwd_silence_frames_upto(tls_fwd_t *, AVFrame *frame, int64_t upto);
-
 #endif

recording-daemon/types.h

@@ -53,6 +53,7 @@ struct sink_s {
 	union {
 		output_t *output;
 		ssrc_t *ssrc;
+		tls_fwd_t **tls_fwd;
 	};
 
 	resample_t resampler;
@@ -92,8 +93,8 @@ struct packet_s {
 
 
 struct tls_fwd_s {
+	sink_t sink;
 	format_t format;
-	resample_t resampler;
 	socket_t sock;
 	uint64_t in_pts;
 	AVFrame *silence_frame;
@@ -101,6 +102,8 @@ struct tls_fwd_s {
 	SSL *ssl;
 	struct streambuf *stream;
 	struct poller poller;
+	ssrc_t *ssrc;
+	metafile_t *metafile;
 	unsigned int sent_intro:1;
 };