Fix data parse error

src/interface/x.h

@@ -19,7 +19,9 @@ typedef void (*NetStatusReport)(const unsigned short, const unsigned short);
 
 typedef struct {
   const char* cfg_path;
-  OnReceiveBuffer on_receive_buffer;
+  OnReceiveBuffer on_receive_video_buffer;
+  OnReceiveBuffer on_receive_audio_buffer;
+  OnReceiveBuffer on_receive_data_buffer;
   NetStatusReport net_status_report;
 } Params;
 

src/pc/peer_connection.cpp

@@ -9,8 +9,7 @@
 
 using nlohmann::json;
 
-PeerConnection::PeerConnection(OnReceiveBuffer on_receive_buffer)
-    : on_receive_buffer_(on_receive_buffer) {}
+PeerConnection::PeerConnection() {}
 
 PeerConnection::~PeerConnection() {
   if (nv12_data_) {
@@ -40,10 +39,14 @@ int PeerConnection::Init(PeerConnectionParams params,
   LOG_INFO("stun server ip [{}] port [{}]", cfg_stun_server_ip_,
            stun_server_port_);
 
+  on_receive_video_buffer_ = params.on_receive_video_buffer;
+  on_receive_audio_buffer_ = params.on_receive_audio_buffer;
+  on_receive_data_buffer_ = params.on_receive_data_buffer;
+
   on_receive_ws_msg_ = [this](const std::string &msg) { ProcessSignal(msg); };
 
-  on_receive_ice_msg_ = [this](const char *data, size_t size,
-                               const char *user_id, size_t user_id_size) {
+  on_receive_video_ = [this](const char *data, size_t size, const char *user_id,
+                             size_t user_id_size) {
     int num_frame_returned = Decode((uint8_t *)data, size);
     uint32_t width = 0;
     uint32_t height = 0;
@@ -51,8 +54,24 @@ int PeerConnection::Init(PeerConnectionParams params,
     for (size_t i = 0; i < num_frame_returned; ++i) {
       int ret = GetFrame((uint8_t *)nv12_data_, width, height, frame_size);
 
-      on_receive_buffer_(nv12_data_, width * height * 3 / 2, user_id,
-                         user_id_size);
+      if (on_receive_video_buffer_) {
+        on_receive_video_buffer_(nv12_data_, width * height * 3 / 2, user_id,
+                                 user_id_size);
+      }
+    }
+  };
+
+  on_receive_audio_ = [this](const char *data, size_t size, const char *user_id,
+                             size_t user_id_size) {
+    if (on_receive_audio_buffer_) {
+      on_receive_audio_buffer_(data, size, user_id, user_id_size);
+    }
+  };
+
+  on_receive_data_ = [this](const char *data, size_t size, const char *user_id,
+                            size_t user_id_size) {
+    if (on_receive_data_buffer_) {
+      on_receive_data_buffer_(data, size, user_id, user_id_size);
     }
   };
 
@@ -152,9 +171,17 @@ void PeerConnection::ProcessSignal(const std::string &signal) {
 
       for (auto &remote_user_id : user_id_list_) {
         ice_transmission_list_[remote_user_id] =
-            std::make_unique<IceTransmission>(
-                true, transmission_id, user_id_, remote_user_id, ws_transport_,
-                on_receive_ice_msg_, on_ice_status_change_);
+            std::make_unique<IceTransmission>(true, transmission_id, user_id_,
+                                              remote_user_id, ws_transport_,
+                                              on_ice_status_change_);
+
+        ice_transmission_list_[remote_user_id]->SetOnReceiveVideoFunc(
+            on_receive_video_);
+        ice_transmission_list_[remote_user_id]->SetOnReceiveAudioFunc(
+            on_receive_audio_);
+        ice_transmission_list_[remote_user_id]->SetOnReceiveDataFunc(
+            on_receive_data_);
+
         ice_transmission_list_[remote_user_id]->InitIceTransmission(
             cfg_stun_server_ip_, stun_server_port_);
         ice_transmission_list_[remote_user_id]->JoinTransmission();
@@ -186,9 +213,16 @@ void PeerConnection::ProcessSignal(const std::string &signal) {
         LOG_INFO("[{}] receive offer from [{}]", user_id_, remote_user_id);
 
         ice_transmission_list_[remote_user_id] =
-            std::make_unique<IceTransmission>(
-                false, transmission_id, user_id_, remote_user_id, ws_transport_,
-                on_receive_ice_msg_, on_ice_status_change_);
+            std::make_unique<IceTransmission>(false, transmission_id, user_id_,
+                                              remote_user_id, ws_transport_,
+                                              on_ice_status_change_);
+
+        ice_transmission_list_[remote_user_id]->SetOnReceiveVideoFunc(
+            on_receive_video_);
+        ice_transmission_list_[remote_user_id]->SetOnReceiveAudioFunc(
+            on_receive_audio_);
+        ice_transmission_list_[remote_user_id]->SetOnReceiveDataFunc(
+            on_receive_data_);
 
         ice_transmission_list_[remote_user_id]->InitIceTransmission(
             cfg_stun_server_ip_, stun_server_port_);

src/pc/peer_connection.h

@@ -18,13 +18,15 @@ typedef void (*NetStatusReport)(const unsigned short, const unsigned short);
 
 typedef struct {
   const char *cfg_path;
-  OnReceiveBuffer on_receive_buffer;
+  OnReceiveBuffer on_receive_video_buffer;
+  OnReceiveBuffer on_receive_audio_buffer;
+  OnReceiveBuffer on_receive_data_buffer;
   NetStatusReport net_status_report;
 } PeerConnectionParams;
 
 class PeerConnection : public VideoEncoder, VideoDecoder {
  public:
-  PeerConnection(OnReceiveBuffer on_receive_buffer);
+  PeerConnection();
   ~PeerConnection();
 
  public:
@@ -73,11 +75,17 @@ class PeerConnection : public VideoEncoder, VideoDecoder {
   std::map<std::string, std::unique_ptr<IceTransmission>>
       ice_transmission_list_;
   std::function<void(const char *, size_t, const char *, size_t)>
-      on_receive_ice_msg_ = nullptr;
+      on_receive_video_ = nullptr;
+  std::function<void(const char *, size_t, const char *, size_t)>
+      on_receive_audio_ = nullptr;
+  std::function<void(const char *, size_t, const char *, size_t)>
+      on_receive_data_ = nullptr;
   std::function<void(std::string)> on_ice_status_change_ = nullptr;
   bool ice_ready_ = false;
 
-  OnReceiveBuffer on_receive_buffer_;
+  OnReceiveBuffer on_receive_video_buffer_;
+  OnReceiveBuffer on_receive_audio_buffer_;
+  OnReceiveBuffer on_receive_data_buffer_;
   char *nv12_data_ = nullptr;
 };
 

src/rtc/x_inner.cpp

@@ -14,9 +14,11 @@ static PeerConnection *peer_connection;
 
 PeerPtr *CreatePeer(const Params *params) {
   PeerPtr *peer_ptr = new PeerPtr;
-  peer_ptr->peer_connection = new PeerConnection(params->on_receive_buffer);
+  peer_ptr->peer_connection = new PeerConnection();
   peer_ptr->pc_params.cfg_path = params->cfg_path;
-  peer_ptr->pc_params.on_receive_buffer = params->on_receive_buffer;
+  peer_ptr->pc_params.on_receive_video_buffer = params->on_receive_video_buffer;
+  peer_ptr->pc_params.on_receive_audio_buffer = params->on_receive_audio_buffer;
+  peer_ptr->pc_params.on_receive_data_buffer = params->on_receive_data_buffer;
   peer_ptr->pc_params.net_status_report = params->net_status_report;
 
   return peer_ptr;

src/rtp/rtp_codec.cpp

@@ -154,6 +154,6 @@ size_t RtpCodec::Decode(RtpPacket& packet, uint8_t* payload) {
     return packet.DecodeH264Fua(payload);
   } else {
     LOG_ERROR("Default");
-    return packet.Decode(payload);
+    return packet.DecodeData(payload);
   }
 }

src/rtp/rtp_data_receiver.cpp

@@ -1,5 +1,90 @@
 #include "rtp_data_receiver.h"
 
+#define RTCP_RR_INTERVAL 1000
+
 RtpDataReceiver::RtpDataReceiver() {}
 
-RtpDataReceiver::~RtpDataReceiver() {}
+RtpDataReceiver::~RtpDataReceiver() {
+  if (rtp_statistics_) {
+    rtp_statistics_->Stop();
+  }
+}
+
+void RtpDataReceiver::InsertRtpPacket(RtpPacket& rtp_packet) {
+  if (!rtp_statistics_) {
+    rtp_statistics_ = std::make_unique<RtpStatistics>();
+    rtp_statistics_->Start();
+  }
+
+  if (rtp_statistics_) {
+    rtp_statistics_->UpdateReceiveBytes(rtp_packet.Size());
+  }
+
+  if (CheckIsTimeSendRR()) {
+    RtcpReceiverReport rtcp_rr;
+    RtcpReportBlock report;
+
+    auto duration = std::chrono::system_clock::now().time_since_epoch();
+    auto seconds = std::chrono::duration_cast<std::chrono::seconds>(duration);
+    uint32_t seconds_u32 = static_cast<uint32_t>(
+        std::chrono::duration_cast<std::chrono::seconds>(duration).count());
+
+    uint32_t fraction_u32 = static_cast<uint32_t>(
+        std::chrono::duration_cast<std::chrono::nanoseconds>(duration - seconds)
+            .count());
+
+    report.source_ssrc = 0x00;
+    report.fraction_lost = 0;
+    report.cumulative_lost = 0;
+    report.extended_high_seq_num = 0;
+    report.jitter = 0;
+    report.lsr = 0;
+    report.dlsr = 0;
+
+    rtcp_rr.SetReportBlock(report);
+
+    rtcp_rr.Encode();
+
+    SendRtcpRR(rtcp_rr);
+  }
+
+  if (on_receive_data_) {
+    on_receive_data_((const char*)rtp_packet.Payload(),
+                     rtp_packet.PayloadSize());
+  }
+}
+
+void RtpDataReceiver::SetSendDataFunc(
+    std::function<int(const char*, size_t)> data_send_func) {
+  data_send_func_ = data_send_func;
+}
+
+int RtpDataReceiver::SendRtcpRR(RtcpReceiverReport& rtcp_rr) {
+  if (!data_send_func_) {
+    LOG_ERROR("data_send_func_ is nullptr");
+    return -1;
+  }
+
+  if (data_send_func_((const char*)rtcp_rr.Buffer(), rtcp_rr.Size())) {
+    LOG_ERROR("Send RR failed");
+    return -1;
+  }
+
+  // LOG_ERROR("Send RR");
+
+  return 0;
+}
+
+bool RtpDataReceiver::CheckIsTimeSendRR() {
+  uint32_t now_ts = static_cast<uint32_t>(
+      std::chrono::duration_cast<std::chrono::milliseconds>(
+          std::chrono::high_resolution_clock::now().time_since_epoch())
+          .count());
+
+  if (now_ts - last_send_rtcp_rr_packet_ts_ >= RTCP_RR_INTERVAL) {
+    last_send_rtcp_rr_packet_ts_ = now_ts;
+    return true;
+  } else {
+    return false;
+  }
+}

src/rtp/rtp_data_receiver.h

@@ -1,13 +1,39 @@
 #ifndef _RTP_DATA_RECEIVER_H_
 #define _RTP_DATA_RECEIVER_H_
 
+#include <functional>
+
+#include "rtcp_receiver_report.h"
+#include "rtp_codec.h"
+#include "rtp_statistics.h"
+
 class RtpDataReceiver {
  public:
   RtpDataReceiver();
   ~RtpDataReceiver();
 
+ public:
+  void InsertRtpPacket(RtpPacket& rtp_packet);
+
+  void SetSendDataFunc(std::function<int(const char*, size_t)> data_send_func);
+
+  void SetOnReceiveData(
+      std::function<void(const char*, size_t)> on_receive_data) {
+    on_receive_data_ = on_receive_data;
+  }
+
  private:
-  /* data */
+  bool CheckIsTimeSendRR();
+  int SendRtcpRR(RtcpReceiverReport& rtcp_rr);
+
+ private:
+  std::function<void(const char*, size_t)> on_receive_data_ = nullptr;
+  uint32_t last_complete_frame_ts_ = 0;
+
+ private:
+  std::unique_ptr<RtpStatistics> rtp_statistics_ = nullptr;
+  uint32_t last_send_rtcp_rr_packet_ts_ = 0;
+  std::function<int(const char*, size_t)> data_send_func_ = nullptr;
 };
 
 #endif

src/rtp/rtp_data_sender.cpp

@@ -102,7 +102,7 @@ int RtpDataSender::SendRtcpSR(RtcpSenderReport& rtcp_sr) {
     return -1;
   }
 
-  LOG_ERROR("Send SR");
+  // LOG_ERROR("Send SR");
 
   return 0;
 }

src/rtp/rtp_packet.cpp

@@ -4,14 +4,25 @@
 
 #include "log.h"
 
-inline void RtpPacket::TryToDecodeH264RtpPacket(uint8_t *buffer) {
-  if (PAYLOAD_TYPE::H264 == NAL_UNIT_TYPE(buffer[1] & 0x7f)) {
-    nal_unit_type_ = NAL_UNIT_TYPE(buffer[12] & 0x1F);
+void RtpPacket::TryToDecodeRtpPacket() {
+  if (PAYLOAD_TYPE::H264 == NAL_UNIT_TYPE(buffer_[1] & 0x7F)) {
+    nal_unit_type_ = NAL_UNIT_TYPE(buffer_[12] & 0x1F);
     if (NAL_UNIT_TYPE::NALU == nal_unit_type_) {
       DecodeH264Nalu();
     } else if (NAL_UNIT_TYPE::FU_A == nal_unit_type_) {
       DecodeH264Fua();
     }
+  } else if (PAYLOAD_TYPE::DATA == NAL_UNIT_TYPE(buffer_[1] & 0x7F)) {
+    DecodeData();
+  } else {
+    LOG_ERROR("Unknown pt: {}", NAL_UNIT_TYPE(buffer_[1] & 0x7F));
+  }
+}
+
+void RtpPacket::ParseRtpData() {
+  if (!parsed_) {
+    TryToDecodeRtpPacket();
+    parsed_ = true;
   }
 }
 
@@ -25,7 +36,7 @@ RtpPacket::RtpPacket(const uint8_t *buffer, size_t size) {
     memcpy(buffer_, buffer, size);
     size_ = size;
 
-    TryToDecodeH264RtpPacket(buffer_);
+    // TryToDecodeH264RtpPacket(buffer_);
   }
 }
 
@@ -35,7 +46,7 @@ RtpPacket::RtpPacket(const RtpPacket &rtp_packet) {
     memcpy(buffer_, rtp_packet.buffer_, rtp_packet.size_);
     size_ = rtp_packet.size_;
 
-    TryToDecodeH264RtpPacket(buffer_);
+    // TryToDecodeH264RtpPacket(buffer_);
   }
 }
 
@@ -45,7 +56,7 @@ RtpPacket::RtpPacket(RtpPacket &&rtp_packet)
   rtp_packet.buffer_ = nullptr;
   rtp_packet.size_ = 0;
 
-  TryToDecodeH264RtpPacket(buffer_);
+  // TryToDecodeH264RtpPacket(buffer_);
 }
 
 RtpPacket &RtpPacket::operator=(const RtpPacket &rtp_packet) {
@@ -58,7 +69,7 @@ RtpPacket &RtpPacket::operator=(const RtpPacket &rtp_packet) {
     memcpy(buffer_, rtp_packet.buffer_, rtp_packet.size_);
     size_ = rtp_packet.size_;
 
-    TryToDecodeH264RtpPacket(buffer_);
+    // TryToDecodeH264RtpPacket(buffer_);
   }
   return *this;
 }
@@ -70,7 +81,7 @@ RtpPacket &RtpPacket::operator=(RtpPacket &&rtp_packet) {
     size_ = rtp_packet.size_;
     rtp_packet.size_ = 0;
 
-    TryToDecodeH264RtpPacket(buffer_);
+    // TryToDecodeH264RtpPacket(buffer_);
   }
   return *this;
 }
@@ -232,7 +243,7 @@ const uint8_t *RtpPacket::EncodeH264Fua(uint8_t *payload, size_t payload_size) {
   return buffer_;
 }
 
-size_t RtpPacket::Decode(uint8_t *payload) {
+size_t RtpPacket::DecodeData(uint8_t *payload) {
   version_ = (buffer_[0] >> 6) & 0x03;
   has_padding_ = (buffer_[0] >> 5) & 0x01;
   has_extension_ = (buffer_[0] >> 4) & 0x01;
@@ -259,7 +270,8 @@ size_t RtpPacket::Decode(uint8_t *payload) {
         (buffer_[14 + extension_offset] << 8) | buffer_[15 + extension_offset];
 
     // extension_data_ = new uint8_t[extension_len_];
-    // memcpy(extension_data_, buffer_ + 16 + extension_offset, extension_len_);
+    // memcpy(extension_data_, buffer_ + 16 + extension_offset,
+    // extension_len_);
     extension_data_ = buffer_ + 16 + extension_offset;
   }
 
@@ -268,7 +280,9 @@ size_t RtpPacket::Decode(uint8_t *payload) {
 
   payload_size_ = size_ - (12 + payload_offset);
   payload_ = buffer_ + 12 + payload_offset;
-  memcpy(payload, payload_, payload_size_);
+  if (payload) {
+    memcpy(payload, payload_, payload_size_);
+  }
 
   return payload_size_;
 }
@@ -300,7 +314,8 @@ size_t RtpPacket::DecodeH264Nalu(uint8_t *payload) {
         (buffer_[14 + extension_offset] << 8) | buffer_[15 + extension_offset];
 
     // extension_data_ = new uint8_t[extension_len_];
-    // memcpy(extension_data_, buffer_ + 16 + extension_offset, extension_len_);
+    // memcpy(extension_data_, buffer_ + 16 + extension_offset,
+    // extension_len_);
     extension_data_ = buffer_ + 16 + extension_offset;
   }
 
@@ -346,7 +361,8 @@ size_t RtpPacket::DecodeH264Fua(uint8_t *payload) {
         (buffer_[14 + extension_offset] << 8) | buffer_[15 + extension_offset];
 
     // extension_data_ = new uint8_t[extension_len_];
-    // memcpy(extension_data_, buffer_ + 16 + extension_offset, extension_len_);
+    // memcpy(extension_data_, buffer_ + 16 + extension_offset,
+    // extension_len_);
     extension_data_ = buffer_ + 16 + extension_offset;
   }
 

src/rtp/rtp_packet.h

@@ -135,39 +135,88 @@ class RtpPacket {
   const uint8_t *Encode(uint8_t *payload, size_t payload_size);
   const uint8_t *EncodeH264Nalu(uint8_t *payload, size_t payload_size);
   const uint8_t *EncodeH264Fua(uint8_t *payload, size_t payload_size);
-  size_t Decode(uint8_t *payload);
+  size_t DecodeData(uint8_t *payload = nullptr);
   size_t DecodeH264Nalu(uint8_t *payload = nullptr);
   size_t DecodeH264Fua(uint8_t *payload = nullptr);
 
  public:
   // Get Header
-  uint32_t Verion() const { return version_; }
-  bool HasPadding() const { return has_padding_; }
-  bool HasExtension() const { return has_extension_; }
-  bool Marker() const { return marker_; }
-  PAYLOAD_TYPE PayloadType() const { return PAYLOAD_TYPE(payload_type_); }
-  uint16_t SequenceNumber() const { return sequence_number_; }
-  uint32_t Timestamp() const { return timestamp_; }
-  uint32_t Ssrc() const { return ssrc_; }
-  std::vector<uint32_t> Csrcs() const { return csrcs_; };
-  uint16_t ExtensionProfile() const { return extension_profile_; }
-  const uint8_t *ExtensionData() const { return extension_data_; }
+  uint32_t Verion() {
+    ParseRtpData();
+    return version_;
+  }
+  bool HasPadding() {
+    ParseRtpData();
+    return has_padding_;
+  }
+  bool HasExtension() {
+    ParseRtpData();
+    return has_extension_;
+  }
+  bool Marker() {
+    ParseRtpData();
+    return marker_;
+  }
+  PAYLOAD_TYPE PayloadType() {
+    ParseRtpData();
+    return PAYLOAD_TYPE(payload_type_);
+  }
+  uint16_t SequenceNumber() {
+    ParseRtpData();
+    return sequence_number_;
+  }
+  uint32_t Timestamp() {
+    ParseRtpData();
+    return timestamp_;
+  }
+  uint32_t Ssrc() {
+    ParseRtpData();
+    return ssrc_;
+  }
+  std::vector<uint32_t> Csrcs() {
+    ParseRtpData();
+    return csrcs_;
+  };
+  uint16_t ExtensionProfile() {
+    ParseRtpData();
+    return extension_profile_;
+  }
+  const uint8_t *ExtensionData() {
+    ParseRtpData();
+    return extension_data_;
+  }
 
   // Payload
-  const uint8_t *Payload() const { return payload_; };
-  size_t PayloadSize() const { return payload_size_; }
+  const uint8_t *Payload() {
+    ParseRtpData();
+    return payload_;
+  };
+  size_t PayloadSize() {
+    ParseRtpData();
+    return payload_size_;
+  }
 
   // Entire RTP buffer
   const uint8_t *Buffer() const { return buffer_; }
   size_t Size() const { return size_; }
 
   // NAL
-  NAL_UNIT_TYPE NalUnitType() const { return nal_unit_type_; }
-  bool FuAStart() const { return fu_header_.start; }
-  bool FuAEnd() const { return fu_header_.end; }
+  NAL_UNIT_TYPE NalUnitType() {
+    ParseRtpData();
+    return nal_unit_type_;
+  }
+  bool FuAStart() {
+    ParseRtpData();
+    return fu_header_.start;
+  }
+  bool FuAEnd() {
+    ParseRtpData();
+    return fu_header_.end;
+  }
 
  private:
-  inline void TryToDecodeH264RtpPacket(uint8_t *buffer);
+  void TryToDecodeRtpPacket();
+  void ParseRtpData();
 
  private:
   // Header
@@ -198,6 +247,8 @@ class RtpPacket {
 
   // NAL
   NAL_UNIT_TYPE nal_unit_type_ = NAL_UNIT_TYPE::UNKNOWN;
+
+  bool parsed_ = false;
 };
 
 #endif

src/rtp/rtp_statistics.cpp

@@ -16,11 +16,11 @@ void RtpStatistics::UpdateReceiveBytes(uint32_t received_bytes) {
 
 bool RtpStatistics::Process() {
   if (!sent_bytes_) {
-    LOG_INFO("rtp statistics: Send [{} bps]", sent_bytes_);
+    // LOG_INFO("rtp statistics: Send [{} bps]", sent_bytes_);
   }
 
   if (!received_bytes_) {
-    LOG_INFO("rtp statistics: Receive [{} bps]", received_bytes_);
+    // LOG_INFO("rtp statistics: Receive [{} bps]", received_bytes_);
   }
 
   sent_bytes_ = 0;

src/rtp/rtp_video_receiver.cpp

@@ -166,7 +166,7 @@ int RtpVideoReceiver::SendRtcpRR(RtcpReceiverReport& rtcp_rr) {
     return -1;
   }
 
-  LOG_ERROR("Send RR");
+  // LOG_ERROR("Send RR");
 
   return 0;
 }

src/rtp/rtp_video_sender.cpp

@@ -102,7 +102,7 @@ int RtpVideoSender::SendRtcpSR(RtcpSenderReport& rtcp_sr) {
     return -1;
   }
 
-  LOG_ERROR("Send SR");
+  // LOG_ERROR("Send SR");
 
   return 0;
 }

src/transmission/ice_transmission.cpp

@@ -20,15 +20,12 @@ IceTransmission::IceTransmission(
     bool offer_peer, std::string &transmission_id, std::string &user_id,
     std::string &remote_user_id,
     std::shared_ptr<WsTransmission> ice_ws_transmission,
-    std::function<void(const char *, size_t, const char *, size_t)>
-        on_receive_ice_msg,
     std::function<void(std::string)> on_ice_status_change)
     : offer_peer_(offer_peer),
       transmission_id_(transmission_id),
       user_id_(user_id),
       remote_user_id_(remote_user_id),
       ice_ws_transport_(ice_ws_transmission),
-      on_receive_ice_msg_cb_(on_receive_ice_msg),
       on_ice_status_change_(on_ice_status_change) {}
 
 IceTransmission::~IceTransmission() {
@@ -67,9 +64,9 @@ int IceTransmission::InitIceTransmission(std::string &ip, int port) {
   rtp_video_receiver_->SetOnReceiveCompleteFrame(
       [this](VideoFrame &video_frame) -> void {
         // LOG_ERROR("OnReceiveCompleteFrame {}", video_frame.Size());
-        on_receive_ice_msg_cb_((const char *)video_frame.Buffer(),
-                               video_frame.Size(), remote_user_id_.data(),
-                               remote_user_id_.size());
+        on_receive_video_((const char *)video_frame.Buffer(),
+                          video_frame.Size(), remote_user_id_.data(),
+                          remote_user_id_.size());
       });
 
   rtp_video_receiver_->Start();
@@ -100,6 +97,22 @@ int IceTransmission::InitIceTransmission(std::string &ip, int port) {
 
   rtp_data_sender_->Start();
 
+  rtp_data_receiver_ = std::make_unique<RtpDataReceiver>();
+  rtp_data_receiver_->SetSendDataFunc(
+      [this](const char *data, size_t size) -> int {
+        if (!ice_agent_) {
+          LOG_ERROR("ice_agent_ is nullptr");
+          return -1;
+        }
+
+        return ice_agent_->Send(data, size);
+      });
+  rtp_data_receiver_->SetOnReceiveData(
+      [this](const char *data, size_t size) -> void {
+        on_receive_data_(data, size, remote_user_id_.data(),
+                         remote_user_id_.size());
+      });
+
   ice_agent_ = std::make_unique<IceAgent>(ip, port);
 
   ice_agent_->CreateIceAgent(
@@ -148,8 +161,11 @@ int IceTransmission::InitIceTransmission(std::string &ip, int port) {
               RtpPacket packet((uint8_t *)buffer, size);
               ice_transmission_obj->rtp_video_receiver_->InsertRtpPacket(
                   packet);
+            } else if (ice_transmission_obj->CheckIsDataPacket(buffer, size)) {
+              RtpPacket packet((uint8_t *)buffer, size);
+              ice_transmission_obj->rtp_data_receiver_->InsertRtpPacket(packet);
             } else if (ice_transmission_obj->CheckIsRtcpPacket(buffer, size)) {
-              LOG_ERROR("Rtcp packet [{}]", (uint8_t)(buffer[1]));
+              // LOG_ERROR("Rtcp packet [{}]", (uint8_t)(buffer[1]));
             }
           }
         }
@@ -317,4 +333,17 @@ uint8_t IceTransmission::CheckIsAudioPacket(const char *buffer, size_t size) {
   } else {
     return 0;
   }
-}
+}
+
+uint8_t IceTransmission::CheckIsDataPacket(const char *buffer, size_t size) {
+  if (size < 4) {
+    return 0;
+  }
+
+  uint8_t pt = buffer[1] & 0x7F;
+  if (RtpPacket::PAYLOAD_TYPE::DATA == pt) {
+    return pt;
+  } else {
+    return 0;
+  }
+}

src/transmission/ice_transmission.h

@@ -19,13 +19,10 @@ class IceTransmission {
   typedef enum { VIDEO = 96, AUDIO = 97, DATA = 127 } DATA_TYPE;
 
  public:
-  IceTransmission(
-      bool offer_peer, std::string &transmission_id, std::string &user_id,
-      std::string &remote_user_id,
-      std::shared_ptr<WsTransmission> ice_ws_transmission,
-      std::function<void(const char *, size_t, const char *, size_t)>
-          on_receive_ice_msg,
-      std::function<void(std::string)> on_ice_status_change);
+  IceTransmission(bool offer_peer, std::string &transmission_id,
+                  std::string &user_id, std::string &remote_user_id,
+                  std::shared_ptr<WsTransmission> ice_ws_transmission,
+                  std::function<void(std::string)> on_ice_status_change);
   ~IceTransmission();
 
  public:
@@ -33,6 +30,24 @@ class IceTransmission {
 
   int DestroyIceTransmission();
 
+  void SetOnReceiveVideoFunc(
+      std::function<void(const char *, size_t, const char *, size_t)>
+          on_receive_video) {
+    on_receive_video_ = on_receive_video;
+  }
+
+  void SetOnReceiveAudioFunc(
+      std::function<void(const char *, size_t, const char *, size_t)>
+          on_receive_audio) {
+    on_receive_audio_ = on_receive_audio;
+  }
+
+  void SetOnReceiveDataFunc(
+      std::function<void(const char *, size_t, const char *, size_t)>
+          on_receive_data) {
+    on_receive_data_ = on_receive_data;
+  }
+
   int JoinTransmission();
 
   int SetTransmissionId(const std::string &transmission_id);
@@ -60,13 +75,18 @@ class IceTransmission {
   uint8_t CheckIsRtcpPacket(const char *buffer, size_t size);
   uint8_t CheckIsVideoPacket(const char *buffer, size_t size);
   uint8_t CheckIsAudioPacket(const char *buffer, size_t size);
+  uint8_t CheckIsDataPacket(const char *buffer, size_t size);
 
  private:
   std::unique_ptr<IceAgent> ice_agent_ = nullptr;
   std::shared_ptr<WsTransmission> ice_ws_transport_ = nullptr;
   CongestionControl *congestion_control_ = nullptr;
   std::function<void(const char *, size_t, const char *, size_t)>
-      on_receive_ice_msg_cb_ = nullptr;
+      on_receive_video_ = nullptr;
+  std::function<void(const char *, size_t, const char *, size_t)>
+      on_receive_audio_ = nullptr;
+  std::function<void(const char *, size_t, const char *, size_t)>
+      on_receive_data_ = nullptr;
   std::function<void(std::string)> on_ice_status_change_ = nullptr;
   std::string local_sdp_;
   std::string remote_sdp_;