rtspsrc: reset-sync before play
[platform/upstream/gstreamer.git] / gst / rtsp / gstrtspsrc.c
index 902dc2c..e8b00c1 100644 (file)
@@ -89,6 +89,7 @@
 #include <stdio.h>
 #include <stdarg.h>
 
+#include <gst/net/gstnet.h>
 #include <gst/sdp/gstsdpmessage.h>
 #include <gst/rtp/gstrtppayloads.h>
 
@@ -119,7 +120,7 @@ GST_STATIC_PAD_TEMPLATE ("internalsink_%u",
 
 enum
 {
-  /* FILL ME */
+  SIGNAL_HANDLE_REQUEST,
   LAST_SIGNAL
 };
 
@@ -181,6 +182,8 @@ gst_rtsp_src_buffer_mode_get_type (void)
 #define DEFAULT_PROBATION        2
 #define DEFAULT_UDP_RECONNECT    TRUE
 #define DEFAULT_MULTICAST_IFACE  NULL
+#define DEFAULT_NTP_SYNC         FALSE
+#define DEFAULT_USE_PIPELINE_CLOCK      FALSE
 
 enum
 {
@@ -210,6 +213,8 @@ enum
   PROP_PROBATION,
   PROP_UDP_RECONNECT,
   PROP_MULTICAST_IFACE,
+  PROP_NTP_SYNC,
+  PROP_USE_PIPELINE_CLOCK,
   PROP_LAST
 };
 
@@ -238,6 +243,8 @@ static void gst_rtspsrc_set_property (GObject * object, guint prop_id,
 static void gst_rtspsrc_get_property (GObject * object, guint prop_id,
     GValue * value, GParamSpec * pspec);
 
+static GstClock *gst_rtspsrc_provide_clock (GstElement * element);
+
 static void gst_rtspsrc_uri_handler_init (gpointer g_iface,
     gpointer iface_data);
 
@@ -299,7 +306,8 @@ G_STMT_START {                                          \
   g_free (__txt);                                       \
 } G_STMT_END
 
-/*static guint gst_rtspsrc_signals[LAST_SIGNAL] = { 0 }; */
+static guint gst_rtspsrc_signals[LAST_SIGNAL] = { 0 };
+
 #define gst_rtspsrc_parent_class parent_class
 G_DEFINE_TYPE_WITH_CODE (GstRTSPSrc, gst_rtspsrc, GST_TYPE_BIN,
     G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_rtspsrc_uri_handler_init));
@@ -533,7 +541,40 @@ gst_rtspsrc_class_init (GstRTSPSrcClass * klass)
           "The network interface on which to join the multicast group",
           DEFAULT_MULTICAST_IFACE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
+  g_object_class_install_property (gobject_class, PROP_NTP_SYNC,
+      g_param_spec_boolean ("ntp-sync", "Sync on NTP clock",
+          "Synchronize received streams to the NTP clock", DEFAULT_NTP_SYNC,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class, PROP_USE_PIPELINE_CLOCK,
+      g_param_spec_boolean ("use-pipeline-clock", "Use pipeline clock",
+          "Use the pipeline running-time to set the NTP time in the RTCP SR messages",
+          DEFAULT_USE_PIPELINE_CLOCK,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
+  /**
+   * GstRTSPSrc::handle-request:
+   * @rtspsrc: a #GstRTSPSrc
+   * @request: a #GstRTSPMessage
+   * @response: a #GstRTSPMessage
+   *
+   * Handle a server request in @request and prepare @response.
+   *
+   * This signal is called from the streaming thread, you should therefore not
+   * do any state changes on @rtspsrc because this might deadlock. If you want
+   * to modify the state as a result of this signal, post a
+   * #GST_MESSAGE_REQUEST_STATE message on the bus or signal the main thread
+   * in some other way.
+   *
+   * Since: 1.2
+   */
+  gst_rtspsrc_signals[SIGNAL_HANDLE_REQUEST] =
+      g_signal_new ("handle-request", G_TYPE_FROM_CLASS (klass), 0,
+      0, NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2,
+      G_TYPE_POINTER, G_TYPE_POINTER);
+
   gstelement_class->send_event = gst_rtspsrc_send_event;
+  gstelement_class->provide_clock = gst_rtspsrc_provide_clock;
   gstelement_class->change_state = gst_rtspsrc_change_state;
 
   gst_element_class_add_pad_template (gstelement_class,
@@ -579,6 +620,8 @@ gst_rtspsrc_init (GstRTSPSrc * src)
   src->probation = DEFAULT_PROBATION;
   src->udp_reconnect = DEFAULT_UDP_RECONNECT;
   src->multi_iface = g_strdup (DEFAULT_MULTICAST_IFACE);
+  src->ntp_sync = DEFAULT_NTP_SYNC;
+  src->use_pipeline_clock = DEFAULT_USE_PIPELINE_CLOCK;
 
   /* get a list of all extensions */
   src->extensions = gst_rtsp_ext_list_get ();
@@ -618,6 +661,8 @@ gst_rtspsrc_finalize (GObject * object)
     gst_sdp_message_free (rtspsrc->sdp);
     rtspsrc->sdp = NULL;
   }
+  if (rtspsrc->provided_clock)
+    gst_object_unref (rtspsrc->provided_clock);
 
   /* free locks */
   g_rec_mutex_clear (&rtspsrc->stream_rec_lock);
@@ -626,6 +671,18 @@ gst_rtspsrc_finalize (GObject * object)
   G_OBJECT_CLASS (parent_class)->finalize (object);
 }
 
+static GstClock *
+gst_rtspsrc_provide_clock (GstElement * element)
+{
+  GstRTSPSrc *src = GST_RTSPSRC (element);
+  GstClock *clock;
+
+  if ((clock = src->provided_clock) != NULL)
+    gst_object_ref (clock);
+
+  return clock;
+}
+
 /* a proxy string of the format [user:passwd@]host[:port] */
 static gboolean
 gst_rtspsrc_set_proxy (GstRTSPSrc * rtsp, const gchar * proxy)
@@ -807,6 +864,12 @@ gst_rtspsrc_set_property (GObject * object, guint prop_id, const GValue * value,
       else
         rtspsrc->multi_iface = g_value_dup_string (value);
       break;
+    case PROP_NTP_SYNC:
+      rtspsrc->ntp_sync = g_value_get_boolean (value);
+      break;
+    case PROP_USE_PIPELINE_CLOCK:
+      rtspsrc->use_pipeline_clock = g_value_get_boolean (value);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
@@ -923,6 +986,12 @@ gst_rtspsrc_get_property (GObject * object, guint prop_id, GValue * value,
     case PROP_MULTICAST_IFACE:
       g_value_set_string (value, rtspsrc->multi_iface);
       break;
+    case PROP_NTP_SYNC:
+      g_value_set_boolean (value, rtspsrc->ntp_sync);
+      break;
+    case PROP_USE_PIPELINE_CLOCK:
+      g_value_set_boolean (value, rtspsrc->use_pipeline_clock);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
@@ -1325,6 +1394,10 @@ gst_rtspsrc_cleanup (GstRTSPSrc * src)
     gst_event_unref (src->start_segment);
     src->start_segment = NULL;
   }
+  if (src->provided_clock) {
+    gst_object_unref (src->provided_clock);
+    src->provided_clock = NULL;
+  }
 }
 
 #define PARSE_INT(p, del, res)          \
@@ -1737,6 +1810,8 @@ again:
    * server told us to really use the UDP ports. */
   stream->udpsrc[0] = gst_object_ref_sink (udpsrc0);
   stream->udpsrc[1] = gst_object_ref_sink (udpsrc1);
+  gst_element_set_locked_state (stream->udpsrc[0], TRUE);
+  gst_element_set_locked_state (stream->udpsrc[1], TRUE);
 
   /* keep track of next available port number when we have a range
    * configured */
@@ -1783,12 +1858,30 @@ cleanup:
 }
 
 static void
+gst_rtspsrc_set_state (GstRTSPSrc * src, GstState state)
+{
+  GList *walk;
+
+  if (src->manager)
+    gst_element_set_state (GST_ELEMENT_CAST (src->manager), state);
+
+  for (walk = src->streams; walk; walk = g_list_next (walk)) {
+    GstRTSPStream *stream = (GstRTSPStream *) walk->data;
+    gint i;
+
+    for (i = 0; i < 2; i++) {
+      if (stream->udpsrc[i])
+        gst_element_set_state (stream->udpsrc[i], state);
+    }
+  }
+}
+
+static void
 gst_rtspsrc_flush (GstRTSPSrc * src, gboolean flush, gboolean playing)
 {
   GstEvent *event;
-  gint cmd, i;
+  gint cmd;
   GstState state;
-  GList *walk;
 
   if (flush) {
     event = gst_event_new_flush_start ();
@@ -1806,22 +1899,7 @@ gst_rtspsrc_flush (GstRTSPSrc * src, gboolean flush, gboolean playing)
   }
   gst_rtspsrc_push_event (src, event);
   gst_rtspsrc_loop_send_cmd (src, cmd, CMD_LOOP);
-
-  /* to manage jitterbuffer buffer mode */
-  if (src->manager)
-    gst_element_set_state (GST_ELEMENT_CAST (src->manager), state);
-
-  /* make running time start start at 0 again */
-  for (walk = src->streams; walk; walk = g_list_next (walk)) {
-    GstRTSPStream *stream = (GstRTSPStream *) walk->data;
-
-    for (i = 0; i < 2; i++) {
-      /* for udp case */
-      if (stream->udpsrc[i]) {
-        gst_element_set_state (stream->udpsrc[i], state);
-      }
-    }
-  }
+  gst_rtspsrc_set_state (src, state);
 }
 
 static GstRTSPResult
@@ -2489,9 +2567,8 @@ gst_rtspsrc_stream_configure_manager (GstRTSPSrc * src, GstRTSPStream * stream,
     /* configure the manager */
     if (src->manager == NULL) {
       GObjectClass *klass;
-      GstState target;
 
-      if (!(src->manager = gst_element_factory_make (manager, NULL))) {
+      if (!(src->manager = gst_element_factory_make (manager, "manager"))) {
         /* fallback */
         if (gst_rtsp_transport_get_manager (transport->trans, &manager, 1) < 0)
           goto no_manager;
@@ -2499,24 +2576,31 @@ gst_rtspsrc_stream_configure_manager (GstRTSPSrc * src, GstRTSPStream * stream,
         if (!manager)
           goto use_no_manager;
 
-        if (!(src->manager = gst_element_factory_make (manager, NULL)))
+        if (!(src->manager = gst_element_factory_make (manager, "manager")))
           goto manager_failed;
       }
 
       /* we manage this element */
+      gst_element_set_locked_state (src->manager, TRUE);
       gst_bin_add (GST_BIN_CAST (src), src->manager);
 
-      GST_OBJECT_LOCK (src);
-      target = GST_STATE_TARGET (src);
-      GST_OBJECT_UNLOCK (src);
-
-      ret = gst_element_set_state (src->manager, target);
+      ret = gst_element_set_state (src->manager, GST_STATE_PAUSED);
       if (ret == GST_STATE_CHANGE_FAILURE)
         goto start_manager_failure;
 
       g_object_set (src->manager, "latency", src->latency, NULL);
 
       klass = G_OBJECT_GET_CLASS (G_OBJECT (src->manager));
+
+      if (g_object_class_find_property (klass, "ntp-sync")) {
+        g_object_set (src->manager, "ntp-sync", src->ntp_sync, NULL);
+      }
+
+      if (g_object_class_find_property (klass, "use-pipeline-clock")) {
+        g_object_set (src->manager, "use-pipeline-clock",
+            src->use_pipeline_clock, NULL);
+      }
+
       if (g_object_class_find_property (klass, "drop-on-latency")) {
         g_object_set (src->manager, "drop-on-latency", src->drop_on_latency,
             NULL);
@@ -2849,11 +2933,14 @@ gst_rtspsrc_stream_configure_mcast (GstRTSPSrc * src, GstRTSPStream * stream,
           src->multi_iface, NULL);
 
     /* change state */
+    gst_element_set_locked_state (stream->udpsrc[0], TRUE);
     gst_element_set_state (stream->udpsrc[0], GST_STATE_PAUSED);
   }
 
   /* creating another UDP source for RTCP */
   if (max != -1) {
+    GstCaps *caps;
+
     uri = g_strdup_printf ("udp://%s:%d", destination, max);
     stream->udpsrc[1] =
         gst_element_make_from_uri (GST_URI_SRC, uri, NULL, NULL);
@@ -2861,6 +2948,10 @@ gst_rtspsrc_stream_configure_mcast (GstRTSPSrc * src, GstRTSPStream * stream,
     if (stream->udpsrc[1] == NULL)
       goto no_element;
 
+    caps = gst_caps_new_empty_simple ("application/x-rtcp");
+    g_object_set (stream->udpsrc[1], "caps", caps, NULL);
+    gst_caps_unref (caps);
+
     /* take ownership */
     gst_object_ref_sink (stream->udpsrc[1]);
 
@@ -2898,6 +2989,7 @@ gst_rtspsrc_stream_configure_udp (GstRTSPSrc * src, GstRTSPStream * stream,
   /* we manage the UDP elements now. For unicast, the UDP sources where
    * allocated in the stream when we suggested a transport. */
   if (stream->udpsrc[0]) {
+    gst_element_set_locked_state (stream->udpsrc[0], TRUE);
     gst_bin_add (GST_BIN_CAST (src), stream->udpsrc[0]);
 
     GST_DEBUG_OBJECT (src, "setting up UDP source");
@@ -2936,8 +3028,15 @@ gst_rtspsrc_stream_configure_udp (GstRTSPSrc * src, GstRTSPStream * stream,
 
   /* RTCP port */
   if (stream->udpsrc[1]) {
+    GstCaps *caps;
+
+    gst_element_set_locked_state (stream->udpsrc[1], TRUE);
     gst_bin_add (GST_BIN_CAST (src), stream->udpsrc[1]);
 
+    caps = gst_caps_new_empty_simple ("application/x-rtcp");
+    g_object_set (stream->udpsrc[1], "caps", caps, NULL);
+    gst_caps_unref (caps);
+
     if (stream->channelpad[1]) {
       GstPad *pad;
 
@@ -3532,15 +3631,18 @@ gst_rtspsrc_connection_flush (GstRTSPSrc * src, gboolean flush)
 
   GST_DEBUG_OBJECT (src, "set flushing %d", flush);
   GST_RTSP_STATE_LOCK (src);
-  if (src->conninfo.connection) {
+  if (src->conninfo.connection && src->conninfo.flushing != flush) {
     GST_DEBUG_OBJECT (src, "connection flush");
     gst_rtsp_connection_flush (src->conninfo.connection, flush);
+    src->conninfo.flushing = flush;
   }
   for (walk = src->streams; walk; walk = g_list_next (walk)) {
     GstRTSPStream *stream = (GstRTSPStream *) walk->data;
-    GST_DEBUG_OBJECT (src, "stream %p flush", stream);
-    if (stream->conninfo.connection)
+    if (stream->conninfo.connection && stream->conninfo.flushing != flush) {
+      GST_DEBUG_OBJECT (src, "stream %p flush", stream);
       gst_rtsp_connection_flush (stream->conninfo.connection, flush);
+      stream->conninfo.flushing = flush;
+    }
   }
   GST_RTSP_STATE_UNLOCK (src);
 }
@@ -3562,13 +3664,16 @@ gst_rtspsrc_handle_request (GstRTSPSrc * src, GstRTSPConnection * conn,
 
   if (res == GST_RTSP_ENOTIMPL) {
     /* default implementation, send OK */
+    GST_DEBUG_OBJECT (src, "prepare OK reply");
     res =
         gst_rtsp_message_init_response (&response, GST_RTSP_STS_OK, "OK",
         request);
     if (res < 0)
       goto send_error;
 
-    GST_DEBUG_OBJECT (src, "replying with OK");
+    /* let app parse and reply */
+    g_signal_emit (src, gst_rtspsrc_signals[SIGNAL_HANDLE_REQUEST],
+        0, request, response);
 
     if (src->debug)
       gst_rtsp_message_dump (&response);
@@ -3659,98 +3764,19 @@ send_error:
 }
 
 static GstFlowReturn
-gst_rtspsrc_loop_interleaved (GstRTSPSrc * src)
+gst_rtspsrc_handle_data (GstRTSPSrc * src, GstRTSPMessage * message)
 {
-  GstRTSPMessage message = { 0 };
-  GstRTSPResult res;
+  GstFlowReturn ret = GST_FLOW_OK;
   gint channel;
   GstRTSPStream *stream;
   GstPad *outpad = NULL;
   guint8 *data;
   guint size;
-  GstFlowReturn ret = GST_FLOW_OK;
   GstBuffer *buf;
-  gboolean is_rtcp, have_data;
+  gboolean is_rtcp;
   GstEvent *event;
 
-  /* here we are only interested in data messages */
-  have_data = FALSE;
-  do {
-    GTimeVal tv_timeout;
-
-    /* get the next timeout interval */
-    gst_rtsp_connection_next_timeout (src->conninfo.connection, &tv_timeout);
-
-    /* see if the timeout period expired */
-    if ((tv_timeout.tv_sec | tv_timeout.tv_usec) == 0) {
-      GST_DEBUG_OBJECT (src, "timout, sending keep-alive");
-      /* send keep-alive, only act on interrupt, a warning will be posted for
-       * other errors. */
-      if ((res = gst_rtspsrc_send_keep_alive (src)) == GST_RTSP_EINTR)
-        goto interrupt;
-      /* get new timeout */
-      gst_rtsp_connection_next_timeout (src->conninfo.connection, &tv_timeout);
-    }
-
-    GST_DEBUG_OBJECT (src, "doing receive with timeout %ld seconds, %ld usec",
-        tv_timeout.tv_sec, tv_timeout.tv_usec);
-
-    /* protect the connection with the connection lock so that we can see when
-     * we are finished doing server communication */
-    res =
-        gst_rtspsrc_connection_receive (src, src->conninfo.connection,
-        &message, src->ptcp_timeout);
-
-    switch (res) {
-      case GST_RTSP_OK:
-        GST_DEBUG_OBJECT (src, "we received a server message");
-        break;
-      case GST_RTSP_EINTR:
-        /* we got interrupted this means we need to stop */
-        goto interrupt;
-      case GST_RTSP_ETIMEOUT:
-        /* no reply, send keep alive */
-        GST_DEBUG_OBJECT (src, "timeout, sending keep-alive");
-        if ((res = gst_rtspsrc_send_keep_alive (src)) == GST_RTSP_EINTR)
-          goto interrupt;
-        continue;
-      case GST_RTSP_EEOF:
-        /* go EOS when the server closed the connection */
-        goto server_eof;
-      default:
-        goto receive_error;
-    }
-
-    switch (message.type) {
-      case GST_RTSP_MESSAGE_REQUEST:
-        /* server sends us a request message, handle it */
-        res =
-            gst_rtspsrc_handle_request (src, src->conninfo.connection,
-            &message);
-        if (res == GST_RTSP_EEOF)
-          goto server_eof;
-        else if (res < 0)
-          goto handle_request_failed;
-        break;
-      case GST_RTSP_MESSAGE_RESPONSE:
-        /* we ignore response messages */
-        GST_DEBUG_OBJECT (src, "ignoring response message");
-        if (src->debug)
-          gst_rtsp_message_dump (&message);
-        break;
-      case GST_RTSP_MESSAGE_DATA:
-        GST_DEBUG_OBJECT (src, "got data message");
-        have_data = TRUE;
-        break;
-      default:
-        GST_WARNING_OBJECT (src, "ignoring unknown message type %d",
-            message.type);
-        break;
-    }
-  }
-  while (!have_data);
-
-  channel = message.type_data.data.channel;
+  channel = message->type_data.data.channel;
 
   stream = find_stream (src, &channel, (gpointer) find_stream_by_channel);
   if (!stream)
@@ -3767,7 +3793,7 @@ gst_rtspsrc_loop_interleaved (GstRTSPSrc * src)
   }
 
   /* take a look at the body to figure out what we have */
-  gst_rtsp_message_get_body (&message, &data, &size);
+  gst_rtsp_message_get_body (message, &data, &size);
   if (size < 2)
     goto invalid_length;
 
@@ -3783,7 +3809,7 @@ gst_rtspsrc_loop_interleaved (GstRTSPSrc * src)
     goto unknown_stream;
 
   /* take the message body for further processing */
-  gst_rtsp_message_steal_body (&message, &data, &size);
+  gst_rtsp_message_steal_body (message, &data, &size);
 
   /* strip the trailing \0 */
   size -= 1;
@@ -3793,12 +3819,28 @@ gst_rtspsrc_loop_interleaved (GstRTSPSrc * src)
       gst_memory_new_wrapped (0, data, size, 0, size, data, g_free));
 
   /* don't need message anymore */
-  gst_rtsp_message_unset (&message);
+  gst_rtsp_message_unset (message);
 
   GST_DEBUG_OBJECT (src, "pushing data of size %d on channel %d", size,
       channel);
 
   if (src->need_activate) {
+    gchar *stream_id;
+    GstEvent *event;
+    GChecksum *cs;
+    gchar *uri;
+
+    /* generate an SHA256 sum of the URI */
+    cs = g_checksum_new (G_CHECKSUM_SHA256);
+    uri = src->conninfo.location;
+    g_checksum_update (cs, (const guchar *) uri, strlen (uri));
+    stream_id =
+        g_strdup_printf ("%s/%d", g_checksum_get_string (cs), stream->id);
+    g_checksum_free (cs);
+    event = gst_event_new_stream_start (stream_id);
+    g_free (stream_id);
+    gst_rtspsrc_push_event (src, event);
+
     gst_rtspsrc_activate_streams (src);
     src->need_activate = FALSE;
   }
@@ -3859,9 +3901,102 @@ gst_rtspsrc_loop_interleaved (GstRTSPSrc * src)
 unknown_stream:
   {
     GST_DEBUG_OBJECT (src, "unknown stream on channel %d, ignored", channel);
-    gst_rtsp_message_unset (&message);
+    gst_rtsp_message_unset (message);
+    return GST_FLOW_OK;
+  }
+invalid_length:
+  {
+    GST_ELEMENT_WARNING (src, RESOURCE, READ, (NULL),
+        ("Short message received, ignoring."));
+    gst_rtsp_message_unset (message);
     return GST_FLOW_OK;
   }
+}
+
+static GstFlowReturn
+gst_rtspsrc_loop_interleaved (GstRTSPSrc * src)
+{
+  GstRTSPMessage message = { 0 };
+  GstRTSPResult res;
+  GstFlowReturn ret = GST_FLOW_OK;
+  GTimeVal tv_timeout;
+
+  while (TRUE) {
+    /* get the next timeout interval */
+    gst_rtsp_connection_next_timeout (src->conninfo.connection, &tv_timeout);
+
+    /* see if the timeout period expired */
+    if ((tv_timeout.tv_sec | tv_timeout.tv_usec) == 0) {
+      GST_DEBUG_OBJECT (src, "timout, sending keep-alive");
+      /* send keep-alive, only act on interrupt, a warning will be posted for
+       * other errors. */
+      if ((res = gst_rtspsrc_send_keep_alive (src)) == GST_RTSP_EINTR)
+        goto interrupt;
+      /* get new timeout */
+      gst_rtsp_connection_next_timeout (src->conninfo.connection, &tv_timeout);
+    }
+
+    GST_DEBUG_OBJECT (src, "doing receive with timeout %ld seconds, %ld usec",
+        tv_timeout.tv_sec, tv_timeout.tv_usec);
+
+    /* protect the connection with the connection lock so that we can see when
+     * we are finished doing server communication */
+    res =
+        gst_rtspsrc_connection_receive (src, src->conninfo.connection,
+        &message, src->ptcp_timeout);
+
+    switch (res) {
+      case GST_RTSP_OK:
+        GST_DEBUG_OBJECT (src, "we received a server message");
+        break;
+      case GST_RTSP_EINTR:
+        /* we got interrupted this means we need to stop */
+        goto interrupt;
+      case GST_RTSP_ETIMEOUT:
+        /* no reply, send keep alive */
+        GST_DEBUG_OBJECT (src, "timeout, sending keep-alive");
+        if ((res = gst_rtspsrc_send_keep_alive (src)) == GST_RTSP_EINTR)
+          goto interrupt;
+        continue;
+      case GST_RTSP_EEOF:
+        /* go EOS when the server closed the connection */
+        goto server_eof;
+      default:
+        goto receive_error;
+    }
+
+    switch (message.type) {
+      case GST_RTSP_MESSAGE_REQUEST:
+        /* server sends us a request message, handle it */
+        res =
+            gst_rtspsrc_handle_request (src, src->conninfo.connection,
+            &message);
+        if (res == GST_RTSP_EEOF)
+          goto server_eof;
+        else if (res < 0)
+          goto handle_request_failed;
+        break;
+      case GST_RTSP_MESSAGE_RESPONSE:
+        /* we ignore response messages */
+        GST_DEBUG_OBJECT (src, "ignoring response message");
+        if (src->debug)
+          gst_rtsp_message_dump (&message);
+        break;
+      case GST_RTSP_MESSAGE_DATA:
+        GST_DEBUG_OBJECT (src, "got data message");
+        ret = gst_rtspsrc_handle_data (src, &message);
+        if (ret != GST_FLOW_OK)
+          goto handle_data_failed;
+        break;
+      default:
+        GST_WARNING_OBJECT (src, "ignoring unknown message type %d",
+            message.type);
+        break;
+    }
+  }
+  g_assert_not_reached ();
+
+  /* ERRORS */
 server_eof:
   {
     GST_DEBUG_OBJECT (src, "we got an eof from the server");
@@ -3874,8 +4009,7 @@ server_eof:
 interrupt:
   {
     gst_rtsp_message_unset (&message);
-    GST_DEBUG_OBJECT (src, "got interrupted: stop connection flush");
-    gst_rtspsrc_connection_flush (src, FALSE);
+    GST_DEBUG_OBJECT (src, "got interrupted");
     return GST_FLOW_FLUSHING;
   }
 receive_error:
@@ -3899,12 +4033,10 @@ handle_request_failed:
     gst_rtsp_message_unset (&message);
     return GST_FLOW_ERROR;
   }
-invalid_length:
+handle_data_failed:
   {
-    GST_ELEMENT_WARNING (src, RESOURCE, READ, (NULL),
-        ("Short message received, ignoring."));
-    gst_rtsp_message_unset (&message);
-    return GST_FLOW_OK;
+    GST_DEBUG_OBJECT (src, "could no handle data message");
+    return ret;
   }
 }
 
@@ -4003,13 +4135,13 @@ gst_rtspsrc_loop_udp (GstRTSPSrc * src)
         break;
     }
   }
+  g_assert_not_reached ();
 
   /* we get here when the connection got interrupted */
 interrupt:
   {
     gst_rtsp_message_unset (&message);
-    GST_DEBUG_OBJECT (src, "got interrupted: stop connection flush");
-    gst_rtspsrc_connection_flush (src, FALSE);
+    GST_DEBUG_OBJECT (src, "got interrupted");
     return GST_FLOW_FLUSHING;
   }
 connect_error:
@@ -4250,7 +4382,7 @@ gst_rtspsrc_loop_send_cmd (GstRTSPSrc * src, gint cmd, gint mask)
     src->pending_cmd = CMD_WAIT;
     GST_OBJECT_UNLOCK (src);
     /* cancel previous request */
-    GST_DEBUG_OBJECT (src, "cancel previous request");
+    GST_DEBUG_OBJECT (src, "cancel previous request %d", old);
     gst_rtspsrc_loop_cancel_cmd (src, old);
     GST_OBJECT_LOCK (src);
   }
@@ -4656,7 +4788,8 @@ next:
       break;
     case GST_RTSP_MESSAGE_DATA:
       /* get next response */
-      GST_DEBUG_OBJECT (src, "ignoring data response message");
+      GST_DEBUG_OBJECT (src, "handle data response message");
+      gst_rtspsrc_handle_data (src, response);
       goto next;
     default:
       GST_WARNING_OBJECT (src, "ignoring unknown message type %d",
@@ -5600,6 +5733,70 @@ gst_rtspsrc_parse_range (GstRTSPSrc * src, const gchar * range,
   return TRUE;
 }
 
+/* Parse clock profived by the server with following syntax:
+ *
+ * "GstNetTimeProvider <wrapped-clock> <server-IP:port> <clock-time>"
+ */
+static gboolean
+gst_rtspsrc_parse_gst_clock (GstRTSPSrc * src, const gchar * gstclock)
+{
+  gboolean res = FALSE;
+
+  if (g_str_has_prefix (gstclock, "GstNetTimeProvider ")) {
+    gchar **fields = NULL, **parts = NULL;
+    gchar *remote_ip, *str;
+    gint port;
+    GstClockTime base_time;
+    GstClock *netclock;
+
+    fields = g_strsplit (gstclock, " ", 0);
+
+    /* wrapped clock, not very interesting for now */
+    if (fields[1] == NULL)
+      goto cleanup;
+
+    /* remote IP address and port */
+    if ((str = fields[2]) == NULL)
+      goto cleanup;
+
+    parts = g_strsplit (str, ":", 0);
+
+    if ((remote_ip = parts[0]) == NULL)
+      goto cleanup;
+
+    if ((str = parts[1]) == NULL)
+      goto cleanup;
+
+    port = atoi (str);
+    if (port == 0)
+      goto cleanup;
+
+    /* base-time */
+    if ((str = fields[3]) == NULL)
+      goto cleanup;
+
+    base_time = g_ascii_strtoull (str, NULL, 10);
+
+    netclock =
+        gst_net_client_clock_new ((gchar *) "GstRTSPClock", remote_ip, port,
+        base_time);
+
+    if (src->provided_clock)
+      gst_object_unref (src->provided_clock);
+    src->provided_clock = netclock;
+
+    gst_element_post_message (GST_ELEMENT_CAST (src),
+        gst_message_new_clock_provide (GST_OBJECT_CAST (src),
+            src->provided_clock, TRUE));
+
+    res = TRUE;
+  cleanup:
+    g_strfreev (fields);
+    g_strfreev (parts);
+  }
+  return res;
+}
+
 /* must be called with the RTSP state lock */
 static GstRTSPResult
 gst_rtspsrc_open_from_sdp (GstRTSPSrc * src, GstSDPMessage * sdp,
@@ -5635,6 +5832,22 @@ gst_rtspsrc_open_from_sdp (GstRTSPSrc * src, GstSDPMessage * sdp,
         break;
     }
   }
+  /* parse clock information. This is GStreamer specific, a server can tell the
+   * client what clock it is using and wrap that in a network clock. The
+   * advantage of that is that we can slave to it. */
+  {
+    const gchar *gstclock;
+
+    for (i = 0;; i++) {
+      gstclock = gst_sdp_message_get_attribute_val_n (sdp, "x-gst-clock", i);
+      if (gstclock == NULL)
+        break;
+
+      /* parse the clock and expose it in the provide_clock method */
+      if (gst_rtspsrc_parse_gst_clock (src, gstclock))
+        break;
+    }
+  }
   /* try to find a global control attribute. Note that a '*' means that we should
    * do aggregate control with the current url (so we don't do anything and
    * leave the current connection as is) */
@@ -5921,6 +6134,8 @@ gst_rtspsrc_close (GstRTSPSrc * src, gboolean async, gboolean only_close)
 
   GST_DEBUG_OBJECT (src, "TEARDOWN...");
 
+  gst_rtspsrc_set_state (src, GST_STATE_READY);
+
   if (src->state < GST_RTSP_STATE_READY) {
     GST_DEBUG_OBJECT (src, "not ready, doing cleanup");
     goto close;
@@ -6240,27 +6455,11 @@ gst_rtspsrc_play (GstRTSPSrc * src, GstSegment * segment, gboolean async)
    * udp sources */
   gst_rtspsrc_send_dummy_packets (src);
 
-  /* activate receive elements;
-   * only in async case, since receive elements may not have been affected
-   * by overall state change (e.g. not around yet),
-   * do not mess with state in sync case (e.g. seeking) */
-  if (async) {
-    /* state change might be happening in the application thread. A
-     * specific case is when chaging state to NULL where we will wait
-     * for this task to finish (gst_rtspsrc_stop). However this task
-     * will try to change the state to PLAYING causing a deadlock. */
-
-    /* make sure we are not in the middle of a state change. The
-     * state lock is a recursive lock so it's safe to lock twice from
-     * the same thread */
-    if (GST_STATE_TRYLOCK (src)) {
-      gst_element_set_state (GST_ELEMENT_CAST (src), GST_STATE_PLAYING);
-      GST_STATE_UNLOCK (src);
-    } else {
-      res = GST_RTSP_ERROR;
-      goto changing_state;
-    }
-  }
+  /* require new SR packets */
+  if (src->manager)
+    g_signal_emit_by_name (src->manager, "reset-sync", NULL);
+
+  gst_rtspsrc_set_state (src, GST_STATE_PLAYING);
 
   /* construct a control url */
   if (src->control)
@@ -6423,11 +6622,6 @@ was_playing:
     GST_DEBUG_OBJECT (src, "we were already PLAYING");
     goto done;
   }
-changing_state:
-  {
-    GST_DEBUG_OBJECT (src, "failed going to PLAYING, already changing state");
-    goto done;
-  }
 create_request_failed:
   {
     gchar *str = gst_rtsp_strresult (res);
@@ -6523,6 +6717,9 @@ gst_rtspsrc_pause (GstRTSPSrc * src, gboolean async)
       break;
   }
 
+  /* change element states now */
+  gst_rtspsrc_set_state (src, GST_STATE_PAUSED);
+
 no_connection:
   src->state = GST_RTSP_STATE_READY;
 
@@ -6661,7 +6858,8 @@ gst_rtspsrc_thread (GstRTSPSrc * src)
 
   GST_OBJECT_LOCK (src);
   cmd = src->pending_cmd;
-  if (cmd == CMD_RECONNECT || cmd == CMD_PLAY || cmd == CMD_LOOP)
+  if (cmd == CMD_RECONNECT || cmd == CMD_PLAY || cmd == CMD_PAUSE
+      || cmd == CMD_LOOP)
     src->pending_cmd = CMD_LOOP;
   else
     src->pending_cmd = CMD_WAIT;
@@ -6797,6 +6995,9 @@ gst_rtspsrc_change_state (GstElement * element, GstStateChange transition)
     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
       /* unblock the tcp tasks and make the loop waiting */
       gst_rtspsrc_loop_send_cmd (rtspsrc, CMD_WAIT, CMD_LOOP);
+      /* make sure it is waiting before we send PAUSE or PLAY below */
+      GST_RTSP_STREAM_LOCK (rtspsrc);
+      GST_RTSP_STREAM_UNLOCK (rtspsrc);
       break;
     case GST_STATE_CHANGE_PAUSED_TO_READY:
       break;
@@ -6809,22 +7010,28 @@ gst_rtspsrc_change_state (GstElement * element, GstStateChange transition)
     goto done;
 
   switch (transition) {
+    case GST_STATE_CHANGE_NULL_TO_READY:
+      ret = GST_STATE_CHANGE_SUCCESS;
+      break;
+    case GST_STATE_CHANGE_READY_TO_PAUSED:
+      ret = GST_STATE_CHANGE_NO_PREROLL;
+      break;
     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
       gst_rtspsrc_loop_send_cmd (rtspsrc, CMD_PLAY, 0);
+      ret = GST_STATE_CHANGE_SUCCESS;
       break;
     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
       /* send pause request and keep the idle task around */
       gst_rtspsrc_loop_send_cmd (rtspsrc, CMD_PAUSE, CMD_LOOP);
       ret = GST_STATE_CHANGE_NO_PREROLL;
       break;
-    case GST_STATE_CHANGE_READY_TO_PAUSED:
-      ret = GST_STATE_CHANGE_NO_PREROLL;
-      break;
     case GST_STATE_CHANGE_PAUSED_TO_READY:
       gst_rtspsrc_loop_send_cmd (rtspsrc, CMD_CLOSE, CMD_PAUSE);
+      ret = GST_STATE_CHANGE_SUCCESS;
       break;
     case GST_STATE_CHANGE_READY_TO_NULL:
       gst_rtspsrc_stop (rtspsrc);
+      ret = GST_STATE_CHANGE_SUCCESS;
       break;
     default:
       break;
@@ -6870,7 +7077,9 @@ static const gchar *const *
 gst_rtspsrc_uri_get_protocols (GType type)
 {
   static const gchar *protocols[] =
-      { "rtsp", "rtspu", "rtspt", "rtsph", "rtsp-sdp", NULL };
+      { "rtsp", "rtspu", "rtspt", "rtsph", "rtsp-sdp",
+    "rtsps", "rtspsu", "rtspst", "rtspsh", NULL
+  };
 
   return protocols;
 }