avdevice/pulse_audio_dec: reimplement using the non simple API

This fixes timestamps

Based-on: code from pulseaudio
Reviewed-by: Lukasz Marek <lukasz.m.luki2@gmail.com>
Signed-off-by: Michael Niedermayer <michaelni@gmx.at>

2 files changed, 235 insertions, 54 deletions

diff --git a/libavdevice/pulse_audio_dec.c b/libavdevice/pulse_audio_dec.c
index 414d73b..01ff6d1 100644
--- a/libavdevice/pulse_audio_dec.c
+++ b/libavdevice/pulse_audio_dec.c
@@ -1,6 +1,8 @@
 /*
  * Pulseaudio input
  * Copyright (c) 2011 Luca Barbato <lu_zero@gentoo.org>
+ * Copyright 2004-2006 Lennart Poettering
+ * Copyright (c) 2014 Michael Niedermayer <michaelni@gmx.at>
  *
  * This file is part of FFmpeg.
  *
@@ -19,19 +21,14 @@
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  */
 
-/**
- * @file
- * PulseAudio input using the simple API.
- * @author Luca Barbato <lu_zero@gentoo.org>
- */
-
-#include <pulse/simple.h>
 #include <pulse/rtclock.h>
 #include <pulse/error.h>
 #include "libavformat/avformat.h"
 #include "libavformat/internal.h"
 #include "libavutil/opt.h"
+#include "libavutil/time.h"
 #include "pulse_audio_common.h"
+#include "timefilter.h"
 
 #define DEFAULT_CODEC_ID AV_NE(AV_CODEC_ID_PCM_S16BE, AV_CODEC_ID_PCM_S16LE)
 
@@ -44,17 +41,102 @@ typedef struct PulseData {
     int  channels;
     int  frame_size;
     int  fragment_size;
-    pa_simple *s;
-    int64_t pts;
-    int64_t frame_duration;
+
+    pa_threaded_mainloop *mainloop;
+    pa_context *context;
+    pa_stream *stream;
+
+    TimeFilter *timefilter;
+    int last_period;
 } PulseData;
 
+
+#define CHECK_SUCCESS_GOTO(rerror, expression, label)        \
+    do {                                                        \
+        if (!(expression)) {                                    \
+            rerror = AVERROR_EXTERNAL;                          \
+            goto label;                                         \
+        }                                                       \
+    } while(0);
+
+#define CHECK_DEAD_GOTO(p, rerror, label)                               \
+    do {                                                                \
+        if (!(p)->context || !PA_CONTEXT_IS_GOOD(pa_context_get_state((p)->context)) || \
+            !(p)->stream || !PA_STREAM_IS_GOOD(pa_stream_get_state((p)->stream))) { \
+            rerror = AVERROR_EXTERNAL;                                  \
+            goto label;                                                 \
+        }                                                               \
+    } while(0);
+
+static void context_state_cb(pa_context *c, void *userdata) {
+    PulseData *p = userdata;
+
+    switch (pa_context_get_state(c)) {
+        case PA_CONTEXT_READY:
+        case PA_CONTEXT_TERMINATED:
+        case PA_CONTEXT_FAILED:
+            pa_threaded_mainloop_signal(p->mainloop, 0);
+            break;
+    }
+}
+
+static void stream_state_cb(pa_stream *s, void * userdata) {
+    PulseData *p = userdata;
+
+    switch (pa_stream_get_state(s)) {
+        case PA_STREAM_READY:
+        case PA_STREAM_FAILED:
+        case PA_STREAM_TERMINATED:
+            pa_threaded_mainloop_signal(p->mainloop, 0);
+            break;
+    }
+}
+
+static void stream_request_cb(pa_stream *s, size_t length, void *userdata) {
+    PulseData *p = userdata;
+
+    pa_threaded_mainloop_signal(p->mainloop, 0);
+}
+
+static void stream_latency_update_cb(pa_stream *s, void *userdata) {
+    PulseData *p = userdata;
+
+    pa_threaded_mainloop_signal(p->mainloop, 0);
+}
+
+static av_cold int pulse_close(AVFormatContext *s)
+{
+    PulseData *pd = s->priv_data;
+
+    if (pd->mainloop)
+        pa_threaded_mainloop_stop(pd->mainloop);
+
+    if (pd->stream)
+        pa_stream_unref(pd->stream);
+    pd->stream = NULL;
+
+    if (pd->context) {
+        pa_context_disconnect(pd->context);
+        pa_context_unref(pd->context);
+    }
+    pd->context = NULL;
+
+    if (pd->mainloop)
+        pa_threaded_mainloop_free(pd->mainloop);
+    pd->mainloop = NULL;
+
+    ff_timefilter_destroy(pd->timefilter);
+    pd->timefilter = NULL;
+
+    return 0;
+}
+
 static av_cold int pulse_read_header(AVFormatContext *s)
 {
     PulseData *pd = s->priv_data;
     AVStream *st;
     char *device = NULL;
-    int ret, sample_bytes;
+    int ret;
     enum AVCodecID codec_id =
         s->audio_codec_id == AV_CODEC_ID_NONE ? DEFAULT_CODEC_ID : s->audio_codec_id;
     const pa_sample_spec ss = { ff_codec_id_to_pulse_format(codec_id),
@@ -75,77 +157,176 @@ static av_cold int pulse_read_header(AVFormatContext *s)
     if (strcmp(s->filename, "default"))
         device = s->filename;
 
-    pd->s = pa_simple_new(pd->server, pd->name,
-                          PA_STREAM_RECORD,
-                          device, pd->stream_name, &ss,
-                          NULL, &attr, &ret);
+    if (!(pd->mainloop = pa_threaded_mainloop_new())) {
+        pulse_close(s);
+        return AVERROR_EXTERNAL;
+    }
+
+    if (!(pd->context = pa_context_new(pa_threaded_mainloop_get_api(pd->mainloop), pd->name))) {
+        pulse_close(s);
+        return AVERROR_EXTERNAL;
+    }
+
+    pa_context_set_state_callback(pd->context, context_state_cb, pd);
+
+    if (pa_context_connect(pd->context, pd->server, 0, NULL) < 0) {
+        pulse_close(s);
+        return AVERROR(pa_context_errno(pd->context));
+    }
+
+    pa_threaded_mainloop_lock(pd->mainloop);
 
-    if (!pd->s) {
-        av_log(s, AV_LOG_ERROR, "pa_simple_new failed: %s\n",
-               pa_strerror(ret));
-        return AVERROR(EIO);
+    if (pa_threaded_mainloop_start(pd->mainloop) < 0) {
+        ret = -1;
+        goto unlock_and_fail;
     }
+
+    for (;;) {
+        pa_context_state_t state;
+
+        state = pa_context_get_state(pd->context);
+
+        if (state == PA_CONTEXT_READY)
+            break;
+
+        if (!PA_CONTEXT_IS_GOOD(state)) {
+            ret = AVERROR(pa_context_errno(pd->context));
+            goto unlock_and_fail;
+        }
+
+        /* Wait until the context is ready */
+        pa_threaded_mainloop_wait(pd->mainloop);
+    }
+
+    if (!(pd->stream = pa_stream_new(pd->context, pd->stream_name, &ss, NULL))) {
+        ret = AVERROR(pa_context_errno(pd->context));
+        goto unlock_and_fail;
+    }
+
+    pa_stream_set_state_callback(pd->stream, stream_state_cb, pd);
+    pa_stream_set_read_callback(pd->stream, stream_request_cb, pd);
+    pa_stream_set_write_callback(pd->stream, stream_request_cb, pd);
+    pa_stream_set_latency_update_callback(pd->stream, stream_latency_update_cb, pd);
+
+    ret = pa_stream_connect_record(pd->stream, device, &attr,
+                                    PA_STREAM_INTERPOLATE_TIMING
+                                    |PA_STREAM_ADJUST_LATENCY
+                                    |PA_STREAM_AUTO_TIMING_UPDATE);
+
+    if (ret < 0) {
+        ret = AVERROR(pa_context_errno(pd->context));
+        goto unlock_and_fail;
+    }
+
+    for (;;) {
+        pa_stream_state_t state;
+
+        state = pa_stream_get_state(pd->stream);
+
+        if (state == PA_STREAM_READY)
+            break;
+
+        if (!PA_STREAM_IS_GOOD(state)) {
+            ret = AVERROR(pa_context_errno(pd->context));
+            goto unlock_and_fail;
+        }
+
+        /* Wait until the stream is ready */
+        pa_threaded_mainloop_wait(pd->mainloop);
+    }
+
+    pa_threaded_mainloop_unlock(pd->mainloop);
+
     /* take real parameters */
     st->codec->codec_type  = AVMEDIA_TYPE_AUDIO;
     st->codec->codec_id    = codec_id;
     st->codec->sample_rate = pd->sample_rate;
     st->codec->channels    = pd->channels;
-    avpriv_set_pts_info(st, 64, 1, pd->sample_rate);  /* 64 bits pts in us */
+    avpriv_set_pts_info(st, 64, 1, 1000000);  /* 64 bits pts in us */
 
-    pd->pts = AV_NOPTS_VALUE;
-    sample_bytes = (av_get_bits_per_sample(codec_id) >> 3) * pd->channels;
+    pd->timefilter = ff_timefilter_new(1000000.0 / pd->sample_rate,
+                                       1000, 1.5E-6);
 
-    if (pd->frame_size % sample_bytes) {
-        av_log(s, AV_LOG_WARNING, "frame_size %i is not divisible by %i "
-            "(channels * bytes_per_sample) \n", pd->frame_size, sample_bytes);
+    if (!pd->timefilter) {
+        pulse_close(s);
+        return AVERROR(ENOMEM);
     }
 
-    pd->frame_duration = pd->frame_size / sample_bytes;
-
     return 0;
+
+unlock_and_fail:
+    pa_threaded_mainloop_unlock(pd->mainloop);
+
+    pulse_close(s);
+    return ret;
 }
 
 static int pulse_read_packet(AVFormatContext *s, AVPacket *pkt)
 {
     PulseData *pd  = s->priv_data;
-    int res;
+    int ret;
+    size_t read_length;
+    const void *read_data = NULL;
+    int64_t dts;
+    pa_usec_t latency;
+    int negative;
 
-    if (av_new_packet(pkt, pd->frame_size) < 0) {
-        return AVERROR(ENOMEM);
-    }
+    pa_threaded_mainloop_lock(pd->mainloop);
 
-    if ((pa_simple_read(pd->s, pkt->data, pkt->size, &res)) < 0) {
-        av_log(s, AV_LOG_ERROR, "pa_simple_read failed: %s\n",
-               pa_strerror(res));
-        av_free_packet(pkt);
-        return AVERROR(EIO);
-    }
+    CHECK_DEAD_GOTO(pd, ret, unlock_and_fail);
 
-    if (pd->pts == AV_NOPTS_VALUE) {
-        pa_usec_t latency;
+    while (!read_data) {
+        int r;
 
-        if ((latency = pa_simple_get_latency(pd->s, &res)) == (pa_usec_t) -1) {
-            av_log(s, AV_LOG_ERROR, "pa_simple_get_latency() failed: %s\n",
-                   pa_strerror(res));
-            return AVERROR(EIO);
+        r = pa_stream_peek(pd->stream, &read_data, &read_length);
+        CHECK_SUCCESS_GOTO(ret, r == 0, unlock_and_fail);
+
+        if (read_length <= 0) {
+            pa_threaded_mainloop_wait(pd->mainloop);
+            CHECK_DEAD_GOTO(pd, ret, unlock_and_fail);
+        } else if (!read_data) {
+            /* There's a hole in the stream, skip it. We could generate
+                * silence, but that wouldn't work for compressed streams. */
+            r = pa_stream_drop(pd->stream);
+            CHECK_SUCCESS_GOTO(ret, r == 0, unlock_and_fail);
         }
+    }
 
-        pd->pts = -latency;
+    if (av_new_packet(pkt, read_length) < 0) {
+        ret = AVERROR(ENOMEM);
+        goto unlock_and_fail;
     }
 
-    pkt->pts = pd->pts;
+    dts = av_gettime();
+    pa_operation_unref(pa_stream_update_timing_info(pd->stream, NULL, NULL));
 
-    pd->pts += pd->frame_duration;
+    if (pa_stream_get_latency(pd->stream, &latency, &negative) >= 0) {
+        enum AVCodecID codec_id =
+            s->audio_codec_id == AV_CODEC_ID_NONE ? DEFAULT_CODEC_ID : s->audio_codec_id;
+        int frame_size = ((av_get_bits_per_sample(codec_id) >> 3) * pd->channels);
+        int frame_duration = read_length / frame_size;
 
-    return 0;
-}
 
-static av_cold int pulse_close(AVFormatContext *s)
-{
-    PulseData *pd = s->priv_data;
-    pa_simple_free(pd->s);
-    pd->s = NULL;
+        if (negative) {
+            dts += latency;
+        } else
+            dts -= latency;
+        pkt->pts = ff_timefilter_update(pd->timefilter, dts, pd->last_period);
+
+        pd->last_period = frame_duration;
+    } else {
+        av_log(s, AV_LOG_WARNING, "pa_stream_get_latency() failed\n");
+    }
+
+    memcpy(pkt->data, read_data, read_length);
+    pa_stream_drop(pd->stream);
+
+    pa_threaded_mainloop_unlock(pd->mainloop);
     return 0;
+
+unlock_and_fail:
+    pa_threaded_mainloop_unlock(pd->mainloop);
+    return ret;
 }
 
 static int pulse_get_device_list(AVFormatContext *h, AVDeviceInfoList *device_list)
diff --git a/libavdevice/version.h b/libavdevice/version.h
index 163a4c6..a098549 100644
--- a/libavdevice/version.h
+++ b/libavdevice/version.h
@@ -29,7 +29,7 @@
 
 #define LIBAVDEVICE_VERSION_MAJOR  55
 #define LIBAVDEVICE_VERSION_MINOR  13
-#define LIBAVDEVICE_VERSION_MICRO 101
+#define LIBAVDEVICE_VERSION_MICRO 102
 
 #define LIBAVDEVICE_VERSION_INT AV_VERSION_INT(LIBAVDEVICE_VERSION_MAJOR, \
                                                LIBAVDEVICE_VERSION_MINOR, \