diff options
Diffstat (limited to 'src/modules/rtp/rtp-gstreamer.c')
| -rw-r--r-- | src/modules/rtp/rtp-gstreamer.c | 665 |
1 files changed, 665 insertions, 0 deletions
diff --git a/src/modules/rtp/rtp-gstreamer.c b/src/modules/rtp/rtp-gstreamer.c new file mode 100644 index 0000000..28d367b --- /dev/null +++ b/src/modules/rtp/rtp-gstreamer.c @@ -0,0 +1,665 @@ +/*** + 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/base/gstadapter.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) +#define RTP_HEADER_SIZE 12 + +struct pa_rtp_context { + pa_fdsem *fdsem; + pa_sample_spec ss; + + GstElement *pipeline; + GstElement *appsrc; + GstElement *appsink; + GstCaps *meta_reference; + + bool first_buffer; + uint32_t last_timestamp; + + uint8_t *send_buf; + size_t mtu; +}; + +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; + GSocket *socket; + GInetSocketAddress *addr; + GInetAddress *iaddr; + guint16 port; + gchar *addr_str; + + MAKE_ELEMENT(appsrc, "appsrc"); + MAKE_ELEMENT(pay, "rtpL16pay"); + MAKE_ELEMENT(capsf, "capsfilter"); + MAKE_ELEMENT(rtpbin, "rtpbin"); + MAKE_ELEMENT(sink, "udpsink"); + + 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; + } + + socket = g_socket_new_from_fd(fd, NULL); + if (!socket) { + pa_log("Failed to create socket"); + goto fail; + } + + addr = G_INET_SOCKET_ADDRESS(g_socket_get_remote_address(socket, NULL)); + iaddr = g_inet_socket_address_get_address(addr); + addr_str = g_inet_address_to_string(iaddr); + port = g_inet_socket_address_get_port(addr); + + 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, "socket", socket, "host", addr_str, "port", port, + "enable-last-sample", FALSE, "sync", FALSE, "loop", + g_socket_get_multicast_loopback(socket), "ttl", + g_socket_get_ttl(socket), "ttl-mc", + g_socket_get_multicast_ttl(socket), "auto-multicast", FALSE, + NULL); + + g_free(addr_str); + g_object_unref(addr); + g_object_unref(socket); + + 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); + + pa_log_info("Initialising GStreamer RTP backend for send"); + + c = pa_xnew0(pa_rtp_context, 1); + + c->ss = *ss; + c->mtu = mtu - RTP_HEADER_SIZE; + c->send_buf = pa_xmalloc(c->mtu); + + 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; +} + +/* Called from I/O thread context */ +int pa_rtp_send(pa_rtp_context *c, pa_memblockq *q) { + GstBuffer *buf; + size_t n = 0; + + pa_assert(c); + pa_assert(q); + + if (!process_bus_messages(c)) + return -1; + + /* + * While we check here for atleast MTU worth of data being available in + * memblockq, we might not have exact equivalent to MTU. Hence, we walk + * over the memchunks in memblockq and accumulate MTU bytes next. + */ + if (pa_memblockq_get_length(q) < c->mtu) + return 0; + + for (;;) { + pa_memchunk chunk; + int r; + + pa_memchunk_reset(&chunk); + + if ((r = pa_memblockq_peek(q, &chunk)) >= 0) { + /* + * Accumulate MTU bytes of data before sending. If the current + * chunk length + accumulated bytes exceeds MTU, we drop bytes + * considered for transfer in this iteration from memblockq. + * + * The remaining bytes will be available in the next iteration, + * as these will be tracked and maintained by memblockq. + */ + size_t k = n + chunk.length > c->mtu ? c->mtu - n : chunk.length; + + pa_assert(chunk.memblock); + + memcpy(c->send_buf + n, pa_memblock_acquire_chunk(&chunk), k); + pa_memblock_release(chunk.memblock); + pa_memblock_unref(chunk.memblock); + + n += k; + pa_memblockq_drop(q, k); + } + + if (r < 0 || n >= c->mtu) { + GstClock *clock; + GstClockTime timestamp, clock_time; + GstMapInfo info; + + if (n > 0) { + clock = gst_element_get_clock(c->pipeline); + clock_time = gst_clock_get_time(clock); + gst_object_unref(clock); + + timestamp = gst_element_get_base_time(c->pipeline); + if (timestamp > clock_time) + timestamp -= clock_time; + else + timestamp = 0; + + buf = gst_buffer_new_allocate(NULL, n, NULL); + pa_assert(buf); + + GST_BUFFER_PTS(buf) = timestamp; + + pa_assert_se(gst_buffer_map(buf, &info, GST_MAP_WRITE)); + + memcpy(info.data, c->send_buf, n); + gst_buffer_unmap(buf, &info); + + if (gst_app_src_push_buffer(GST_APP_SRC(c->appsrc), buf) != GST_FLOW_OK) { + pa_log_error("Could not push buffer"); + return -1; + } + } + + if (r < 0 || pa_memblockq_get_length(q) < c->mtu) + break; + + n = 0; + } + } + + return 0; +} + +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 GstPadProbeReturn udpsrc_buffer_probe(GstPad *pad, GstPadProbeInfo *info, gpointer userdata) { + struct timeval tv; + pa_usec_t timestamp; + pa_rtp_context *c = (pa_rtp_context *) userdata; + + pa_assert(info->type & GST_PAD_PROBE_TYPE_BUFFER); + + pa_gettimeofday(&tv); + timestamp = pa_timeval_load(&tv); + + gst_buffer_add_reference_timestamp_meta(GST_BUFFER(info->data), c->meta_reference, timestamp * GST_USECOND, + GST_CLOCK_TIME_NONE); + + return GST_PAD_PROBE_OK; +} + +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; + GstPad *pad; + 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); + + /* This logic should go into udpsrc, and we should be populating the + * receive timestamp using SCM_TIMESTAMP, but until we have that ... */ + c->meta_reference = gst_caps_new_empty_simple("timestamp/x-pulseaudio-wallclock"); + + pad = gst_element_get_static_pad(udpsrc, "src"); + gst_pad_add_probe(pad, GST_PAD_PROBE_TYPE_BUFFER, udpsrc_buffer_probe, c, NULL); + gst_object_unref(pad); + + 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); + + pa_log_info("Initialising GStreamer RTP backend for receive"); + + c = pa_xnew0(pa_rtp_context, 1); + + c->fdsem = pa_fdsem_new(); + c->ss = *ss; + c->send_buf = NULL; + c->first_buffer = true; + + 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; + GstBufferList *buf_list; + GstAdapter *adapter; + GstBuffer *buf; + GstMapInfo info; + GstClockTime timestamp = GST_CLOCK_TIME_NONE; + uint8_t *data; + uint64_t data_len = 0; + + if (!process_bus_messages(c)) + goto fail; + + adapter = gst_adapter_new(); + pa_assert(adapter); + + while (true) { + sample = gst_app_sink_try_pull_sample(GST_APP_SINK(c->appsink), 0); + if (!sample) + break; + + buf = gst_sample_get_buffer(sample); + + /* Get the timestamp from the first buffer */ + if (timestamp == GST_CLOCK_TIME_NONE) { + GstReferenceTimestampMeta *meta = gst_buffer_get_reference_timestamp_meta(buf, c->meta_reference); + + /* Use the meta if we were able to insert it and it came through, + * else try to fallback to the DTS, which is only available in + * GStreamer 1.16 and earlier. */ + if (meta) + timestamp = meta->timestamp; + else if (GST_BUFFER_DTS(buf) != GST_CLOCK_TIME_NONE) + timestamp = GST_BUFFER_DTS(buf); + else + timestamp = 0; + } + + if (GST_BUFFER_IS_DISCONT(buf)) + pa_log_info("Discontinuity detected, possibly lost some packets"); + + if (!gst_buffer_map(buf, &info, GST_MAP_READ)) { + pa_log_info("Failed to map buffer"); + gst_sample_unref(sample); + goto fail; + } + + data_len += info.size; + /* We need the buffer to be valid longer than the sample, which will + * be valid only for the duration of this loop. + * + * To do this, increase the ref count. Ownership is transferred to the + * adapter in gst_adapter_push. + */ + gst_buffer_ref(buf); + gst_adapter_push(adapter, buf); + gst_buffer_unmap(buf, &info); + + gst_sample_unref(sample); + } + + buf_list = gst_adapter_take_buffer_list(adapter, data_len); + pa_assert(buf_list); + + pa_assert(pa_mempool_block_size_max(pool) >= data_len); + + chunk->memblock = pa_memblock_new(pool, data_len); + chunk->index = 0; + chunk->length = data_len; + + data = (uint8_t *) pa_memblock_acquire_chunk(chunk); + + for (int i = 0; i < gst_buffer_list_length(buf_list); i++) { + buf = gst_buffer_list_get(buf_list, i); + + if (!gst_buffer_map(buf, &info, GST_MAP_READ)) { + gst_buffer_list_unref(buf_list); + goto fail; + } + + memcpy(data, info.data, info.size); + data += info.size; + gst_buffer_unmap(buf, &info); + } + + 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. */ + *rtp_tstamp = gst_util_uint64_scale_int(GST_BUFFER_PTS(gst_buffer_list_get(buf_list, 0)), c->ss.rate, GST_SECOND) & 0xFFFFFFFFU; + if (timestamp != GST_CLOCK_TIME_NONE) + pa_timeval_rtstore(tstamp, timestamp / PA_NSEC_PER_USEC, false); + + if (c->first_buffer) { + c->first_buffer = false; + c->last_timestamp = *rtp_tstamp; + } else { + /* The RTP clock -> time domain -> RTP clock transformation above might + * add a ±1 rounding error, so let's get rid of that */ + uint32_t expected = c->last_timestamp + (uint32_t) (data_len / pa_rtp_context_get_frame_size(c)); + int delta = *rtp_tstamp - expected; + + if (delta == 1 || delta == -1) + *rtp_tstamp -= delta; + + c->last_timestamp = *rtp_tstamp; + } + + gst_buffer_list_unref(buf_list); + gst_object_unref(adapter); + + return 0; + +fail: + if (adapter) + gst_object_unref(adapter); + + if (chunk->memblock) + pa_memblock_unref(chunk->memblock); + + return -1; +} + +void pa_rtp_context_free(pa_rtp_context *c) { + pa_assert(c); + + if (c->meta_reference) + gst_caps_unref(c->meta_reference); + + if (c->appsrc) { + gst_app_src_end_of_stream(GST_APP_SRC(c->appsrc)); + gst_object_unref(c->appsrc); + pa_xfree(c->send_buf); + } + + 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); +} |