session: handle NACK feedback and generate events

Handle and parse the feedback NACK packets and generate a Retransmission
event for each NACKed packet

diff --git a/gst/rtpmanager/gstrtpsession.c b/gst/rtpmanager/gstrtpsession.c
index 9a9c4e1..8dbcaba 100644 (file)
--- a/gst/rtpmanager/gstrtpsession.c
+++ b/gst/rtpmanager/gstrtpsession.c
@@ -267,6 +267,8 @@ static void gst_rtp_session_request_key_unit (RTPSession * sess,
     gboolean all_headers, gpointer user_data);
 static GstClockTime gst_rtp_session_request_time (RTPSession * session,
     gpointer user_data);
+static void gst_rtp_session_notify_nack (RTPSession * sess,
+    guint16 seqnum, guint16 blp, gpointer user_data);
 
 static RTPSessionCallbacks callbacks = {
   gst_rtp_session_process_rtp,
@@ -276,7 +278,8 @@ static RTPSessionCallbacks callbacks = {
   gst_rtp_session_clock_rate,
   gst_rtp_session_reconsider,
   gst_rtp_session_request_key_unit,
-  gst_rtp_session_request_time
+  gst_rtp_session_request_time,
+  gst_rtp_session_notify_nack
 };
 
 /* GObject vmethods */
@@ -2332,3 +2335,36 @@ gst_rtp_session_request_time (RTPSession * session, gpointer user_data)
 
   return gst_clock_get_time (rtpsession->priv->sysclock);
 }
+
+static void
+gst_rtp_session_notify_nack (RTPSession * sess, guint16 seqnum,
+    guint16 blp, gpointer user_data)
+{
+  GstRtpSession *rtpsession = GST_RTP_SESSION (user_data);
+  GstEvent *event;
+  GstPad *send_rtp_sink;
+
+  GST_RTP_SESSION_LOCK (rtpsession);
+  if ((send_rtp_sink = rtpsession->send_rtp_sink))
+    gst_object_ref (send_rtp_sink);
+  GST_RTP_SESSION_UNLOCK (rtpsession);
+
+  if (send_rtp_sink) {
+    while (TRUE) {
+      event = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
+          gst_structure_new ("GstRTPRetransmissionRequest",
+              "seqnum", G_TYPE_UINT, (guint) seqnum, NULL));
+      gst_pad_push_event (send_rtp_sink, event);
+
+      if (blp == 0)
+        break;
+
+      seqnum++;
+      while ((blp & 1) == 0) {
+        seqnum++;
+        blp >>= 1;
+      }
+    }
+    gst_object_unref (send_rtp_sink);
+  }
+}

diff --git a/gst/rtpmanager/rtpsession.c b/gst/rtpmanager/rtpsession.c
index 1344074..4e978c1 100644 (file)
--- a/gst/rtpmanager/rtpsession.c
+++ b/gst/rtpmanager/rtpsession.c
@@ -847,6 +847,10 @@ rtp_session_set_callbacks (RTPSession * sess, RTPSessionCallbacks * callbacks,
     sess->callbacks.request_time = callbacks->request_time;
     sess->request_time_user_data = user_data;
   }
+  if (callbacks->notify_nack) {
+    sess->callbacks.notify_nack = callbacks->notify_nack;
+    sess->notify_nack_user_data = user_data;
+  }
 }
 
 /**
@@ -2169,6 +2173,32 @@ rtp_session_process_fir (RTPSession * sess, guint32 sender_ssrc,
 }
 
 static void
+rtp_session_process_nack (RTPSession * sess, guint32 sender_ssrc,
+    guint32 media_ssrc, guint8 * fci_data, guint fci_length,
+    GstClockTime current_time)
+{
+  if (!sess->callbacks.notify_nack)
+    return;
+
+  while (fci_length > 0) {
+    guint16 seqnum, blp;
+
+    seqnum = GST_READ_UINT16_BE (fci_data);
+    blp = GST_READ_UINT16_BE (fci_data + 2);
+
+    GST_DEBUG ("NACK #%u, blp %04x", seqnum, blp);
+
+    RTP_SESSION_UNLOCK (sess);
+    sess->callbacks.notify_nack (sess, seqnum, blp,
+        sess->notify_nack_user_data);
+    RTP_SESSION_LOCK (sess);
+
+    fci_data += 4;
+    fci_length -= 4;
+  }
+}
+
+static void
 rtp_session_process_feedback (RTPSession * sess, GstRTCPPacket * packet,
     RTPArrivalStats * arrival, GstClockTime current_time)
 {
@@ -2230,6 +2260,14 @@ rtp_session_process_feedback (RTPSession * sess, GstRTCPPacket * packet,
         }
         break;
       case GST_RTCP_TYPE_RTPFB:
+        switch (fbtype) {
+          case GST_RTCP_RTPFB_TYPE_NACK:
+            rtp_session_process_nack (sess, sender_ssrc, media_ssrc,
+                fci_data, fci_length, current_time);
+            break;
+          default:
+            break;
+        }
       default:
         break;
     }

diff --git a/gst/rtpmanager/rtpsession.h b/gst/rtpmanager/rtpsession.h
index 7867bac..c8dee20 100644 (file)
--- a/gst/rtpmanager/rtpsession.h
+++ b/gst/rtpmanager/rtpsession.h
@@ -124,7 +124,7 @@ typedef void (*RTPSessionReconsider) (RTPSession *sess, gpointer user_data);
  * @all_headers: %TRUE if "all-headers" property should be set on the key unit
  *  request
  * @user_data: user data specified when registering
-*
+ *
  * Asks the encoder to produce a key unit as soon as possibly within the
  * bandwidth constraints
  */
@@ -143,6 +143,18 @@ typedef GstClockTime (*RTPSessionRequestTime) (RTPSession *sess,
     gpointer user_data);
 
 /**
+ * RTPSessionNotifyNACK:
+ * @sess: an #RTPSession
+ * @seqnum: the missing seqnum
+ * @blp: other missing seqnums
+ * @user_data: user data specified when registering
+ *
+ * Notifies of NACKed frames.
+ */
+typedef void (*RTPSessionNotifyNACK) (RTPSession *sess,
+    guint16 seqnum, guint16 blp, gpointer user_data);
+
+/**
  * RTPSessionCallbacks:
  * @RTPSessionProcessRTP: callback to process RTP packets
  * @RTPSessionSendRTP: callback for sending RTP packets
@@ -150,6 +162,8 @@ typedef GstClockTime (*RTPSessionRequestTime) (RTPSession *sess,
  * @RTPSessionSyncRTCP: callback for handling SR packets
  * @RTPSessionReconsider: callback for reconsidering the timeout
  * @RTPSessionRequestKeyUnit: callback for requesting a new key unit
+ * @RTPSessionRequestTime: callback for requesting the current time
+ * @RTPSessionNotifyNACK: callback for notifying NACK
  *
  * These callbacks can be installed on the session manager to get notification
  * when RTP and RTCP packets are ready for further processing. These callbacks
@@ -164,6 +178,7 @@ typedef struct {
   RTPSessionReconsider  reconsider;
   RTPSessionRequestKeyUnit request_key_unit;
   RTPSessionRequestTime request_time;
+  RTPSessionNotifyNACK  notify_nack;
 } RTPSessionCallbacks;
 
 /**
@@ -227,6 +242,7 @@ struct _RTPSession {
   gpointer              reconsider_user_data;
   gpointer              request_key_unit_user_data;
   gpointer              request_time_user_data;
+  gpointer              notify_nack_user_data;
 
   RTPSessionStats stats;