TT#5566 support mp3 audio output

Change-Id: Ife4001f5a9fcf8951c7c2b93d47fa2dcf2750a7b

recording-daemon/decoder.c

@@ -1,6 +1,8 @@
 #include "decoder.h"
 #include <libavcodec/avcodec.h>
 #include <libavformat/avformat.h>
+#include <libavutil/audio_fifo.h>
+#include <libavutil/channel_layout.h>
 #include <glib.h>
 #include <stdint.h>
 #include "types.h"
@@ -28,6 +30,10 @@ struct output_s {
 	AVFormatContext *fmtctx;
 	AVStream *avst;
 	AVPacket avpkt;
+	AVAudioFifo *fifo;
+	int64_t fifo_pts; // pts of first data in fifo
+	int64_t mux_dts; // last dts passed to muxer
+	AVFrame *frame;
 };
 
 
@@ -77,6 +83,11 @@ typedef struct decoder_def_s decoder_def_t;
 
 
 
+static int output_codec_id;
+static const char *output_file_format;
+
+
+
 static void output_shutdown(output_t *output);
 static int output_config(output_t *output, unsigned int clockrate, unsigned int channels);
 
@@ -151,38 +162,80 @@ err:
 }
 
 
-static int output_add(output_t *output, AVFrame *frame) {
-	if (!output)
-		return -1;
+static int output_flush(output_t *output) {
+	while (av_audio_fifo_size(output->fifo) >= output->frame->nb_samples) {
 
-#if LIBAVCODEC_VERSION_INT >= AV_VERSION_INT(57, 0, 0)
-	int ret = avcodec_send_frame(output->avcctx, frame);
-	dbg("send frame ret %i", ret);
-	if (ret)
-		return -1;
+		if (av_audio_fifo_read(output->fifo, (void **) output->frame->data,
+					output->frame->nb_samples) <= 0)
+			abort();
 
-	ret = avcodec_receive_packet(output->avcctx, &output->avpkt);
-	dbg("receive packet ret %i", ret);
-	if (ret)
-		return -1;
+		dbg("%p output fifo pts %lu", output, (unsigned long) output->fifo_pts);
+		output->frame->pts = output->fifo_pts;
+
+#if LIBAVCODEC_VERSION_INT >= AV_VERSION_INT(57, 0, 0)
+		int ret = avcodec_send_frame(output->avcctx, output->frame);
+		dbg("%p send frame ret %i", output, ret);
+		if (ret)
+			return -1;
+
+		ret = avcodec_receive_packet(output->avcctx, &output->avpkt);
+		dbg("%p receive packet ret %i", output, ret);
+		if (ret)
+			return -1;
 #else
-	int got_packet = 0;
-	int ret = avcodec_encode_audio2(output->avcctx, &output->avpkt, frame, &got_packet);
-	dbg("encode frame ret %i, got packet %i", ret, got_packet);
-	if (!got_packet)
-		return 0;
+		int got_packet = 0;
+		int ret = avcodec_encode_audio2(output->avcctx, &output->avpkt, output->frame, &got_packet);
+		dbg("%p encode frame ret %i, got packet %i", output, ret, got_packet);
+		if (!got_packet)
+			return 0;
 #endif
 
-	av_write_frame(output->fmtctx, &output->avpkt);
+		dbg("%p output avpkt size is %i", output, (int) output->avpkt.size);
+		dbg("%p output pkt pts/dts is %li/%li", output, (long) output->avpkt.pts,
+				(long) output->avpkt.dts);
+		dbg("%p output dts %li", output, (long) output->mux_dts);
+
+		// the encoder may return frames with the same dts multiple consecutive times.
+		// the muxer may not like this, so ensure monotonically increasing dts.
+		if (output->mux_dts > output->avpkt.dts)
+			output->avpkt.dts = output->mux_dts;
+		if (output->avpkt.pts < output->avpkt.dts)
+			output->avpkt.pts = output->avpkt.dts;
+
+		av_write_frame(output->fmtctx, &output->avpkt);
+
+		output->fifo_pts += output->frame->nb_samples;
+		output->mux_dts = output->avpkt.dts + 1; // min next expected dts
+	}
 
 	return 0;
 }
 
 
+static int output_add(output_t *output, AVFrame *frame) {
+	if (!output)
+		return -1;
+
+	dbg("%p output fifo size %u fifo_pts %lu", output, (unsigned int) av_audio_fifo_size(output->fifo),
+			(unsigned long) output->fifo_pts);
+	// fix up output pts
+	if (av_audio_fifo_size(output->fifo) == 0)
+		output->fifo_pts = frame->pts;
+
+	if (av_audio_fifo_write(output->fifo, (void **) frame->extended_data, frame->nb_samples) < 0)
+		return -1;
+
+	return output_flush(output);
+}
+
+
 int decoder_input(decoder_t *dec, const str *data, unsigned long ts, output_t *output) {
 	if (G_UNLIKELY(!dec))
 		return -1;
 
+	dbg("%p dec pts %lu rtp_ts %lu incoming ts %lu", dec, (unsigned long) dec->pts,
+			(unsigned long) dec->rtp_ts, (unsigned long) ts);
+
 	if (G_UNLIKELY(dec->rtp_ts == (unsigned long) -1L)) {
 		// initialize pts
 		dec->pts = 0;
@@ -217,9 +270,12 @@ int decoder_input(decoder_t *dec, const str *data, unsigned long ts, output_t *o
 #endif
 
 	dec->frame->pts = dec->frame->pkt_pts;
+	dbg("%p dec frame pts %lu pkt_pts %lu", dec, (unsigned long) dec->frame->pts,
+			(unsigned long) dec->frame->pkt_dts);
 
 	output_config(output, dec->avcctx->sample_rate, dec->avcctx->channels);
-	output_add(output, dec->frame);
+	if (output_add(output, dec->frame))
+		return -1;
 
 	return 0;
 }
@@ -227,9 +283,11 @@ int decoder_input(decoder_t *dec, const str *data, unsigned long ts, output_t *o
 
 output_t *output_new(const char *filename) {
 	output_t *ret = g_slice_alloc0(sizeof(*ret));
-	ret->filename = strdup(filename);
+	if (asprintf(&ret->filename, "%s.%s", filename, output_file_format) <= 0)
+		abort();
 	ret->clockrate = -1;
 	ret->channels = -1;
+	ret->frame = av_frame_alloc();
 	return ret;
 }
 
@@ -255,11 +313,11 @@ format_mismatch:
 	output->fmtctx = avformat_alloc_context();
 	if (!output->fmtctx)
 		goto err;
-	output->fmtctx->oformat = av_guess_format("wav", NULL, NULL); // XXX better way?
+	output->fmtctx->oformat = av_guess_format(output_file_format, NULL, NULL);
 	if (!output->fmtctx->oformat)
 		goto err;
 
-	AVCodec *codec = avcodec_find_encoder(AV_CODEC_ID_PCM_S16LE);
+	AVCodec *codec = avcodec_find_encoder(output_codec_id);
 	// XXX error handling
 	output->avst = avformat_new_stream(output->fmtctx, codec);
 	if (!output->avst)
@@ -273,6 +331,7 @@ format_mismatch:
 #endif
 
 	output->avcctx->channels = output->channels;
+	output->avcctx->channel_layout = av_get_default_channel_layout(output->channels);
 	output->avcctx->sample_rate = output->clockrate;
 	output->avcctx->sample_fmt = AV_SAMPLE_FMT_S16;
 	output->avcctx->time_base = (AVRational){output->clockrate,1};
@@ -294,6 +353,19 @@ format_mismatch:
 
 	av_init_packet(&output->avpkt);
 
+	// output frame and fifo
+	output->frame->nb_samples = output->avcctx->frame_size ? : 256;
+	output->frame->format = output->avcctx->sample_fmt;
+	output->frame->sample_rate = output->avcctx->sample_rate;
+	output->frame->channel_layout = output->avcctx->channel_layout;
+	if (!output->frame->channel_layout)
+		output->frame->channel_layout = av_get_default_channel_layout(output->avcctx->channels);
+	if (av_frame_get_buffer(output->frame, 0) < 0)
+		abort();
+
+	output->fifo = av_audio_fifo_alloc(output->avcctx->sample_fmt, output->avcctx->channels,
+			output->frame->nb_samples);
+
 	return 0;
 
 err:
@@ -318,10 +390,13 @@ static void output_shutdown(output_t *output) {
 	avcodec_close(output->avcctx);
 	avio_closep(&output->fmtctx->pb);
 	avformat_free_context(output->fmtctx);
+	av_audio_fifo_free(output->fifo);
+	av_frame_free(&output->frame);
 
 	output->avcctx = NULL;
 	output->fmtctx = NULL;
 	output->avst = NULL;
+	output->fifo = NULL;
 }
 
 
@@ -332,3 +407,17 @@ void output_close(output_t *output) {
 	free(output->filename);
 	g_slice_free1(sizeof(*output), output);
 }
+
+
+void output_init(const char *format) {
+	if (!strcmp(format, "wav")) {
+		output_codec_id = AV_CODEC_ID_PCM_S16LE;
+		output_file_format = "wav";
+	}
+	else if (!strcmp(format, "mp3")) {
+		output_codec_id = AV_CODEC_ID_MP3;
+		output_file_format = "mp3";
+	}
+	else
+		die("Unknown output format '%s'", format);
+}

recording-daemon/decoder.h

@@ -5,6 +5,8 @@
 #include "str.h"
 
 
+void output_init(const char *format);
+
 decoder_t *decoder_new(const char *payload_str);
 int decoder_input(decoder_t *, const str *, unsigned long ts, output_t *);
 void decoder_close(decoder_t *);

recording-daemon/epoll.c

@@ -42,8 +42,11 @@ void *poller_thread(void *ptr) {
 		int ret = epoll_wait(epoll_fd, &epev, 1, 10000);
 		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
 
-		if (ret == -1)
+		if (ret == -1) {
+			if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
+				continue;
 			die_errno("epoll_wait failed");
+		}
 
 		if (ret > 0) {
 			dbg("thread %u handling event", me_num);

recording-daemon/main.c

@@ -17,6 +17,7 @@
 #include "garbage.h"
 #include "loglib.h"
 #include "auxlib.h"
+#include "decoder.h"
 
 
 
@@ -24,6 +25,7 @@ int ktable = 0;
 int num_threads = 8;
 const char *spool_dir = "/var/spool/rtpengine";
 const char *output_dir = "/var/lib/rtpengine-recording";
+static const char *output_format = "wav";
 
 
 static GQueue threads = G_QUEUE_INIT; // only accessed from main thread
@@ -44,11 +46,19 @@ static void signals(void) {
 
 
 static void avlog_ilog(void *ptr, int loglevel, const char *fmt, va_list ap) {
-	__vpilog(loglevel, NULL, fmt, ap);
+	char *msg;
+	if (vasprintf(&msg, fmt, ap) <= 0)
+		ilog(LOG_ERR, "av_log message dropped");
+	else {
+		ilog(MAX(LOG_ERR, loglevel), "av_log: %s", msg);
+		free(msg);
+	}
 }
 
 
 static void setup(void) {
+	openlog("rtpengine-recording", LOG_PID | LOG_NDELAY, LOG_DAEMON);
+
 	log_init();
 	av_register_all();
 	avcodec_register_all();
@@ -58,7 +68,7 @@ static void setup(void) {
 	epoll_setup();
 	inotify_setup();
 	av_log_set_callback(avlog_ilog);
-	openlog("rtpengine-recording", LOG_PID | LOG_NDELAY, LOG_DAEMON);
+	output_init(output_format);
 
 	if (!g_file_test(output_dir, G_FILE_TEST_IS_DIR)) {
 		ilog(LOG_INFO, "Creating output dir '%s'", output_dir);
@@ -123,6 +133,7 @@ static void options(int *argc, char ***argv) {
 		{ "table",		't', 0, G_OPTION_ARG_INT,	&ktable,	"Kernel table rtpengine uses",		"INT"		},
 		{ "spool-dir",		0,   0, G_OPTION_ARG_STRING,	&spool_dir,	"Directory containing rtpengine metadata files", "PATH" },
 		{ "output-dir",		0,   0, G_OPTION_ARG_STRING,	&output_dir,	"Where to write media files to",	"PATH"		},
+		{ "output-format",	0,   0, G_OPTION_ARG_STRING,	&output_format,	"Write audio files of this type",	"wav|mp3"	},
 		{ "num-threads",	0,   0, G_OPTION_ARG_INT,	&num_threads,	"Number of worker threads",		"INT"		},
 		{ NULL, }
 	};

recording-daemon/packet.c

@@ -56,7 +56,7 @@ static ssrc_t *ssrc_get(metafile_t *mf, unsigned long ssrc) {
 	ret->seq = -1;
 
 	char buf[256];
-	snprintf(buf, sizeof(buf), "%s/%s-%08lx.wav", output_dir, mf->parent, ssrc);
+	snprintf(buf, sizeof(buf), "%s/%s-%08lx", output_dir, mf->parent, ssrc);
 	ret->output = output_new(buf);
 
 	g_hash_table_insert(mf->ssrc_hash, GUINT_TO_POINTER(ssrc), ret);
@@ -161,8 +161,9 @@ have_packet:;
 			}
 		}
 
-		decoder_input(ssrc->decoders[payload_type], &packet->payload, ntohl(packet->rtp->timestamp),
-				ssrc->output);
+		if (decoder_input(ssrc->decoders[payload_type], &packet->payload, ntohl(packet->rtp->timestamp),
+				ssrc->output))
+			ilog(LOG_ERR, "Failed to decode media packet");
 
 next_packet:
 		ssrc->seq = (packet->seq + 1) & 0xffff;