rtp: Add a GStreamer-based RTP implementation

This adds a GStreamer-based RTP implementation to replace our own. The
original implementation is retained for cases where it is not possible
to include GStreamer as a dependency.

The idea with this is to be able to start supporting more advanced RTP
features such as RTCP, non-PCM audio, and potentially synchronised
playback.

Signed-off-by: Arun Raghavan <arun@arunraghavan.net>

configure.ac

@@ -1310,6 +1310,22 @@ AC_SUBST(HAVE_SYSTEMD_JOURNAL)
 AM_CONDITIONAL([HAVE_SYSTEMD_JOURNAL], [test "x$HAVE_SYSTEMD_JOURNAL" = x1])
 AS_IF([test "x$HAVE_SYSTEMD_JOURNAL" = "x1"], AC_DEFINE([HAVE_SYSTEMD_JOURNAL], 1, [Have SYSTEMDJOURNAL?]))
 
+#### GStreamer-based RTP support (optional) ####
+
+AC_ARG_ENABLE([gstreamer],
+    AS_HELP_STRING([--disable-gstreamer],[Disable optional GStreamer-based RTP support]))
+
+AS_IF([test "x$enable_gstreamer" != "xno"],
+    [PKG_CHECK_MODULES(GSTREAMER, [ gstreamer-1.0 gstreamer-app-1.0 gstreamer-rtp-1.0 gio-2.0 ],
+                       HAVE_GSTREAMER=1, HAVE_GSTREAMER=0)],
+    HAVE_GSTREAMER=0)
+
+AS_IF([test "x$enable_gstreamer" = "xyes" && test "x$HAVE_GSTREAMER" = "x0"],
+    [AC_MSG_ERROR([*** GStreamer 1.0 support not found])])
+
+AM_CONDITIONAL([HAVE_GSTREAMER], [test "x$HAVE_GSTREAMER" = x1])
+AS_IF([test "x$HAVE_GSTREAMER" = "x1"], AC_DEFINE([HAVE_GSTREAMER], 1, [Have GStreamer?]))
+
 #### Build and Install man pages ####
 
 AC_ARG_ENABLE([manpages],
@@ -1614,6 +1630,7 @@ AS_IF([test "x$HAVE_ADRIAN_EC" = "x1"], ENABLE_ADRIAN_EC=yes, ENABLE_ADRIAN_EC=n
 AS_IF([test "x$HAVE_SPEEX" = "x1"], ENABLE_SPEEX=yes, ENABLE_SPEEX=no)
 AS_IF([test "x$HAVE_SOXR" = "x1"], ENABLE_SOXR=yes, ENABLE_SOXR=no)
 AS_IF([test "x$HAVE_WEBRTC" = "x1"], ENABLE_WEBRTC=yes, ENABLE_WEBRTC=no)
+AS_IF([test "x$HAVE_GSTREAMER" = "x1"], ENABLE_GSTREAMER=yes, ENABLE_GSTREAMER=no)
 AS_IF([test "x$HAVE_TDB" = "x1"], ENABLE_TDB=yes, ENABLE_TDB=no)
 AS_IF([test "x$HAVE_GDBM" = "x1"], ENABLE_GDBM=yes, ENABLE_GDBM=no)
 AS_IF([test "x$HAVE_SIMPLEDB" = "x1"], ENABLE_SIMPLEDB=yes, ENABLE_SIMPLEDB=no)
@@ -1677,6 +1694,7 @@ echo "
     Enable speex (resampler, AEC): ${ENABLE_SPEEX}
     Enable soxr (resampler):       ${ENABLE_SOXR}
     Enable WebRTC echo canceller:  ${ENABLE_WEBRTC}
+    Enable GStreamer-based RTP:    ${ENABLE_GSTREAMER}
     Enable gcov coverage:          ${ENABLE_GCOV}
     Enable unit tests:             ${ENABLE_TESTS}
     Database

meson.build

@@ -669,6 +669,15 @@ if webrtc_dep.found()
   cdata.set('HAVE_WEBRTC', 1)
 endif
 
+gst_dep = dependency('gstreamer-1.0', required : get_option('gstreamer'))
+gstapp_dep = dependency('gstreamer-app-1.0', required : get_option('gstreamer'))
+gstrtp_dep = dependency('gstreamer-rtp-1.0', required : get_option('gstreamer'))
+
+have_gstreamer = false
+if gst_dep.found() and gstapp_dep.found() and gstrtp_dep.found()
+  have_gstreamer = true
+endif
+
 # These are required for the CMake file generation
 cdata.set('PA_LIBDIR', libdir)
 cdata.set('PA_INCDIR', includedir)
@@ -815,6 +824,7 @@ summary = [
   'Enable OpenSSL (for Airtunes): @0@'.format(openssl_dep.found()),
   'Enable FFTW:                   @0@'.format(fftw_dep.found()),
   'Enable ORC:                    @0@'.format(have_orcc),
+  'Enable GStreamer:              @0@'.format(have_gstreamer),
   'Enable Adrian echo canceller:  @0@'.format(get_option('adrian-aec')),
   'Enable Speex (resampler, AEC): @0@'.format(speex_dep.found()),
   'Enable SoXR (resampler):       @0@'.format(soxr_dep.found()),

meson_options.txt

@@ -93,6 +93,9 @@ option('glib',
 option('gsettings',
        type : 'feature', value : 'auto',
        description : 'Optional GSettings support')
+option('gstreamer',
+       type : 'feature', value : 'auto',
+       description : 'Optional GStreamer dependency for media-related functionality')
 option('gtk',
        type : 'feature', value : 'auto',
        description : 'Optional Gtk+ 3 support')

po/POTFILES.in

@@ -66,7 +66,9 @@ src/modules/raop/raop-sink.c
 src/modules/reserve-wrap.c
 src/modules/rtp/module-rtp-recv.c
 src/modules/rtp/module-rtp-send.c
-src/modules/rtp/rtp.c
+src/modules/rtp/rtp-common.c
+src/modules/rtp/rtp-native.c
+src/modules/rtp/rtp-gstreamer.c
 src/modules/rtp/sap.c
 src/modules/rtp/sdp.c
 src/modules/x11/module-x11-bell.c

src/Makefile.am

@@ -1176,13 +1176,21 @@ libprotocol_esound_la_LIBADD = $(AM_LIBADD) libpulsecore-@PA_MAJORMINOR@.la libp
 endif
 
 librtp_la_SOURCES = \
-		modules/rtp/rtp.c modules/rtp/rtp.h \
+		modules/rtp/rtp-common.c modules/rtp/rtp.h \
 		modules/rtp/sdp.c modules/rtp/sdp.h \
 		modules/rtp/sap.c modules/rtp/sap.h \
 		modules/rtp/rtsp_client.c modules/rtp/rtsp_client.h \
 		modules/rtp/headerlist.c modules/rtp/headerlist.h
+librtp_la_CFLAGS = $(AM_CFLAGS)
 librtp_la_LDFLAGS = $(AM_LDFLAGS) $(AM_LIBLDFLAGS) -avoid-version
 librtp_la_LIBADD = $(AM_LIBADD) libpulsecore-@PA_MAJORMINOR@.la libpulsecommon-@PA_MAJORMINOR@.la libpulse.la
+if HAVE_GSTREAMER
+librtp_la_SOURCES += modules/rtp/rtp-gstreamer.c
+librtp_la_CFLAGS += $(GSTREAMER_CFLAGS)
+librtp_la_LIBADD += $(GSTREAMER_LIBS)
+else
+librtp_la_SOURCES += modules/rtp/rtp-native.c
+endif
 
 libraop_la_SOURCES = \
         modules/raop/raop-util.c modules/raop/raop-util.h \
@@ -2049,12 +2057,12 @@ endif
 module_rtp_send_la_SOURCES = modules/rtp/module-rtp-send.c
 module_rtp_send_la_LDFLAGS = $(MODULE_LDFLAGS)
 module_rtp_send_la_LIBADD = $(MODULE_LIBADD) librtp.la
-module_rtp_send_la_CFLAGS = $(AM_CFLAGS) -DPA_MODULE_NAME=module_rtp_send
+module_rtp_send_la_CFLAGS = $(AM_CFLAGS) $(GSTREAMER_CFLAGS) -DPA_MODULE_NAME=module_rtp_send
 
 module_rtp_recv_la_SOURCES = modules/rtp/module-rtp-recv.c
 module_rtp_recv_la_LDFLAGS = $(MODULE_LDFLAGS)
 module_rtp_recv_la_LIBADD = $(MODULE_LIBADD) librtp.la
-module_rtp_recv_la_CFLAGS = $(AM_CFLAGS) -DPA_MODULE_NAME=module_rtp_recv
+module_rtp_recv_la_CFLAGS = $(AM_CFLAGS) $(GSTREAMER_CFLAGS) -DPA_MODULE_NAME=module_rtp_recv
 
 # JACK
 

src/modules/rtp/meson.build

@@ -1,5 +1,5 @@
 librtp_sources = [
-  'rtp.c',
+  'rtp-common.c',
   'sdp.c',
   'sap.c',
   'rtsp_client.c',
@@ -14,13 +14,19 @@ librtp_headers = [
   'headerlist.h',
 ]
 
+if have_gstreamer
+  librtp_sources += 'rtp-gstreamer.c'
+else
+  librtp_sources += 'rtp-native.c'
+endif
+
 librtp = shared_library('rtp',
   librtp_sources,
   librtp_headers,
   c_args : [pa_c_args, server_c_args],
   link_args : [nodelete_link_args],
   include_directories : [configinc, topinc],
-  dependencies : [libpulse_dep, libpulsecommon_dep, libpulsecore_dep, libatomic_ops_dep],
+  dependencies : [libpulse_dep, libpulsecommon_dep, libpulsecore_dep, libatomic_ops_dep, gst_dep, gstapp_dep, gstrtp_dep, gio_dep],
   install : true,
   install_rpath : privlibdir,
   install_dir : modlibexecdir,

src/modules/rtp/module-rtp-recv.c

@@ -568,7 +568,7 @@ static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_in
 
     pa_memblock_unref(silence.memblock);
 
-    if (!(s->rtp_context = pa_rtp_context_new_recv(fd, sdp_info->payload, pa_frame_size(&s->sdp_info.sample_spec))))
+    if (!(s->rtp_context = pa_rtp_context_new_recv(fd, sdp_info->payload, &s->sdp_info.sample_spec)))
         goto fail;
 
     pa_hashmap_put(s->userdata->by_origin, s->sdp_info.origin, s);

src/modules/rtp/module-rtp-send.c

@@ -488,7 +488,7 @@ int pa__init(pa_module*m) {
 
     pa_xfree(n);
 
-    if (!(u->rtp_context = pa_rtp_context_new_send(fd, payload, mtu, pa_frame_size(&ss))))
+    if (!(u->rtp_context = pa_rtp_context_new_send(fd, payload, mtu, &ss)))
         goto fail;
     pa_sap_context_init_send(&u->sap_context, sap_fd, p);
 

src/modules/rtp/rtp-common.c

@@ -0,0 +1,97 @@
+/***
+  This file is part of PulseAudio.
+
+  Copyright 2006 Lennart Poettering
+
+  PulseAudio is free software; you can redistribute it and/or modify
+  it under the terms of the GNU Lesser General Public License as published
+  by the Free Software Foundation; either version 2.1 of the License,
+  or (at your option) any later version.
+
+  PulseAudio is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+  General Public License for more details.
+
+  You should have received a copy of the GNU Lesser General Public License
+  along with PulseAudio; if not, see <http://www.gnu.org/licenses/>.
+***/
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include "rtp.h"
+
+#include <pulsecore/core-util.h>
+
+uint8_t pa_rtp_payload_from_sample_spec(const pa_sample_spec *ss) {
+    pa_assert(ss);
+
+    if (ss->format == PA_SAMPLE_S16BE && ss->rate == 44100 && ss->channels == 2)
+        return 10;
+    if (ss->format == PA_SAMPLE_S16BE && ss->rate == 44100 && ss->channels == 1)
+        return 11;
+
+    return 127;
+}
+
+pa_sample_spec *pa_rtp_sample_spec_from_payload(uint8_t payload, pa_sample_spec *ss) {
+    pa_assert(ss);
+
+    switch (payload) {
+        case 10:
+            ss->channels = 2;
+            ss->format = PA_SAMPLE_S16BE;
+            ss->rate = 44100;
+            break;
+
+        case 11:
+            ss->channels = 1;
+            ss->format = PA_SAMPLE_S16BE;
+            ss->rate = 44100;
+            break;
+
+        default:
+            return NULL;
+    }
+
+    return ss;
+}
+
+pa_sample_spec *pa_rtp_sample_spec_fixup(pa_sample_spec * ss) {
+    pa_assert(ss);
+
+    if (!pa_rtp_sample_spec_valid(ss))
+        ss->format = PA_SAMPLE_S16BE;
+
+    pa_assert(pa_rtp_sample_spec_valid(ss));
+    return ss;
+}
+
+int pa_rtp_sample_spec_valid(const pa_sample_spec *ss) {
+    pa_assert(ss);
+
+    if (!pa_sample_spec_valid(ss))
+        return 0;
+
+    return ss->format == PA_SAMPLE_S16BE;
+}
+
+const char* pa_rtp_format_to_string(pa_sample_format_t f) {
+    switch (f) {
+        case PA_SAMPLE_S16BE:
+            return "L16";
+        default:
+            return NULL;
+    }
+}
+
+pa_sample_format_t pa_rtp_string_to_format(const char *s) {
+    pa_assert(s);
+
+    if (pa_streq(s, "L16"))
+        return PA_SAMPLE_S16BE;
+    else
+        return PA_SAMPLE_INVALID;
+}

src/modules/rtp/rtp-gstreamer.c

@@ -0,0 +1,480 @@
+/***
+  This file is part of PulseAudio.
+
+  Copyright 2016 Arun Raghavan <mail@arunraghavan.net>
+
+  PulseAudio is free software; you can redistribute it and/or modify
+  it under the terms of the GNU Lesser General Public License as published
+  by the Free Software Foundation; either version 2.1 of the License,
+  or (at your option) any later version.
+
+  PulseAudio is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+  General Public License for more details.
+
+  You should have received a copy of the GNU Lesser General Public License
+  along with PulseAudio; if not, see <http://www.gnu.org/licenses/>.
+***/
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include <pulse/timeval.h>
+#include <pulsecore/fdsem.h>
+#include <pulsecore/core-rtclock.h>
+
+#include "rtp.h"
+
+#include <gio/gio.h>
+
+#include <gst/gst.h>
+#include <gst/app/gstappsrc.h>
+#include <gst/app/gstappsink.h>
+#include <gst/rtp/gstrtpbuffer.h>
+
+#define MAKE_ELEMENT_NAMED(v, e, n)                     \
+    v = gst_element_factory_make(e, n);                 \
+    if (!v) {                                           \
+        pa_log("Could not create %s element", e);       \
+        goto fail;                                      \
+    }
+
+#define MAKE_ELEMENT(v, e) MAKE_ELEMENT_NAMED((v), (e), NULL)
+
+struct pa_rtp_context {
+    pa_fdsem *fdsem;
+    pa_sample_spec ss;
+
+    GstElement *pipeline;
+    GstElement *appsrc;
+    GstElement *appsink;
+
+    uint32_t last_timestamp;
+};
+
+static GstCaps* caps_from_sample_spec(const pa_sample_spec *ss) {
+    if (ss->format != PA_SAMPLE_S16BE)
+        return NULL;
+
+    return gst_caps_new_simple("audio/x-raw",
+            "format", G_TYPE_STRING, "S16BE",
+            "rate", G_TYPE_INT, (int) ss->rate,
+            "channels", G_TYPE_INT, (int) ss->channels,
+            "layout", G_TYPE_STRING, "interleaved",
+            NULL);
+}
+static bool init_send_pipeline(pa_rtp_context *c, int fd, uint8_t payload, size_t mtu, const pa_sample_spec *ss) {
+    GstElement *appsrc = NULL, *pay = NULL, *capsf = NULL, *rtpbin = NULL, *sink = NULL;
+    GstCaps *caps;
+
+    MAKE_ELEMENT(appsrc, "appsrc");
+    MAKE_ELEMENT(pay, "rtpL16pay");
+    MAKE_ELEMENT(capsf, "capsfilter");
+    MAKE_ELEMENT(rtpbin, "rtpbin");
+    MAKE_ELEMENT(sink, "fdsink");
+
+    c->pipeline = gst_pipeline_new(NULL);
+
+    gst_bin_add_many(GST_BIN(c->pipeline), appsrc, pay, capsf, rtpbin, sink, NULL);
+
+    caps = caps_from_sample_spec(ss);
+    if (!caps) {
+        pa_log("Unsupported format to payload");
+        goto fail;
+    }
+
+    g_object_set(appsrc, "caps", caps, "is-live", TRUE, "blocksize", mtu, "format", 3 /* time */, NULL);
+    g_object_set(pay, "mtu", mtu, NULL);
+    g_object_set(sink, "fd", fd, "enable-last-sample", FALSE, NULL);
+
+    gst_caps_unref(caps);
+
+    /* Force the payload type that we want */
+    caps = gst_caps_new_simple("application/x-rtp", "payload", G_TYPE_INT, (int) payload, NULL);
+    g_object_set(capsf, "caps", caps, NULL);
+    gst_caps_unref(caps);
+
+    if (!gst_element_link(appsrc, pay) ||
+        !gst_element_link(pay, capsf) ||
+        !gst_element_link_pads(capsf, "src", rtpbin, "send_rtp_sink_0") ||
+        !gst_element_link_pads(rtpbin, "send_rtp_src_0", sink, "sink")) {
+
+        pa_log("Could not set up send pipeline");
+        goto fail;
+    }
+
+    if (gst_element_set_state(c->pipeline, GST_STATE_PLAYING) == GST_STATE_CHANGE_FAILURE) {
+        pa_log("Could not start pipeline");
+        goto fail;
+    }
+
+    c->appsrc = gst_object_ref(appsrc);
+
+    return true;
+
+fail:
+    if (c->pipeline) {
+        gst_object_unref(c->pipeline);
+    } else {
+        /* These weren't yet added to pipeline, so we still have a ref */
+        if (appsrc)
+            gst_object_unref(appsrc);
+        if (pay)
+            gst_object_unref(pay);
+        if (capsf)
+            gst_object_unref(capsf);
+        if (rtpbin)
+            gst_object_unref(rtpbin);
+        if (sink)
+            gst_object_unref(sink);
+    }
+
+    return false;
+}
+
+pa_rtp_context* pa_rtp_context_new_send(int fd, uint8_t payload, size_t mtu, const pa_sample_spec *ss) {
+    pa_rtp_context *c = NULL;
+    GError *error = NULL;
+
+    pa_assert(fd >= 0);
+
+    c = pa_xnew0(pa_rtp_context, 1);
+
+    c->ss = *ss;
+
+    if (!gst_init_check(NULL, NULL, &error)) {
+        pa_log_error("Could not initialise GStreamer: %s", error->message);
+        g_error_free(error);
+        goto fail;
+    }
+
+    if (!init_send_pipeline(c, fd, payload, mtu, ss))
+        goto fail;
+
+    return c;
+
+fail:
+    pa_rtp_context_free(c);
+    return NULL;
+}
+
+/* Called from I/O thread context */
+static bool process_bus_messages(pa_rtp_context *c) {
+    GstBus *bus;
+    GstMessage *message;
+    bool ret = true;
+
+    bus = gst_pipeline_get_bus(GST_PIPELINE(c->pipeline));
+
+    while (ret && (message = gst_bus_pop(bus))) {
+        if (GST_MESSAGE_TYPE(message) == GST_MESSAGE_ERROR) {
+            GError *error = NULL;
+
+            ret = false;
+
+            gst_message_parse_error(message, &error, NULL);
+            pa_log("Got an error: %s", error->message);
+
+            g_error_free(error);
+        }
+
+        gst_message_unref(message);
+    }
+
+    gst_object_unref(bus);
+
+    return ret;
+}
+
+static void free_buffer(pa_memblock *memblock) {
+    pa_memblock_release(memblock);
+    pa_memblock_unref(memblock);
+}
+
+/* Called from I/O thread context */
+int pa_rtp_send(pa_rtp_context *c, pa_memblockq *q) {
+    pa_memchunk chunk = { 0, };
+    GstBuffer *buf;
+    void *data;
+    bool stop = false;
+    int ret = 0;
+
+    pa_assert(c);
+    pa_assert(q);
+
+    if (!process_bus_messages(c))
+        return -1;
+
+    while (!stop && pa_memblockq_peek(q, &chunk) == 0) {
+        pa_assert(chunk.memblock);
+
+        data = pa_memblock_acquire(chunk.memblock);
+
+        buf = gst_buffer_new_wrapped_full(GST_MEMORY_FLAG_READONLY | GST_MEMORY_FLAG_PHYSICALLY_CONTIGUOUS,
+                                          data, chunk.length, chunk.index, chunk.length, chunk.memblock,
+                                          (GDestroyNotify) free_buffer);
+
+        if (gst_app_src_push_buffer(GST_APP_SRC(c->appsrc), buf) != GST_FLOW_OK) {
+            pa_log_error("Could not push buffer");
+            stop = true;
+            ret = -1;
+        }
+
+        pa_memblockq_drop(q, chunk.length);
+    }
+
+    return ret;
+}
+
+static GstCaps* rtp_caps_from_sample_spec(const pa_sample_spec *ss) {
+    if (ss->format != PA_SAMPLE_S16BE)
+        return NULL;
+
+    return gst_caps_new_simple("application/x-rtp",
+            "media", G_TYPE_STRING, "audio",
+            "encoding-name", G_TYPE_STRING, "L16",
+            "clock-rate", G_TYPE_INT, (int) ss->rate,
+            "payload", G_TYPE_INT, (int) pa_rtp_payload_from_sample_spec(ss),
+            "layout", G_TYPE_STRING, "interleaved",
+            NULL);
+}
+
+static void on_pad_added(GstElement *element, GstPad *pad, gpointer userdata) {
+    pa_rtp_context *c = (pa_rtp_context *) userdata;
+    GstElement *depay;
+    GstPad *sinkpad;
+    GstPadLinkReturn ret;
+
+    depay = gst_bin_get_by_name(GST_BIN(c->pipeline), "depay");
+    pa_assert(depay);
+
+    sinkpad = gst_element_get_static_pad(depay, "sink");
+
+    ret = gst_pad_link(pad, sinkpad);
+    if (ret != GST_PAD_LINK_OK) {
+        GstBus *bus;
+        GError *error;
+
+        bus = gst_pipeline_get_bus(GST_PIPELINE(c->pipeline));
+        error = g_error_new(GST_CORE_ERROR, GST_CORE_ERROR_PAD, "Could not link rtpbin to depayloader");
+        gst_bus_post(bus, gst_message_new_error(GST_OBJECT(c->pipeline), error, NULL));
+
+        /* Actually cause the I/O thread to wake up and process the error */
+        pa_fdsem_post(c->fdsem);
+
+        g_error_free(error);
+        gst_object_unref(bus);
+    }
+
+    gst_object_unref(sinkpad);
+    gst_object_unref(depay);
+}
+
+static bool init_receive_pipeline(pa_rtp_context *c, int fd, const pa_sample_spec *ss) {
+    GstElement *udpsrc = NULL, *rtpbin = NULL, *depay = NULL, *appsink = NULL;
+    GstCaps *caps;
+    GSocket *socket;
+    GError *error = NULL;
+
+    MAKE_ELEMENT(udpsrc, "udpsrc");
+    MAKE_ELEMENT(rtpbin, "rtpbin");
+    MAKE_ELEMENT_NAMED(depay, "rtpL16depay", "depay");
+    MAKE_ELEMENT(appsink, "appsink");
+
+    c->pipeline = gst_pipeline_new(NULL);
+
+    gst_bin_add_many(GST_BIN(c->pipeline), udpsrc, rtpbin, depay, appsink, NULL);
+
+    socket = g_socket_new_from_fd(fd, &error);
+    if (error) {
+        pa_log("Could not create socket: %s", error->message);
+        g_error_free(error);
+        goto fail;
+    }
+
+    caps = rtp_caps_from_sample_spec(ss);
+    if (!caps) {
+        pa_log("Unsupported format to payload");
+        goto fail;
+    }
+
+    g_object_set(udpsrc, "socket", socket, "caps", caps, "auto-multicast" /* caller handles this */, FALSE, NULL);
+    g_object_set(rtpbin, "latency", 0, "buffer-mode", 0 /* none */, NULL);
+    g_object_set(appsink, "sync", FALSE, "enable-last-sample", FALSE, NULL);
+
+    gst_caps_unref(caps);
+    g_object_unref(socket);
+
+    if (!gst_element_link_pads(udpsrc, "src", rtpbin, "recv_rtp_sink_0") ||
+        !gst_element_link(depay, appsink)) {
+
+        pa_log("Could not set up receive pipeline");
+        goto fail;
+    }
+
+    g_signal_connect(G_OBJECT(rtpbin), "pad-added", G_CALLBACK(on_pad_added), c);
+
+    if (gst_element_set_state(c->pipeline, GST_STATE_PLAYING) == GST_STATE_CHANGE_FAILURE) {
+        pa_log("Could not start pipeline");
+        goto fail;
+    }
+
+    c->appsink = gst_object_ref(appsink);
+
+    return true;
+
+fail:
+    if (c->pipeline) {
+        gst_object_unref(c->pipeline);
+    } else {
+        /* These weren't yet added to pipeline, so we still have a ref */
+        if (udpsrc)
+            gst_object_unref(udpsrc);
+        if (depay)
+            gst_object_unref(depay);
+        if (rtpbin)
+            gst_object_unref(rtpbin);
+        if (appsink)
+            gst_object_unref(appsink);
+    }
+
+    return false;
+}
+
+/* Called from the GStreamer streaming thread */
+static void appsink_eos(GstAppSink *appsink, gpointer userdata) {
+    pa_rtp_context *c = (pa_rtp_context *) userdata;
+
+    pa_fdsem_post(c->fdsem);
+}
+
+/* Called from the GStreamer streaming thread */
+static GstFlowReturn appsink_new_sample(GstAppSink *appsink, gpointer userdata) {
+    pa_rtp_context *c = (pa_rtp_context *) userdata;
+
+    pa_fdsem_post(c->fdsem);
+
+    return GST_FLOW_OK;
+}
+
+pa_rtp_context* pa_rtp_context_new_recv(int fd, uint8_t payload, const pa_sample_spec *ss) {
+    pa_rtp_context *c = NULL;
+    GstAppSinkCallbacks callbacks = { 0, };
+    GError *error = NULL;
+
+    pa_assert(fd >= 0);
+
+    c = pa_xnew0(pa_rtp_context, 1);
+
+    c->fdsem = pa_fdsem_new();
+    c->ss = *ss;
+
+    if (!gst_init_check(NULL, NULL, &error)) {
+        pa_log_error("Could not initialise GStreamer: %s", error->message);
+        g_error_free(error);
+        goto fail;
+    }
+
+    if (!init_receive_pipeline(c, fd, ss))
+        goto fail;
+
+    callbacks.eos = appsink_eos;
+    callbacks.new_sample = appsink_new_sample;
+    gst_app_sink_set_callbacks(GST_APP_SINK(c->appsink), &callbacks, c, NULL);
+
+    return c;
+
+fail:
+    pa_rtp_context_free(c);
+    return NULL;
+}
+
+/* Called from I/O thread context */
+int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool, uint32_t *rtp_tstamp, struct timeval *tstamp) {
+    GstSample *sample = NULL;
+    GstBuffer *buf;
+    GstMapInfo info;
+    void *data;
+
+    if (!process_bus_messages(c))
+        goto fail;
+
+    sample = gst_app_sink_pull_sample(GST_APP_SINK(c->appsink));
+    if (!sample) {
+        pa_log_warn("Could not get any more data");
+        goto fail;
+    }
+
+    buf = gst_sample_get_buffer(sample);
+
+    if (GST_BUFFER_IS_DISCONT(buf))
+        pa_log_info("Discontinuity detected, possibly lost some packets");
+
+    if (!gst_buffer_map(buf, &info, GST_MAP_READ))
+        goto fail;
+
+    pa_assert(pa_mempool_block_size_max(pool) >= info.size);
+
+    chunk->memblock = pa_memblock_new(pool, info.size);
+    chunk->index = 0;
+    chunk->length = info.size;
+
+    data = pa_memblock_acquire_chunk(chunk);
+    /* TODO: we could probably just provide an allocator and avoid a memcpy */
+    memcpy(data, info.data, info.size);
+    pa_memblock_release(chunk->memblock);
+
+    /* When buffer-mode = none, the buffer PTS is the RTP timestamp, converted
+     * to time units (instead of clock-rate units as is in the header) and
+     * wraparound-corrected, and the DTS is the pipeline clock timestamp from
+     * when the buffer was acquired at the source (this is actually the running
+     * time which is why we need to add base time). */
+    *rtp_tstamp = gst_util_uint64_scale_int(GST_BUFFER_PTS(buf), c->ss.rate, GST_SECOND) & 0xFFFFFFFFU;
+    pa_timeval_rtstore(tstamp, (GST_BUFFER_DTS(buf) + gst_element_get_base_time(c->pipeline)) / GST_USECOND, false);
+
+    gst_buffer_unmap(buf, &info);
+    gst_sample_unref(sample);
+
+    return 0;
+
+fail:
+    if (sample)
+        gst_sample_unref(sample);
+
+    if (chunk->memblock)
+        pa_memblock_unref(chunk->memblock);
+
+    return -1;
+}
+
+void pa_rtp_context_free(pa_rtp_context *c) {
+    pa_assert(c);
+
+    if (c->appsrc) {
+        gst_app_src_end_of_stream(GST_APP_SRC(c->appsrc));
+        gst_object_unref(c->appsrc);
+    }
+
+    if (c->appsink)
+        gst_object_unref(c->appsink);
+
+    if (c->pipeline) {
+        gst_element_set_state(c->pipeline, GST_STATE_NULL);
+        gst_object_unref(c->pipeline);
+    }
+
+    if (c->fdsem)
+        pa_fdsem_free(c->fdsem);
+
+    pa_xfree(c);
+}
+
+pa_rtpoll_item* pa_rtp_context_get_rtpoll_item(pa_rtp_context *c, pa_rtpoll *rtpoll) {
+    return pa_rtpoll_item_new_fdsem(rtpoll, PA_RTPOLL_LATE, c->fdsem);
+}
+
+size_t pa_rtp_context_get_frame_size(pa_rtp_context *c) {
+    return pa_frame_size(&c->ss);
+}

src/modules/rtp/rtp-native.c

@@ -58,7 +58,7 @@ typedef struct pa_rtp_context {
     pa_memchunk memchunk;
 } pa_rtp_context;
 
-pa_rtp_context* pa_rtp_context_new_send(int fd, uint8_t payload, size_t mtu, size_t frame_size) {
+pa_rtp_context* pa_rtp_context_new_send(int fd, uint8_t payload, size_t mtu, const pa_sample_spec *ss) {
     pa_rtp_context *c;
 
     pa_assert(fd >= 0);
@@ -70,7 +70,7 @@ pa_rtp_context* pa_rtp_context_new_send(int fd, uint8_t payload, size_t mtu, siz
     c->timestamp = 0;
     c->ssrc = (uint32_t) (rand()*rand());
     c->payload = (uint8_t) (payload & 127U);
-    c->frame_size = frame_size;
+    c->frame_size = pa_frame_size(ss);
     c->mtu = mtu;
 
     c->recv_buf = NULL;
@@ -169,14 +169,14 @@ int pa_rtp_send(pa_rtp_context *c, pa_memblockq *q) {
     return 0;
 }
 
-pa_rtp_context* pa_rtp_context_new_recv(int fd, uint8_t payload, size_t frame_size) {
+pa_rtp_context* pa_rtp_context_new_recv(int fd, uint8_t payload, const pa_sample_spec *ss) {
     pa_rtp_context *c;
 
     c = pa_xnew0(pa_rtp_context, 1);
 
     c->fd = fd;
     c->payload = payload;
-    c->frame_size = frame_size;
+    c->frame_size = pa_frame_size(ss);
 
     c->recv_buf_size = 2000;
     c->recv_buf = pa_xmalloc(c->recv_buf_size);
@@ -369,59 +369,6 @@ fail:
     return -1;
 }
 
-uint8_t pa_rtp_payload_from_sample_spec(const pa_sample_spec *ss) {
-    pa_assert(ss);
-
-    if (ss->format == PA_SAMPLE_S16BE && ss->rate == 44100 && ss->channels == 2)
-        return 10;
-    if (ss->format == PA_SAMPLE_S16BE && ss->rate == 44100 && ss->channels == 1)
-        return 11;
-
-    return 127;
-}
-
-pa_sample_spec *pa_rtp_sample_spec_from_payload(uint8_t payload, pa_sample_spec *ss) {
-    pa_assert(ss);
-
-    switch (payload) {
-        case 10:
-            ss->channels = 2;
-            ss->format = PA_SAMPLE_S16BE;
-            ss->rate = 44100;
-            break;
-
-        case 11:
-            ss->channels = 1;
-            ss->format = PA_SAMPLE_S16BE;
-            ss->rate = 44100;
-            break;
-
-        default:
-            return NULL;
-    }
-
-    return ss;
-}
-
-pa_sample_spec *pa_rtp_sample_spec_fixup(pa_sample_spec * ss) {
-    pa_assert(ss);
-
-    if (!pa_rtp_sample_spec_valid(ss))
-        ss->format = PA_SAMPLE_S16BE;
-
-    pa_assert(pa_rtp_sample_spec_valid(ss));
-    return ss;
-}
-
-int pa_rtp_sample_spec_valid(const pa_sample_spec *ss) {
-    pa_assert(ss);
-
-    if (!pa_sample_spec_valid(ss))
-        return 0;
-
-    return ss->format == PA_SAMPLE_S16BE;
-}
-
 void pa_rtp_context_free(pa_rtp_context *c) {
     pa_assert(c);
 
@@ -434,24 +381,6 @@ void pa_rtp_context_free(pa_rtp_context *c) {
     pa_xfree(c);
 }
 
-const char* pa_rtp_format_to_string(pa_sample_format_t f) {
-    switch (f) {
-        case PA_SAMPLE_S16BE:
-            return "L16";
-        default:
-            return NULL;
-    }
-}
-
-pa_sample_format_t pa_rtp_string_to_format(const char *s) {
-    pa_assert(s);
-
-    if (pa_streq(s, "L16"))
-        return PA_SAMPLE_S16BE;
-    else
-        return PA_SAMPLE_INVALID;
-}
-
 size_t pa_rtp_context_get_frame_size(pa_rtp_context *c) {
     return c->frame_size;
 }

src/modules/rtp/rtp.h

@@ -30,13 +30,13 @@
 typedef struct pa_rtp_context pa_rtp_context;
 
 int pa_rtp_context_init_send(pa_rtp_context *c, int fd, uint8_t payload, size_t mtu, size_t frame_size);
-pa_rtp_context* pa_rtp_context_new_send(int fd, uint8_t payload, size_t mtu, size_t frame_size);
+pa_rtp_context* pa_rtp_context_new_send(int fd, uint8_t payload, size_t mtu, const pa_sample_spec *ss);
 
 /* If the memblockq doesn't have a silence memchunk set, then the caller must
  * guarantee that the current read index doesn't point to a hole. */
 int pa_rtp_send(pa_rtp_context *c, pa_memblockq *q);
 
-pa_rtp_context* pa_rtp_context_new_recv(int fd, uint8_t payload, size_t frame_size);
+pa_rtp_context* pa_rtp_context_new_recv(int fd, uint8_t payload, const pa_sample_spec *ss);
 int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool, uint32_t *rtp_tstamp, struct timeval *tstamp);
 
 void pa_rtp_context_free(pa_rtp_context *c);