licode學習之erizo篇--MediaStream

MediaStream是erizo進行流數據處理的核心模塊。當網絡數據，經過DtlsTransport進行srtp解密后，得到的rtp裸數據與rtcp裸數據，都要進入MediaStream進行處理；需要發送給對方的rtp數據與rtcp裸數據也要經過MediaStream處理后，才會給DtlsTransport進行加密並發送。

MediaStream也是個人認為erizo的最為復雜的一個部分。先看一看其集成體系：

class MediaStream: public MediaSink, public MediaSource, public FeedbackSink,
                        public FeedbackSource, public LogContext, public HandlerManagerListener,
                        public std::enable_shared_from_this<MediaStream>, public Service 

整個繼承體系里面，涉及處理的基類有：MediaSink，MediaSource，FeedbackSink，FeedbackSource， HandlerManagerListener，Service

MediaStream同時承載着收數據處理，和發送數據處理兩部分內容。其中和丟包重傳等結合起來，就變為：接收rtp數據，發送rtcp重傳信息；發送rtp數據，接收rtcp重傳信息。

在接口上面，需要對外提供發送和接收到的裸數據回調。

這里面erizo的實現，是將MediaSink與MediaSource糾纏到一起的。但是從宏觀上，我的理解是：

MediaSink：負責發送數據（write to client）

FeedbackSink：負責發送數據（write to client）

MediaSource：負責read出來rtp數據 （read from client）

FeedbackSource：負責read出來數據（read from client）

 

MediaStream繼承MediaSink和FeedbackSink，所以直接調用MediaStream對象的deliverVideoData，deliverAudioData，deliverFeedback即可直接向對端發送數據。

要想接收對方的數據，需要MediaSource，FeedbackSource進行數據回調，怎么回調，需要進一步看一下MediaSource的定義：

**
 * A MediaSource is any class that produces audio or video data.
 */
class MediaSource: public virtual Monitor {
 protected:
    // SSRCs coming from the source
    uint32_t audio_source_ssrc_;
    std::vector<uint32_t> video_source_ssrc_list_;
    MediaSink* video_sink_;
    MediaSink* audio_sink_;
    MediaSink* event_sink_;
    // can it accept feedback
    FeedbackSink* source_fb_sink_;

 public:
    void setAudioSink(MediaSink* audio_sink) { boost::mutex::scoped_lock lock(monitor_mutex_); this->audio_sink_ = audio_sink; } void setVideoSink(MediaSink* video_sink) { boost::mutex::scoped_lock lock(monitor_mutex_); this->video_sink_ = video_sink; } void setEventSink(MediaSink* event_sink) { boost::mutex::scoped_lock lock(monitor_mutex_); this->event_sink_ = event_sink; }

    FeedbackSink* getFeedbackSink() {
        boost::mutex::scoped_lock lock(monitor_mutex_);
        return source_fb_sink_;
    }
    virtual int sendPLI() = 0;
    uint32_t getVideoSourceSSRC() {
        boost::mutex::scoped_lock lock(monitor_mutex_);
        if (video_source_ssrc_list_.empty()) {
          return 0;
        }
        return video_source_ssrc_list_[0];
    }
    void setVideoSourceSSRC(uint32_t ssrc) {
        boost::mutex::scoped_lock lock(monitor_mutex_);
        if (video_source_ssrc_list_.empty()) {
          video_source_ssrc_list_.push_back(ssrc);
          return;
        }
        video_source_ssrc_list_[0] = ssrc;
    }
    std::vector<uint32_t> getVideoSourceSSRCList() {
        boost::mutex::scoped_lock lock(monitor_mutex_);
        return video_source_ssrc_list_;  //  return by copy to avoid concurrent access
    }
    void setVideoSourceSSRCList(const std::vector<uint32_t>& new_ssrc_list) {
        boost::mutex::scoped_lock lock(monitor_mutex_);
        video_source_ssrc_list_ = new_ssrc_list;
    }
    uint32_t getAudioSourceSSRC() {
        boost::mutex::scoped_lock lock(monitor_mutex_);
        return audio_source_ssrc_;
    }
    void setAudioSourceSSRC(uint32_t ssrc) {
        boost::mutex::scoped_lock lock(monitor_mutex_);
        audio_source_ssrc_ = ssrc;
    }

    bool isVideoSourceSSRC(uint32_t ssrc) {
      auto found_ssrc = std::find_if(video_source_ssrc_list_.begin(), video_source_ssrc_list_.end(),
          [ssrc](uint32_t known_ssrc) {
          return known_ssrc == ssrc;
          });
      return (found_ssrc != video_source_ssrc_list_.end());
    }

    bool isAudioSourceSSRC(uint32_t ssrc) {
      return audio_source_ssrc_ == ssrc;
    }

    MediaSource() : audio_source_ssrc_{0}, video_source_ssrc_list_{std::vector<uint32_t>(1, 0)},
      video_sink_{nullptr}, audio_sink_{nullptr}, event_sink_{nullptr}, source_fb_sink_{nullptr} {}
    virtual ~MediaSource() {}

    virtual void close() = 0;
};

MediaSource定義了4個MediaSink對象，分別對應video,audio,event,feedback。

MediaStream繼承了MediaSource同樣的4個set接口，讓調用者可以設置MediaSource的這4個對象。當發生讀取事件時，MediaStream會調用這4個設置的MediaSink對象的相應方法，來向外傳遞數據。

FeedbackSource也是同樣的道理。

void MediaStream::read(std::shared_ptr<DataPacket> packet) {
  char* buf = packet->data;
  int len = packet->length;
  // PROCESS RTCP
  RtpHeader *head = reinterpret_cast<RtpHeader*> (buf);
  RtcpHeader *chead = reinterpret_cast<RtcpHeader*> (buf);
  uint32_t recvSSRC = 0;
  if (!chead->isRtcp()) {
    recvSSRC = head->getSSRC();
  } else if (chead->packettype == RTCP_Sender_PT || chead->packettype == RTCP_SDES_PT) {  // Sender Report
    recvSSRC = chead->getSSRC();
  }
  // DELIVER FEEDBACK (RR, FEEDBACK PACKETS)
  if (chead->isFeedback()) {
    if (fb_sink_ != nullptr && should_send_feedback_) {
      fb_sink_->deliverFeedback(std::move(packet));
    }
  } else {
    // RTP or RTCP Sender Report
    if (bundle_) {
      // Check incoming SSRC
      // Deliver data
      if (isVideoSourceSSRC(recvSSRC) && video_sink_) {
        parseIncomingPayloadType(buf, len, VIDEO_PACKET);
        video_sink_->deliverVideoData(std::move(packet));
      } else if (isAudioSourceSSRC(recvSSRC) && audio_sink_) {
        parseIncomingPayloadType(buf, len, AUDIO_PACKET);
        audio_sink_->deliverAudioData(std::move(packet));
      } else {
        ELOG_DEBUG("%s read video unknownSSRC: %u, localVideoSSRC: %u, localAudioSSRC: %u",
                    toLog(), recvSSRC, this->getVideoSourceSSRC(), this->getAudioSourceSSRC());
      }
    } else {
      if (packet->type == AUDIO_PACKET && audio_sink_) {
        parseIncomingPayloadType(buf, len, AUDIO_PACKET);
        // Firefox does not send SSRC in SDP
        if (getAudioSourceSSRC() == 0) {
          ELOG_DEBUG("%s discoveredAudioSourceSSRC:%u", toLog(), recvSSRC);
          this->setAudioSourceSSRC(recvSSRC);
        }
        audio_sink_->deliverAudioData(std::move(packet));
      } else if (packet->type == VIDEO_PACKET && video_sink_) {
        parseIncomingPayloadType(buf, len, VIDEO_PACKET);
        // Firefox does not send SSRC in SDP
        if (getVideoSourceSSRC() == 0) {
          ELOG_DEBUG("%s discoveredVideoSourceSSRC:%u", toLog(), recvSSRC);
          this->setVideoSourceSSRC(recvSSRC);
        }
        // change ssrc for RTP packets, don't touch here if RTCP
        video_sink_->deliverVideoData(std::move(packet));
      }
    }  // if not bundle
  }  // if not Feedback
}

這里，MediaStream的讀寫就清楚了，如果我們需要使用MediaStream，則需要做：

1、定義一個MediaSink的子類，將之設置給MediaStream，用於接收MediaStream的數據

2、直接調用MediaStream的deliver方法，讓其向外發送數據。

class Receiver : public MediaSink
{
public:
    virtual void close() {};

private:
    virtual int deliverAudioData_(std::shared_ptr<DataPacket> data_packet) {
        printf("now receiver audio packet\n");
    };
    virtual int deliverVideoData_(std::shared_ptr<DataPacket> data_packet) {
        printf("now receive video packet\n");
    };
    virtual int deliverEvent_(MediaEventPtr event) {
        printf("now receive event packet \n");
    };
};
class StreamControl
{
public:
    void init() {
        m_workerPool = std::make_shared<ThreadPool>(2);
        m_ioWorkerPool = std::make_shared<IOThreadPool>(2);

        std::string connid = "1";
        IceConfig cfg;//you may need init the cfg value
        std::vector<RtpMap> rtp_mappings;//you may need to init the mappings
        std::vector<erizo::ExtMap> ext_mappings; //you may need to init the ext mappings
        WebRtcConnectionEventListener* listener = new SendOfferEvtListener;
        m_conn = std::make_shared<WebRtcConnection>(workerPool->getLessUsedWorker(),
            m_ioWorkerPool->getLessUsedIOWorker(),
            connid,
            cfg,
            rtp_mappings,
            ext_mappings,
            listener);
        std::string stream_id = "1";
        std::string stream_label = "1";
        m_pStream = new MediaStream(m_workerPool->getLessUsedWorker(),
            m_conn,
            stream_id,
            stream_label,
            true);
        m_conn->addMediaStream(m_pStream);
        m_pStream->setVideoSink(&m_receiver);
        m_pStream->setAudioSink(&m_receiver);
        m_pStream->setEventSink(&m_receiver);
    }

    void sendRtpData(const char* buf, int len, bool isVideo) {
        std::shared_ptr<DataPacket> dp = std::make_shared<DataPacket>();
        dp->length = len;
        memcpy(dp->data, buf);
        dp->comp = 1;
        dp->type = isVideo ? VIDEO_PACKET : AUDIO_PACKET;
        if (isVideo)
        {
            m_pStream->deliverVideoData(dp);
        }
        else
        {
            m_pStream->deliverAudioData(dp);
        }
    }
private:
    Receiver m_receiver;
    MediaStream* m_pStream;
    std::shared_ptr<ThreadPool> m_workerPool;
    std::shared_ptr<IOThreadPool> m_ioWorkerPool;
    WebRtcConnection* m_conn;
};

 

MediaStream還繼承了HandlerManagerListener，Service這兩部分是媒體處理的核心模塊，pipeline使用的內容，之后學習時再寫吧。pipeline更是復雜，代碼好難讀。