From 74f8456acb6e419f01a325655b18f8a45083e26f Mon Sep 17 00:00:00 2001 From: Arun Raghavan Date: Thu, 12 May 2016 19:26:55 +0530 Subject: 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 --- configure.ac | 18 ++ meson.build | 10 + meson_options.txt | 3 + po/POTFILES.in | 4 +- src/Makefile.am | 14 +- src/modules/rtp/meson.build | 10 +- src/modules/rtp/module-rtp-recv.c | 2 +- src/modules/rtp/module-rtp-send.c | 2 +- src/modules/rtp/rtp-common.c | 97 ++++++++ src/modules/rtp/rtp-gstreamer.c | 480 ++++++++++++++++++++++++++++++++++++++ src/modules/rtp/rtp-native.c | 400 +++++++++++++++++++++++++++++++ src/modules/rtp/rtp.c | 471 ------------------------------------- src/modules/rtp/rtp.h | 4 +- 13 files changed, 1034 insertions(+), 481 deletions(-) create mode 100644 src/modules/rtp/rtp-common.c create mode 100644 src/modules/rtp/rtp-gstreamer.c create mode 100644 src/modules/rtp/rtp-native.c delete mode 100644 src/modules/rtp/rtp.c diff --git a/configure.ac b/configure.ac index 8278353d4..21571a05c 100644 --- a/configure.ac +++ b/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 diff --git a/meson.build b/meson.build index 59cba00cb..e24eccc28 100644 --- a/meson.build +++ b/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()), diff --git a/meson_options.txt b/meson_options.txt index 491653e7a..817889271 100644 --- a/meson_options.txt +++ b/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') diff --git a/po/POTFILES.in b/po/POTFILES.in index 0b519464a..33f7ae322 100644 --- a/po/POTFILES.in +++ b/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 diff --git a/src/Makefile.am b/src/Makefile.am index b84c778cc..e3baf587a 100644 --- a/src/Makefile.am +++ b/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 diff --git a/src/modules/rtp/meson.build b/src/modules/rtp/meson.build index c3efde6ab..119cf08ce 100644 --- a/src/modules/rtp/meson.build +++ b/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, diff --git a/src/modules/rtp/module-rtp-recv.c b/src/modules/rtp/module-rtp-recv.c index 5733dbae1..a9b42bbc5 100644 --- a/src/modules/rtp/module-rtp-recv.c +++ b/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); diff --git a/src/modules/rtp/module-rtp-send.c b/src/modules/rtp/module-rtp-send.c index e647a9d3b..5a4c6fc06 100644 --- a/src/modules/rtp/module-rtp-send.c +++ b/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); diff --git a/src/modules/rtp/rtp-common.c b/src/modules/rtp/rtp-common.c new file mode 100644 index 000000000..65e2c7acd --- /dev/null +++ b/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 . +***/ + +#ifdef HAVE_CONFIG_H +#include +#endif + +#include "rtp.h" + +#include + +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; +} diff --git a/src/modules/rtp/rtp-gstreamer.c b/src/modules/rtp/rtp-gstreamer.c new file mode 100644 index 000000000..bebd1c80a --- /dev/null +++ b/src/modules/rtp/rtp-gstreamer.c @@ -0,0 +1,480 @@ +/*** + This file is part of PulseAudio. + + Copyright 2016 Arun Raghavan + + 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 . +***/ + +#ifdef HAVE_CONFIG_H +#include +#endif + +#include +#include +#include + +#include "rtp.h" + +#include + +#include +#include +#include +#include + +#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); +} diff --git a/src/modules/rtp/rtp-native.c b/src/modules/rtp/rtp-native.c new file mode 100644 index 000000000..af2bf9fc6 --- /dev/null +++ b/src/modules/rtp/rtp-native.c @@ -0,0 +1,400 @@ +/*** + 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 . +***/ + +#ifdef HAVE_CONFIG_H +#include +#endif + +#include +#include +#include +#include +#include + +#ifdef HAVE_SYS_FILIO_H +#include +#endif + +#ifdef HAVE_SYS_UIO_H +#include +#endif + +#include +#include +#include +#include +#include +#include + +#include "rtp.h" + +typedef struct pa_rtp_context { + int fd; + uint16_t sequence; + uint32_t timestamp; + uint32_t ssrc; + uint8_t payload; + size_t frame_size; + size_t mtu; + + uint8_t *recv_buf; + size_t recv_buf_size; + pa_memchunk memchunk; +} pa_rtp_context; + +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); + + c = pa_xnew0(pa_rtp_context, 1); + + c->fd = fd; + c->sequence = (uint16_t) (rand()*rand()); + c->timestamp = 0; + c->ssrc = (uint32_t) (rand()*rand()); + c->payload = (uint8_t) (payload & 127U); + c->frame_size = pa_frame_size(ss); + c->mtu = mtu; + + c->recv_buf = NULL; + c->recv_buf_size = 0; + pa_memchunk_reset(&c->memchunk); + + return c; +} + +#define MAX_IOVECS 16 + +int pa_rtp_send(pa_rtp_context *c, pa_memblockq *q) { + struct iovec iov[MAX_IOVECS]; + pa_memblock* mb[MAX_IOVECS]; + int iov_idx = 1; + size_t n = 0; + + pa_assert(c); + pa_assert(q); + + if (pa_memblockq_get_length(q) < c->mtu) + return 0; + + for (;;) { + int r; + pa_memchunk chunk; + + pa_memchunk_reset(&chunk); + + if ((r = pa_memblockq_peek(q, &chunk)) >= 0) { + + size_t k = n + chunk.length > c->mtu ? c->mtu - n : chunk.length; + + pa_assert(chunk.memblock); + + iov[iov_idx].iov_base = pa_memblock_acquire_chunk(&chunk); + iov[iov_idx].iov_len = k; + mb[iov_idx] = chunk.memblock; + iov_idx ++; + + n += k; + pa_memblockq_drop(q, k); + } + + pa_assert(n % c->frame_size == 0); + + if (r < 0 || n >= c->mtu || iov_idx >= MAX_IOVECS) { + uint32_t header[3]; + struct msghdr m; + ssize_t k; + int i; + + if (n > 0) { + header[0] = htonl(((uint32_t) 2 << 30) | ((uint32_t) c->payload << 16) | ((uint32_t) c->sequence)); + header[1] = htonl(c->timestamp); + header[2] = htonl(c->ssrc); + + iov[0].iov_base = (void*)header; + iov[0].iov_len = sizeof(header); + + m.msg_name = NULL; + m.msg_namelen = 0; + m.msg_iov = iov; + m.msg_iovlen = (size_t) iov_idx; + m.msg_control = NULL; + m.msg_controllen = 0; + m.msg_flags = 0; + + k = sendmsg(c->fd, &m, MSG_DONTWAIT); + + for (i = 1; i < iov_idx; i++) { + pa_memblock_release(mb[i]); + pa_memblock_unref(mb[i]); + } + + c->sequence++; + } else + k = 0; + + c->timestamp += (unsigned) (n/c->frame_size); + + if (k < 0) { + if (errno != EAGAIN && errno != EINTR) /* If the queue is full, just ignore it */ + pa_log("sendmsg() failed: %s", pa_cstrerror(errno)); + return -1; + } + + if (r < 0 || pa_memblockq_get_length(q) < c->mtu) + break; + + n = 0; + iov_idx = 1; + } + } + + return 0; +} + +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 = pa_frame_size(ss); + + c->recv_buf_size = 2000; + c->recv_buf = pa_xmalloc(c->recv_buf_size); + pa_memchunk_reset(&c->memchunk); + + return c; +} + +int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool, uint32_t *rtp_tstamp, struct timeval *tstamp) { + int size; + size_t audio_length; + size_t metadata_length; + struct msghdr m; + struct cmsghdr *cm; + struct iovec iov; + uint32_t header; + uint32_t ssrc; + uint8_t payload; + unsigned cc; + ssize_t r; + uint8_t aux[1024]; + bool found_tstamp = false; + + pa_assert(c); + pa_assert(chunk); + + pa_memchunk_reset(chunk); + + if (ioctl(c->fd, FIONREAD, &size) < 0) { + pa_log_warn("FIONREAD failed: %s", pa_cstrerror(errno)); + goto fail; + } + + if (size <= 0) { + /* size can be 0 due to any of the following reasons: + * + * 1. Somebody sent us a perfectly valid zero-length UDP packet. + * 2. Somebody sent us a UDP packet with a bad CRC. + * + * It is unknown whether size can actually be less than zero. + * + * In the first case, the packet has to be read out, otherwise the + * kernel will tell us again and again about it, thus preventing + * reception of any further packets. So let's just read it out + * now and discard it later, when comparing the number of bytes + * received (0) with the number of bytes wanted (1, see below). + * + * In the second case, recvmsg() will fail, thus allowing us to + * return the error. + * + * Just to avoid passing zero-sized memchunks and NULL pointers to + * recvmsg(), let's force allocation of at least one byte by setting + * size to 1. + */ + size = 1; + } + + if (c->recv_buf_size < (size_t) size) { + do + c->recv_buf_size *= 2; + while (c->recv_buf_size < (size_t) size); + + c->recv_buf = pa_xrealloc(c->recv_buf, c->recv_buf_size); + } + + pa_assert(c->recv_buf_size >= (size_t) size); + + iov.iov_base = c->recv_buf; + iov.iov_len = (size_t) size; + + m.msg_name = NULL; + m.msg_namelen = 0; + m.msg_iov = &iov; + m.msg_iovlen = 1; + m.msg_control = aux; + m.msg_controllen = sizeof(aux); + m.msg_flags = 0; + + r = recvmsg(c->fd, &m, 0); + + if (r != size) { + if (r < 0 && errno != EAGAIN && errno != EINTR) + pa_log_warn("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch"); + + goto fail; + } + + if (size < 12) { + pa_log_warn("RTP packet too short."); + goto fail; + } + + memcpy(&header, iov.iov_base, sizeof(uint32_t)); + memcpy(rtp_tstamp, (uint8_t*) iov.iov_base + 4, sizeof(uint32_t)); + memcpy(&ssrc, (uint8_t*) iov.iov_base + 8, sizeof(uint32_t)); + + header = ntohl(header); + *rtp_tstamp = ntohl(*rtp_tstamp); + ssrc = ntohl(c->ssrc); + + if ((header >> 30) != 2) { + pa_log_warn("Unsupported RTP version."); + goto fail; + } + + if ((header >> 29) & 1) { + pa_log_warn("RTP padding not supported."); + goto fail; + } + + if ((header >> 28) & 1) { + pa_log_warn("RTP header extensions not supported."); + goto fail; + } + + if (ssrc != c->ssrc) { + pa_log_debug("Got unexpected SSRC"); + goto fail; + } + + cc = (header >> 24) & 0xF; + payload = (uint8_t) ((header >> 16) & 127U); + c->sequence = (uint16_t) (header & 0xFFFFU); + + metadata_length = 12 + cc * 4; + + if (payload != c->payload) { + pa_log_debug("Got unexpected payload: %u", payload); + goto fail; + } + + if (metadata_length > (unsigned) size) { + pa_log_warn("RTP packet too short. (CSRC)"); + goto fail; + } + + audio_length = size - metadata_length; + + if (audio_length % c->frame_size != 0) { + pa_log_warn("Bad RTP packet size."); + goto fail; + } + + if (c->memchunk.length < (unsigned) audio_length) { + size_t l; + + if (c->memchunk.memblock) + pa_memblock_unref(c->memchunk.memblock); + + l = PA_MAX((size_t) audio_length, pa_mempool_block_size_max(pool)); + + c->memchunk.memblock = pa_memblock_new(pool, l); + c->memchunk.index = 0; + c->memchunk.length = pa_memblock_get_length(c->memchunk.memblock); + } + + memcpy(pa_memblock_acquire_chunk(&c->memchunk), c->recv_buf + metadata_length, audio_length); + pa_memblock_release(c->memchunk.memblock); + + chunk->memblock = pa_memblock_ref(c->memchunk.memblock); + chunk->index = c->memchunk.index; + chunk->length = audio_length; + + c->memchunk.index += audio_length; + c->memchunk.length -= audio_length; + + if (c->memchunk.length <= 0) { + pa_memblock_unref(c->memchunk.memblock); + pa_memchunk_reset(&c->memchunk); + } + + for (cm = CMSG_FIRSTHDR(&m); cm; cm = CMSG_NXTHDR(&m, cm)) + if (cm->cmsg_level == SOL_SOCKET && cm->cmsg_type == SCM_TIMESTAMP) { + memcpy(tstamp, CMSG_DATA(cm), sizeof(struct timeval)); + found_tstamp = true; + break; + } + + if (!found_tstamp) { + pa_log_warn("Couldn't find SCM_TIMESTAMP data in auxiliary recvmsg() data!"); + pa_zero(*tstamp); + } + + return 0; + +fail: + if (chunk->memblock) + pa_memblock_unref(chunk->memblock); + + return -1; +} + +void pa_rtp_context_free(pa_rtp_context *c) { + pa_assert(c); + + pa_assert_se(pa_close(c->fd) == 0); + + if (c->memchunk.memblock) + pa_memblock_unref(c->memchunk.memblock); + + pa_xfree(c->recv_buf); + pa_xfree(c); +} + +size_t pa_rtp_context_get_frame_size(pa_rtp_context *c) { + return c->frame_size; +} + +pa_rtpoll_item* pa_rtp_context_get_rtpoll_item(pa_rtp_context *c, pa_rtpoll *rtpoll) { + pa_rtpoll_item *item; + struct pollfd *p; + + item = pa_rtpoll_item_new(rtpoll, PA_RTPOLL_LATE, 1); + + p = pa_rtpoll_item_get_pollfd(item, NULL); + p->fd = c->fd; + p->events = POLLIN; + p->revents = 0; + + return item; +} diff --git a/src/modules/rtp/rtp.c b/src/modules/rtp/rtp.c deleted file mode 100644 index 5a066d92b..000000000 --- a/src/modules/rtp/rtp.c +++ /dev/null @@ -1,471 +0,0 @@ -/*** - 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 . -***/ - -#ifdef HAVE_CONFIG_H -#include -#endif - -#include -#include -#include -#include -#include - -#ifdef HAVE_SYS_FILIO_H -#include -#endif - -#ifdef HAVE_SYS_UIO_H -#include -#endif - -#include -#include -#include -#include -#include -#include - -#include "rtp.h" - -typedef struct pa_rtp_context { - int fd; - uint16_t sequence; - uint32_t timestamp; - uint32_t ssrc; - uint8_t payload; - size_t frame_size; - size_t mtu; - - uint8_t *recv_buf; - size_t recv_buf_size; - 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 *c; - - pa_assert(fd >= 0); - - c = pa_xnew0(pa_rtp_context, 1); - - c->fd = fd; - c->sequence = (uint16_t) (rand()*rand()); - c->timestamp = 0; - c->ssrc = (uint32_t) (rand()*rand()); - c->payload = (uint8_t) (payload & 127U); - c->frame_size = frame_size; - c->mtu = mtu; - - c->recv_buf = NULL; - c->recv_buf_size = 0; - pa_memchunk_reset(&c->memchunk); - - return c; -} - -#define MAX_IOVECS 16 - -int pa_rtp_send(pa_rtp_context *c, pa_memblockq *q) { - struct iovec iov[MAX_IOVECS]; - pa_memblock* mb[MAX_IOVECS]; - int iov_idx = 1; - size_t n = 0; - - pa_assert(c); - pa_assert(q); - - if (pa_memblockq_get_length(q) < c->mtu) - return 0; - - for (;;) { - int r; - pa_memchunk chunk; - - pa_memchunk_reset(&chunk); - - if ((r = pa_memblockq_peek(q, &chunk)) >= 0) { - - size_t k = n + chunk.length > c->mtu ? c->mtu - n : chunk.length; - - pa_assert(chunk.memblock); - - iov[iov_idx].iov_base = pa_memblock_acquire_chunk(&chunk); - iov[iov_idx].iov_len = k; - mb[iov_idx] = chunk.memblock; - iov_idx ++; - - n += k; - pa_memblockq_drop(q, k); - } - - pa_assert(n % c->frame_size == 0); - - if (r < 0 || n >= c->mtu || iov_idx >= MAX_IOVECS) { - uint32_t header[3]; - struct msghdr m; - ssize_t k; - int i; - - if (n > 0) { - header[0] = htonl(((uint32_t) 2 << 30) | ((uint32_t) c->payload << 16) | ((uint32_t) c->sequence)); - header[1] = htonl(c->timestamp); - header[2] = htonl(c->ssrc); - - iov[0].iov_base = (void*)header; - iov[0].iov_len = sizeof(header); - - m.msg_name = NULL; - m.msg_namelen = 0; - m.msg_iov = iov; - m.msg_iovlen = (size_t) iov_idx; - m.msg_control = NULL; - m.msg_controllen = 0; - m.msg_flags = 0; - - k = sendmsg(c->fd, &m, MSG_DONTWAIT); - - for (i = 1; i < iov_idx; i++) { - pa_memblock_release(mb[i]); - pa_memblock_unref(mb[i]); - } - - c->sequence++; - } else - k = 0; - - c->timestamp += (unsigned) (n/c->frame_size); - - if (k < 0) { - if (errno != EAGAIN && errno != EINTR) /* If the queue is full, just ignore it */ - pa_log("sendmsg() failed: %s", pa_cstrerror(errno)); - return -1; - } - - if (r < 0 || pa_memblockq_get_length(q) < c->mtu) - break; - - n = 0; - iov_idx = 1; - } - } - - return 0; -} - -pa_rtp_context* pa_rtp_context_new_recv(int fd, uint8_t payload, size_t frame_size) { - pa_rtp_context *c; - - c = pa_xnew0(pa_rtp_context, 1); - - c->fd = fd; - c->payload = payload; - c->frame_size = frame_size; - - c->recv_buf_size = 2000; - c->recv_buf = pa_xmalloc(c->recv_buf_size); - pa_memchunk_reset(&c->memchunk); - - return c; -} - -int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool, uint32_t *rtp_tstamp, struct timeval *tstamp) { - int size; - size_t audio_length; - size_t metadata_length; - struct msghdr m; - struct cmsghdr *cm; - struct iovec iov; - uint32_t header; - uint32_t ssrc; - uint8_t payload; - unsigned cc; - ssize_t r; - uint8_t aux[1024]; - bool found_tstamp = false; - - pa_assert(c); - pa_assert(chunk); - - pa_memchunk_reset(chunk); - - if (ioctl(c->fd, FIONREAD, &size) < 0) { - pa_log_warn("FIONREAD failed: %s", pa_cstrerror(errno)); - goto fail; - } - - if (size <= 0) { - /* size can be 0 due to any of the following reasons: - * - * 1. Somebody sent us a perfectly valid zero-length UDP packet. - * 2. Somebody sent us a UDP packet with a bad CRC. - * - * It is unknown whether size can actually be less than zero. - * - * In the first case, the packet has to be read out, otherwise the - * kernel will tell us again and again about it, thus preventing - * reception of any further packets. So let's just read it out - * now and discard it later, when comparing the number of bytes - * received (0) with the number of bytes wanted (1, see below). - * - * In the second case, recvmsg() will fail, thus allowing us to - * return the error. - * - * Just to avoid passing zero-sized memchunks and NULL pointers to - * recvmsg(), let's force allocation of at least one byte by setting - * size to 1. - */ - size = 1; - } - - if (c->recv_buf_size < (size_t) size) { - do - c->recv_buf_size *= 2; - while (c->recv_buf_size < (size_t) size); - - c->recv_buf = pa_xrealloc(c->recv_buf, c->recv_buf_size); - } - - pa_assert(c->recv_buf_size >= (size_t) size); - - iov.iov_base = c->recv_buf; - iov.iov_len = (size_t) size; - - m.msg_name = NULL; - m.msg_namelen = 0; - m.msg_iov = &iov; - m.msg_iovlen = 1; - m.msg_control = aux; - m.msg_controllen = sizeof(aux); - m.msg_flags = 0; - - r = recvmsg(c->fd, &m, 0); - - if (r != size) { - if (r < 0 && errno != EAGAIN && errno != EINTR) - pa_log_warn("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch"); - - goto fail; - } - - if (size < 12) { - pa_log_warn("RTP packet too short."); - goto fail; - } - - memcpy(&header, iov.iov_base, sizeof(uint32_t)); - memcpy(rtp_tstamp, (uint8_t*) iov.iov_base + 4, sizeof(uint32_t)); - memcpy(&ssrc, (uint8_t*) iov.iov_base + 8, sizeof(uint32_t)); - - header = ntohl(header); - *rtp_tstamp = ntohl(*rtp_tstamp); - ssrc = ntohl(c->ssrc); - - if ((header >> 30) != 2) { - pa_log_warn("Unsupported RTP version."); - goto fail; - } - - if ((header >> 29) & 1) { - pa_log_warn("RTP padding not supported."); - goto fail; - } - - if ((header >> 28) & 1) { - pa_log_warn("RTP header extensions not supported."); - goto fail; - } - - if (ssrc != c->ssrc) { - pa_log_debug("Got unexpected SSRC"); - goto fail; - } - - cc = (header >> 24) & 0xF; - payload = (uint8_t) ((header >> 16) & 127U); - c->sequence = (uint16_t) (header & 0xFFFFU); - - metadata_length = 12 + cc * 4; - - if (payload != c->payload) { - pa_log_debug("Got unexpected payload: %u", payload); - goto fail; - } - - if (metadata_length > (unsigned) size) { - pa_log_warn("RTP packet too short. (CSRC)"); - goto fail; - } - - audio_length = size - metadata_length; - - if (audio_length % c->frame_size != 0) { - pa_log_warn("Bad RTP packet size."); - goto fail; - } - - if (c->memchunk.length < (unsigned) audio_length) { - size_t l; - - if (c->memchunk.memblock) - pa_memblock_unref(c->memchunk.memblock); - - l = PA_MAX((size_t) audio_length, pa_mempool_block_size_max(pool)); - - c->memchunk.memblock = pa_memblock_new(pool, l); - c->memchunk.index = 0; - c->memchunk.length = pa_memblock_get_length(c->memchunk.memblock); - } - - memcpy(pa_memblock_acquire_chunk(&c->memchunk), c->recv_buf + metadata_length, audio_length); - pa_memblock_release(c->memchunk.memblock); - - chunk->memblock = pa_memblock_ref(c->memchunk.memblock); - chunk->index = c->memchunk.index; - chunk->length = audio_length; - - c->memchunk.index += audio_length; - c->memchunk.length -= audio_length; - - if (c->memchunk.length <= 0) { - pa_memblock_unref(c->memchunk.memblock); - pa_memchunk_reset(&c->memchunk); - } - - for (cm = CMSG_FIRSTHDR(&m); cm; cm = CMSG_NXTHDR(&m, cm)) - if (cm->cmsg_level == SOL_SOCKET && cm->cmsg_type == SCM_TIMESTAMP) { - memcpy(tstamp, CMSG_DATA(cm), sizeof(struct timeval)); - found_tstamp = true; - break; - } - - if (!found_tstamp) { - pa_log_warn("Couldn't find SCM_TIMESTAMP data in auxiliary recvmsg() data!"); - pa_zero(*tstamp); - } - - return 0; - -fail: - if (chunk->memblock) - pa_memblock_unref(chunk->memblock); - - 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); - - pa_assert_se(pa_close(c->fd) == 0); - - if (c->memchunk.memblock) - pa_memblock_unref(c->memchunk.memblock); - - pa_xfree(c->recv_buf); - 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; -} - -pa_rtpoll_item* pa_rtp_context_get_rtpoll_item(pa_rtp_context *c, pa_rtpoll *rtpoll) { - pa_rtpoll_item *item; - struct pollfd *p; - - item = pa_rtpoll_item_new(rtpoll, PA_RTPOLL_LATE, 1); - - p = pa_rtpoll_item_get_pollfd(item, NULL); - p->fd = c->fd; - p->events = POLLIN; - p->revents = 0; - - return item; -} diff --git a/src/modules/rtp/rtp.h b/src/modules/rtp/rtp.h index e3146ec07..372df75be 100644 --- a/src/modules/rtp/rtp.h +++ b/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); -- cgit v1.2.3