2 * Copyright (C) 2008 Wim Taymans <wim.taymans at gmail.com>
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Library General Public
6 * License as published by the Free Software Foundation; either
7 * version 2 of the License, or (at your option) any later version.
9 * This library is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
12 * Library General Public License for more details.
14 * You should have received a copy of the GNU Library General Public
15 * License along with this library; if not, write to the
16 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
17 * Boston, MA 02110-1301, USA.
25 #include <gst/app/gstappsrc.h>
26 #include <gst/app/gstappsink.h>
28 #include "rtsp-stream.h"
30 #define GST_RTSP_STREAM_GET_PRIVATE(obj) \
31 (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_RTSP_STREAM, GstRTSPStreamPrivate))
33 struct _GstRTSPStreamPrivate
38 GstElement *payloader;
43 /* pads on the rtpbin */
44 GstPad *send_rtp_sink;
48 /* the RTPSession object */
51 /* sinks used for sending and receiving RTP and RTCP, they share
53 GstElement *udpsrc[2];
54 GstElement *udpsink[2];
55 /* for TCP transport */
56 GstElement *appsrc[2];
57 GstElement *appqueue[2];
58 GstElement *appsink[2];
61 GstElement *funnel[2];
63 /* server ports for sending/receiving */
64 GstRTSPRange server_port;
66 /* multicast addresses */
67 GstRTSPAddressPool *pool;
70 /* the caps of the stream */
74 /* transports we stream to */
86 GST_DEBUG_CATEGORY_STATIC (rtsp_stream_debug);
87 #define GST_CAT_DEFAULT rtsp_stream_debug
89 static GQuark ssrc_stream_map_key;
91 static void gst_rtsp_stream_finalize (GObject * obj);
93 G_DEFINE_TYPE (GstRTSPStream, gst_rtsp_stream, G_TYPE_OBJECT);
96 gst_rtsp_stream_class_init (GstRTSPStreamClass * klass)
98 GObjectClass *gobject_class;
100 g_type_class_add_private (klass, sizeof (GstRTSPStreamPrivate));
102 gobject_class = G_OBJECT_CLASS (klass);
104 gobject_class->finalize = gst_rtsp_stream_finalize;
106 GST_DEBUG_CATEGORY_INIT (rtsp_stream_debug, "rtspstream", 0, "GstRTSPStream");
108 ssrc_stream_map_key = g_quark_from_static_string ("GstRTSPServer.stream");
112 gst_rtsp_stream_init (GstRTSPStream * stream)
114 GstRTSPStreamPrivate *priv = GST_RTSP_STREAM_GET_PRIVATE (stream);
116 GST_DEBUG ("new stream %p", stream);
120 g_mutex_init (&priv->lock);
124 gst_rtsp_stream_finalize (GObject * obj)
126 GstRTSPStream *stream;
127 GstRTSPStreamPrivate *priv;
129 stream = GST_RTSP_STREAM (obj);
132 GST_DEBUG ("finalize stream %p", stream);
134 /* we really need to be unjoined now */
135 g_return_if_fail (!priv->is_joined);
138 gst_rtsp_address_free (priv->addr);
140 g_object_unref (priv->pool);
141 gst_object_unref (priv->payloader);
142 gst_object_unref (priv->srcpad);
143 g_mutex_clear (&priv->lock);
145 G_OBJECT_CLASS (gst_rtsp_stream_parent_class)->finalize (obj);
149 * gst_rtsp_stream_new:
152 * @payloader: a #GstElement
154 * Create a new media stream with index @idx that handles RTP data on
155 * @srcpad and has a payloader element @payloader.
157 * Returns: a new #GstRTSPStream
160 gst_rtsp_stream_new (guint idx, GstElement * payloader, GstPad * srcpad)
162 GstRTSPStreamPrivate *priv;
163 GstRTSPStream *stream;
165 g_return_val_if_fail (GST_IS_ELEMENT (payloader), NULL);
166 g_return_val_if_fail (GST_IS_PAD (srcpad), NULL);
167 g_return_val_if_fail (GST_PAD_IS_SRC (srcpad), NULL);
169 stream = g_object_new (GST_TYPE_RTSP_STREAM, NULL);
172 priv->payloader = gst_object_ref (payloader);
173 priv->srcpad = gst_object_ref (srcpad);
179 * gst_rtsp_stream_get_index:
180 * @stream: a #GstRTSPStream
182 * Get the stream index.
184 * Return: the stream index.
187 gst_rtsp_stream_get_index (GstRTSPStream * stream)
189 g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), -1);
191 return stream->priv->idx;
195 * gst_rtsp_stream_set_mtu:
196 * @stream: a #GstRTSPStream
199 * Configure the mtu in the payloader of @stream to @mtu.
202 gst_rtsp_stream_set_mtu (GstRTSPStream * stream, guint mtu)
204 GstRTSPStreamPrivate *priv;
206 g_return_if_fail (GST_IS_RTSP_STREAM (stream));
210 GST_LOG_OBJECT (stream, "set MTU %u", mtu);
212 g_object_set (G_OBJECT (priv->payloader), "mtu", mtu, NULL);
216 * gst_rtsp_stream_get_mtu:
217 * @stream: a #GstRTSPStream
219 * Get the configured MTU in the payloader of @stream.
221 * Returns: the MTU of the payloader.
224 gst_rtsp_stream_get_mtu (GstRTSPStream * stream)
226 GstRTSPStreamPrivate *priv;
229 g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), 0);
233 g_object_get (G_OBJECT (priv->payloader), "mtu", &mtu, NULL);
239 * gst_rtsp_stream_set_address_pool:
240 * @stream: a #GstRTSPStream
241 * @pool: a #GstRTSPAddressPool
243 * configure @pool to be used as the address pool of @stream.
246 gst_rtsp_stream_set_address_pool (GstRTSPStream * stream,
247 GstRTSPAddressPool * pool)
249 GstRTSPStreamPrivate *priv;
250 GstRTSPAddressPool *old;
252 g_return_if_fail (GST_IS_RTSP_STREAM (stream));
256 GST_LOG_OBJECT (stream, "set address pool %p", pool);
258 g_mutex_lock (&priv->lock);
259 if ((old = priv->pool) != pool)
260 priv->pool = pool ? g_object_ref (pool) : NULL;
263 g_mutex_unlock (&priv->lock);
266 g_object_unref (old);
270 * gst_rtsp_stream_get_address_pool:
271 * @stream: a #GstRTSPStream
273 * Get the #GstRTSPAddressPool used as the address pool of @stream.
275 * Returns: (transfer full): the #GstRTSPAddressPool of @stream. g_object_unref() after
279 gst_rtsp_stream_get_address_pool (GstRTSPStream * stream)
281 GstRTSPStreamPrivate *priv;
282 GstRTSPAddressPool *result;
284 g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
288 g_mutex_lock (&priv->lock);
289 if ((result = priv->pool))
290 g_object_ref (result);
291 g_mutex_unlock (&priv->lock);
297 * gst_rtsp_stream_get_address:
298 * @stream: a #GstRTSPStream
300 * Get the multicast address of @stream.
302 * Returns: the #GstRTSPAddress of @stream or %NULL when no address could be
303 * allocated. gst_rtsp_address_free() after usage.
306 gst_rtsp_stream_get_address (GstRTSPStream * stream)
308 GstRTSPStreamPrivate *priv;
309 GstRTSPAddress *result;
311 g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
315 g_mutex_lock (&priv->lock);
316 if (priv->addr == NULL) {
317 if (priv->pool == NULL)
320 priv->addr = gst_rtsp_address_pool_acquire_address (priv->pool,
321 GST_RTSP_ADDRESS_FLAG_EVEN_PORT, 2);
322 if (priv->addr == NULL)
325 result = gst_rtsp_address_copy (priv->addr);
326 g_mutex_unlock (&priv->lock);
333 GST_ERROR_OBJECT (stream, "no address pool specified");
334 g_mutex_unlock (&priv->lock);
339 GST_ERROR_OBJECT (stream, "failed to acquire address from pool");
340 g_mutex_unlock (&priv->lock);
345 /* must be called with lock */
347 alloc_ports (GstRTSPStream * stream)
349 GstRTSPStreamPrivate *priv = stream->priv;
350 GstStateChangeReturn ret;
351 GstElement *udpsrc0, *udpsrc1;
352 GstElement *udpsink0, *udpsink1;
353 gint tmp_rtp, tmp_rtcp;
355 gint rtpport, rtcpport;
365 /* Start with random port */
369 host = "udp://[::0]";
371 host = "udp://0.0.0.0";
373 /* try to allocate 2 UDP ports, the RTP port should be an even
374 * number and the RTCP port should be the next (uneven) port */
376 udpsrc0 = gst_element_make_from_uri (GST_URI_SRC, host, NULL, NULL);
378 goto no_udp_protocol;
379 g_object_set (G_OBJECT (udpsrc0), "port", tmp_rtp, NULL);
381 ret = gst_element_set_state (udpsrc0, GST_STATE_PAUSED);
382 if (ret == GST_STATE_CHANGE_FAILURE) {
388 gst_element_set_state (udpsrc0, GST_STATE_NULL);
389 gst_object_unref (udpsrc0);
393 goto no_udp_protocol;
396 g_object_get (G_OBJECT (udpsrc0), "port", &tmp_rtp, NULL);
398 /* check if port is even */
399 if ((tmp_rtp & 1) != 0) {
400 /* port not even, close and allocate another */
404 gst_element_set_state (udpsrc0, GST_STATE_NULL);
405 gst_object_unref (udpsrc0);
411 /* allocate port+1 for RTCP now */
412 udpsrc1 = gst_element_make_from_uri (GST_URI_SRC, host, NULL, NULL);
414 goto no_udp_rtcp_protocol;
417 tmp_rtcp = tmp_rtp + 1;
418 g_object_set (G_OBJECT (udpsrc1), "port", tmp_rtcp, NULL);
420 ret = gst_element_set_state (udpsrc1, GST_STATE_PAUSED);
421 /* tmp_rtcp port is busy already : retry to make rtp/rtcp pair */
422 if (ret == GST_STATE_CHANGE_FAILURE) {
427 gst_element_set_state (udpsrc0, GST_STATE_NULL);
428 gst_object_unref (udpsrc0);
430 gst_element_set_state (udpsrc1, GST_STATE_NULL);
431 gst_object_unref (udpsrc1);
436 /* all fine, do port check */
437 g_object_get (G_OBJECT (udpsrc0), "port", &rtpport, NULL);
438 g_object_get (G_OBJECT (udpsrc1), "port", &rtcpport, NULL);
440 /* this should not happen... */
441 if (rtpport != tmp_rtp || rtcpport != tmp_rtcp)
444 udpsink0 = gst_element_factory_make ("multiudpsink", NULL);
446 goto no_udp_protocol;
448 g_object_get (G_OBJECT (udpsrc0), "used-socket", &socket, NULL);
449 g_object_set (G_OBJECT (udpsink0), "socket", socket, NULL);
450 g_object_unref (socket);
451 g_object_set (G_OBJECT (udpsink0), "close-socket", FALSE, NULL);
453 udpsink1 = gst_element_factory_make ("multiudpsink", NULL);
455 goto no_udp_protocol;
457 if (g_object_class_find_property (G_OBJECT_GET_CLASS (udpsink0),
458 "send-duplicates")) {
459 g_object_set (G_OBJECT (udpsink0), "send-duplicates", FALSE, NULL);
460 g_object_set (G_OBJECT (udpsink1), "send-duplicates", FALSE, NULL);
463 ("old multiudpsink version found without send-duplicates property");
466 if (g_object_class_find_property (G_OBJECT_GET_CLASS (udpsink0),
468 g_object_set (G_OBJECT (udpsink0), "buffer-size", priv->buffer_size, NULL);
470 GST_WARNING ("multiudpsink version found without buffer-size property");
473 g_object_get (G_OBJECT (udpsrc1), "used-socket", &socket, NULL);
474 g_object_set (G_OBJECT (udpsink1), "socket", socket, NULL);
475 g_object_unref (socket);
476 g_object_set (G_OBJECT (udpsink1), "close-socket", FALSE, NULL);
477 g_object_set (G_OBJECT (udpsink1), "sync", FALSE, NULL);
478 g_object_set (G_OBJECT (udpsink1), "async", FALSE, NULL);
479 g_object_set (G_OBJECT (udpsink0), "auto-multicast", FALSE, NULL);
480 g_object_set (G_OBJECT (udpsink0), "loop", FALSE, NULL);
481 g_object_set (G_OBJECT (udpsink1), "auto-multicast", FALSE, NULL);
482 g_object_set (G_OBJECT (udpsink1), "loop", FALSE, NULL);
484 /* we keep these elements, we will further configure them when the
485 * client told us to really use the UDP ports. */
486 priv->udpsrc[0] = udpsrc0;
487 priv->udpsrc[1] = udpsrc1;
488 priv->udpsink[0] = udpsink0;
489 priv->udpsink[1] = udpsink1;
490 priv->server_port.min = rtpport;
491 priv->server_port.max = rtcpport;
504 no_udp_rtcp_protocol:
515 gst_element_set_state (udpsrc0, GST_STATE_NULL);
516 gst_object_unref (udpsrc0);
519 gst_element_set_state (udpsrc1, GST_STATE_NULL);
520 gst_object_unref (udpsrc1);
523 gst_element_set_state (udpsink0, GST_STATE_NULL);
524 gst_object_unref (udpsink0);
527 gst_element_set_state (udpsink1, GST_STATE_NULL);
528 gst_object_unref (udpsink1);
535 * gst_rtsp_stream_get_server_port:
536 * @stream: a #GstRTSPStream
537 * @server_port: (out): result server port
539 * Fill @server_port with the port pair used by the server. This function can
540 * only be called when @stream has been joined.
543 gst_rtsp_stream_get_server_port (GstRTSPStream * stream,
544 GstRTSPRange * server_port)
546 GstRTSPStreamPrivate *priv;
548 g_return_if_fail (GST_IS_RTSP_STREAM (stream));
550 g_return_if_fail (priv->is_joined);
552 g_mutex_lock (&priv->lock);
554 *server_port = priv->server_port;
555 g_mutex_unlock (&priv->lock);
559 * gst_rtsp_stream_get_ssrc:
560 * @stream: a #GstRTSPStream
561 * @ssrc: (out): result ssrc
563 * Get the SSRC used by the RTP session of this stream. This function can only
564 * be called when @stream has been joined.
567 gst_rtsp_stream_get_ssrc (GstRTSPStream * stream, guint * ssrc)
569 GstRTSPStreamPrivate *priv;
571 g_return_if_fail (GST_IS_RTSP_STREAM (stream));
573 g_return_if_fail (priv->is_joined);
575 g_mutex_lock (&priv->lock);
576 if (ssrc && priv->session)
577 g_object_get (priv->session, "internal-ssrc", ssrc, NULL);
578 g_mutex_unlock (&priv->lock);
581 /* executed from streaming thread */
583 caps_notify (GstPad * pad, GParamSpec * unused, GstRTSPStream * stream)
585 GstRTSPStreamPrivate *priv = stream->priv;
586 GstCaps *newcaps, *oldcaps;
588 newcaps = gst_pad_get_current_caps (pad);
590 GST_INFO ("stream %p received caps %p, %" GST_PTR_FORMAT, stream, newcaps,
593 g_mutex_lock (&priv->lock);
594 oldcaps = priv->caps;
595 priv->caps = newcaps;
596 g_mutex_unlock (&priv->lock);
599 gst_caps_unref (oldcaps);
603 dump_structure (const GstStructure * s)
607 sstr = gst_structure_to_string (s);
608 GST_INFO ("structure: %s", sstr);
612 static GstRTSPStreamTransport *
613 find_transport (GstRTSPStream * stream, const gchar * rtcp_from)
615 GstRTSPStreamPrivate *priv = stream->priv;
617 GstRTSPStreamTransport *result = NULL;
622 if (rtcp_from == NULL)
625 tmp = g_strrstr (rtcp_from, ":");
629 port = atoi (tmp + 1);
630 dest = g_strndup (rtcp_from, tmp - rtcp_from);
632 g_mutex_lock (&priv->lock);
633 GST_INFO ("finding %s:%d in %d transports", dest, port,
634 g_list_length (priv->transports));
636 for (walk = priv->transports; walk; walk = g_list_next (walk)) {
637 GstRTSPStreamTransport *trans = walk->data;
638 const GstRTSPTransport *tr;
641 tr = gst_rtsp_stream_transport_get_transport (trans);
643 min = tr->client_port.min;
644 max = tr->client_port.max;
646 if ((strcmp (tr->destination, dest) == 0) && (min == port || max == port)) {
651 g_mutex_unlock (&priv->lock);
658 static GstRTSPStreamTransport *
659 check_transport (GObject * source, GstRTSPStream * stream)
662 GstRTSPStreamTransport *trans;
664 /* see if we have a stream to match with the origin of the RTCP packet */
665 trans = g_object_get_qdata (source, ssrc_stream_map_key);
667 g_object_get (source, "stats", &stats, NULL);
669 const gchar *rtcp_from;
671 dump_structure (stats);
673 rtcp_from = gst_structure_get_string (stats, "rtcp-from");
674 if ((trans = find_transport (stream, rtcp_from))) {
675 GST_INFO ("%p: found transport %p for source %p", stream, trans,
677 g_object_set_qdata (source, ssrc_stream_map_key, trans);
679 gst_structure_free (stats);
687 on_new_ssrc (GObject * session, GObject * source, GstRTSPStream * stream)
689 GstRTSPStreamTransport *trans;
691 GST_INFO ("%p: new source %p", stream, source);
693 trans = check_transport (source, stream);
696 GST_INFO ("%p: source %p for transport %p", stream, source, trans);
700 on_ssrc_sdes (GObject * session, GObject * source, GstRTSPStream * stream)
702 GST_INFO ("%p: new SDES %p", stream, source);
706 on_ssrc_active (GObject * session, GObject * source, GstRTSPStream * stream)
708 GstRTSPStreamTransport *trans;
710 trans = check_transport (source, stream);
713 GST_INFO ("%p: source %p in transport %p is active", stream, source, trans);
714 gst_rtsp_stream_transport_keep_alive (trans);
719 g_object_get (source, "stats", &stats, NULL);
721 dump_structure (stats);
722 gst_structure_free (stats);
729 on_bye_ssrc (GObject * session, GObject * source, GstRTSPStream * stream)
731 GST_INFO ("%p: source %p bye", stream, source);
735 on_bye_timeout (GObject * session, GObject * source, GstRTSPStream * stream)
737 GstRTSPStreamTransport *trans;
739 GST_INFO ("%p: source %p bye timeout", stream, source);
741 if ((trans = g_object_get_qdata (source, ssrc_stream_map_key))) {
742 gst_rtsp_stream_transport_set_timed_out (trans, TRUE);
747 on_timeout (GObject * session, GObject * source, GstRTSPStream * stream)
749 GstRTSPStreamTransport *trans;
751 GST_INFO ("%p: source %p timeout", stream, source);
753 if ((trans = g_object_get_qdata (source, ssrc_stream_map_key))) {
754 gst_rtsp_stream_transport_set_timed_out (trans, TRUE);
759 handle_new_sample (GstAppSink * sink, gpointer user_data)
761 GstRTSPStreamPrivate *priv;
765 GstRTSPStream *stream;
767 sample = gst_app_sink_pull_sample (sink);
771 stream = (GstRTSPStream *) user_data;
773 buffer = gst_sample_get_buffer (sample);
775 g_mutex_lock (&priv->lock);
776 for (walk = priv->transports; walk; walk = g_list_next (walk)) {
777 GstRTSPStreamTransport *tr = (GstRTSPStreamTransport *) walk->data;
779 if (GST_ELEMENT_CAST (sink) == priv->appsink[0]) {
780 gst_rtsp_stream_transport_send_rtp (tr, buffer);
782 gst_rtsp_stream_transport_send_rtcp (tr, buffer);
785 g_mutex_unlock (&priv->lock);
787 gst_sample_unref (sample);
792 static GstAppSinkCallbacks sink_cb = {
793 NULL, /* not interested in EOS */
794 NULL, /* not interested in preroll samples */
799 * gst_rtsp_stream_join_bin:
800 * @stream: a #GstRTSPStream
801 * @bin: a #GstBin to join
802 * @rtpbin: a rtpbin element in @bin
803 * @state: the target state of the new elements
805 * Join the #Gstbin @bin that contains the element @rtpbin.
807 * @stream will link to @rtpbin, which must be inside @bin. The elements
808 * added to @bin will be set to the state given in @state.
810 * Returns: %TRUE on success.
813 gst_rtsp_stream_join_bin (GstRTSPStream * stream, GstBin * bin,
814 GstElement * rtpbin, GstState state)
816 GstRTSPStreamPrivate *priv;
819 GstPad *pad, *teepad, *queuepad, *selpad;
820 GstPadLinkReturn ret;
822 g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
823 g_return_val_if_fail (GST_IS_BIN (bin), FALSE);
824 g_return_val_if_fail (GST_IS_ELEMENT (rtpbin), FALSE);
828 g_mutex_lock (&priv->lock);
832 /* create a session with the same index as the stream */
835 GST_INFO ("stream %p joining bin as session %d", stream, idx);
837 if (!alloc_ports (stream))
840 /* get a pad for sending RTP */
841 name = g_strdup_printf ("send_rtp_sink_%u", idx);
842 priv->send_rtp_sink = gst_element_get_request_pad (rtpbin, name);
844 /* link the RTP pad to the session manager, it should not really fail unless
845 * this is not really an RTP pad */
846 ret = gst_pad_link (priv->srcpad, priv->send_rtp_sink);
847 if (ret != GST_PAD_LINK_OK)
850 /* get pads from the RTP session element for sending and receiving
852 name = g_strdup_printf ("send_rtp_src_%u", idx);
853 priv->send_src[0] = gst_element_get_static_pad (rtpbin, name);
855 name = g_strdup_printf ("send_rtcp_src_%u", idx);
856 priv->send_src[1] = gst_element_get_request_pad (rtpbin, name);
858 name = g_strdup_printf ("recv_rtp_sink_%u", idx);
859 priv->recv_sink[0] = gst_element_get_request_pad (rtpbin, name);
861 name = g_strdup_printf ("recv_rtcp_sink_%u", idx);
862 priv->recv_sink[1] = gst_element_get_request_pad (rtpbin, name);
865 /* get the session */
866 g_signal_emit_by_name (rtpbin, "get-internal-session", idx, &priv->session);
868 g_signal_connect (priv->session, "on-new-ssrc", (GCallback) on_new_ssrc,
870 g_signal_connect (priv->session, "on-ssrc-sdes", (GCallback) on_ssrc_sdes,
872 g_signal_connect (priv->session, "on-ssrc-active",
873 (GCallback) on_ssrc_active, stream);
874 g_signal_connect (priv->session, "on-bye-ssrc", (GCallback) on_bye_ssrc,
876 g_signal_connect (priv->session, "on-bye-timeout",
877 (GCallback) on_bye_timeout, stream);
878 g_signal_connect (priv->session, "on-timeout", (GCallback) on_timeout,
881 for (i = 0; i < 2; i++) {
882 /* For the sender we create this bit of pipeline for both
883 * RTP and RTCP. Sync and preroll are enabled on udpsink so
884 * we need to add a queue before appsink to make the pipeline
885 * not block. For the TCP case, we want to pump data to the
886 * client as fast as possible anyway.
888 * .--------. .-----. .---------.
889 * | rtpbin | | tee | | udpsink |
890 * | send->sink src->sink |
891 * '--------' | | '---------'
892 * | | .---------. .---------.
893 * | | | queue | | appsink |
894 * | src->sink src->sink |
895 * '-----' '---------' '---------'
897 /* make tee for RTP/RTCP */
898 priv->tee[i] = gst_element_factory_make ("tee", NULL);
899 gst_bin_add (bin, priv->tee[i]);
901 /* and link to rtpbin send pad */
902 pad = gst_element_get_static_pad (priv->tee[i], "sink");
903 gst_pad_link (priv->send_src[i], pad);
904 gst_object_unref (pad);
907 gst_bin_add (bin, priv->udpsink[i]);
909 /* link tee to udpsink */
910 teepad = gst_element_get_request_pad (priv->tee[i], "src_%u");
911 pad = gst_element_get_static_pad (priv->udpsink[i], "sink");
912 gst_pad_link (teepad, pad);
913 gst_object_unref (pad);
914 gst_object_unref (teepad);
917 priv->appqueue[i] = gst_element_factory_make ("queue", NULL);
918 gst_bin_add (bin, priv->appqueue[i]);
919 /* and link to tee */
920 teepad = gst_element_get_request_pad (priv->tee[i], "src_%u");
921 pad = gst_element_get_static_pad (priv->appqueue[i], "sink");
922 gst_pad_link (teepad, pad);
923 gst_object_unref (pad);
924 gst_object_unref (teepad);
927 priv->appsink[i] = gst_element_factory_make ("appsink", NULL);
928 g_object_set (priv->appsink[i], "async", FALSE, "sync", FALSE, NULL);
929 g_object_set (priv->appsink[i], "emit-signals", FALSE, NULL);
930 gst_bin_add (bin, priv->appsink[i]);
931 gst_app_sink_set_callbacks (GST_APP_SINK_CAST (priv->appsink[i]),
932 &sink_cb, stream, NULL);
933 /* and link to queue */
934 queuepad = gst_element_get_static_pad (priv->appqueue[i], "src");
935 pad = gst_element_get_static_pad (priv->appsink[i], "sink");
936 gst_pad_link (queuepad, pad);
937 gst_object_unref (pad);
938 gst_object_unref (queuepad);
940 /* For the receiver we create this bit of pipeline for both
941 * RTP and RTCP. We receive RTP/RTCP on appsrc and udpsrc
942 * and it is all funneled into the rtpbin receive pad.
944 * .--------. .--------. .--------.
945 * | udpsrc | | funnel | | rtpbin |
946 * | src->sink src->sink |
947 * '--------' | | '--------'
951 * '--------' '--------'
953 /* make funnel for the RTP/RTCP receivers */
954 priv->funnel[i] = gst_element_factory_make ("funnel", NULL);
955 gst_bin_add (bin, priv->funnel[i]);
957 pad = gst_element_get_static_pad (priv->funnel[i], "src");
958 gst_pad_link (pad, priv->recv_sink[i]);
959 gst_object_unref (pad);
961 /* we set and keep these to playing so that they don't cause NO_PREROLL return
963 gst_element_set_state (priv->udpsrc[i], GST_STATE_PLAYING);
964 gst_element_set_locked_state (priv->udpsrc[i], TRUE);
966 gst_bin_add (bin, priv->udpsrc[i]);
967 /* and link to the funnel */
968 selpad = gst_element_get_request_pad (priv->funnel[i], "sink_%u");
969 pad = gst_element_get_static_pad (priv->udpsrc[i], "src");
970 gst_pad_link (pad, selpad);
971 gst_object_unref (pad);
972 gst_object_unref (selpad);
974 /* make and add appsrc */
975 priv->appsrc[i] = gst_element_factory_make ("appsrc", NULL);
976 gst_bin_add (bin, priv->appsrc[i]);
977 /* and link to the funnel */
978 selpad = gst_element_get_request_pad (priv->funnel[i], "sink_%u");
979 pad = gst_element_get_static_pad (priv->appsrc[i], "src");
980 gst_pad_link (pad, selpad);
981 gst_object_unref (pad);
982 gst_object_unref (selpad);
984 /* check if we need to set to a special state */
985 if (state != GST_STATE_NULL) {
986 gst_element_set_state (priv->udpsink[i], state);
987 gst_element_set_state (priv->appsink[i], state);
988 gst_element_set_state (priv->appqueue[i], state);
989 gst_element_set_state (priv->tee[i], state);
990 gst_element_set_state (priv->funnel[i], state);
991 gst_element_set_state (priv->appsrc[i], state);
995 /* be notified of caps changes */
996 priv->caps_sig = g_signal_connect (priv->send_rtp_sink, "notify::caps",
997 (GCallback) caps_notify, stream);
999 priv->is_joined = TRUE;
1000 g_mutex_unlock (&priv->lock);
1007 g_mutex_unlock (&priv->lock);
1012 g_mutex_unlock (&priv->lock);
1013 GST_WARNING ("failed to allocate ports %d", idx);
1018 GST_WARNING ("failed to link stream %d", idx);
1019 gst_object_unref (priv->send_rtp_sink);
1020 priv->send_rtp_sink = NULL;
1021 g_mutex_unlock (&priv->lock);
1027 * gst_rtsp_stream_leave_bin:
1028 * @stream: a #GstRTSPStream
1030 * @rtpbin: a rtpbin #GstElement
1032 * Remove the elements of @stream from @bin. @bin must be set
1033 * to the NULL state before calling this.
1035 * Return: %TRUE on success.
1038 gst_rtsp_stream_leave_bin (GstRTSPStream * stream, GstBin * bin,
1039 GstElement * rtpbin)
1041 GstRTSPStreamPrivate *priv;
1044 g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
1045 g_return_val_if_fail (GST_IS_BIN (bin), FALSE);
1046 g_return_val_if_fail (GST_IS_ELEMENT (rtpbin), FALSE);
1048 priv = stream->priv;
1050 g_mutex_lock (&priv->lock);
1051 if (!priv->is_joined)
1052 goto was_not_joined;
1054 /* all transports must be removed by now */
1055 g_return_val_if_fail (priv->transports == NULL, FALSE);
1057 GST_INFO ("stream %p leaving bin", stream);
1059 gst_pad_unlink (priv->srcpad, priv->send_rtp_sink);
1060 g_signal_handler_disconnect (priv->send_rtp_sink, priv->caps_sig);
1061 gst_element_release_request_pad (rtpbin, priv->send_rtp_sink);
1062 gst_object_unref (priv->send_rtp_sink);
1063 priv->send_rtp_sink = NULL;
1065 for (i = 0; i < 2; i++) {
1066 /* and set udpsrc to NULL now before removing */
1067 gst_element_set_locked_state (priv->udpsrc[i], FALSE);
1068 gst_element_set_state (priv->udpsrc[i], GST_STATE_NULL);
1070 /* removing them should also nicely release the request
1071 * pads when they finalize */
1072 gst_bin_remove (bin, priv->udpsrc[i]);
1073 gst_bin_remove (bin, priv->udpsink[i]);
1074 gst_bin_remove (bin, priv->appsrc[i]);
1075 gst_bin_remove (bin, priv->appsink[i]);
1076 gst_bin_remove (bin, priv->appqueue[i]);
1077 gst_bin_remove (bin, priv->tee[i]);
1078 gst_bin_remove (bin, priv->funnel[i]);
1080 gst_element_release_request_pad (rtpbin, priv->recv_sink[i]);
1081 gst_object_unref (priv->recv_sink[i]);
1082 priv->recv_sink[i] = NULL;
1084 priv->udpsrc[i] = NULL;
1085 priv->udpsink[i] = NULL;
1086 priv->appsrc[i] = NULL;
1087 priv->appsink[i] = NULL;
1088 priv->appqueue[i] = NULL;
1089 priv->tee[i] = NULL;
1090 priv->funnel[i] = NULL;
1092 gst_object_unref (priv->send_src[0]);
1093 priv->send_src[0] = NULL;
1095 gst_element_release_request_pad (rtpbin, priv->send_src[1]);
1096 gst_object_unref (priv->send_src[1]);
1097 priv->send_src[1] = NULL;
1099 g_object_unref (priv->session);
1101 gst_caps_unref (priv->caps);
1103 priv->is_joined = FALSE;
1104 g_mutex_unlock (&priv->lock);
1115 * gst_rtsp_stream_get_rtpinfo:
1116 * @stream: a #GstRTSPStream
1117 * @rtptime: result RTP timestamp
1118 * @seq: result RTP seqnum
1120 * Retrieve the current rtptime and seq. This is used to
1121 * construct a RTPInfo reply header.
1123 * Returns: %TRUE when rtptime and seq could be determined.
1126 gst_rtsp_stream_get_rtpinfo (GstRTSPStream * stream,
1127 guint * rtptime, guint * seq)
1129 GstRTSPStreamPrivate *priv;
1130 GObjectClass *payobjclass;
1132 g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
1133 g_return_val_if_fail (rtptime != NULL, FALSE);
1134 g_return_val_if_fail (seq != NULL, FALSE);
1136 priv = stream->priv;
1138 payobjclass = G_OBJECT_GET_CLASS (priv->payloader);
1140 if (!g_object_class_find_property (payobjclass, "seqnum") ||
1141 !g_object_class_find_property (payobjclass, "timestamp"))
1144 g_object_get (priv->payloader, "seqnum", seq, "timestamp", rtptime, NULL);
1150 * gst_rtsp_stream_get_caps:
1151 * @stream: a #GstRTSPStream
1153 * Retrieve the current caps of @stream.
1155 * Returns: (transfer full): the #GstCaps of @stream. use gst_caps_unref()
1159 gst_rtsp_stream_get_caps (GstRTSPStream * stream)
1161 GstRTSPStreamPrivate *priv;
1164 g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
1166 priv = stream->priv;
1168 g_mutex_lock (&priv->lock);
1169 if ((result = priv->caps))
1170 gst_caps_ref (result);
1171 g_mutex_unlock (&priv->lock);
1177 * gst_rtsp_stream_recv_rtp:
1178 * @stream: a #GstRTSPStream
1179 * @buffer: (transfer full): a #GstBuffer
1181 * Handle an RTP buffer for the stream. This method is usually called when a
1182 * message has been received from a client using the TCP transport.
1184 * This function takes ownership of @buffer.
1186 * Returns: a GstFlowReturn.
1189 gst_rtsp_stream_recv_rtp (GstRTSPStream * stream, GstBuffer * buffer)
1191 GstRTSPStreamPrivate *priv;
1193 GstElement *element;
1195 g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), GST_FLOW_ERROR);
1196 priv = stream->priv;
1197 g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
1198 g_return_val_if_fail (priv->is_joined, FALSE);
1200 g_mutex_lock (&priv->lock);
1201 element = gst_object_ref (priv->appsrc[0]);
1202 g_mutex_unlock (&priv->lock);
1204 ret = gst_app_src_push_buffer (GST_APP_SRC_CAST (element), buffer);
1206 gst_object_unref (element);
1212 * gst_rtsp_stream_recv_rtcp:
1213 * @stream: a #GstRTSPStream
1214 * @buffer: (transfer full): a #GstBuffer
1216 * Handle an RTCP buffer for the stream. This method is usually called when a
1217 * message has been received from a client using the TCP transport.
1219 * This function takes ownership of @buffer.
1221 * Returns: a GstFlowReturn.
1224 gst_rtsp_stream_recv_rtcp (GstRTSPStream * stream, GstBuffer * buffer)
1226 GstRTSPStreamPrivate *priv;
1228 GstElement *element;
1230 g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), GST_FLOW_ERROR);
1231 priv = stream->priv;
1232 g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
1233 g_return_val_if_fail (priv->is_joined, FALSE);
1235 g_mutex_lock (&priv->lock);
1236 element = gst_object_ref (priv->appsrc[1]);
1237 g_mutex_unlock (&priv->lock);
1239 ret = gst_app_src_push_buffer (GST_APP_SRC_CAST (element), buffer);
1241 gst_object_unref (element);
1246 /* must be called with lock */
1248 update_transport (GstRTSPStream * stream, GstRTSPStreamTransport * trans,
1251 GstRTSPStreamPrivate *priv = stream->priv;
1252 const GstRTSPTransport *tr;
1254 tr = gst_rtsp_stream_transport_get_transport (trans);
1256 switch (tr->lower_transport) {
1257 case GST_RTSP_LOWER_TRANS_UDP:
1258 case GST_RTSP_LOWER_TRANS_UDP_MCAST:
1264 dest = tr->destination;
1265 if (tr->lower_transport == GST_RTSP_LOWER_TRANS_UDP_MCAST) {
1270 min = tr->client_port.min;
1271 max = tr->client_port.max;
1275 GST_INFO ("adding %s:%d-%d", dest, min, max);
1276 g_signal_emit_by_name (priv->udpsink[0], "add", dest, min, NULL);
1277 g_signal_emit_by_name (priv->udpsink[1], "add", dest, max, NULL);
1279 GST_INFO ("setting ttl-mc %d", ttl);
1280 g_object_set (G_OBJECT (priv->udpsink[0]), "ttl-mc", ttl, NULL);
1281 g_object_set (G_OBJECT (priv->udpsink[1]), "ttl-mc", ttl, NULL);
1283 priv->transports = g_list_prepend (priv->transports, trans);
1285 GST_INFO ("removing %s:%d-%d", dest, min, max);
1286 g_signal_emit_by_name (priv->udpsink[0], "remove", dest, min, NULL);
1287 g_signal_emit_by_name (priv->udpsink[1], "remove", dest, max, NULL);
1288 priv->transports = g_list_remove (priv->transports, trans);
1292 case GST_RTSP_LOWER_TRANS_TCP:
1294 GST_INFO ("adding TCP %s", tr->destination);
1295 priv->transports = g_list_prepend (priv->transports, trans);
1297 GST_INFO ("removing TCP %s", tr->destination);
1298 priv->transports = g_list_remove (priv->transports, trans);
1302 goto unknown_transport;
1309 GST_INFO ("Unknown transport %d", tr->lower_transport);
1316 * gst_rtsp_stream_add_transport:
1317 * @stream: a #GstRTSPStream
1318 * @trans: a #GstRTSPStreamTransport
1320 * Add the transport in @trans to @stream. The media of @stream will
1321 * then also be send to the values configured in @trans.
1323 * @stream must be joined to a bin.
1325 * @trans must contain a valid #GstRTSPTransport.
1327 * Returns: %TRUE if @trans was added
1330 gst_rtsp_stream_add_transport (GstRTSPStream * stream,
1331 GstRTSPStreamTransport * trans)
1333 GstRTSPStreamPrivate *priv;
1336 g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
1337 priv = stream->priv;
1338 g_return_val_if_fail (GST_IS_RTSP_STREAM_TRANSPORT (trans), FALSE);
1339 g_return_val_if_fail (priv->is_joined, FALSE);
1341 g_mutex_lock (&priv->lock);
1342 res = update_transport (stream, trans, TRUE);
1343 g_mutex_unlock (&priv->lock);
1349 * gst_rtsp_stream_remove_transport:
1350 * @stream: a #GstRTSPStream
1351 * @trans: a #GstRTSPStreamTransport
1353 * Remove the transport in @trans from @stream. The media of @stream will
1354 * not be sent to the values configured in @trans.
1356 * @stream must be joined to a bin.
1358 * @trans must contain a valid #GstRTSPTransport.
1360 * Returns: %TRUE if @trans was removed
1363 gst_rtsp_stream_remove_transport (GstRTSPStream * stream,
1364 GstRTSPStreamTransport * trans)
1366 GstRTSPStreamPrivate *priv;
1369 g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
1370 priv = stream->priv;
1371 g_return_val_if_fail (GST_IS_RTSP_STREAM_TRANSPORT (trans), FALSE);
1372 g_return_val_if_fail (priv->is_joined, FALSE);
1374 g_mutex_lock (&priv->lock);
1375 res = update_transport (stream, trans, FALSE);
1376 g_mutex_unlock (&priv->lock);