Protection against early RTCP packets.

When receiving RTCP packets early the funnel is not ready yet and
GST_FLOW_FLUSHING will be returned when pushing data to it's srcpad.
This causes the thread that handle RTCP packets to go to pause mode.
Since this thread is in pause mode there will be no further callbacks to
handle keep-alive for incoming RTCP packets. This will make the session
time out if the client is not using another keep-alive mechanism.

Change-Id: Idb29db05f59c06423fa693a2aeeacbe3a1883fc5
Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-rtsp-server/-/merge_requests/211>

gst/rtsp-server/rtsp-media.c

@@ -4683,6 +4683,21 @@ preroll_failed:
   }
 }
 
+static void
+gst_rtsp_media_unblock_rtcp (GstRTSPMedia * media)
+{
+  GstRTSPMediaPrivate *priv;
+  guint i;
+
+  priv = media->priv;
+  g_mutex_lock (&priv->lock);
+  for (i = 0; i < priv->streams->len; i++) {
+    GstRTSPStream *stream = g_ptr_array_index (priv->streams, i);
+    gst_rtsp_stream_unblock_rtcp (stream);
+  }
+  g_mutex_unlock (&priv->lock);
+}
+
 /**
  * gst_rtsp_media_unsuspend:
  * @media: a #GstRTSPMedia
@@ -4711,6 +4726,7 @@ gst_rtsp_media_unsuspend (GstRTSPMedia * media)
   }
 
 done:
+  gst_rtsp_media_unblock_rtcp (media);
   g_rec_mutex_unlock (&priv->state_lock);
 
   return TRUE;

gst/rtsp-server/rtsp-stream.c

@@ -224,6 +224,12 @@ struct _GstRTSPStreamPrivate
 
   /* Whether we should send and receive RTCP */
   gboolean enable_rtcp;
+
+  /* blocking early rtcp packets */
+  GstPad *block_early_rtcp_pad;
+  gulong block_early_rtcp_probe;
+  GstPad *block_early_rtcp_pad_ipv6;
+  gulong block_early_rtcp_probe_ipv6;
 };
 
 #define DEFAULT_CONTROL         NULL
@@ -347,6 +353,10 @@ gst_rtsp_stream_init (GstRTSPStream * stream)
   priv->ptmap = g_hash_table_new_full (NULL, NULL, NULL,
       (GDestroyNotify) gst_caps_unref);
   priv->send_pool = NULL;
+  priv->block_early_rtcp_pad = NULL;
+  priv->block_early_rtcp_probe = 0;
+  priv->block_early_rtcp_pad_ipv6 = NULL;
+  priv->block_early_rtcp_probe_ipv6 = 0;
 }
 
 typedef struct _UdpClientAddrInfo UdpClientAddrInfo;
@@ -431,6 +441,18 @@ gst_rtsp_stream_finalize (GObject * obj)
   g_mutex_clear (&priv->send_lock);
   g_cond_clear (&priv->send_cond);
 
+  if (priv->block_early_rtcp_probe != 0) {
+    gst_pad_remove_probe
+        (priv->block_early_rtcp_pad, priv->block_early_rtcp_probe);
+    gst_object_unref (priv->block_early_rtcp_pad);
+  }
+
+  if (priv->block_early_rtcp_probe_ipv6 != 0) {
+    gst_pad_remove_probe
+        (priv->block_early_rtcp_pad_ipv6, priv->block_early_rtcp_probe_ipv6);
+    gst_object_unref (priv->block_early_rtcp_pad_ipv6);
+  }
+
   G_OBJECT_CLASS (gst_rtsp_stream_parent_class)->finalize (obj);
 }
 
@@ -3755,6 +3777,15 @@ create_receiver_part (GstRTSPStream * stream, const GstRTSPTransport *
         g_object_set (priv->udpsrc_v4[i], "caps", rtp_caps, NULL);
       } else {
         g_object_set (priv->udpsrc_v4[i], "caps", rtcp_caps, NULL);
+
+        /* block early rtcp packets, pipeline not ready */
+        g_assert (priv->block_early_rtcp_pad == NULL);
+        priv->block_early_rtcp_pad = gst_element_get_static_pad
+            (priv->udpsrc_v4[i], "src");
+        priv->block_early_rtcp_probe = gst_pad_add_probe
+            (priv->block_early_rtcp_pad,
+            GST_PAD_PROBE_TYPE_BLOCK | GST_PAD_PROBE_TYPE_BUFFER, NULL, NULL,
+            NULL);
       }
 
       plug_src (stream, bin, priv->udpsrc_v4[i], priv->funnel[i]);
@@ -3770,6 +3801,15 @@ create_receiver_part (GstRTSPStream * stream, const GstRTSPTransport *
         g_object_set (priv->udpsrc_v6[i], "caps", rtp_caps, NULL);
       } else {
         g_object_set (priv->udpsrc_v6[i], "caps", rtcp_caps, NULL);
+
+        /* block early rtcp packets, pipeline not ready */
+        g_assert (priv->block_early_rtcp_pad_ipv6 == NULL);
+        priv->block_early_rtcp_pad_ipv6 = gst_element_get_static_pad
+            (priv->udpsrc_v6[i], "src");
+        priv->block_early_rtcp_probe_ipv6 = gst_pad_add_probe
+            (priv->block_early_rtcp_pad_ipv6,
+            GST_PAD_PROBE_TYPE_BLOCK | GST_PAD_PROBE_TYPE_BUFFER, NULL, NULL,
+            NULL);
       }
 
       plug_src (stream, bin, priv->udpsrc_v6[i], priv->funnel[i]);
@@ -6290,3 +6330,37 @@ gst_rtsp_stream_get_rate_control (GstRTSPStream * stream)
 
   return ret;
 }
+
+/**
+ * gst_rtsp_stream_unblock_rtcp:
+ *
+ * Remove blocking probe from the RTCP source. When creating an UDP source for
+ * RTCP it is initially blocked until this function is called.
+ * This functions should be called once the pipeline is ready for handling RTCP
+ * packets.
+ *
+ * Since: 1.20
+ */
+void
+gst_rtsp_stream_unblock_rtcp (GstRTSPStream * stream)
+{
+  GstRTSPStreamPrivate *priv;
+
+  priv = stream->priv;
+  g_mutex_lock (&priv->lock);
+  if (priv->block_early_rtcp_probe != 0) {
+    gst_pad_remove_probe
+        (priv->block_early_rtcp_pad, priv->block_early_rtcp_probe);
+    priv->block_early_rtcp_probe = 0;
+    gst_object_unref (priv->block_early_rtcp_pad);
+    priv->block_early_rtcp_pad = NULL;
+  }
+  if (priv->block_early_rtcp_probe_ipv6 != 0) {
+    gst_pad_remove_probe
+        (priv->block_early_rtcp_pad_ipv6, priv->block_early_rtcp_probe_ipv6);
+    priv->block_early_rtcp_probe_ipv6 = 0;
+    gst_object_unref (priv->block_early_rtcp_pad_ipv6);
+    priv->block_early_rtcp_pad_ipv6 = NULL;
+  }
+  g_mutex_unlock (&priv->lock);
+}

gst/rtsp-server/rtsp-stream.h

@@ -365,6 +365,9 @@ void               gst_rtsp_stream_set_rate_control (GstRTSPStream * stream, gbo
 GST_RTSP_SERVER_API
 gboolean           gst_rtsp_stream_get_rate_control (GstRTSPStream * stream);
 
+GST_RTSP_SERVER_API
+void               gst_rtsp_stream_unblock_rtcp (GstRTSPStream * stream);
+
 /**
  * GstRTSPStreamTransportFilterFunc:
  * @stream: a #GstRTSPStream object

gst/rtsp-sink/gstrtspclientsink.c

@@ -4580,6 +4580,11 @@ gst_rtsp_client_sink_record (GstRTSPClientSink * sink, gboolean async)
 
   gst_rtsp_client_sink_set_state (sink, GST_STATE_PLAYING);
   sink->state = GST_RTSP_STATE_PLAYING;
+  for (walk = sink->contexts; walk; walk = g_list_next (walk)) {
+    GstRTSPStreamContext *context = (GstRTSPStreamContext *) walk->data;
+
+    gst_rtsp_stream_unblock_rtcp (context->stream);
+  }
 
   /* clean up any messages */
   gst_rtsp_message_unset (&request);