Revert "rtpbin: pipeline gets an EOS when any rtpsources byes"

This reverts commit eeea2a7fe88a17b15318d5b6ae6e190b2f777030.

It breaks EOS in some sender pipelines, see
https://bugzilla.gnome.org/show_bug.cgi?id=773218#c20

gst/rtpmanager/gstrtpsession.c

@@ -293,7 +293,7 @@ static GstFlowReturn gst_rtp_session_process_rtp (RTPSession * sess,
 static GstFlowReturn gst_rtp_session_send_rtp (RTPSession * sess,
     RTPSource * src, gpointer data, gpointer user_data);
 static GstFlowReturn gst_rtp_session_send_rtcp (RTPSession * sess,
-    RTPSource * src, GstBuffer * buffer, gpointer user_data);
+    RTPSource * src, GstBuffer * buffer, gboolean eos, gpointer user_data);
 static GstFlowReturn gst_rtp_session_sync_rtcp (RTPSession * sess,
     GstBuffer * buffer, gpointer user_data);
 static gint gst_rtp_session_clock_rate (RTPSession * sess, guint8 payload,
@@ -1156,22 +1156,6 @@ rtcp_thread (GstRtpSession * rtpsession)
     GST_RTP_SESSION_UNLOCK (rtpsession);
     rtp_session_on_timeout (session, current_time, ntpnstime, running_time);
     GST_RTP_SESSION_LOCK (rtpsession);
-
-    if (!rtp_session_get_num_sources (session)) {
-      /* when no sources left in the session, all of the them have went
-       * BYE at some point and removed, we can send EOS to the
-       * pipeline. */
-      GstPad *rtcp_src = rtpsession->send_rtcp_src;
-
-      if (rtcp_src) {
-        gst_object_ref (rtcp_src);
-        GST_LOG_OBJECT (rtpsession, "sending EOS");
-        GST_RTP_SESSION_UNLOCK (rtpsession);
-        gst_pad_push_event (rtpsession->send_rtcp_src, gst_event_new_eos ());
-        GST_RTP_SESSION_LOCK (rtpsession);
-        gst_object_unref (rtcp_src);
-      }
-    }
   }
   /* mark the thread as stopped now */
   rtpsession->priv->thread_stopped = TRUE;
@@ -1429,10 +1413,11 @@ do_rtcp_events (GstRtpSession * rtpsession, GstPad * srcpad)
 }
 
 /* called when the session manager has an RTCP packet ready for further
- * sending. */
+ * sending. The eos flag is set when an EOS event should be sent downstream as
+ * well. */
 static GstFlowReturn
 gst_rtp_session_send_rtcp (RTPSession * sess, RTPSource * src,
-    GstBuffer * buffer, gpointer user_data)
+    GstBuffer * buffer, gboolean eos, gpointer user_data)
 {
   GstFlowReturn result;
   GstRtpSession *rtpsession;
@@ -1455,6 +1440,11 @@ gst_rtp_session_send_rtcp (RTPSession * sess, RTPSource * src,
     GST_LOG_OBJECT (rtpsession, "sending RTCP");
     result = gst_pad_push (rtcp_src, buffer);
 
+    /* we have to send EOS after this packet */
+    if (eos) {
+      GST_LOG_OBJECT (rtpsession, "sending EOS");
+      gst_pad_push_event (rtcp_src, gst_event_new_eos ());
+    }
     gst_object_unref (rtcp_src);
   } else {
     GST_RTP_SESSION_UNLOCK (rtpsession);
@@ -2066,6 +2056,7 @@ gst_rtp_session_event_send_rtcp_src (GstPad * pad, GstObject * parent,
   return ret;
 }
 
+
 static gboolean
 gst_rtp_session_event_send_rtp_sink (GstPad * pad, GstObject * parent,
     GstEvent * event)

gst/rtpmanager/rtpsession.c

@@ -3266,6 +3266,7 @@ early_exit:
 typedef struct
 {
   RTPSource *source;
+  gboolean is_bye;
   GstBuffer *buffer;
 } ReportOutput;
 
@@ -3877,6 +3878,7 @@ static void
 generate_rtcp (const gchar * key, RTPSource * source, ReportData * data)
 {
   RTPSession *sess = data->sess;
+  gboolean is_bye = FALSE;
   ReportOutput *output;
 
   /* only generate RTCP for active internal sources */
@@ -3895,6 +3897,7 @@ generate_rtcp (const gchar * key, RTPSource * source, ReportData * data)
   if (source->marked_bye) {
     /* send BYE */
     make_source_bye (sess, source, data);
+    is_bye = TRUE;
   } else if (!data->is_early) {
     /* loop over all known sources and add report blocks. If we are early, we
      * just make a minimal RTCP packet and skip this step */
@@ -3919,6 +3922,7 @@ generate_rtcp (const gchar * key, RTPSource * source, ReportData * data)
 
   output = g_slice_new (ReportOutput);
   output->source = g_object_ref (source);
+  output->is_bye = is_bye;
   output->buffer = data->rtcp;
   /* queue the RTCP packet to push later */
   g_queue_push_tail (&data->output, output);
@@ -4102,7 +4106,7 @@ done:
       GST_DEBUG ("%p, sending RTCP packet, avg size %u, %u", &sess->stats,
           sess->stats.avg_rtcp_packet_size, packet_size);
       result =
-          sess->callbacks.send_rtcp (sess, source, buffer,
+          sess->callbacks.send_rtcp (sess, source, buffer, output->is_bye,
           sess->send_rtcp_user_data);
 
       RTP_SESSION_LOCK (sess);

gst/rtpmanager/rtpsession.h

@@ -71,6 +71,7 @@ typedef GstFlowReturn (*RTPSessionSendRTP) (RTPSession *sess, RTPSource *src, gp
  * @sess: an #RTPSession
  * @src: the #RTPSource
  * @buffer: the RTCP buffer ready for sending
+ * @eos: if an EOS event should be pushed
  * @user_data: user data specified when registering
  *
  * This callback will be called when @sess has @buffer ready for sending to
@@ -79,7 +80,7 @@ typedef GstFlowReturn (*RTPSessionSendRTP) (RTPSession *sess, RTPSource *src, gp
  * Returns: a #GstFlowReturn.
  */
 typedef GstFlowReturn (*RTPSessionSendRTCP) (RTPSession *sess, RTPSource *src, GstBuffer *buffer,
-    gpointer user_data);
+    gboolean eos, gpointer user_data);
 
 /**
  * RTPSessionSyncRTCP: