gstreamer分析-rtp流的接收

 关于数据流的接收过程，[1]有很好的分析。这个问题，令人困惑，依然有些东西没有搞明白。gst_pad_start_task()创建任务，使得gst_base_src_loop()被循环调用。

gst_base_src_set_playing
{
    if (start)
      gst_pad_start_task (basesrc->srcpad, (GstTaskFunction) gst_base_src_loop,
          basesrc->srcpad, NULL);   
}
static void
gst_base_src_loop (GstPad * pad)
{
  ret = gst_base_src_get_range (src, position, blocksize, &buf);
}
static GstFlowReturn
gst_base_src_get_range (GstBaseSrc * src, guint64 offset, guint length,
    GstBuffer ** buf)
{
  ret = bclass->create (src, offset, length, &res_buf);
//GstBaseClass中的函数指针klass->create = GST_DEBUG_FUNCPTR (gst_base_src_default_create);
  if (src->priv->pending_bufferlist != NULL) {
    ret = gst_pad_push_list (pad, src->priv->pending_bufferlist);
    src->priv->pending_bufferlist = NULL;
  } else {
    ret = gst_pad_push (pad, buf);//数据在pipeline中流动,调用下一个element的chainfun，推送数据。
  }
}
static GstFlowReturn
gst_base_src_default_create (GstBaseSrc * src, guint64 offset,
    guint size, GstBuffer ** buffer)
{
    ret = bclass->fill (src, offset, size, res_buf);
//这个函数在子类中被覆盖gstbasesrc_class->fill = GST_DEBUG_FUNCPTR (gst_push_src_fill);
}
static GstFlowReturn
gst_push_src_fill (GstBaseSrc * bsrc, guint64 offset, guint length,
    GstBuffer * ret)
{
  pclass = GST_PUSH_SRC_GET_CLASS (src);
  if (pclass->fill)//GstPushSrcClass有同名的方法，与bclass->fill不同
    fret = pclass->fill (src, ret);//这个方法再次被子类覆盖
}
//其中GstUDPSrcClass继承了GstPushSrc
//gstpushsrc_class->fill = gst_udpsrc_fill;
static GstFlowReturn
gst_udpsrc_fill (GstPushSrc * psrc, GstBuffer * outbuf)
{
//问题出现，向socket读取数据，数据流就一定到来了吗？
//g_socket_condition_timed_wait说明这个socket并不是一个阻塞的socket。那么要是没有数据该怎么处理？
  res =
      g_socket_receive_message (udpsrc->used_socket, p_saddr, ivec, 2,
      p_msgs, &n_msgs, &flags, udpsrc->cancellable, &err);
}

 udosrc怎么同rtpbin关联起来呢？gst-lpugins-good/tests/examples/client-rtpaux.c下有个例子，有个图：

example

 udosrc和rtpbin都继承了element。gst_element_link_pads将两者连起来。

rtpSrc = gst_element_factory_make ("udpsrc", NULL);
gst_element_link_pads (rtpSrc, "src", rtpBin, padName);

 rtpbin中的pad是如何接收数据呢？

static GstPad *
gst_rtp_bin_request_new_pad (GstElement * element,
    GstPadTemplate * templ, const gchar * name, const GstCaps * caps)
{
  if (templ == gst_element_class_get_pad_template (klass, "recv_rtp_sink_%u")) {
    result = create_recv_rtp (rtpbin, templ, pad_name);
  }
}
static GstPad *
create_recv_rtp (GstRtpBin * rtpbin, GstPadTemplate * templ, const gchar * name)
{
  recv_rtp_sink = complete_session_sink (rtpbin, session, TRUE);
}
static GstPad *
complete_session_sink (GstRtpBin * rtpbin, GstRtpBinSession * session,
    gboolean bundle_demuxer_needed)
{
  session->recv_rtp_sink =
      gst_element_get_request_pad (session->session, "recv_rtp_sink");
}
static GstPad *
gst_rtp_session_request_new_pad (GstElement * element,
    GstPadTemplate * templ, const gchar * name, const GstCaps * caps)
{
  /* figure out the template */
  if (templ == gst_element_class_get_pad_template (klass, "recv_rtp_sink")) {
    if (rtpsession->recv_rtp_sink != NULL)
      goto exists;

    result = create_recv_rtp_sink (rtpsession);
  }
}
static GstPad *
create_recv_rtp_sink (GstRtpSession * rtpsession)
{
  gst_pad_set_chain_function (rtpsession->recv_rtp_sink,
      gst_rtp_session_chain_recv_rtp);
  return rtpsession->recv_rtp_sink;
}

gst_pad_push中调用的chainfun就是这个gst_rtp_session_chain_recv_rtp。
 接下来分析rtp数据流的处理，以on-feedback-rtcp这个类型的消息为例。

static GstFlowReturn
gst_rtp_session_chain_recv_rtp (GstPad * pad, GstObject * parent,
    GstBuffer * buffer)
{
  ret = rtp_session_process_rtp (priv->session, buffer, current_time,
      running_time, ntpnstime);
}
GstFlowReturn
rtp_session_process_rtp (RTPSession * sess, GstBuffer * buffer,
    GstClockTime current_time, GstClockTime running_time, guint64 ntpnstime)
{
  /* update pinfo stats */
  if (!update_packet_info (sess, &pinfo, FALSE, TRUE, FALSE, buffer,
          current_time, running_time, ntpnstime)) {
    GST_DEBUG ("invalid RTP packet received");
    RTP_SESSION_UNLOCK (sess);
    return rtp_session_process_rtcp (sess, buffer, current_time, ntpnstime);
  }
}
GstFlowReturn
rtp_session_process_rtcp (RTPSession * sess, GstBuffer * buffer,
    GstClockTime current_time, guint64 ntpnstime)
{
    switch (type) {
      case GST_RTCP_TYPE_PSFB:
        rtp_session_process_feedback (sess, &packet, &pinfo, current_time);
        break;
}
}
static void
rtp_session_process_feedback (RTPSession * sess, GstRTCPPacket * packet,
    RTPPacketInfo * pinfo, GstClockTime current_time)
{
  if (g_signal_has_handler_pending (sess,
          rtp_session_signals[SIGNAL_ON_FEEDBACK_RTCP], 0, TRUE)) {
    g_signal_emit (sess, rtp_session_signals[SIGNAL_ON_FEEDBACK_RTCP], 0,
        type, fbtype, sender_ssrc, media_ssrc, fci_buffer);
}
}
//需要查找下程序哪里注册了"on-feedback-rtcp"这个信号。
//以此为例，信号的注册是这个形式g_signal_connect (rtpsession, "on-feedback-rtcp", G_CALLBACK (callback_fun), self);
//信号发射g_signal_emit，就会调用callback_fun进行处理。
//这样的注册函数在kurento里有。
static void
kms_remb_remote_on_feedback_rtcp (GObject *rtpsession,
    guint type, guint fbtype, guint sender_ssrc, guint media_ssrc,
    GstBuffer *fci)
{
  GST_LOG_OBJECT (rtpsession, "Signal 'on-feedback-rtcp'");
  if (type == GST_RTCP_TYPE_PSFB
      && fbtype == GST_RTCP_PSFB_TYPE_AFB) {
    process_psfb_afb (rtpsession, sender_ssrc, fci);
  }
}

 前前后后一共浪费了六天时间，好奇心害死猫。以后应该研究一些数学密集型的程序，而不是逻辑密集型的程序。
 剧终。
[1]GStreamer插件架构简析