rtpbin: pipeline gets an EOS when any rtpsources byes

Instead of sending EOS when a source byes we have to wait for
all the sources to be gone, which means they already sent BYE and
were removed from the session. We now handle the EOS in the rtcp
loop checking the amount of sources in the session.

https://bugzilla.gnome.org/show_bug.cgi?id=773218

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, gboolean eos, gpointer user_data);
+    RTPSource * src, GstBuffer * buffer, 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,6 +1156,22 @@ 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;
@@ -1413,11 +1429,10 @@ do_rtcp_events (GstRtpSession * rtpsession, GstPad * srcpad)
 }
 
 /* called when the session manager has an RTCP packet ready for further
- * sending. The eos flag is set when an EOS event should be sent downstream as
- * well. */
+ * sending. */
 static GstFlowReturn
 gst_rtp_session_send_rtcp (RTPSession * sess, RTPSource * src,
-    GstBuffer * buffer, gboolean eos, gpointer user_data)
+    GstBuffer * buffer, gpointer user_data)
 {
   GstFlowReturn result;
   GstRtpSession *rtpsession;
@@ -1440,11 +1455,6 @@ 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);
@@ -2056,7 +2066,6 @@ 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

@@ -3263,7 +3263,6 @@ early_exit:
 typedef struct
 {
   RTPSource *source;
-  gboolean is_bye;
   GstBuffer *buffer;
 } ReportOutput;
 
@@ -3874,7 +3873,6 @@ 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 */
@@ -3893,7 +3891,6 @@ 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 */
@@ -3918,7 +3915,6 @@ 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);
@@ -4098,7 +4094,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, output->is_bye,
+          sess->callbacks.send_rtcp (sess, source, buffer,
           sess->send_rtcp_user_data);
 
       RTP_SESSION_LOCK (sess);

gst/rtpmanager/rtpsession.h

@@ -71,7 +71,6 @@ 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
@@ -80,7 +79,7 @@ typedef GstFlowReturn (*RTPSessionSendRTP) (RTPSession *sess, RTPSource *src, gp
  * Returns: a #GstFlowReturn.
  */
 typedef GstFlowReturn (*RTPSessionSendRTCP) (RTPSession *sess, RTPSource *src, GstBuffer *buffer,
-    gboolean eos, gpointer user_data);
+    gpointer user_data);
 
 /**
  * RTPSessionSyncRTCP: