path: root/src/audio.c

#include "audio.h"
#include "varint.h"
#include "cmumble.h"
#include <string.h>

#include <gst/tag/tag.h>

#define SAMPLERATE 48000
#define FRAMESIZE 480 /* SAMPLERATE/100 */
#define CHANNELS 1
#define BUFFER_TIME (gst_util_uint64_scale_int(1, GST_SECOND, 100))
#define CELT_CAPS "audio/x-celt,channels=" G_STRINGIFY(CHANNELS) "," \
	"rate=" G_STRINGIFY(SAMPLERATE) ",frame-size=" G_STRINGIFY(FRAMESIZE)
#define AUDIO_CAPS "audio/x-raw,format=S16LE,channels=" \
	G_STRINGIFY(CHANNELS) ",rate=" G_STRINGIFY(SAMPLERATE)

static void
add_celt_streamheader(struct cmumble *cm, GstAppSrc *src)
{
	GValue streamheader = { 0, }, val = { 0, };
	GstTagList *tags;
	GstBuffer *buf[2];
	GstStructure *s;
	GstCaps *caps;
	gint i;

	buf[0] = gst_buffer_new_allocate(NULL, sizeof(CELTHeader), NULL);
	gst_buffer_fill(buf[0], 0, cm->audio.celt_header_packet,
			sizeof(CELTHeader));

	tags = gst_tag_list_new_empty();
	buf[1] = gst_tag_list_to_vorbiscomment_buffer(tags, NULL, 0, "mumble");
	gst_tag_list_unref(tags);

	g_value_init(&streamheader, GST_TYPE_ARRAY);
	for (i = 0; i < G_N_ELEMENTS(buf); ++i) {
		GST_BUFFER_FLAG_SET(buf[i], GST_BUFFER_FLAG_HEADER);
		GST_BUFFER_OFFSET(buf[i]) = 0;
		GST_BUFFER_OFFSET_END(buf[i]) = 0;
		g_value_init(&val, GST_TYPE_BUFFER);
		gst_value_take_buffer(&val, buf[i]);
		gst_value_array_append_value(&streamheader, &val);
		g_value_unset(&val);
	}

	caps = gst_app_src_get_caps(src);
	caps = gst_caps_make_writable(caps);
	s = gst_caps_get_structure(caps, 0);
	gst_structure_set_value(s, "streamheader", &streamheader);
	g_value_unset(&streamheader);

	gst_app_src_set_caps(src, caps);
	gst_caps_unref(caps);
}

static void
set_pulse_sink_states(GstElement *el, struct cmumble_user *user)
{
	GstStructure *props;
	gchar *name;

	/* FIXME: Move this into a man-page or so:
	 * Dear User: Add the following to the pulseaudio configuration:
	 * load-module module-device-manager "do_routing=1"
	 * This is to let new join users default to e.g. a headset output.
	 * Also consider setting device.intended_roles = "phone" for your
	 * output to be marked as headset (if you dont have a usb headset dev). */

	name = g_strdup_printf("cmumble [%s]", user->name);

	props = gst_structure_new("props",
				  "application.name", G_TYPE_STRING, name,
				  "media.role", G_TYPE_STRING, "phone",
				  NULL);

	g_object_set(el, "stream-properties", props, NULL);
	gst_structure_free(props);
	g_free(name);
}

static void
on_sink_bin_element_added (GstBin *bin, GstElement *el, gpointer user_data)
{
	struct cmumble_user *user = user_data;

	if (g_strcmp0(G_OBJECT_TYPE_NAME(el), "GstPulseSink") == 0 &&
	    g_object_class_find_property(G_OBJECT_GET_CLASS(el),
					 "stream-properties") != NULL) {
		set_pulse_sink_states(el, user);
	}
}


int
cmumble_audio_create_playback_pipeline(struct cmumble *cm,
				       struct cmumble_user *user)
{
	GstElement *pipeline, *sink_bin;
	GError *error = NULL;
	char *desc = "appsrc name=src format=GST_FORMAT_TIME caps="CELT_CAPS" "
		     "! celtdec name=dec "
		     "! audioresample ! audioconvert ! autoaudiosink name=sink";

	pipeline = gst_parse_launch(desc, &error);
	if (error) {
		g_printerr("Failed to create pipeline: %s\n", error->message);
		return -1;
	}

	user->pipeline = pipeline;
	user->src = GST_APP_SRC(gst_bin_get_by_name(GST_BIN(pipeline), "src"));
	add_celt_streamheader(cm, user->src);
	sink_bin = gst_bin_get_by_name(GST_BIN(pipeline), "sink");
	g_signal_connect(G_OBJECT(sink_bin), "element-added",
			 G_CALLBACK(on_sink_bin_element_added), user);

	gst_element_set_state(pipeline, GST_STATE_PLAYING);
	user->last_sequence = -1;

	return 0;
}

void
cmumble_audio_push(struct cmumble *cm, struct cmumble_user *user,
		   const guint8 *data, gsize size, gint64 sequence)
{
	GstBuffer *gstbuf;
	GstClock *clock;
	GstClockTime time = 0;
	GstClockTime base, now = 0;

	if (user->src == NULL)
		if (cmumble_audio_create_playback_pipeline(cm, user) < 0)
			return;

	if (cm->verbose)
		g_print("%s: sequence: %ld\n", __func__, sequence);

	base = gst_element_get_base_time(GST_ELEMENT(user->src));

	clock = gst_element_get_clock(GST_ELEMENT(user->src));
	if (clock) {
		now = gst_clock_get_time(clock);
		g_object_unref(clock);
	}

	/* FIXME: What to do when sequence is a bad value?
	 * e.g to little in value, to be uptodate?
	 *  - just drop?
	 *  - enqueue as now?
	 */

	gstbuf = gst_buffer_new_wrapped(g_memdup(data, size), size);
	GST_BUFFER_FLAG_SET(gstbuf, GST_BUFFER_FLAG_LIVE);

	/* Asume packets are in order, since we're using tcp tunnel only atm.
	 * FIXME: This assumption is probably wrong, since the packets may have
	 *        been received out of order at the server? */

	if (user->last_sequence < 0 || sequence == 0 ||
	    sequence < (user->last_sequence + 1)) {
		GST_BUFFER_FLAG_SET(gstbuf, GST_BUFFER_FLAG_DISCONT);
		GST_BUFFER_FLAG_SET(gstbuf, GST_BUFFER_FLAG_RESYNC);
		time = now - base;
		if (cm->verbose)
			g_print("%s: set time to now\n", __func__);
	} else if (sequence >= user->last_sequence + 1) {
		gint64 num = sequence - (user->last_sequence + 1);
		time = user->last_time_end;
		if (num > 0) {
			time += gst_util_uint64_scale_int(num, GST_SECOND, 100);
			GST_BUFFER_FLAG_SET(gstbuf, GST_BUFFER_FLAG_DISCONT);
		}
		if (cm->verbose)
			g_print("%s: set time by sequence: %lu, now: %lu\n",
				__func__, time, now - base);

	}

	if (time < (now - base)) {
		GST_BUFFER_FLAG_SET(gstbuf, GST_BUFFER_FLAG_DISCONT);
		GST_BUFFER_FLAG_SET(gstbuf, GST_BUFFER_FLAG_RESYNC);
		time = now - base;
		if (cm->verbose)
			g_print("%s: time is in the past, setting to now\n",
				__func__);
	}

	GST_BUFFER_DTS(gstbuf) = now - base;
	GST_BUFFER_PTS(gstbuf) = time;
	GST_BUFFER_DURATION(gstbuf) = BUFFER_TIME;

	user->last_time_end = time + GST_BUFFER_DURATION(gstbuf);
	user->last_sequence = sequence;

	gst_app_src_push_buffer(user->src, gstbuf);
}

static GstFlowReturn
send_queued_celt_buffers(struct cmumble *cm)
{
	uint8_t data[1024];
	uint32_t written = 0, pos = 0;
	mumble_udptunnel_t tunnel;
	GstSample *sample;
	GstBuffer *buf;
	int i;

	if (g_queue_is_empty(cm->audio.buffer_queue))
		return GST_FLOW_ERROR;

	data[pos++] = (udp_voice_celt_alpha << 5) | (udp_normal_talking);
	encode_varint(&data[pos], &written, cm->sequence, sizeof(data)-pos);
	pos += written;

	for (i = 0; !g_queue_is_empty(cm->audio.buffer_queue); ++i) {
		sample = g_queue_pop_head(cm->audio.buffer_queue);
		buf = gst_sample_get_buffer(sample);

		data[pos] = gst_buffer_get_size(buf) & 0x7F;
		if (!g_queue_is_empty(cm->audio.buffer_queue))
			data[pos] |= 0x80;
		pos += 1;
		gst_buffer_extract(buf, 0, &data[pos], gst_buffer_get_size(buf));
		pos += gst_buffer_get_size(buf);
		gst_sample_unref(sample);
	}

	cm->sequence += i;

	cmumble_init_udptunnel(&tunnel);
	tunnel.packet.data = data;
	tunnel.packet.len = pos;
	cmumble_send_udptunnel(cm, &tunnel);

	return GST_FLOW_OK;
}

static GstFlowReturn
pull_buffer(GstAppSink *sink, gpointer user_data)
{
	struct cmumble *cm = user_data;
	GstSample *sample;
	GstBuffer *buf;
	GstClockTime *silence;


	sample = gst_app_sink_pull_sample(cm->audio.sink);
	if (sample == NULL)
		return GST_FLOW_ERROR;

	buf = gst_sample_get_buffer(sample);

	if (GST_BUFFER_FLAG_IS_SET(buf, GST_BUFFER_FLAG_HEADER)) {
		gst_sample_unref(sample);
		return GST_FLOW_OK;
	}

	/* FIXME: Make this more generic/disable pulling
	 * the pipeline completely if not connected?
	 */
	if (cm->con.conn == NULL) {
		gst_sample_unref(sample);
		return GST_FLOW_OK;
	}

	if (gst_buffer_get_size(buf) > 127) {
		g_printerr("error: unexpected buffer size\n");
		gst_sample_unref(sample);
		return GST_FLOW_ERROR;
	}

	g_print("buf: %lu%s\n", GST_BUFFER_PTS(buf),
		GST_BUFFER_FLAG_IS_SET(buf, GST_BUFFER_FLAG_DISCONT) ? " discont" : "");

	if (cm->audio.last_time < GST_BUFFER_PTS(buf) - BUFFER_TIME) {
		if (!g_queue_is_empty(cm->audio.buffer_queue))
			send_queued_celt_buffers(cm);
		cm->sequence = 0;
	}
	cm->audio.last_time = GST_BUFFER_PTS(buf);

	silence = g_queue_peek_head(cm->audio.silence_timestamps);
	//int i = 0;
	//if (silence)
		//g_print("silence %d: %lu, pts: %lu\n", i++, *silence, GST_BUFFER_PTS(buf));
	while (silence && GST_BUFFER_PTS(buf) > *silence) {
		g_queue_remove(cm->audio.silence_timestamps, silence);
		g_free(silence);
		silence = g_queue_peek_head(cm->audio.silence_timestamps);
	}

	g_queue_push_tail(cm->audio.buffer_queue, sample);

	if (silence) {
		//g_print("silence: %lu, pts: %lu\n", *silence, GST_BUFFER_PTS(buf));
	}
	if (silence && *silence == (GST_BUFFER_PTS(buf) + BUFFER_TIME))
		return send_queued_celt_buffers(cm);

	/* FIXME: This should not be hardcoded, but derived from bitrate */
	if (g_queue_get_length(cm->audio.buffer_queue) == 4)
		return send_queued_celt_buffers(cm);

	return GST_FLOW_OK;
}

static gboolean
idle(gpointer user_data)
{
	struct cmumble *cm = user_data;
	GstAppSink *sink;

	sink = cm->audio.sink;
	while ((sink = g_async_queue_try_pop(cm->async_queue)) != NULL)
		pull_buffer(sink, cm);

	return FALSE;
}

static GstFlowReturn
new_sample(GstAppSink *sink, gpointer user_data)
{
	struct cmumble *cm = user_data;

	g_print("new_sample\n");
	g_async_queue_push(cm->async_queue, sink);
	g_idle_add(idle, cm);

	return GST_FLOW_OK;
}

GstAppSinkCallbacks sink_cbs = {
	.new_sample = new_sample
};

static void
handle_cutter_message(struct cmumble *cm, GstMessage *message)
{
	const GstStructure *s;
	gboolean above;
	GstClockTime *time;

	s = gst_message_get_structure(message);
	if (!gst_structure_get_boolean(s, "above", &above))
		return;

	g_print("cutter message. above: %d\n", above);
	/* We are only intrested in below state */
	if (above)
		return;

	time = g_new(GstClockTime, 1);
	if (!time)
		return;

	if (!gst_structure_get_clock_time(s, "timestamp", time))
		return;
	g_print("cutter message. ts: %lu\n", *time);

	send_queued_celt_buffers(cm);
	g_queue_push_tail(cm->audio.silence_timestamps, time);
}

static gboolean
record_pipe_bus_message(GstBus *bus, GstMessage *message, gpointer data)
{
	struct cmumble *cm = data;

	switch (GST_MESSAGE_TYPE(message)) {
	case GST_MESSAGE_ELEMENT:
		if (GST_MESSAGE_SRC(message) == GST_OBJECT(cm->audio.cutter))
			handle_cutter_message(cm, message);
	default:
		break;
	}

	return TRUE;
}

static int
setup_recording_gst_pipeline(struct cmumble *cm)
{
	GstElement *pipeline, *sink;
	GError *error = NULL;
	GstBus *bus;
	char *desc = "autoaudiosrc name=src ! cutter name=cutter ! "
		     "audioresample ! audioconvert ! "AUDIO_CAPS" ! "
		     "celtenc name=enc  " /*perfect-timestamp=true hard-resync=true" */" ! "
		     "appsink name=sink caps="CELT_CAPS;

	pipeline = gst_parse_launch(desc, &error);
	if (error) {
		g_printerr("Failed to create pipeline: %s\n", error->message);
		return -1;
	}
	sink = gst_bin_get_by_name(GST_BIN(pipeline), "sink");
	cm->audio.sink = GST_APP_SINK(sink);
	cm->audio.record_pipeline = pipeline;

	cm->audio.src = gst_bin_get_by_name(GST_BIN(pipeline), "src");

	cm->audio.cutter = gst_bin_get_by_name(GST_BIN(pipeline), "cutter");
	/* FIXME: The threshold should be configurable. */
	g_object_set(G_OBJECT(cm->audio.cutter),
		     "threshold_dB", -45.0, "leaky", TRUE, NULL);

	gst_app_sink_set_callbacks(cm->audio.sink, &sink_cbs, cm, NULL);
	gst_app_sink_set_drop(cm->audio.sink, TRUE);

	bus = gst_pipeline_get_bus(GST_PIPELINE(pipeline));
	cm->audio.bus_watch_id =
		gst_bus_add_watch(bus, record_pipe_bus_message, cm);
	g_object_unref(bus);

	cm->audio.buffer_queue = g_queue_new();
	if (!cm->audio.buffer_queue)
		return -1;
	cm->audio.silence_timestamps = g_queue_new();
	if (!cm->audio.silence_timestamps)
		return -1;

	gst_element_set_state(pipeline, GST_STATE_PLAYING);


	cm->audio.enc = gst_bin_get_by_name(GST_BIN(pipeline), "enc");

	cm->sequence = 0;

	return 0;
}

static int
setup_playback_gst_pipeline(struct cmumble *cm)
{
	cm->audio.celt_mode = celt_mode_create(SAMPLERATE,
					       SAMPLERATE / 100, NULL);
	celt_header_init(&cm->audio.celt_header, cm->audio.celt_mode, CHANNELS);
	celt_header_to_packet(&cm->audio.celt_header,
			      cm->audio.celt_header_packet, sizeof(CELTHeader));

	celt_mode_info(cm->audio.celt_mode, CELT_GET_BITSTREAM_VERSION,
		       &cm->audio.celt_bitstream_version);

	return 0;
}

int
cmumble_audio_init(struct cmumble *cm)
{
	if (setup_playback_gst_pipeline(cm) < 0)
		return -1;

	if (setup_recording_gst_pipeline(cm) < 0)
		return -1;

	return 0;
}

static void
destroy_record_pipeline(struct cmumble *cm)
{
	if (cm->audio.record_pipeline == NULL)
		return;

	g_source_remove(cm->audio.bus_watch_id);
	if (cm->audio.silence_timestamps)
		g_queue_free_full(cm->audio.silence_timestamps, g_free);
	if (cm->audio.silence_timestamps)
		g_queue_free_full(cm->audio.buffer_queue, (GDestroyNotify) gst_sample_unref);

	gst_element_set_state(cm->audio.record_pipeline, GST_STATE_NULL);
	gst_object_unref(GST_OBJECT(cm->audio.record_pipeline));
	gst_object_unref(GST_OBJECT(cm->audio.sink));
	gst_object_unref(GST_OBJECT(cm->audio.src));
	gst_object_unref(GST_OBJECT(cm->audio.cutter));
}

static void
destroy_playback_pipelines(struct cmumble *cm)
{
	struct cmumble_user *user = NULL;
	GList *l;

	for (l = cm->users; l; l = l->next) {
		user = l->data;
		if (user->pipeline == NULL || user->src == NULL)
			continue;
		gst_element_set_state(user->pipeline, GST_STATE_NULL);
		gst_object_unref(GST_OBJECT(user->pipeline));
		user->pipeline = NULL;
		gst_object_unref(GST_OBJECT(user->src));
		user->src = NULL;
	}
}

int
cmumble_audio_fini(struct cmumble *cm)
{
	destroy_record_pipeline(cm);
	destroy_playback_pipelines(cm);

	return 0;
}