rtpmanager: fix pad callbacks so they handle when parent goes away

1) We need to lock and get a strong ref to the parent, if still there.
2) If it has gone away, we need to handle that gracefully.

This is necessary in order to safely modify a running pipeline. Has been
observed when a streaming thread is doing a buffer_alloc() while an
application thread sends an event on a pad further downstream, and from
within a pad probe (holding STREAM_LOCK) carries out the pipeline plumbing
while the streaming thread has its buffer_alloc() in progress.

gst/rtpmanager/gstrtpjitterbuffer.c

@@ -1008,6 +1008,8 @@ gst_rtp_jitter_buffer_src_event (GstPad * pad, GstEvent * event)
   GstRtpJitterBufferPrivate *priv;
 
   jitterbuffer = GST_RTP_JITTER_BUFFER (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (jitterbuffer == NULL))
+    return FALSE;
   priv = jitterbuffer->priv;
 
   GST_DEBUG_OBJECT (jitterbuffer, "received %s", GST_EVENT_TYPE_NAME (event));

gst/rtpmanager/gstrtpptdemux.c

@@ -434,6 +434,8 @@ gst_rtp_pt_demux_sink_event (GstPad * pad, GstEvent * event)
   gboolean res = FALSE;
 
   rtpdemux = GST_RTP_PT_DEMUX (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (rtpdemux == NULL))
+    return FALSE;
 
   switch (GST_EVENT_TYPE (event)) {
     case GST_EVENT_CUSTOM_DOWNSTREAM:

gst/rtpmanager/gstrtpsession.c

@@ -1325,6 +1325,8 @@ gst_rtp_session_event_recv_rtp_sink (GstPad * pad, GstEvent * event)
   gboolean ret = FALSE;
 
   rtpsession = GST_RTP_SESSION (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (rtpsession == NULL))
+    return FALSE;
 
   GST_DEBUG_OBJECT (rtpsession, "received event %s",
       GST_EVENT_TYPE_NAME (event));
@@ -1463,10 +1465,14 @@ gst_rtp_session_event_recv_rtp_src (GstPad * pad, GstEvent * event)
 static GstIterator *
 gst_rtp_session_iterate_internal_links (GstPad * pad)
 {
-  GstRtpSession *rtpsession = GST_RTP_SESSION (gst_pad_get_parent (pad));
+  GstRtpSession *rtpsession;
   GstPad *otherpad = NULL;
   GstIterator *it = NULL;
 
+  rtpsession = GST_RTP_SESSION (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (rtpsession == NULL))
+    return NULL;
+
   GST_RTP_SESSION_LOCK (rtpsession);
   if (pad == rtpsession->recv_rtp_src) {
     otherpad = gst_object_ref (rtpsession->recv_rtp_sink);
@@ -1633,12 +1639,17 @@ static gboolean
 gst_rtp_session_event_send_rtcp_src (GstPad * pad, GstEvent * event)
 {
   GstRtpSession *rtpsession;
-  gboolean ret;
+  gboolean ret = TRUE;
+  gboolean handled = FALSE;
 
   rtpsession = GST_RTP_SESSION (gst_pad_get_parent (pad));
 
   GST_DEBUG_OBJECT (rtpsession, "received EVENT");
 
+  rtpsession = GST_RTP_SESSION (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (rtpsession == NULL))
+    return FALSE;
+
   switch (GST_EVENT_TYPE (event)) {
     case GST_EVENT_SEEK:
     case GST_EVENT_LATENCY:
@@ -1654,6 +1665,24 @@ gst_rtp_session_event_send_rtcp_src (GstPad * pad, GstEvent * event)
 
   gst_object_unref (rtpsession);
 
+  if (handled)
+    gst_event_unref (event);
+  else {
+    GstPad *recv_rtp_sink = NULL;
+    GST_RTP_SESSION_LOCK (rtpsession);
+    if (rtpsession->recv_rtp_sink)
+      recv_rtp_sink = gst_object_ref (rtpsession->recv_rtp_sink);
+    GST_RTP_SESSION_UNLOCK (rtpsession);
+
+    if (recv_rtp_sink) {
+      ret = gst_pad_push_event (recv_rtp_sink, event);
+      gst_object_unref (recv_rtp_sink);
+    } else
+      gst_event_unref (event);
+  }
+
+  gst_object_unref (rtpsession);
+
   return ret;
 }
 
@@ -1716,9 +1745,21 @@ gst_rtp_session_event_send_rtp_sink (GstPad * pad, GstEvent * event)
           current_time);
       break;
     }
-    default:
-      ret = gst_pad_push_event (rtpsession->send_rtp_src, event);
+    default:{
+      GstPad *send_rtp_src = NULL;
+      GST_RTP_SESSION_LOCK (rtpsession);
+      if (rtpsession->send_rtp_src)
+        send_rtp_src = gst_object_ref (rtpsession->send_rtp_src);
+      GST_RTP_SESSION_UNLOCK (rtpsession);
+
+      if (send_rtp_src) {
+        ret = gst_pad_push_event (send_rtp_src, event);
+        gst_object_unref (send_rtp_src);
+      } else
+        gst_event_unref (event);
+
       break;
+    }
   }
   gst_object_unref (rtpsession);
 

gst/rtpmanager/gstrtpssrcdemux.c

@@ -420,6 +420,8 @@ gst_rtp_ssrc_demux_sink_event (GstPad * pad, GstEvent * event)
   gboolean res = FALSE;
 
   demux = GST_RTP_SSRC_DEMUX (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (demux == NULL))
+    return FALSE;
 
   switch (GST_EVENT_TYPE (event)) {
     case GST_EVENT_FLUSH_STOP:
@@ -699,6 +701,8 @@ gst_rtp_ssrc_demux_src_query (GstPad * pad, GstQuery * query)
   gboolean res = FALSE;
 
   demux = GST_RTP_SSRC_DEMUX (gst_pad_get_parent (pad));
+  if (G_UNLIKELY (demux == NULL))
+    return FALSE;
 
   switch (GST_QUERY_TYPE (query)) {
     case GST_QUERY_LATENCY: