rtpsession: Implement sending PLI packets in response to GstForceKeyUnit

gst/rtpmanager/gstrtpsession.c

@@ -1370,6 +1370,77 @@ gst_rtp_session_event_recv_rtp_sink (GstPad * pad, GstEvent * event)
 
 }
 
+static gboolean
+gst_rtp_session_request_remote_key_unit (GstRtpSession * rtpsession,
+    guint32 ssrc, guint payload, gboolean all_headers)
+{
+  GstCaps *caps;
+  gboolean requested = FALSE;
+
+  caps = gst_rtp_session_get_caps_for_pt (rtpsession, payload);
+
+  if (caps) {
+    gboolean fir, pli;
+    const GstStructure *s = gst_caps_get_structure (caps, 0);
+
+    if (all_headers &&
+        gst_structure_get_boolean (s, "rtcp-fb-nack-fir", &fir) && fir) {
+      /* 500 ms acceptable delay for FIR request is a guesstimate, it could
+       * be made configurable if needed
+       */
+      rtp_session_request_early_rtcp (rtpsession->priv->session,
+          gst_clock_get_time (rtpsession->priv->sysclock), 500 * GST_MSECOND);
+      rtp_session_request_key_unit (rtpsession->priv->session, ssrc, TRUE);
+      requested = TRUE;
+    } else if (gst_structure_get_boolean (s, "rtcp-fb-nack-pli", &pli) && pli) {
+      rtp_session_request_key_unit (rtpsession->priv->session, ssrc, FALSE);
+      requested = TRUE;
+    }
+    gst_caps_unref (caps);
+  }
+
+  return requested;
+}
+
+static gboolean
+gst_rtp_session_event_recv_rtp_src (GstPad * pad, GstEvent * event)
+{
+  GstRtpSession *rtpsession;
+  gboolean forward = TRUE;
+  gboolean ret = TRUE;
+  const GstStructure *s;
+  guint32 ssrc;
+  guint pt;
+
+  rtpsession = GST_RTP_SESSION (gst_pad_get_parent (pad));
+
+  switch (GST_EVENT_TYPE (event)) {
+    case GST_EVENT_CUSTOM_UPSTREAM:
+      s = gst_event_get_structure (event);
+      if (gst_structure_has_name (s, "GstForceKeyUnit") &&
+          gst_structure_get_uint (s, "ssrc", &ssrc) &&
+          gst_structure_get_uint (s, "payload", &pt)) {
+        gboolean all_headers = FALSE;
+
+        gst_structure_get_boolean (s, "all-headers", &all_headers);
+        if (gst_rtp_session_request_remote_key_unit (rtpsession, ssrc, pt,
+                all_headers))
+          forward = FALSE;
+      }
+      break;
+    default:
+      break;
+  }
+
+  if (forward)
+    ret = gst_pad_push_event (rtpsession->recv_rtp_sink, event);
+
+  gst_object_unref (rtpsession);
+
+  return ret;
+}
+
+
 static GstIterator *
 gst_rtp_session_iterate_internal_links (GstPad * pad)
 {
@@ -1780,6 +1851,8 @@ create_recv_rtp_sink (GstRtpSession * rtpsession)
   rtpsession->recv_rtp_src =
       gst_pad_new_from_static_template (&rtpsession_recv_rtp_src_template,
       "recv_rtp_src");
+  gst_pad_set_event_function (rtpsession->recv_rtp_src,
+      (GstPadEventFunction) gst_rtp_session_event_recv_rtp_src);
   gst_pad_set_iterate_internal_links_function (rtpsession->recv_rtp_src,
       gst_rtp_session_iterate_internal_links);
   gst_pad_use_fixed_caps (rtpsession->recv_rtp_src);

gst/rtpmanager/rtpsession.c

@@ -102,6 +102,10 @@ static void rtp_session_set_property (GObject * object, guint prop_id,
 static void rtp_session_get_property (GObject * object, guint prop_id,
     GValue * value, GParamSpec * pspec);
 
+static gboolean rtp_session_on_sending_rtcp (RTPSession * sess,
+    GstBuffer * buffer, gboolean early);
+
+
 static guint rtp_session_signals[LAST_SIGNAL] = { 0 };
 
 G_DEFINE_TYPE (RTPSession, rtp_session, G_TYPE_OBJECT);
@@ -403,6 +407,7 @@ rtp_session_class_init (RTPSessionClass * klass)
 
   klass->get_source_by_ssrc =
       GST_DEBUG_FUNCPTR (rtp_session_get_source_by_ssrc);
+  klass->on_sending_rtcp = GST_DEBUG_FUNCPTR (rtp_session_on_sending_rtcp);
 
   GST_DEBUG_CATEGORY_INIT (rtp_session_debug, "rtpsession", 0, "RTP Session");
 }
@@ -457,6 +462,8 @@ rtp_session_init (RTPSession * sess)
   sess->allow_early = TRUE;
   sess->rtcp_feedback_retention_window = DEFAULT_RTCP_FEEDBACK_RETENTION_WINDOW;
 
+  sess->rtcp_pli_requests = g_array_new (FALSE, FALSE, sizeof (guint32));
+
   GST_DEBUG ("%p: session using SSRC: %08x", sess, sess->source->ssrc);
 }
 
@@ -477,6 +484,8 @@ rtp_session_finalize (GObject * object)
   g_hash_table_destroy (sess->cnames);
   g_object_unref (sess->source);
 
+  g_array_free (sess->rtcp_pli_requests, TRUE);
+
   G_OBJECT_CLASS (rtp_session_parent_class)->finalize (object);
 }
 
@@ -2973,3 +2982,71 @@ dont_send:
   RTP_SESSION_UNLOCK (sess);
 
 }
+
+void
+rtp_session_request_key_unit (RTPSession * sess, guint32 ssrc, gboolean fir)
+{
+  guint i;
+
+  if (fir)
+    return;
+
+  for (i = 0; i < sess->rtcp_pli_requests->len; i++)
+    if (ssrc == g_array_index (sess->rtcp_pli_requests, guint32, i))
+      return;
+
+  g_array_append_val (sess->rtcp_pli_requests, ssrc);
+}
+
+static gboolean
+has_pli_compare_func (gconstpointer a, gconstpointer ignored)
+{
+  GstRTCPPacket packet;
+
+  packet.buffer = (GstBuffer *) a;
+  packet.offset = 0;
+
+  if (gst_rtcp_packet_get_type (&packet) == GST_RTCP_TYPE_PSFB &&
+      gst_rtcp_packet_fb_get_type (&packet) == GST_RTCP_PSFB_TYPE_PLI)
+    return TRUE;
+  else
+    return FALSE;
+}
+
+static gboolean
+rtp_session_on_sending_rtcp (RTPSession * sess, GstBuffer * buffer,
+    gboolean early)
+{
+  gboolean ret = FALSE;
+
+  RTP_SESSION_LOCK (sess);
+
+  while (sess->rtcp_pli_requests->len) {
+    GstRTCPPacket rtcppacket;
+    guint media_ssrc = g_array_index (sess->rtcp_pli_requests, guint32, 0);
+    RTPSource *media_src = g_hash_table_lookup (sess->ssrcs[sess->mask_idx],
+        GUINT_TO_POINTER (media_ssrc));
+
+    if (media_src && !rtp_source_has_retained (media_src,
+            has_pli_compare_func, NULL)) {
+      if (gst_rtcp_buffer_add_packet (buffer, GST_RTCP_TYPE_PSFB, &rtcppacket)) {
+        gst_rtcp_packet_fb_set_type (&rtcppacket, GST_RTCP_PSFB_TYPE_PLI);
+        gst_rtcp_packet_fb_set_sender_ssrc (&rtcppacket,
+            rtp_source_get_ssrc (sess->source));
+        gst_rtcp_packet_fb_set_media_ssrc (&rtcppacket, media_ssrc);
+        ret = TRUE;
+      } else {
+        /* Break because the packet is full, will put next request in a
+         * further packet
+         */
+        break;
+      }
+    }
+
+    g_array_remove_index (sess->rtcp_pli_requests, 0);
+  }
+
+  RTP_SESSION_UNLOCK (sess);
+
+  return ret;
+}

gst/rtpmanager/rtpsession.h

@@ -203,6 +203,8 @@ struct _RTPSession {
   gboolean      change_ssrc;
   gboolean      favor_new;
   GstClockTime  rtcp_feedback_retention_window;
+
+  GArray       *rtcp_pli_requests;
 };
 
 /**
@@ -308,5 +310,9 @@ GstFlowReturn   rtp_session_on_timeout             (RTPSession *sess, GstClockTi
 void            rtp_session_request_early_rtcp     (RTPSession * sess, GstClockTime current_time,
                                                     GstClockTimeDiff max_delay);
 
+/* Notify session of a request for a new key unit */
+void            rtp_session_request_key_unit       (RTPSession * sess,
+                                                    guint32 ssrc,
+                                                    gboolean fir);
 
 #endif /* __RTP_SESSION_H__ */