1 /* GStreamer RIST plugin
2 * Copyright (C) 2019 Net Insight AB
3 * Author: Nicolas Dufresne <nicolas.dufresne@collabora.com>
5 * This library is free software; you can redistribute it and/or
6 * modify it under the terms of the GNU Library General Public
7 * License as published by the Free Software Foundation; either
8 * version 2 of the License, or (at your option) any later version.
10 * This library is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 * Library General Public License for more details.
15 * You should have received a copy of the GNU Library General Public
16 * License along with this library; if not, write to the
17 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
18 * Boston, MA 02110-1301, USA.
22 * SECTION:element-ristsink
26 * This element implements RIST TR-06-1 Simple Profile transmitter. It
27 * currently supports any registered RTP static payload types such as
28 * MPEG TS. The stream passed to this element must be already RTP
29 * payloaded. Even though RTP SSRC collision are rare in
30 * unidirectional streaming, this element expects the upstream elements
31 * to obey to collision events and change the SSRC in use. Collisions
32 * will occur when transmitting and receiving over multicast on the
33 * same host, and will be properly ignored.
35 * ## Example gst-launch line
37 * gst-launch-1.0 udpsrc ! tsparse set-timestamps=1 smoothing-latency=40000 ! \
38 * rtpmp2tpay ! ristsink address=10.0.0.1 port=5004
41 * Additionally, this element supports bonding, which consist of using
42 * multiple links in order to transmit the streams. The address of
43 * each link is configured through the "bonding-addresses"
44 * property. When set, this will replace the value that might have
45 * been set on the "address" and "port" properties. Each link will be
46 * mapped to its own RTP session. RTX request are only replied to on the
47 * link the NACK was received from.
49 * There are currently two bonding methods in place: "broadcast" and "round-robin".
50 * In "broadcast" mode, all the packets are duplicated over all sessions.
51 * While in "round-robin" mode, packets are evenly distributed over the links. One
52 * can also implement its own dispatcher element and configure it using the
53 * "dispatcher" property. As a reference, "broadcast" mode is implemented with
54 * the "tee" element, while "round-robin" mode is implemented with the
55 * "round-robin" element.
57 * ## Example gst-launch line for bonding
59 * gst-launch-1.0 udpsrc ! tsparse set-timestamps=1 smoothing-latency=40000 ! \
60 * rtpmp2tpay ! ristsink bonding-addresses="10.0.0.1:5004,11.0.0.1:5006"
64 /* using GValueArray, which has not replacement */
65 #define GLIB_DISABLE_DEPRECATION_WARNINGS
72 #include <gst/rtp/rtp.h>
79 GST_DEBUG_CATEGORY_STATIC (gst_rist_sink_debug);
80 #define GST_CAT_DEFAULT gst_rist_sink_debug
87 PROP_MIN_RTCP_INTERVAL,
88 PROP_MAX_RTCP_BANDWIDTH,
89 PROP_STATS_UPDATE_INTERVAL,
92 PROP_MULTICAST_LOOPBACK,
95 PROP_BONDING_ADDRESSES,
102 GST_RIST_BONDING_METHOD_BROADCAST,
103 GST_RIST_BONDING_METHOD_ROUND_ROBIN,
104 } GstRistBondingMethod;
106 static GstStaticPadTemplate sink_templ = GST_STATIC_PAD_TEMPLATE ("sink",
109 GST_STATIC_CAPS ("application/x-rtp"));
115 gchar *multicast_iface;
117 GstElement *rtcp_src;
118 GstElement *rtp_sink;
119 GstElement *rtcp_sink;
120 GstElement *rtx_send;
121 GstElement *rtx_queue;
129 /* Common elements in the pipeline */
131 GstElement *ssrc_filter;
134 GstElement *dispatcher;
136 /* Common properties, protected by bonds_lock */
138 gboolean multicast_loopback;
139 GstClockTime min_rtcp_interval;
140 gdouble max_rtcp_bandwidth;
141 GstRistBondingMethod bonding_method;
145 /* this is needed as setting sibling properties will try to take the object
146 * lock. Thus, any properties that affects the bonds will be protected with
147 * that lock instead of the object lock. */
151 guint stats_interval;
153 GstClockID stats_cid;
155 /* This is set whenever there is a pipeline construction failure, and used
156 * to fail state changes later */
157 gboolean construct_failed;
158 const gchar *missing_plugin;
162 gst_rist_bonding_method_get_type (void)
165 static const GEnumValue values[] = {
166 {GST_RIST_BONDING_METHOD_BROADCAST,
167 "GST_RIST_BONDING_METHOD_BROADCAST", "broadcast"},
168 {GST_RIST_BONDING_METHOD_ROUND_ROBIN,
169 "GST_RIST_BONDING_METHOD_ROUND_ROBIN", "round-robin"},
173 if (g_once_init_enter (&id)) {
174 GType tmp = g_enum_register_static ("GstRistBondingMethodType", values);
175 g_once_init_leave (&id, tmp);
181 G_DEFINE_TYPE_WITH_CODE (GstRistSink, gst_rist_sink, GST_TYPE_BIN,
182 GST_DEBUG_CATEGORY_INIT (gst_rist_sink_debug, "ristsink", 0, "RIST Sink"));
184 static RistSenderBond *
185 gst_rist_sink_add_bond (GstRistSink * sink)
187 RistSenderBond *bond = g_slice_new0 (RistSenderBond);
191 bond->session = sink->bonds->len;
192 bond->address = g_strdup ("localhost");
194 g_snprintf (name, 32, "rist_rtp_udpsink%u", bond->session);
195 bond->rtp_sink = gst_element_factory_make ("udpsink", name);
196 if (!bond->rtp_sink) {
197 g_slice_free (RistSenderBond, bond);
198 sink->missing_plugin = "udp";
202 /* these are all from UDP plugin, so they cannot fail */
203 g_snprintf (name, 32, "rist_rtcp_udpsrc%u", bond->session);
204 bond->rtcp_src = gst_element_factory_make ("udpsrc", name);
205 g_snprintf (name, 32, "rist_rtcp_udpsink%u", bond->session);
206 bond->rtcp_sink = gst_element_factory_make ("udpsink", name);
207 g_object_set (bond->rtcp_sink, "async", FALSE, NULL);
209 gst_bin_add_many (GST_BIN (sink), bond->rtp_sink, bond->rtcp_src,
210 bond->rtcp_sink, NULL);
211 gst_element_set_locked_state (bond->rtcp_src, TRUE);
212 gst_element_set_locked_state (bond->rtcp_sink, TRUE);
214 g_snprintf (name, 32, "rist_rtx_queue%u", bond->session);
215 bond->rtx_queue = gst_element_factory_make ("queue", name);
216 gst_bin_add (GST_BIN (sink->rtxbin), bond->rtx_queue);
218 g_snprintf (name, 32, "rist_rtx_send%u", bond->session);
219 bond->rtx_send = gst_element_factory_make ("ristrtxsend", name);
220 if (!bond->rtx_send) {
221 sink->missing_plugin = "rtpmanager";
222 g_slice_free (RistSenderBond, bond);
225 gst_bin_add (GST_BIN (sink->rtxbin), bond->rtx_send);
227 gst_element_link (bond->rtx_queue, bond->rtx_send);
229 pad = gst_element_get_static_pad (bond->rtx_send, "src");
230 g_snprintf (name, 32, "src_%u", bond->session);
231 gpad = gst_ghost_pad_new (name, pad);
232 gst_object_unref (pad);
233 gst_element_add_pad (sink->rtxbin, gpad);
235 g_object_set (bond->rtx_send, "max-size-packets", 0, NULL);
237 g_snprintf (name, 32, "send_rtp_sink_%u", bond->session);
238 if (bond->session == 0) {
239 gst_element_link_pads (sink->ssrc_filter, "src", sink->rtpbin, name);
243 /* to make a sender, we need to create an unused pad on rtpbin, which will
244 * require an unused pad on the rtxbin */
245 g_snprintf (name, 32, "sink_%u", bond->session);
246 pad = gst_ghost_pad_new_no_target (name, GST_PAD_SINK);
247 gst_element_add_pad (sink->rtxbin, pad);
249 g_snprintf (name, 32, "send_rtp_sink_%u", bond->session);
250 pad = gst_element_get_request_pad (sink->rtpbin, name);
251 gst_object_unref (pad);
254 g_snprintf (name, 32, "send_rtp_src_%u", bond->session);
255 gst_element_link_pads (sink->rtpbin, name, bond->rtp_sink, "sink");
257 g_snprintf (name, 32, "recv_rtcp_sink_%u", bond->session);
258 gst_element_link_pads (bond->rtcp_src, "src", sink->rtpbin, name);
260 g_snprintf (name, 32, "send_rtcp_src_%u", bond->session);
261 gst_element_link_pads (sink->rtpbin, name, bond->rtcp_sink, "sink");
263 g_ptr_array_add (sink->bonds, bond);
268 gst_rist_sink_request_pt_map (GstRistSrc * sink, guint session_id, guint pt,
271 const GstRTPPayloadInfo *pt_info;
274 pt_info = gst_rtp_payload_info_for_pt (pt);
275 if (!pt_info || !pt_info->clock_rate)
278 ret = gst_caps_new_simple ("application/x-rtp",
279 "media", G_TYPE_STRING, pt_info->media,
280 "encoding_name", G_TYPE_STRING, pt_info->encoding_name,
281 "clock-rate", G_TYPE_INT, (gint) pt_info->clock_rate, NULL);
283 /* FIXME add sprop-parameter-set if any */
284 g_warn_if_fail (pt_info->encoding_parameters == NULL);
290 gst_rist_sink_request_aux_sender (GstRistSink * sink, guint session_id,
293 return gst_object_ref (sink->rtxbin);
297 on_app_rtcp (GObject * session, guint32 subtype, guint32 ssrc,
298 const gchar * name, GstBuffer * data, GstElement * rtpsession)
300 if (g_str_equal (name, "RIST")) {
303 GstPad *send_rtp_sink;
307 send_rtp_sink = gst_element_get_static_pad (rtpsession, "send_rtp_sink");
309 gst_buffer_map (data, &map, GST_MAP_READ);
311 for (i = 0; i < map.size; i += sizeof (guint32)) {
312 guint32 dword = GST_READ_UINT32_BE (map.data + i);
313 guint16 seqnum = dword >> 16;
314 guint16 num = dword & 0x0000FFFF;
317 GST_DEBUG ("got RIST nack packet, #%u %u", seqnum, num);
319 /* num is inclusive, i.e. it can be 0, which means exactly 1 seqnum */
320 for (j = 0; j <= num; j++) {
321 event = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
322 gst_structure_new ("GstRTPRetransmissionRequest",
323 "seqnum", G_TYPE_UINT, (guint) seqnum + j,
324 "ssrc", G_TYPE_UINT, (guint) ssrc, NULL));
325 gst_pad_push_event (send_rtp_sink, event);
329 gst_buffer_unmap (data, &map);
330 gst_object_unref (send_rtp_sink);
337 gst_rist_sink_on_new_sender_ssrc (GstRistSink * sink, guint session_id,
338 guint ssrc, GstElement * rtpbin)
340 GObject *gstsession = NULL;
341 GObject *session = NULL;
342 GObject *source = NULL;
347 g_signal_emit_by_name (rtpbin, "get-session", session_id, &gstsession);
348 g_signal_emit_by_name (rtpbin, "get-internal-session", session_id, &session);
349 g_signal_emit_by_name (session, "get-source-by-ssrc", ssrc, &source);
352 g_object_set (source, "disable-rtcp", TRUE, NULL);
354 g_signal_connect (session, "on-app-rtcp", (GCallback) on_app_rtcp,
357 g_object_unref (source);
358 g_object_unref (session);
362 gst_rist_sink_on_new_receiver_ssrc (GstRistSink * sink, guint session_id,
363 guint ssrc, GstElement * rtpbin)
365 RistSenderBond *bond;
370 GST_INFO_OBJECT (sink, "Got RTCP remote SSRC %u", ssrc);
371 bond = g_ptr_array_index (sink->bonds, session_id);
372 bond->rtcp_ssrc = ssrc;
375 static GstPadProbeReturn
376 gst_rist_sink_fix_collision (GstPad * pad, GstPadProbeInfo * info,
379 GstEvent *event = info->data;
380 const GstStructure *cs;
384 /* We simply ignore collisions */
385 if (GST_EVENT_TYPE (event) != GST_EVENT_CUSTOM_UPSTREAM)
386 return GST_PAD_PROBE_OK;
388 cs = gst_event_get_structure (event);
389 if (!gst_structure_has_name (cs, "GstRTPCollision"))
390 return GST_PAD_PROBE_OK;
392 gst_structure_get_uint (cs, "suggested-ssrc", &ssrc);
394 return GST_PAD_PROBE_OK;
396 event = info->data = gst_event_make_writable (event);
397 /* we can drop the const qualifier as we ensured writability */
398 s = (GstStructure *) gst_event_get_structure (event);
399 gst_structure_set (s, "suggested-ssrc", G_TYPE_UINT, ssrc - 1, NULL);
401 return GST_PAD_PROBE_OK;
405 gst_rist_sink_set_caps (GstRistSink * sink, GstCaps * caps)
407 const GstStructure *s = gst_caps_get_structure (caps, 0);
409 if (!gst_structure_get_uint (s, "ssrc", &sink->rtp_ssrc)) {
410 GST_ELEMENT_ERROR (sink, CORE, NEGOTIATION, ("No 'ssrc' field in caps."),
415 if (sink->rtp_ssrc & 1) {
416 GST_ELEMENT_ERROR (sink, CORE, NEGOTIATION,
417 ("Invalid RIST SSRC, LSB must be zero."), (NULL));
425 gst_rist_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
427 GstRistSink *sink = GST_RIST_SINK (parent);
431 switch (GST_EVENT_TYPE (event)) {
433 gst_event_parse_caps (event, &caps);
434 ret = gst_rist_sink_set_caps (sink, caps);
441 ret = gst_pad_event_default (pad, parent, event);
443 gst_event_unref (event);
449 gst_rist_sink_init (GstRistSink * sink)
451 GstPad *ssrc_filter_sinkpad, *rtxbin_gpad;
453 GstStructure *sdes = NULL;
454 RistSenderBond *bond;
456 g_mutex_init (&sink->bonds_lock);
457 sink->bonds = g_ptr_array_new ();
459 /* Construct the RIST RTP sender pipeline.
461 * capsfilter*-> [send_rtp_sink_%u] -------- [send_rtp_src_%u] -> udpsink
463 * udpsrc -> [recv_rtcp_sink_%u] -------- [send_rtcp_src_%u] -> * udpsink
465 * * To select RIST compatible SSRC
467 sink->rtpbin = gst_element_factory_make ("rtpbin", "rist_send_rtpbin");
469 sink->missing_plugin = "rtpmanager";
473 /* RIST specification says the SDES should only contain the CNAME */
474 g_object_get (sink->rtpbin, "sdes", &sdes, NULL);
475 gst_structure_remove_field (sdes, "tool");
477 gst_bin_add (GST_BIN (sink), sink->rtpbin);
478 g_object_set (sink->rtpbin, "do-retransmission", TRUE,
479 "rtp-profile", 3 /* AVFP */ ,
481 gst_structure_free (sdes);
483 g_signal_connect_swapped (sink->rtpbin, "request-pt-map",
484 G_CALLBACK (gst_rist_sink_request_pt_map), sink);
485 g_signal_connect_swapped (sink->rtpbin, "request-aux-sender",
486 G_CALLBACK (gst_rist_sink_request_aux_sender), sink);
487 g_signal_connect_swapped (sink->rtpbin, "on-new-sender-ssrc",
488 G_CALLBACK (gst_rist_sink_on_new_sender_ssrc), sink);
489 g_signal_connect_swapped (sink->rtpbin, "on-new-ssrc",
490 G_CALLBACK (gst_rist_sink_on_new_receiver_ssrc), sink);
492 sink->rtxbin = gst_bin_new ("rist_send_rtxbin");
493 g_object_ref_sink (sink->rtxbin);
495 rtxbin_gpad = gst_ghost_pad_new_no_target ("sink_0", GST_PAD_SINK);
496 gst_element_add_pad (sink->rtxbin, rtxbin_gpad);
498 sink->ssrc_filter = gst_element_factory_make ("capsfilter",
500 gst_bin_add (GST_BIN (sink), sink->ssrc_filter);
502 /* RIST RTP SSRC should have LSB set to 0 */
503 sink->rtp_ssrc = g_random_int () & ~1;
504 ssrc_caps = gst_caps_new_simple ("application/x-rtp",
505 "ssrc", G_TYPE_UINT, sink->rtp_ssrc, NULL);
506 gst_caps_append_structure (ssrc_caps,
507 gst_structure_new_empty ("application/x-rtp"));
508 g_object_set (sink->ssrc_filter, "caps", ssrc_caps, NULL);
509 gst_caps_unref (ssrc_caps);
511 ssrc_filter_sinkpad = gst_element_get_static_pad (sink->ssrc_filter, "sink");
512 sink->sinkpad = gst_ghost_pad_new_from_template ("sink", ssrc_filter_sinkpad,
513 gst_static_pad_template_get (&sink_templ));
514 gst_pad_set_event_function (sink->sinkpad, gst_rist_sink_event);
515 gst_element_add_pad (GST_ELEMENT (sink), sink->sinkpad);
516 gst_object_unref (ssrc_filter_sinkpad);
518 gst_pad_add_probe (sink->sinkpad, GST_PAD_PROBE_TYPE_EVENT_UPSTREAM,
519 gst_rist_sink_fix_collision, sink, NULL);
521 bond = gst_rist_sink_add_bond (sink);
529 GST_ERROR_OBJECT (sink, "'%s' plugin is missing.", sink->missing_plugin);
530 sink->construct_failed = TRUE;
531 /* Just make our element valid, so we fail cleanly */
532 gst_element_add_pad (GST_ELEMENT (sink),
533 gst_pad_new_from_static_template (&sink_templ, "sink"));
538 gst_rist_sink_setup_rtcp_socket (GstRistSink * sink, RistSenderBond * bond)
540 GSocket *socket = NULL;
541 GInetAddress *iaddr = NULL;
542 gchar *remote_addr = NULL;
543 guint port = bond->port + 1;
544 GError *error = NULL;
546 iaddr = g_inet_address_new_from_string (bond->address);
549 GResolver *resolver = NULL;
551 resolver = g_resolver_get_default ();
552 results = g_resolver_lookup_by_name (resolver, bond->address, NULL, &error);
555 g_object_unref (resolver);
556 goto dns_resolve_failed;
559 iaddr = G_INET_ADDRESS (g_object_ref (results->data));
561 g_resolver_free_addresses (results);
562 g_object_unref (resolver);
564 remote_addr = g_inet_address_to_string (iaddr);
566 if (g_inet_address_get_is_multicast (iaddr)) {
567 g_object_set (bond->rtcp_src, "address", remote_addr, "port", port, NULL);
569 const gchar *any_addr;
571 if (g_inet_address_get_family (iaddr) == G_SOCKET_FAMILY_IPV6)
574 any_addr = "0.0.0.0";
576 g_object_set (bond->rtcp_src, "address", any_addr, "port", 0, NULL);
578 g_object_unref (iaddr);
580 gst_element_set_locked_state (bond->rtcp_src, FALSE);
581 gst_element_sync_state_with_parent (bond->rtcp_src);
583 /* share the socket created by the sink */
584 g_object_get (bond->rtcp_src, "used-socket", &socket, NULL);
585 g_object_set (bond->rtcp_sink, "socket", socket, "auto-multicast", FALSE,
586 "close-socket", FALSE, NULL);
587 g_object_unref (socket);
589 gst_element_set_locked_state (bond->rtcp_sink, FALSE);
590 gst_element_sync_state_with_parent (bond->rtcp_sink);
592 return GST_STATE_CHANGE_SUCCESS;
595 GST_ELEMENT_ERROR (sink, RESOURCE, NOT_FOUND,
596 ("Could not resolve hostname '%s'", GST_STR_NULL (remote_addr)),
597 ("DNS resolver reported: %s", error->message));
598 g_free (remote_addr);
599 g_error_free (error);
600 return GST_STATE_CHANGE_FAILURE;
603 static GstStateChangeReturn
604 gst_rist_sink_start (GstRistSink * sink)
606 GstPad *dispatcher_sinkpad, *rtxbin_gpad;
609 /* Unless a custom dispatcher was provided, use the specified bonding method
611 if (!sink->dispatcher) {
612 switch (sink->bonding_method) {
613 case GST_RIST_BONDING_METHOD_BROADCAST:
614 sink->dispatcher = gst_element_factory_make ("tee", "rist_dispatcher");
615 if (!sink->dispatcher) {
616 sink->missing_plugin = "coreelements";
617 sink->construct_failed = TRUE;
620 case GST_RIST_BONDING_METHOD_ROUND_ROBIN:
621 sink->dispatcher = gst_element_factory_make ("roundrobin",
623 g_assert (sink->dispatcher);
628 if (sink->construct_failed) {
629 GST_ELEMENT_ERROR (sink, CORE, MISSING_PLUGIN,
630 ("Your GStreamer installation is missing plugin '%s'",
631 sink->missing_plugin), (NULL));
632 return GST_STATE_CHANGE_FAILURE;
635 gst_bin_add (GST_BIN (sink->rtxbin), sink->dispatcher);
636 dispatcher_sinkpad = gst_element_get_static_pad (sink->dispatcher, "sink");
637 rtxbin_gpad = gst_element_get_static_pad (sink->rtxbin, "sink_0");
638 gst_ghost_pad_set_target (GST_GHOST_PAD (rtxbin_gpad), dispatcher_sinkpad);
639 gst_object_unref (dispatcher_sinkpad);
640 gst_object_unref (rtxbin_gpad);
642 for (i = 0; i < sink->bonds->len; i++) {
643 RistSenderBond *bond = g_ptr_array_index (sink->bonds, i);
644 GObject *session = NULL;
648 g_signal_emit_by_name (sink->rtpbin, "get-session", i, &session);
649 g_object_set (session, "rtcp-min-interval", sink->min_rtcp_interval,
650 "rtcp-fraction", sink->max_rtcp_bandwidth, NULL);
651 g_object_unref (session);
653 g_snprintf (name, 32, "src_%u", bond->session);
654 pad = gst_element_get_request_pad (sink->dispatcher, name);
655 gst_element_link_pads (sink->dispatcher, name, bond->rtx_queue, "sink");
656 gst_object_unref (pad);
658 if (!gst_rist_sink_setup_rtcp_socket (sink, bond))
659 return GST_STATE_CHANGE_FAILURE;
662 return GST_STATE_CHANGE_SUCCESS;
666 static GstStructure *
667 gst_rist_sink_create_stats (GstRistSink * sink)
669 RistSenderBond *bond;
671 GValueArray *session_stats;
672 guint64 total_pkt_sent = 0, total_rtx_sent = 0;
675 ret = gst_structure_new_empty ("rist/x-sender-stats");
676 session_stats = g_value_array_new (sink->bonds->len);
678 for (i = 0; i < sink->bonds->len; i++) {
679 GObject *session = NULL, *source = NULL;
680 GstStructure *sstats = NULL, *stats;
681 guint64 pkt_sent = 0, rtx_sent = 0, rtt;
683 GValue value = G_VALUE_INIT;
685 g_signal_emit_by_name (sink->rtpbin, "get-internal-session", i, &session);
689 stats = gst_structure_new_empty ("rist/x-sender-session-stats");
690 bond = g_ptr_array_index (sink->bonds, i);
692 g_signal_emit_by_name (session, "get-source-by-ssrc", sink->rtp_ssrc,
695 g_object_get (source, "stats", &sstats, NULL);
696 gst_structure_get_uint64 (sstats, "packets-sent", &pkt_sent);
697 gst_structure_free (sstats);
698 g_clear_object (&source);
701 g_signal_emit_by_name (session, "get-source-by-ssrc", bond->rtcp_ssrc,
704 g_object_get (source, "stats", &sstats, NULL);
705 gst_structure_get_uint (sstats, "rb-round-trip", &rb_rtt);
706 gst_structure_free (sstats);
707 g_clear_object (&source);
709 g_object_unref (session);
711 g_object_get (bond->rtx_send, "num-rtx-packets", &rtx_sent, NULL);
713 /* rb_rtt is in Q16 in NTP time */
714 rtt = gst_util_uint64_scale (rb_rtt, GST_SECOND, 65536);
716 gst_structure_set (stats, "session-id", G_TYPE_INT, i,
717 "sent-original-packets", G_TYPE_UINT64, pkt_sent,
718 "sent-retransmitted-packets", G_TYPE_UINT64, rtx_sent,
719 "round-trip-time", G_TYPE_UINT64, rtt, NULL);
721 g_value_init (&value, GST_TYPE_STRUCTURE);
722 g_value_take_boxed (&value, stats);
723 g_value_array_append (session_stats, &value);
724 g_value_unset (&value);
726 total_pkt_sent += pkt_sent;
727 total_rtx_sent += rtx_sent;
730 gst_structure_set (ret,
731 "sent-original-packets", G_TYPE_UINT64, total_pkt_sent,
732 "sent-retransmitted-packets", G_TYPE_UINT64, total_rtx_sent,
733 "session-stats", G_TYPE_VALUE_ARRAY, session_stats, NULL);
734 g_value_array_free (session_stats);
740 gst_rist_sink_dump_stats (GstClock * clock, GstClockTime time, GstClockID id,
743 GstRistSink *sink = GST_RIST_SINK (user_data);
744 GstStructure *stats = gst_rist_sink_create_stats (sink);
746 gst_println ("%s: %" GST_PTR_FORMAT, GST_OBJECT_NAME (sink), stats);
748 gst_structure_free (stats);
753 gst_rist_sink_enable_stats_interval (GstRistSink * sink)
756 GstClockTime start, interval;
758 if (sink->stats_interval == 0)
761 interval = sink->stats_interval * GST_MSECOND;
762 clock = gst_system_clock_obtain ();
763 start = gst_clock_get_time (clock) + interval;
765 sink->stats_cid = gst_clock_new_periodic_id (clock, start, interval);
766 gst_clock_id_wait_async (sink->stats_cid, gst_rist_sink_dump_stats,
767 gst_object_ref (sink), (GDestroyNotify) gst_object_unref);
769 gst_object_unref (clock);
773 gst_rist_sink_disable_stats_interval (GstRistSink * sink)
775 if (sink->stats_cid) {
776 gst_clock_id_unschedule (sink->stats_cid);
777 gst_clock_id_unref (sink->stats_cid);
778 sink->stats_cid = NULL;
782 static GstStateChangeReturn
783 gst_rist_sink_change_state (GstElement * element, GstStateChange transition)
785 GstRistSink *sink = GST_RIST_SINK (element);
786 GstStateChangeReturn ret;
788 switch (transition) {
789 case GST_STATE_CHANGE_PAUSED_TO_READY:
790 gst_rist_sink_disable_stats_interval (sink);
796 ret = GST_ELEMENT_CLASS (gst_rist_sink_parent_class)->change_state (element,
799 switch (transition) {
800 case GST_STATE_CHANGE_NULL_TO_READY:
801 ret = gst_rist_sink_start (sink);
803 case GST_STATE_CHANGE_READY_TO_PAUSED:
804 gst_rist_sink_enable_stats_interval (sink);
813 /* called with bonds lock */
815 gst_rist_sink_update_bond_address (GstRistSink * sink, RistSenderBond * bond,
816 const gchar * address, guint port, const gchar * multicast_iface)
818 g_free (bond->address);
819 g_free (bond->multicast_iface);
820 bond->address = g_strdup (address);
821 bond->multicast_iface = multicast_iface ? g_strdup (multicast_iface) : NULL;
824 g_object_set (G_OBJECT (bond->rtp_sink), "host", address, "port", port,
825 "multicast-iface", bond->multicast_iface, NULL);
826 g_object_set (G_OBJECT (bond->rtcp_sink), "host", address,
827 "port", port + 1, "multicast-iface", bond->multicast_iface, NULL);
829 /* TODO add runtime support
830 * - add blocking the pad probe
831 * - update RTCP socket
832 * - cycle elements through NULL state
836 /* called with bonds lock */
838 gst_rist_sink_get_bonds (GstRistSink * sink)
840 GString *bonds = g_string_new ("");
843 for (i = 0; i < sink->bonds->len; i++) {
844 RistSenderBond *bond = g_ptr_array_index (sink->bonds, i);
846 g_string_append_c (bonds, ':');
848 g_string_append_printf (bonds, "%s:%u", bond->address, bond->port);
850 if (bond->multicast_iface)
851 g_string_append_printf (bonds, "/%s", bond->multicast_iface);
854 return g_string_free (bonds, FALSE);
860 char *multicast_iface;
864 /* called with bonds lock */
866 gst_rist_sink_set_bonds (GstRistSink * sink, const gchar * bonds)
869 struct RistAddress *addrs;
873 goto missing_address;
875 tokens = g_strsplit (bonds, ",", 0);
876 if (tokens[0] == NULL)
877 goto missing_address;
879 addrs = g_new0 (struct RistAddress, g_strv_length (tokens));
881 /* parse the address list */
882 for (i = 0; tokens[i]; i++) {
883 gchar *address = tokens[i];
884 char *port_ptr, *iface_ptr, *endptr;
887 port_ptr = g_utf8_strrchr (address, -1, ':');
888 iface_ptr = g_utf8_strrchr (address, -1, '/');
892 if (!g_ascii_isdigit (port_ptr[1]))
896 if (iface_ptr < port_ptr)
901 port = strtol (port_ptr + 1, &endptr, 0);
902 if (endptr[0] != '\0')
905 /* port must be a multiple of 2 between 2 and 65534 */
906 if (port < 2 || (port & 1) || port > G_MAXUINT16)
910 addrs[i].port = port;
911 addrs[i].address = g_strstrip (address);
913 addrs[i].multicast_iface = g_strstrip (iface_ptr + 1);
916 /* configure the bonds */
917 for (i = 0; tokens[i]; i++) {
918 RistSenderBond *bond = NULL;
920 if (i < sink->bonds->len)
921 bond = g_ptr_array_index (sink->bonds, i);
923 bond = gst_rist_sink_add_bond (sink);
925 gst_rist_sink_update_bond_address (sink, bond, addrs[i].address,
926 addrs[i].port, addrs[i].multicast_iface);
933 g_warning ("'bonding-addresses' cannot be empty");
938 g_warning ("Failed to parse address '%s", tokens[i]);
944 g_warning ("RIST port must valid UDP port and a multiple of 2.");
951 gst_rist_sink_set_multicast_loopback (GstRistSink * sink, gboolean loop)
955 sink->multicast_loopback = loop;
956 for (i = 0; i < sink->bonds->len; i++) {
957 RistSenderBond *bond = g_ptr_array_index (sink->bonds, i);
958 g_object_set (G_OBJECT (bond->rtp_sink), "loop", loop, NULL);
959 g_object_set (G_OBJECT (bond->rtcp_sink), "loop", loop, NULL);
963 /* called with bonds lock */
965 gst_rist_sink_set_multicast_ttl (GstRistSink * sink, gint ttl)
969 sink->multicast_ttl = ttl;
970 for (i = 0; i < sink->bonds->len; i++) {
971 RistSenderBond *bond = g_ptr_array_index (sink->bonds, i);
972 g_object_set (G_OBJECT (bond->rtp_sink), "ttl-mc", ttl, NULL);
973 g_object_set (G_OBJECT (bond->rtcp_sink), "ttl-mc", ttl, NULL);
978 gst_rist_sink_get_property (GObject * object, guint prop_id,
979 GValue * value, GParamSpec * pspec)
981 GstRistSink *sink = GST_RIST_SINK (object);
983 RistSenderBond *bond;
985 if (sink->construct_failed)
988 g_mutex_lock (&sink->bonds_lock);
990 bond = g_ptr_array_index (sink->bonds, 0);
994 g_value_set_string (value, bond->address);
998 g_value_set_uint (value, bond->port);
1001 case PROP_SENDER_BUFFER:
1002 g_object_get_property (G_OBJECT (bond->rtx_send), "max-size-time", value);
1005 case PROP_MIN_RTCP_INTERVAL:
1006 g_value_set_uint (value, (guint) (sink->min_rtcp_interval / GST_MSECOND));
1009 case PROP_MAX_RTCP_BANDWIDTH:
1010 g_value_set_double (value, sink->max_rtcp_bandwidth);
1013 case PROP_STATS_UPDATE_INTERVAL:
1014 g_value_set_uint (value, sink->stats_interval);
1018 g_value_take_boxed (value, gst_rist_sink_create_stats (sink));
1022 g_object_get (sink->rtpbin, "sdes", &sdes, NULL);
1023 g_value_set_string (value, gst_structure_get_string (sdes, "cname"));
1024 gst_structure_free (sdes);
1027 case PROP_MULTICAST_LOOPBACK:
1028 g_value_set_boolean (value, sink->multicast_loopback);
1031 case PROP_MULTICAST_IFACE:
1032 g_value_set_string (value, bond->multicast_iface);
1035 case PROP_MULTICAST_TTL:
1036 g_value_set_int (value, sink->multicast_ttl);
1039 case PROP_BONDING_ADDRESSES:
1040 g_value_take_string (value, gst_rist_sink_get_bonds (sink));
1043 case PROP_BONDING_METHOD:
1044 g_value_set_enum (value, sink->bonding_method);
1047 case PROP_DISPATCHER:
1048 g_value_set_object (value, sink->dispatcher);
1052 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1056 g_mutex_unlock (&sink->bonds_lock);
1060 gst_rist_sink_set_property (GObject * object, guint prop_id,
1061 const GValue * value, GParamSpec * pspec)
1063 GstRistSink *sink = GST_RIST_SINK (object);
1065 RistSenderBond *bond;
1067 if (sink->construct_failed)
1070 g_mutex_lock (&sink->bonds_lock);
1072 bond = g_ptr_array_index (sink->bonds, 0);
1076 g_free (bond->address);
1077 bond->address = g_value_dup_string (value);
1078 g_object_set_property (G_OBJECT (bond->rtp_sink), "host", value);
1079 g_object_set_property (G_OBJECT (bond->rtcp_sink), "host", value);
1083 guint port = g_value_get_uint (value);
1085 /* According to 5.1.1, RTCP receiver port most be event number and RTCP
1086 * port should be the RTP port + 1 */
1089 g_warning ("Invalid RIST port %u, should be an even number.", port);
1094 g_object_set (bond->rtp_sink, "port", port, NULL);
1095 g_object_set (bond->rtcp_sink, "port", port + 1, NULL);
1099 case PROP_SENDER_BUFFER:
1100 g_object_set (bond->rtx_send,
1101 "max-size-time", g_value_get_uint (value), NULL);
1104 case PROP_MIN_RTCP_INTERVAL:
1105 sink->min_rtcp_interval = g_value_get_uint (value) * GST_MSECOND;
1108 case PROP_MAX_RTCP_BANDWIDTH:
1109 sink->max_rtcp_bandwidth = g_value_get_double (value);
1112 case PROP_STATS_UPDATE_INTERVAL:
1113 sink->stats_interval = g_value_get_uint (value);
1117 g_object_get (sink->rtpbin, "sdes", &sdes, NULL);
1118 gst_structure_set_value (sdes, "cname", value);
1119 g_object_set (sink->rtpbin, "sdes", sdes, NULL);
1120 gst_structure_free (sdes);
1123 case PROP_MULTICAST_LOOPBACK:
1124 gst_rist_sink_set_multicast_loopback (sink, g_value_get_boolean (value));
1127 case PROP_MULTICAST_IFACE:
1128 g_free (bond->multicast_iface);
1129 bond->multicast_iface = g_value_dup_string (value);
1130 g_object_set_property (G_OBJECT (bond->rtp_sink),
1131 "multicast-iface", value);
1132 g_object_set_property (G_OBJECT (bond->rtcp_sink),
1133 "multicast-iface", value);
1136 case PROP_MULTICAST_TTL:
1137 gst_rist_sink_set_multicast_ttl (sink, g_value_get_int (value));
1140 case PROP_BONDING_ADDRESSES:
1141 gst_rist_sink_set_bonds (sink, g_value_get_string (value));
1144 case PROP_BONDING_METHOD:
1145 sink->bonding_method = g_value_get_enum (value);
1148 case PROP_DISPATCHER:
1149 if (sink->dispatcher)
1150 g_object_unref (sink->dispatcher);
1151 sink->dispatcher = g_object_ref_sink (g_value_get_object (value));
1155 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1159 g_mutex_unlock (&sink->bonds_lock);
1163 gst_rist_sink_finalize (GObject * object)
1165 GstRistSink *sink = GST_RIST_SINK (object);
1168 g_mutex_lock (&sink->bonds_lock);
1170 for (i = 0; i < sink->bonds->len; i++) {
1171 RistSenderBond *bond = g_ptr_array_index (sink->bonds, i);
1172 g_free (bond->address);
1173 g_free (bond->multicast_iface);
1174 g_slice_free (RistSenderBond, bond);
1176 g_ptr_array_free (sink->bonds, TRUE);
1178 g_clear_object (&sink->rtxbin);
1180 g_mutex_unlock (&sink->bonds_lock);
1181 g_mutex_clear (&sink->bonds_lock);
1183 G_OBJECT_CLASS (gst_rist_sink_parent_class)->finalize (object);
1187 gst_rist_sink_class_init (GstRistSinkClass * klass)
1189 GstElementClass *element_class = (GstElementClass *) klass;
1190 GObjectClass *object_class = (GObjectClass *) klass;
1192 gst_element_class_set_metadata (element_class,
1193 "RIST Sink", "Source/Network",
1194 "Sink that implements RIST TR-06-1 streaming specification",
1195 "Nicolas Dufresne <nicolas.dufresne@collabora.com");
1196 gst_element_class_add_static_pad_template (element_class, &sink_templ);
1198 element_class->change_state = gst_rist_sink_change_state;
1200 object_class->get_property = gst_rist_sink_get_property;
1201 object_class->set_property = gst_rist_sink_set_property;
1202 object_class->finalize = gst_rist_sink_finalize;
1204 g_object_class_install_property (object_class, PROP_ADDRESS,
1205 g_param_spec_string ("address", "Address",
1206 "Address to send packets to (can be IPv4 or IPv6).", "0.0.0.0",
1207 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
1209 g_object_class_install_property (object_class, PROP_PORT,
1210 g_param_spec_uint ("port", "Port", "The port RTP packets will be sent, "
1211 "the RTCP port is this value + 1. This port must be an even number.",
1213 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT));
1215 g_object_class_install_property (object_class, PROP_SENDER_BUFFER,
1216 g_param_spec_uint ("sender-buffer", "Sender Buffer",
1217 "Size of the retransmission queue (in ms)", 0, G_MAXUINT, 1200,
1218 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT));
1220 g_object_class_install_property (object_class, PROP_MIN_RTCP_INTERVAL,
1221 g_param_spec_uint ("min-rtcp-interval", "Minimum RTCP Intercal",
1222 "The minimum interval (in ms) between two regular successive RTCP "
1223 "packets.", 0, 100, 100,
1224 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT));
1226 g_object_class_install_property (object_class, PROP_MAX_RTCP_BANDWIDTH,
1227 g_param_spec_double ("max-rtcp-bandwidth", "Maximum RTCP Bandwidth",
1228 "The maximum bandwidth used for RTCP as a fraction of RTP bandwdith",
1230 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT));
1232 g_object_class_install_property (object_class, PROP_STATS_UPDATE_INTERVAL,
1233 g_param_spec_uint ("stats-update-interval", "Statistics Update Interval",
1234 "The interval between 'stats' update notification (in ms) (0 disabled)",
1236 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT));
1238 g_object_class_install_property (object_class, PROP_STATS,
1239 g_param_spec_boxed ("stats", "Statistics",
1240 "Statistic in a GstStructure named 'rist/x-sender-stats'",
1241 GST_TYPE_STRUCTURE, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
1243 g_object_class_install_property (object_class, PROP_CNAME,
1244 g_param_spec_string ("cname", "CName",
1245 "Set the CNAME in the SDES block of the sender report.", NULL,
1246 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS |
1247 GST_PARAM_DOC_SHOW_DEFAULT));
1249 g_object_class_install_property (object_class, PROP_MULTICAST_LOOPBACK,
1250 g_param_spec_boolean ("multicast-loopback", "Multicast Loopback",
1251 "When enabled, the packet will be received locally.", FALSE,
1252 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT));
1254 g_object_class_install_property (object_class, PROP_MULTICAST_IFACE,
1255 g_param_spec_string ("multicast-iface", "multicast-iface",
1256 "The multicast interface to use to send packets.", NULL,
1257 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
1259 g_object_class_install_property (object_class, PROP_MULTICAST_TTL,
1260 g_param_spec_int ("multicast-ttl", "Multicast TTL",
1261 "The multicast time-to-live parameter.", 0, 255, 1,
1262 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT));
1264 g_object_class_install_property (object_class, PROP_BONDING_ADDRESSES,
1265 g_param_spec_string ("bonding-addresses", "Bonding Addresses",
1266 "Comma (,) separated list of <address>:<port> to send to. ", NULL,
1267 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
1269 g_object_class_install_property (object_class, PROP_BONDING_METHOD,
1270 g_param_spec_enum ("bonding-method", "Bonding Method",
1271 "Defines the bonding method to use.",
1272 gst_rist_bonding_method_get_type (),
1273 GST_RIST_BONDING_METHOD_BROADCAST,
1274 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_CONSTRUCT));
1276 g_object_class_install_property (object_class, PROP_DISPATCHER,
1277 g_param_spec_object ("dispatcher", "Bonding Dispatcher",
1278 "An element that takes care of multi-plexing bonded links. When set "
1279 "\"bonding-method\" is ignored.",
1281 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS |
1282 GST_PARAM_MUTABLE_READY));