rtspsrc: reset-sync before play
[platform/upstream/gstreamer.git] / gst / rtsp / gstrtspsrc.c
index 4a0e4d1..e8b00c1 100644 (file)
@@ -14,8 +14,8 @@
  *
  * You should have received a copy of the GNU Library General Public
  * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
- * Boston, MA 02111-1307, USA.
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
  */
 /*
  * Unless otherwise indicated, Source Code is licensed under MIT license.
@@ -69,7 +69,7 @@
  * <refsect2>
  * <title>Example launch line</title>
  * |[
- * gst-launch rtspsrc location=rtsp://some.server/url ! fakesink
+ * gst-launch-1.0 rtspsrc location=rtsp://some.server/url ! fakesink
  * ]| Establish a connection to an RTSP server and send the raw RTP packets to a
  * fakesink.
  * </refsect2>
 #endif /* HAVE_UNISTD_H */
 #include <stdlib.h>
 #include <string.h>
-#include <locale.h>
 #include <stdio.h>
 #include <stdarg.h>
 
+#include <gst/net/gstnet.h>
 #include <gst/sdp/gstsdpmessage.h>
 #include <gst/rtp/gstrtppayloads.h>
 
@@ -120,7 +120,7 @@ GST_STATIC_PAD_TEMPLATE ("internalsink_%u",
 
 enum
 {
-  /* FILL ME */
+  SIGNAL_HANDLE_REQUEST,
   LAST_SIGNAL
 };
 
@@ -167,6 +167,7 @@ gst_rtsp_src_buffer_mode_get_type (void)
 #define DEFAULT_UDP_BUFFER_SIZE  0x80000
 #define DEFAULT_TCP_TIMEOUT      20000000
 #define DEFAULT_LATENCY_MS       2000
+#define DEFAULT_DROP_ON_LATENCY  FALSE
 #define DEFAULT_CONNECTION_SPEED 0
 #define DEFAULT_NAT_METHOD       GST_RTSP_NAT_DUMMY
 #define DEFAULT_DO_RTCP          TRUE
@@ -178,6 +179,11 @@ gst_rtsp_src_buffer_mode_get_type (void)
 #define DEFAULT_BUFFER_MODE      BUFFER_MODE_AUTO
 #define DEFAULT_PORT_RANGE       NULL
 #define DEFAULT_SHORT_HEADER     FALSE
+#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
 {
@@ -189,11 +195,14 @@ enum
   PROP_TIMEOUT,
   PROP_TCP_TIMEOUT,
   PROP_LATENCY,
+  PROP_DROP_ON_LATENCY,
   PROP_CONNECTION_SPEED,
   PROP_NAT_METHOD,
   PROP_DO_RTCP,
   PROP_DO_RTSP_KEEP_ALIVE,
   PROP_PROXY,
+  PROP_PROXY_ID,
+  PROP_PROXY_PW,
   PROP_RTP_BLOCKSIZE,
   PROP_USER_ID,
   PROP_USER_PW,
@@ -201,6 +210,11 @@ enum
   PROP_PORT_RANGE,
   PROP_UDP_BUFFER_SIZE,
   PROP_SHORT_HEADER,
+  PROP_PROBATION,
+  PROP_UDP_RECONNECT,
+  PROP_MULTICAST_IFACE,
+  PROP_NTP_SYNC,
+  PROP_USE_PIPELINE_CLOCK,
   PROP_LAST
 };
 
@@ -229,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);
 
@@ -261,6 +277,7 @@ static GstRTSPResult gst_rtspsrc_close (GstRTSPSrc * src, gboolean async,
 
 static gboolean gst_rtspsrc_uri_set_uri (GstURIHandler * handler,
     const gchar * uri, GError ** error);
+static gchar *gst_rtspsrc_uri_get_uri (GstURIHandler * handler);
 
 static gboolean gst_rtspsrc_activate_streams (GstRTSPSrc * src);
 static gboolean gst_rtspsrc_loop (GstRTSPSrc * src);
@@ -277,6 +294,9 @@ static gboolean gst_rtspsrc_push_event (GstRTSPSrc * src, GstEvent * event);
 #define CMD_RECONNECT  (1 << 5)
 #define CMD_LOOP       (1 << 6)
 
+/* mask for all commands */
+#define CMD_ALL         ((CMD_LOOP << 1) - 1)
+
 #define GST_ELEMENT_PROGRESS(el, type, code, text)      \
 G_STMT_START {                                          \
   gchar *__txt = _gst_element_error_printf text;        \
@@ -286,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));
@@ -347,6 +368,12 @@ gst_rtspsrc_class_init (GstRTSPSrcClass * klass)
           "Amount of ms to buffer", 0, G_MAXUINT, DEFAULT_LATENCY_MS,
           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
+  g_object_class_install_property (gobject_class, PROP_DROP_ON_LATENCY,
+      g_param_spec_boolean ("drop-on-latency",
+          "Drop buffers when maximum latency is reached",
+          "Tells the jitterbuffer to never exceed the given latency in size",
+          DEFAULT_DROP_ON_LATENCY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
   g_object_class_install_property (gobject_class, PROP_CONNECTION_SPEED,
       g_param_spec_uint64 ("connection-speed", "Connection Speed",
           "Network connection speed in kbps (0 = unknown)",
@@ -398,6 +425,30 @@ gst_rtspsrc_class_init (GstRTSPSrcClass * klass)
       g_param_spec_string ("proxy", "Proxy",
           "Proxy settings for HTTP tunneling. Format: [http://][user:passwd@]host[:port]",
           DEFAULT_PROXY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+  /**
+   * GstRTSPSrc::proxy-id
+   *
+   * Sets the proxy URI user id for authentication. If the URI set via the
+   * "proxy" property contains a user-id already, that will take precedence.
+   *
+   * Since: 1.2
+   */
+  g_object_class_install_property (gobject_class, PROP_PROXY_ID,
+      g_param_spec_string ("proxy-id", "proxy-id",
+          "HTTP proxy URI user id for authentication", "",
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+  /**
+   * GstRTSPSrc::proxy-pw
+   *
+   * Sets the proxy URI password for authentication. If the URI set via the
+   * "proxy" property contains a password already, that will take precedence.
+   *
+   * Since: 1.2
+   */
+  g_object_class_install_property (gobject_class, PROP_PROXY_PW,
+      g_param_spec_string ("proxy-pw", "proxy-pw",
+          "HTTP proxy URI user password for authentication", "",
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
   /**
    * GstRTSPSrc::rtp_blocksize
@@ -474,7 +525,56 @@ gst_rtspsrc_class_init (GstRTSPSrcClass * klass)
           "Only send the basic RTSP headers for broken encoders",
           DEFAULT_SHORT_HEADER, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
+  g_object_class_install_property (gobject_class, PROP_PROBATION,
+      g_param_spec_uint ("probation", "Number of probations",
+          "Consecutive packet sequence numbers to accept the source",
+          0, G_MAXUINT, DEFAULT_PROBATION,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class, PROP_UDP_RECONNECT,
+      g_param_spec_boolean ("udp-reconnect", "Reconnect to the server",
+          "Reconnect to the server if RTSP connection is closed when doing UDP",
+          DEFAULT_UDP_RECONNECT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class, PROP_MULTICAST_IFACE,
+      g_param_spec_string ("multicast-iface", "Multicast Interface",
+          "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,
@@ -503,6 +603,7 @@ gst_rtspsrc_init (GstRTSPSrc * src)
   src->udp_timeout = DEFAULT_TIMEOUT;
   gst_rtspsrc_set_tcp_timeout (src, DEFAULT_TCP_TIMEOUT);
   src->latency = DEFAULT_LATENCY_MS;
+  src->drop_on_latency = DEFAULT_DROP_ON_LATENCY;
   src->connection_speed = DEFAULT_CONNECTION_SPEED;
   src->nat_method = DEFAULT_NAT_METHOD;
   src->do_rtcp = DEFAULT_DO_RTCP;
@@ -516,6 +617,11 @@ gst_rtspsrc_init (GstRTSPSrc * src)
   src->client_port_range.max = 0;
   src->udp_buffer_size = DEFAULT_UDP_BUFFER_SIZE;
   src->short_header = DEFAULT_SHORT_HEADER;
+  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 ();
@@ -549,11 +655,14 @@ gst_rtspsrc_finalize (GObject * object)
   g_free (rtspsrc->conninfo.url_str);
   g_free (rtspsrc->user_id);
   g_free (rtspsrc->user_pw);
+  g_free (rtspsrc->multi_iface);
 
   if (rtspsrc->sdp) {
     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);
@@ -562,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)
@@ -598,6 +719,15 @@ gst_rtspsrc_set_proxy (GstRTSPSrc * rtsp, const gchar * proxy)
 
     /* move to host */
     p = at + 1;
+  } else {
+    if (rtsp->prop_proxy_id != NULL && *rtsp->prop_proxy_id != '\0')
+      rtsp->proxy_user = g_strdup (rtsp->prop_proxy_id);
+    if (rtsp->prop_proxy_pw != NULL && *rtsp->prop_proxy_pw != '\0')
+      rtsp->proxy_passwd = g_strdup (rtsp->prop_proxy_pw);
+    if (rtsp->proxy_user != NULL || rtsp->proxy_passwd != NULL) {
+      GST_LOG_OBJECT (rtsp, "set proxy user/pw from properties: %s:%s",
+          GST_STR_NULL (rtsp->proxy_user), GST_STR_NULL (rtsp->proxy_passwd));
+    }
   }
   col = strchr (p, ':');
 
@@ -656,6 +786,9 @@ gst_rtspsrc_set_property (GObject * object, guint prop_id, const GValue * value,
     case PROP_LATENCY:
       rtspsrc->latency = g_value_get_uint (value);
       break;
+    case PROP_DROP_ON_LATENCY:
+      rtspsrc->drop_on_latency = g_value_get_boolean (value);
+      break;
     case PROP_CONNECTION_SPEED:
       rtspsrc->connection_speed = g_value_get_uint64 (value);
       break;
@@ -671,6 +804,16 @@ gst_rtspsrc_set_property (GObject * object, guint prop_id, const GValue * value,
     case PROP_PROXY:
       gst_rtspsrc_set_proxy (rtspsrc, g_value_get_string (value));
       break;
+    case PROP_PROXY_ID:
+      if (rtspsrc->prop_proxy_id)
+        g_free (rtspsrc->prop_proxy_id);
+      rtspsrc->prop_proxy_id = g_value_dup_string (value);
+      break;
+    case PROP_PROXY_PW:
+      if (rtspsrc->prop_proxy_pw)
+        g_free (rtspsrc->prop_proxy_pw);
+      rtspsrc->prop_proxy_pw = g_value_dup_string (value);
+      break;
     case PROP_RTP_BLOCKSIZE:
       rtspsrc->rtp_blocksize = g_value_get_uint (value);
       break;
@@ -707,6 +850,26 @@ gst_rtspsrc_set_property (GObject * object, guint prop_id, const GValue * value,
     case PROP_SHORT_HEADER:
       rtspsrc->short_header = g_value_get_boolean (value);
       break;
+    case PROP_PROBATION:
+      rtspsrc->probation = g_value_get_uint (value);
+      break;
+    case PROP_UDP_RECONNECT:
+      rtspsrc->udp_reconnect = g_value_get_boolean (value);
+      break;
+    case PROP_MULTICAST_IFACE:
+      g_free (rtspsrc->multi_iface);
+
+      if (g_value_get_string (value) == NULL)
+        rtspsrc->multi_iface = g_strdup (DEFAULT_MULTICAST_IFACE);
+      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;
@@ -749,6 +912,9 @@ gst_rtspsrc_get_property (GObject * object, guint prop_id, GValue * value,
     case PROP_LATENCY:
       g_value_set_uint (value, rtspsrc->latency);
       break;
+    case PROP_DROP_ON_LATENCY:
+      g_value_set_boolean (value, rtspsrc->drop_on_latency);
+      break;
     case PROP_CONNECTION_SPEED:
       g_value_set_uint64 (value, rtspsrc->connection_speed);
       break;
@@ -774,6 +940,12 @@ gst_rtspsrc_get_property (GObject * object, guint prop_id, GValue * value,
       g_value_take_string (value, str);
       break;
     }
+    case PROP_PROXY_ID:
+      g_value_set_string (value, rtspsrc->prop_proxy_id);
+      break;
+    case PROP_PROXY_PW:
+      g_value_set_string (value, rtspsrc->prop_proxy_pw);
+      break;
     case PROP_RTP_BLOCKSIZE:
       g_value_set_uint (value, rtspsrc->rtp_blocksize);
       break;
@@ -805,6 +977,21 @@ gst_rtspsrc_get_property (GObject * object, guint prop_id, GValue * value,
     case PROP_SHORT_HEADER:
       g_value_set_boolean (value, rtspsrc->short_header);
       break;
+    case PROP_PROBATION:
+      g_value_set_uint (value, rtspsrc->probation);
+      break;
+    case PROP_UDP_RECONNECT:
+      g_value_set_boolean (value, rtspsrc->udp_reconnect);
+      break;
+    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;
@@ -1203,6 +1390,14 @@ gst_rtspsrc_cleanup (GstRTSPSrc * src)
     gst_sdp_message_free (src->sdp);
     src->sdp = NULL;
   }
+  if (src->start_segment) {
+    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)          \
@@ -1538,6 +1733,7 @@ again:
       GST_DEBUG_OBJECT (src, "free RTP udpsrc");
       gst_element_set_state (udpsrc0, GST_STATE_NULL);
       gst_object_unref (udpsrc0);
+      udpsrc0 = NULL;
 
       GST_DEBUG_OBJECT (src, "retry %d", count);
       goto again;
@@ -1559,6 +1755,7 @@ again:
     GST_DEBUG_OBJECT (src, "free RTP udpsrc");
     gst_element_set_state (udpsrc0, GST_STATE_NULL);
     gst_object_unref (udpsrc0);
+    udpsrc0 = NULL;
 
     GST_DEBUG_OBJECT (src, "retry %d", count);
     tmp_rtp++;
@@ -1572,7 +1769,7 @@ again:
 
   /* set port */
   tmp_rtcp = tmp_rtp + 1;
-  if (src->client_port_range.max > 0 && tmp_rtcp >= src->client_port_range.max)
+  if (src->client_port_range.max > 0 && tmp_rtcp > src->client_port_range.max)
     goto no_ports;
 
   g_object_set (G_OBJECT (udpsrc1), "port", tmp_rtcp, "reuse", FALSE, NULL);
@@ -1589,6 +1786,7 @@ again:
     GST_DEBUG_OBJECT (src, "free RTP udpsrc");
     gst_element_set_state (udpsrc0, GST_STATE_NULL);
     gst_object_unref (udpsrc0);
+    udpsrc0 = NULL;
 
     GST_DEBUG_OBJECT (src, "free RTCP udpsrc");
     gst_element_set_state (udpsrc1, GST_STATE_NULL);
@@ -1612,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 */
@@ -1658,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 ();
@@ -1681,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
@@ -1877,11 +2080,6 @@ gst_rtspsrc_perform_seek (GstRTSPSrc * src, GstEvent * event)
   GST_DEBUG_OBJECT (src, "Creating newsegment from %" G_GINT64_FORMAT
       " to %" G_GINT64_FORMAT, src->segment.position, stop);
 
-  /* store the newsegment event so it can be sent from the streaming thread. */
-  if (src->start_segment)
-    gst_event_unref (src->start_segment);
-  src->start_segment = gst_event_new_segment (&src->segment);
-
   /* mark discont */
   GST_DEBUG_OBJECT (src, "mark DISCONT, we did a seek to another position");
   for (walk = src->streams; walk; walk = g_list_next (walk)) {
@@ -2070,6 +2268,18 @@ gst_rtspsrc_handle_src_query (GstPad * pad, GstObject * parent,
       }
       break;
     }
+    case GST_QUERY_URI:
+    {
+      gchar *uri;
+
+      uri = gst_rtspsrc_uri_get_uri (GST_URI_HANDLER (src));
+      if (uri != NULL) {
+        gst_query_set_uri (query, uri);
+        g_free (uri);
+        res = TRUE;
+      }
+      break;
+    }
     default:
     {
       GstPad *target = gst_ghost_pad_get_target (GST_GHOST_PAD_CAST (pad));
@@ -2168,7 +2378,7 @@ new_manager_pad (GstElement * manager, GstPad * pad, GstRTSPSrc * src)
   gchar *name;
   GstPadTemplate *template;
   gint id, ssrc, pt;
-  GList *lstream;
+  GList *ostreams;
   GstRTSPStream *stream;
   gboolean all_added;
 
@@ -2180,41 +2390,46 @@ new_manager_pad (GstElement * manager, GstPad * pad, GstRTSPSrc * src)
   if (sscanf (name, "recv_rtp_src_%u_%u_%u", &id, &ssrc, &pt) != 3)
     goto unknown_stream;
 
-  GST_DEBUG_OBJECT (src, "stream: %u, SSRC %d, PT %d", id, ssrc, pt);
+  GST_DEBUG_OBJECT (src, "stream: %u, SSRC %08x, PT %d", id, ssrc, pt);
 
   stream = find_stream (src, &id, (gpointer) find_stream_by_id);
   if (stream == NULL)
     goto unknown_stream;
 
-  /* create a new pad we will use to stream to */
-  template = gst_static_pad_template_get (&rtptemplate);
-  stream->srcpad = gst_ghost_pad_new_from_template (name, pad, template);
-  gst_object_unref (template);
-  g_free (name);
+  /* save SSRC */
+  stream->ssrc = ssrc;
 
+  /* we'll add it later see below */
   stream->added = TRUE;
-  gst_pad_set_event_function (stream->srcpad, gst_rtspsrc_handle_src_event);
-  gst_pad_set_query_function (stream->srcpad, gst_rtspsrc_handle_src_query);
-  gst_pad_set_active (stream->srcpad, TRUE);
-  gst_element_add_pad (GST_ELEMENT_CAST (src), stream->srcpad);
 
   /* check if we added all streams */
   all_added = TRUE;
-  for (lstream = src->streams; lstream; lstream = g_list_next (lstream)) {
-    stream = (GstRTSPStream *) lstream->data;
+  for (ostreams = src->streams; ostreams; ostreams = g_list_next (ostreams)) {
+    GstRTSPStream *ostream = (GstRTSPStream *) ostreams->data;
 
     GST_DEBUG_OBJECT (src, "stream %p, container %d, disabled %d, added %d",
-        stream, stream->container, stream->disabled, stream->added);
+        ostream, ostream->container, ostream->disabled, ostream->added);
 
     /* a container stream only needs one pad added. Also disabled streams don't
      * count */
-    if (!stream->container && !stream->disabled && !stream->added) {
+    if (!ostream->container && !ostream->disabled && !ostream->added) {
       all_added = FALSE;
       break;
     }
   }
   GST_RTSP_STATE_UNLOCK (src);
 
+  /* create a new pad we will use to stream to */
+  template = gst_static_pad_template_get (&rtptemplate);
+  stream->srcpad = gst_ghost_pad_new_from_template (name, pad, template);
+  gst_object_unref (template);
+  g_free (name);
+
+  gst_pad_set_event_function (stream->srcpad, gst_rtspsrc_handle_src_event);
+  gst_pad_set_query_function (stream->srcpad, gst_rtspsrc_handle_src_query);
+  gst_pad_set_active (stream->srcpad, TRUE);
+  gst_element_add_pad (GST_ELEMENT_CAST (src), stream->srcpad);
+
   if (all_added) {
     GST_DEBUG_OBJECT (src, "We added all streams");
     /* when we get here, all stream are added and we can fire the no-more-pads
@@ -2286,20 +2501,30 @@ static void
 on_bye_ssrc (GObject * session, GObject * source, GstRTSPStream * stream)
 {
   GstRTSPSrc *src = stream->parent;
+  guint ssrc;
+
+  g_object_get (source, "ssrc", &ssrc, NULL);
 
-  GST_DEBUG_OBJECT (src, "source in session %u received BYE", stream->id);
+  GST_DEBUG_OBJECT (src, "source %08x, stream %08x, session %u received BYE",
+      ssrc, stream->ssrc, stream->id);
 
-  gst_rtspsrc_do_stream_eos (src, stream);
+  if (ssrc == stream->ssrc)
+    gst_rtspsrc_do_stream_eos (src, stream);
 }
 
 static void
 on_timeout (GObject * session, GObject * source, GstRTSPStream * stream)
 {
   GstRTSPSrc *src = stream->parent;
+  guint ssrc;
 
-  GST_DEBUG_OBJECT (src, "source in session %u timed out", stream->id);
+  g_object_get (source, "ssrc", &ssrc, NULL);
 
-  gst_rtspsrc_do_stream_eos (src, stream);
+  GST_WARNING_OBJECT (src, "source %08x, stream %08x in session %u timed out",
+      ssrc, stream->ssrc, stream->id);
+
+  if (ssrc == stream->ssrc)
+    gst_rtspsrc_do_stream_eos (src, stream);
 }
 
 static void
@@ -2342,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;
@@ -2352,24 +2576,36 @@ 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);
+      }
+
       if (g_object_class_find_property (klass, "buffer-mode")) {
         if (src->buffer_mode != BUFFER_MODE_AUTO) {
           g_object_set (src->manager, "buffer-mode", src->buffer_mode, NULL);
@@ -2455,6 +2691,9 @@ gst_rtspsrc_stream_configure_manager (GstRTSPSrc * src, GstRTSPStream * stream,
           g_object_set (rtpsession, "rtcp-rs-bandwidth", stream->rs_bandwidth,
               NULL);
         }
+
+        g_object_set (rtpsession, "probation", src->probation, NULL);
+
         g_signal_connect (rtpsession, "on-bye-ssrc", (GCallback) on_bye_ssrc,
             stream);
         g_signal_connect (rtpsession, "on-bye-timeout", (GCallback) on_timeout,
@@ -2498,6 +2737,7 @@ gst_rtspsrc_stream_free_udp (GstRTSPStream * stream)
 
   for (i = 0; i < 2; i++) {
     if (stream->udpsrc[i]) {
+      GST_DEBUG ("free UDP source %d for stream %p", i, stream);
       gst_element_set_state (stream->udpsrc[i], GST_STATE_NULL);
       gst_object_unref (stream->udpsrc[i]);
       stream->udpsrc[i] = NULL;
@@ -2688,12 +2928,19 @@ gst_rtspsrc_stream_configure_mcast (GstRTSPSrc * src, GstRTSPStream * stream,
       g_object_set (G_OBJECT (stream->udpsrc[0]), "buffer-size",
           src->udp_buffer_size, NULL);
 
+    if (src->multi_iface != NULL)
+      g_object_set (G_OBJECT (stream->udpsrc[0]), "multicast-iface",
+          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);
@@ -2701,9 +2948,17 @@ 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]);
 
+    if (src->multi_iface != NULL)
+      g_object_set (G_OBJECT (stream->udpsrc[0]), "multicast-iface",
+          src->multi_iface, NULL);
+
     gst_element_set_state (stream->udpsrc[1], GST_STATE_PAUSED);
   }
   return TRUE;
@@ -2734,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");
@@ -2741,8 +2997,8 @@ gst_rtspsrc_stream_configure_udp (GstRTSPSrc * src, GstRTSPStream * stream,
     /* configure a timeout on the UDP port. When the timeout message is
      * posted, we assume UDP transport is not possible. We reconnect using TCP
      * if we can. */
-    g_object_set (G_OBJECT (stream->udpsrc[0]), "timeout", src->udp_timeout,
-        NULL);
+    g_object_set (G_OBJECT (stream->udpsrc[0]), "timeout",
+        src->udp_timeout * 1000, NULL);
 
     /* get output pad of the UDP source. */
     *outpad = gst_element_get_static_pad (stream->udpsrc[0], "src");
@@ -2772,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;
 
@@ -3208,7 +3471,7 @@ gst_rtspsrc_stream_push_event (GstRTSPSrc * src, GstRTSPStream * stream,
   gboolean res = TRUE;
 
   /* only streams that have a connection to the outside world */
-  if (stream->srcpad == NULL)
+  if (stream->container || stream->disabled)
     goto done;
 
   if (stream->udpsrc[0]) {
@@ -3368,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);
 }
@@ -3398,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);
@@ -3495,97 +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;
-
-  /* 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);
+  gboolean is_rtcp;
+  GstEvent *event;
 
-  channel = message.type_data.data.channel;
+  channel = message->type_data.data.channel;
 
   stream = find_stream (src, &channel, (gpointer) find_stream_by_channel);
   if (!stream)
@@ -3602,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;
 
@@ -3618,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;
@@ -3628,15 +3819,35 @@ 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;
   }
+  if ((event = src->start_segment) != NULL) {
+    src->start_segment = NULL;
+    gst_rtspsrc_push_event (src, event);
+  }
 
   if (src->base_time == -1) {
     /* Take current running_time. This timestamp will be put on
@@ -3690,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");
@@ -3705,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:
@@ -3730,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;
   }
 }
 
@@ -3781,10 +4082,13 @@ gst_rtspsrc_loop_udp (GstRTSPSrc * src)
          * see what happens. */
         GST_ELEMENT_WARNING (src, RESOURCE, READ, (NULL),
             ("The server closed the connection."));
-        if ((res =
-                gst_rtsp_conninfo_reconnect (src, &src->conninfo, FALSE)) < 0)
-          goto connect_error;
-
+        if (src->udp_reconnect) {
+          if ((res =
+                  gst_rtsp_conninfo_reconnect (src, &src->conninfo, FALSE)) < 0)
+            goto connect_error;
+        } else {
+          goto server_eof;
+        }
         continue;
       case GST_RTSP_ENET:
         GST_DEBUG_OBJECT (src, "An ethernet problem occured.");
@@ -3831,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:
@@ -4070,10 +4374,15 @@ gst_rtspsrc_loop_send_cmd (GstRTSPSrc * src, gint cmd, gint mask)
 
   GST_OBJECT_LOCK (src);
   old = src->pending_cmd;
+  if (old == CMD_RECONNECT) {
+    GST_DEBUG_OBJECT (src, "ignore, we were reconnecting");
+    cmd = CMD_RECONNECT;
+  }
   if (old != CMD_WAIT) {
     src->pending_cmd = CMD_WAIT;
     GST_OBJECT_UNLOCK (src);
     /* cancel previous request */
+    GST_DEBUG_OBJECT (src, "cancel previous request %d", old);
     gst_rtspsrc_loop_cancel_cmd (src, old);
     GST_OBJECT_LOCK (src);
   }
@@ -4141,6 +4450,7 @@ pause:
           ("streaming task paused, reason %s (%d)", reason, ret));
       gst_rtspsrc_push_event (src, gst_event_new_eos ());
     }
+    gst_rtspsrc_loop_send_cmd (src, CMD_WAIT, CMD_LOOP);
     return FALSE;
   }
 }
@@ -4478,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",
@@ -4527,8 +4838,8 @@ receive_error:
   {
     switch (res) {
       case GST_RTSP_EEOF:
-        GST_WARNING_OBJECT (src, "server closed connection, doing reconnect");
-        if (try == 0) {
+        GST_WARNING_OBJECT (src, "server closed connection");
+        if ((try == 0) && !src->interleaved && src->udp_reconnect) {
           try++;
           /* if reconnect succeeds, try again */
           if ((res =
@@ -4729,9 +5040,7 @@ gst_rtspsrc_parse_methods (GstRTSPSrc * src, GstRTSPMessage * response)
 {
   GstRTSPHeaderField field;
   gchar *respoptions;
-  gchar **options;
   gint indx = 0;
-  gint i;
 
   /* reset supported methods */
   src->methods = 0;
@@ -4749,25 +5058,7 @@ gst_rtspsrc_parse_methods (GstRTSPSrc * src, GstRTSPMessage * response)
     if (!respoptions)
       break;
 
-    /* If we get here, the server gave a list of supported methods, parse
-     * them here. The string is like:
-     *
-     * OPTIONS, DESCRIBE, ANNOUNCE, PLAY, SETUP, ...
-     */
-    options = g_strsplit (respoptions, ",", 0);
-
-    for (i = 0; options[i]; i++) {
-      gchar *stripped;
-      gint method;
-
-      stripped = g_strstrip (options[i]);
-      method = gst_rtsp_find_method (stripped);
-
-      /* keep bitfield of supported methods */
-      if (method != GST_RTSP_INVALID)
-        src->methods |= method;
-    }
-    g_strfreev (options);
+    src->methods |= gst_rtsp_options_from_text (respoptions);
 
     indx++;
   }
@@ -4863,6 +5154,14 @@ gst_rtspsrc_create_transports_string (GstRTSPSrc * src,
     if (add_udp_str)
       g_string_append (result, "/UDP");
     g_string_append (result, ";multicast");
+    if (src->next_port_num != 0) {
+      if (src->client_port_range.max > 0 &&
+          src->next_port_num >= src->client_port_range.max)
+        goto no_ports;
+
+      g_string_append_printf (result, ";client_port=%d-%d",
+          src->next_port_num, src->next_port_num + 1);
+    }
   } else if (protocols & GST_RTSP_LOWER_TRANS_TCP) {
     GST_DEBUG_OBJECT (src, "adding TCP");
 
@@ -4879,8 +5178,14 @@ gst_rtspsrc_create_transports_string (GstRTSPSrc * src,
   /* ERRORS */
 failed:
   {
+    GST_ERROR ("extension gave error %d", res);
     return res;
   }
+no_ports:
+  {
+    GST_ERROR ("no more ports available");
+    return GST_RTSP_ERROR;
+  }
 }
 
 static GstRTSPResult
@@ -4954,6 +5259,7 @@ done:
   /* ERRORS */
 failed:
   {
+    GST_ERROR ("failed to allocate udp ports");
     return GST_RTSP_ERROR;
   }
 }
@@ -5223,6 +5529,12 @@ gst_rtspsrc_setup_streams (GstRTSPSrc * src, gboolean async)
           /* only allow multicast for other streams */
           GST_DEBUG_OBJECT (src, "stream %p as UDP multicast", stream);
           protocols = GST_RTSP_LOWER_TRANS_UDP_MCAST;
+          /* if the server selected our ports, increment our counters so that
+           * we select a new port later */
+          if (src->next_port_num == transport.port.min &&
+              src->next_port_num + 1 == transport.port.max) {
+            src->next_port_num += 2;
+          }
           break;
         case GST_RTSP_LOWER_TRANS_UDP:
           /* only allow unicast for other streams */
@@ -5421,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,
@@ -5456,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) */
@@ -5510,6 +5902,7 @@ gst_rtspsrc_open_from_sdp (GstRTSPSrc * src, GstSDPMessage * sdp,
 setup_failed:
   {
     GST_ERROR_OBJECT (src, "setup failed");
+    gst_rtspsrc_cleanup (src);
     return res;
   }
 }
@@ -5741,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;
@@ -6060,12 +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)
-    gst_element_set_state (GST_ELEMENT_CAST (src), GST_STATE_PLAYING);
+  /* 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)
@@ -6102,6 +6496,11 @@ gst_rtspsrc_play (GstRTSPSrc * src, GstSegment * segment, gboolean async)
 
       gst_rtsp_message_add_header (&request, GST_RTSP_HDR_RANGE, hval);
       g_free (hval);
+
+      /* store the newsegment event so it can be sent from the streaming thread. */
+      if (src->start_segment)
+        gst_event_unref (src->start_segment);
+      src->start_segment = gst_event_new_segment (&src->segment);
     }
 
     if (segment->rate != 1.0) {
@@ -6318,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;
 
@@ -6456,7 +6858,8 @@ gst_rtspsrc_thread (GstRTSPSrc * src)
 
   GST_OBJECT_LOCK (src);
   cmd = src->pending_cmd;
-  if (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;
@@ -6538,7 +6941,7 @@ gst_rtspsrc_stop (GstRTSPSrc * src)
   GST_DEBUG_OBJECT (src, "stopping");
 
   /* also cancels pending task */
-  gst_rtspsrc_loop_send_cmd (src, CMD_WAIT, CMD_CLOSE);
+  gst_rtspsrc_loop_send_cmd (src, CMD_WAIT, CMD_ALL);
 
   GST_OBJECT_LOCK (src);
   if ((task = src->task)) {
@@ -6592,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;
@@ -6604,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;
@@ -6665,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;
 }