Commit 9bed89c3 authored by Wim Taymans's avatar Wim Taymans

rtsp: use RTCP to keep the session alive

Use the RTCP rtcp-from stats field to find the associated session and use this
to keep the session alive.
parent 7bbdf7bf
This diff is collapsed.
......@@ -17,6 +17,8 @@
* Boston, MA 02111-1307, USA.
*/
#include <string.h>
#include <gst/app/gstappsrc.h>
#include <gst/app/gstappsink.h>
......@@ -39,6 +41,8 @@ enum
SIGNAL_LAST
};
static GQuark ssrc_stream_map_key;
static void gst_rtsp_media_get_property (GObject *object, guint propid,
GValue *value, GParamSpec *pspec);
static void gst_rtsp_media_set_property (GObject *object, guint propid,
......@@ -87,6 +91,8 @@ gst_rtsp_media_class_init (GstRTSPMediaClass * klass)
g_critical ("could not start bus thread: %s", error->message);
}
klass->handle_message = default_handle_message;
ssrc_stream_map_key = g_quark_from_static_string ("GstRTSPServer.stream");
}
static void
......@@ -106,6 +112,8 @@ gst_rtsp_media_stream_free (GstRTSPMediaStream *stream)
if (stream->caps)
gst_caps_unref (stream->caps);
g_list_free (stream->transports);
g_free (stream);
}
......@@ -700,48 +708,112 @@ dump_structure (const GstStructure *s)
g_free (sstr);
}
static GstRTSPMediaTrans *
find_transport (GstRTSPMediaStream *stream, const gchar *rtcp_from)
{
GList *walk;
GstRTSPMediaTrans *result = NULL;
const gchar *dest;
guint port;
if (rtcp_from == NULL)
return NULL;
dest = g_strrstr (rtcp_from, ":");
if (dest == NULL)
return NULL;
port = atoi (dest + 1);
dest = g_strndup (rtcp_from, dest - rtcp_from);
g_message ("finding %s:%d", dest, port);
for (walk = stream->transports; walk; walk = g_list_next (walk)) {
GstRTSPMediaTrans *trans = walk->data;
gint min, max;
min = trans->transport->client_port.min;
max = trans->transport->client_port.max;
if ((strcmp (trans->transport->destination, dest) == 0) && (min == port || max == port)) {
result = trans;
break;
}
}
return result;
}
static void
on_new_ssrc (GObject *session, GObject *source, GstRTSPMedia *media)
on_new_ssrc (GObject *session, GObject *source, GstRTSPMediaStream *stream)
{
g_message ("%p: new source %p", media, source);
GstStructure *stats;
GstRTSPMediaTrans *trans;
g_message ("%p: new source %p", stream, source);
/* see if we have a stream to match with the origin of the RTCP packet */
trans = g_object_get_qdata (source, ssrc_stream_map_key);
if (trans == NULL) {
g_object_get (source, "stats", &stats, NULL);
if (stats) {
const gchar *rtcp_from;
rtcp_from = gst_structure_get_string (stats, "rtcp-from");
if ((trans = find_transport (stream, rtcp_from))) {
g_message ("%p: found transport %p for source %p", stream, trans, source);
g_object_set_qdata (source, ssrc_stream_map_key, trans);
}
}
} else {
g_message ("%p: source %p for transport %p", stream, source, trans);
}
}
static void
on_ssrc_sdes (GObject *session, GObject *source, GstRTSPMedia *media)
on_ssrc_sdes (GObject *session, GObject *source, GstRTSPMediaStream *stream)
{
GstStructure *sdes;
g_message ("%p: new SDES %p", media, source);
g_message ("%p: new SDES %p", stream, source);
g_object_get (source, "sdes", &sdes, NULL);
dump_structure (sdes);
}
static void
on_ssrc_active (GObject *session, GObject *source, GstRTSPMedia *media)
on_ssrc_active (GObject *session, GObject *source, GstRTSPMediaStream *stream)
{
GstStructure *stats;
GstRTSPMediaTrans *trans;
trans = g_object_get_qdata (source, ssrc_stream_map_key);
g_message ("%p: source %p in transport %p is active", stream, trans, source);
if (trans && trans->keep_alive) {
trans->keep_alive (trans->ka_user_data);
}
g_message ("%p: source %p is active", media, source);
g_object_get (source, "stats", &stats, NULL);
dump_structure (stats);
gst_structure_free (stats);
}
static void
on_bye_ssrc (GObject *session, GObject *source, GstRTSPMedia *media)
on_bye_ssrc (GObject *session, GObject *source, GstRTSPMediaStream *stream)
{
g_message ("%p: source %p bye", media, source);
g_message ("%p: source %p bye", stream, source);
}
static void
on_bye_timeout (GObject *session, GObject *source, GstRTSPMedia *media)
on_bye_timeout (GObject *session, GObject *source, GstRTSPMediaStream *stream)
{
g_message ("%p: source %p bye timeout", media, source);
g_message ("%p: source %p bye timeout", stream, source);
}
static void
on_timeout (GObject *session, GObject *source, GstRTSPMedia *media)
on_timeout (GObject *session, GObject *source, GstRTSPMediaStream *stream)
{
g_message ("%p: source %p timeout", media, source);
g_message ("%p: source %p timeout", stream, source);
}
static GstFlowReturn
......@@ -836,17 +908,17 @@ setup_stream (GstRTSPMediaStream *stream, guint idx, GstRTSPMedia *media)
&stream->session);
g_signal_connect (stream->session, "on-new-ssrc", (GCallback) on_new_ssrc,
media);
stream);
g_signal_connect (stream->session, "on-ssrc-sdes", (GCallback) on_ssrc_sdes,
media);
stream);
g_signal_connect (stream->session, "on-ssrc-active", (GCallback) on_ssrc_active,
media);
stream);
g_signal_connect (stream->session, "on-bye-ssrc", (GCallback) on_bye_ssrc,
media);
stream);
g_signal_connect (stream->session, "on-bye-timeout", (GCallback) on_bye_timeout,
media);
stream);
g_signal_connect (stream->session, "on-timeout", (GCallback) on_timeout,
media);
stream);
/* link the RTP pad to the session manager */
ret = gst_pad_link (stream->srcpad, stream->send_rtp_sink);
......@@ -1361,12 +1433,14 @@ gst_rtsp_media_set_state (GstRTSPMedia *media, GstState state, GArray *transport
g_message ("adding %s:%d-%d", dest, min, max);
g_signal_emit_by_name (stream->udpsink[0], "add", dest, min, NULL);
g_signal_emit_by_name (stream->udpsink[1], "add", dest, max, NULL);
stream->transports = g_list_prepend (stream->transports, tr);
tr->active = TRUE;
media->active++;
} else if (remove && tr->active) {
g_message ("removing %s:%d-%d", dest, min, max);
g_signal_emit_by_name (stream->udpsink[0], "remove", dest, min, NULL);
g_signal_emit_by_name (stream->udpsink[1], "remove", dest, max, NULL);
stream->transports = g_list_remove (stream->transports, tr);
tr->active = FALSE;
media->active--;
}
......
......@@ -41,7 +41,8 @@ typedef struct _GstRTSPMedia GstRTSPMedia;
typedef struct _GstRTSPMediaClass GstRTSPMediaClass;
typedef struct _GstRTSPMediaTrans GstRTSPMediaTrans;
typedef gboolean (*GstRTSPSendFunc) (GstBuffer *buffer, guint8 channel, gpointer user_data);
typedef gboolean (*GstRTSPSendFunc) (GstBuffer *buffer, guint8 channel, gpointer user_data);
typedef void (*GstRTSPKeepAliveFunc) (gpointer user_data);
/**
* GstRTSPMediaTrans:
......@@ -50,20 +51,33 @@ typedef gboolean (*GstRTSPSendFunc) (GstBuffer *buffer, guint8 channel, gpointer
* @send_rtcp: callback for sending RTCP messages
* @user_data: user data passed in the callbacks
* @notify: free function for the user_data.
* @keep_alive: keep alive callback
* @ka_user_data: data passed to @keep_alive
* @ka_notify: called when @ka_user_data is freed
* @active: if we are actively sending
* @timeout: if we timed out
* @transport: a transport description
* @rtpsource: the receiver rtp source object
*
* A Transport description for stream @idx
*/
struct _GstRTSPMediaTrans {
guint idx;
GstRTSPSendFunc send_rtp;
GstRTSPSendFunc send_rtcp;
gpointer user_data;
GDestroyNotify notify;
gboolean active;
GstRTSPSendFunc send_rtp;
GstRTSPSendFunc send_rtcp;
gpointer user_data;
GDestroyNotify notify;
GstRTSPTransport *transport;
GstRTSPKeepAliveFunc keep_alive;
gpointer ka_user_data;
GDestroyNotify ka_notify;
gboolean active;
gboolean timeout;
GstRTSPTransport *transport;
GObject *rtpsource;
};
/**
......
......@@ -74,6 +74,10 @@ gst_rtsp_session_free_stream (GstRTSPSessionStream *stream)
{
g_message ("free session stream %p", stream);
/* remove callbacks now */
gst_rtsp_session_stream_set_callbacks (stream, NULL, NULL, NULL, NULL);
gst_rtsp_session_stream_set_keepalive (stream, NULL, NULL, NULL);
if (stream->trans.transport)
gst_rtsp_transport_free (stream->trans.transport);
......@@ -308,7 +312,7 @@ gst_rtsp_session_media_get_stream (GstRTSPSessionMedia *media, guint idx)
result->trans.transport = NULL;
result->media_stream = media_stream;
g_array_insert_val (media->streams, idx, result);
g_array_index (media->streams, GstRTSPSessionStream *, idx) = result;
}
return result;
......@@ -512,6 +516,27 @@ gst_rtsp_session_stream_set_callbacks (GstRTSPSessionStream *stream,
stream->trans.notify = notify;
}
/**
* gst_rtsp_session_stream_set_keepalive:
* @stream: a #GstRTSPSessionStream
* @keep_alive: a callback called when the receiver is active
* @user_data: user data passed to callback
* @notify: called with the user_data when no longer needed.
*
* Install callbacks that will be called when RTCP packets are received from the
* receiver of @stream.
*/
void
gst_rtsp_session_stream_set_keepalive (GstRTSPSessionStream *stream,
GstRTSPKeepAliveFunc keep_alive, gpointer user_data, GDestroyNotify notify)
{
stream->trans.keep_alive = keep_alive;
if (stream->trans.ka_notify)
stream->trans.ka_notify (stream->trans.ka_user_data);
stream->trans.ka_user_data = user_data;
stream->trans.ka_notify = notify;
}
/**
* gst_rtsp_session_media_set_state:
* @media: a #GstRTSPSessionMedia
......
......@@ -146,6 +146,10 @@ void gst_rtsp_session_stream_set_callbacks (GstRTSPSessionStre
GstRTSPSendFunc send_rtcp,
gpointer user_data,
GDestroyNotify notify);
void gst_rtsp_session_stream_set_keepalive (GstRTSPSessionStream *stream,
GstRTSPKeepAliveFunc keep_alive,
gpointer user_data,
GDestroyNotify notify);
G_END_DECLS
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment