Opus codec module test pass

src/rtp/rtp_audio_receiver.cpp

@@ -0,0 +1,90 @@
+#include "rtp_audio_receiver.h"
+
+#define RTCP_RR_INTERVAL 1000
+
+RtpAudioReceiver::RtpAudioReceiver() {}
+
+RtpAudioReceiver::~RtpAudioReceiver() {
+  if (rtp_statistics_) {
+    rtp_statistics_->Stop();
+  }
+}
+
+void RtpAudioReceiver::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 RtpAudioReceiver::SetSendDataFunc(
+    std::function<int(const char*, size_t)> data_send_func) {
+  data_send_func_ = data_send_func;
+}
+
+int RtpAudioReceiver::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 RtpAudioReceiver::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_audio_receiver.h

@@ -0,0 +1,45 @@
+/*
+ * @Author: DI JUNKUN
+ * @Date: 2023-11-24
+ * Copyright (c) 2023 by DI JUNKUN, All Rights Reserved.
+ */
+
+#ifndef _RTP_AUDIO_RECEIVER_H_
+#define _RTP_AUDIO_RECEIVER_H_
+
+#include <functional>
+
+#include "rtcp_receiver_report.h"
+#include "rtp_codec.h"
+#include "rtp_statistics.h"
+
+class RtpAudioReceiver {
+ public:
+  RtpAudioReceiver();
+  ~RtpAudioReceiver();
+
+ 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:
+  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_audio_sender.cpp

@@ -0,0 +1,140 @@
+#include "rtp_audio_sender.h"
+
+#include <chrono>
+
+#include "log.h"
+
+#define RTCP_SR_INTERVAL 1000
+
+RtpAudioSender::RtpAudioSender() {}
+
+RtpAudioSender::~RtpAudioSender() {
+  if (rtp_statistics_) {
+    rtp_statistics_->Stop();
+  }
+}
+
+void RtpAudioSender::Enqueue(std::vector<RtpPacket>& rtp_packets) {
+  if (!rtp_statistics_) {
+    rtp_statistics_ = std::make_unique<RtpStatistics>();
+    rtp_statistics_->Start();
+  }
+
+  for (auto& rtp_packet : rtp_packets) {
+    rtp_packe_queue_.push(rtp_packet);
+  }
+}
+
+void RtpAudioSender::SetSendDataFunc(
+    std::function<int(const char*, size_t)> data_send_func) {
+  data_send_func_ = data_send_func;
+}
+
+int RtpAudioSender::SendRtpPacket(RtpPacket& rtp_packet) {
+  if (!data_send_func_) {
+    LOG_ERROR("data_send_func_ is nullptr");
+    return -1;
+  }
+
+  int ret = 0;
+
+  if (0 !=
+      data_send_func_((const char*)rtp_packet.Buffer(), rtp_packet.Size())) {
+    LOG_ERROR("Send rtp packet failed");
+    return -1;
+  }
+
+  last_send_bytes_ += rtp_packet.Size();
+  total_rtp_packets_sent_++;
+  total_rtp_payload_sent_ += rtp_packet.PayloadSize();
+
+  if (CheckIsTimeSendSR()) {
+    RtcpSenderReport rtcp_sr;
+    SenderInfo sender_info;
+    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());
+
+    sender_info.sender_ssrc = 0x00;
+    sender_info.ntp_ts_msw = (uint32_t)seconds_u32;
+    sender_info.ntp_ts_lsw = (uint32_t)fraction_u32;
+    sender_info.rtp_ts =
+        std::chrono::high_resolution_clock::now().time_since_epoch().count() *
+        1000000;
+    sender_info.sender_packet_count = total_rtp_packets_sent_;
+    sender_info.sender_octet_count = total_rtp_payload_sent_;
+
+    rtcp_sr.SetSenderInfo(sender_info);
+
+    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_sr.SetReportBlock(report);
+
+    rtcp_sr.Encode();
+
+    SendRtcpSR(rtcp_sr);
+  }
+
+  return 0;
+}
+
+int RtpAudioSender::SendRtcpSR(RtcpSenderReport& rtcp_sr) {
+  if (!data_send_func_) {
+    LOG_ERROR("data_send_func_ is nullptr");
+    return -1;
+  }
+
+  if (data_send_func_((const char*)rtcp_sr.Buffer(), rtcp_sr.Size())) {
+    LOG_ERROR("Send SR failed");
+    return -1;
+  }
+
+  // LOG_ERROR("Send SR");
+
+  return 0;
+}
+
+bool RtpAudioSender::CheckIsTimeSendSR() {
+  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_sr_packet_ts_ >= RTCP_SR_INTERVAL) {
+    last_send_rtcp_sr_packet_ts_ = now_ts;
+    return true;
+  } else {
+    return false;
+  }
+}
+
+bool RtpAudioSender::Process() {
+  last_send_bytes_ = 0;
+
+  for (size_t i = 0; i < 10; i++)
+    if (!rtp_packe_queue_.isEmpty()) {
+      RtpPacket rtp_packet;
+      rtp_packe_queue_.pop(rtp_packet);
+      SendRtpPacket(rtp_packet);
+    }
+
+  if (rtp_statistics_) {
+    rtp_statistics_->UpdateSentBytes(last_send_bytes_);
+  }
+
+  std::this_thread::sleep_for(std::chrono::milliseconds(5));
+  return true;
+}

src/rtp/rtp_audio_sender.h

@@ -0,0 +1,47 @@
+/*
+ * @Author: DI JUNKUN
+ * @Date: 2023-11-24
+ * Copyright (c) 2023 by DI JUNKUN, All Rights Reserved.
+ */
+
+#ifndef _RTP_AUDIO_SENDER_H_
+#define _RTP_AUDIO_SENDER_H_
+
+#include <functional>
+
+#include "ringbuffer.h"
+#include "rtcp_sender_report.h"
+#include "rtp_packet.h"
+#include "rtp_statistics.h"
+#include "thread_base.h"
+
+class RtpAudioSender : public ThreadBase {
+ public:
+  RtpAudioSender();
+  ~RtpAudioSender();
+
+ public:
+  void Enqueue(std::vector<RtpPacket> &rtp_packets);
+  void SetSendDataFunc(std::function<int(const char *, size_t)> data_send_func);
+
+ private:
+ private:
+  int SendRtpPacket(RtpPacket &rtp_packet);
+  int SendRtcpSR(RtcpSenderReport &rtcp_sr);
+
+  bool CheckIsTimeSendSR();
+
+ private:
+  bool Process() override;
+
+ private:
+  std::function<int(const char *, size_t)> data_send_func_ = nullptr;
+  RingBuffer<RtpPacket> rtp_packe_queue_;
+  std::unique_ptr<RtpStatistics> rtp_statistics_ = nullptr;
+  uint32_t last_send_bytes_ = 0;
+  uint32_t last_send_rtcp_sr_packet_ts_ = 0;
+  uint32_t total_rtp_packets_sent_ = 0;
+  uint32_t total_rtp_payload_sent_ = 0;
+};
+
+#endif

src/rtp/rtp_codec.cpp

@@ -216,6 +216,22 @@ void RtpCodec::Encode(uint8_t* buffer, size_t size,
         packets.emplace_back(rtp_packet);
       }
     }
+  } else if (RtpPacket::PAYLOAD_TYPE::OPUS == payload_type_) {
+    RtpPacket rtp_packet;
+    rtp_packet.SetVerion(version_);
+    rtp_packet.SetHasPadding(has_padding_);
+    rtp_packet.SetHasExtension(has_extension_);
+    rtp_packet.SetMarker(1);
+    rtp_packet.SetPayloadType(RtpPacket::PAYLOAD_TYPE(payload_type_));
+    rtp_packet.SetSequenceNumber(sequence_number_++);
+
+    timestamp_ =
+        std::chrono::high_resolution_clock::now().time_since_epoch().count();
+    rtp_packet.SetTimestamp(timestamp_);
+    rtp_packet.SetSsrc(ssrc_);
+
+    rtp_packet.Encode(buffer, size);
+    packets.emplace_back(rtp_packet);
   } else if (RtpPacket::PAYLOAD_TYPE::DATA == payload_type_) {
     RtpPacket rtp_packet;
     rtp_packet.SetVerion(version_);

src/rtp/rtp_data_sender.h

@@ -1,3 +1,9 @@
+/*
+ * @Author: DI JUNKUN
+ * @Date: 2023-11-24
+ * Copyright (c) 2023 by DI JUNKUN, All Rights Reserved.
+ */
+
 #ifndef _RTP_DATA_SENDER_H_
 #define _RTP_DATA_SENDER_H_
 

src/rtp/rtp_packet.cpp

@@ -17,6 +17,8 @@ void RtpPacket::TryToDecodeRtpPacket() {
     DecodeH264FecSource();
   } else if (PAYLOAD_TYPE::H264_FEC_REPAIR == PAYLOAD_TYPE(buffer_[1] & 0x7F)) {
     DecodeH264FecRepair();
+  } else if (PAYLOAD_TYPE::OPUS == PAYLOAD_TYPE(buffer_[1] & 0x7F)) {
+    DecodeOpus();
   } else if (PAYLOAD_TYPE::DATA == PAYLOAD_TYPE(buffer_[1] & 0x7F)) {
     DecodeData();
   } else {
@@ -368,6 +370,50 @@ const uint8_t *RtpPacket::EncodeH264FecRepair(
   return buffer_;
 }
 
+size_t RtpPacket::DecodeOpus(uint8_t *payload) {
+  version_ = (buffer_[0] >> 6) & 0x03;
+  has_padding_ = (buffer_[0] >> 5) & 0x01;
+  has_extension_ = (buffer_[0] >> 4) & 0x01;
+  total_csrc_number_ = buffer_[0] & 0x0f;
+  marker_ = (buffer_[1] >> 7) & 0x01;
+  payload_type_ = buffer_[1] & 0x7f;
+  sequence_number_ = (buffer_[2] << 8) | buffer_[3];
+  timestamp_ =
+      (buffer_[4] << 24) | (buffer_[5] << 16) | (buffer_[6] << 8) | buffer_[7];
+  ssrc_ = (buffer_[8] << 24) | (buffer_[9] << 16) | (buffer_[10] << 8) |
+          buffer_[11];
+
+  for (uint32_t index = 0; index < total_csrc_number_; index++) {
+    uint32_t csrc = (buffer_[12 + index] << 24) | (buffer_[13 + index] << 16) |
+                    (buffer_[14 + index] << 8) | buffer_[15 + index];
+    csrcs_.push_back(csrc);
+  }
+
+  uint32_t extension_offset = total_csrc_number_ * 4;
+  if (has_extension_) {
+    extension_profile_ =
+        (buffer_[12 + extension_offset] << 8) | buffer_[13 + extension_offset];
+    extension_len_ =
+        (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_);
+    extension_data_ = buffer_ + 16 + extension_offset;
+  }
+
+  uint32_t payload_offset =
+      (has_extension_ ? extension_len_ : 0) + extension_offset;
+
+  payload_size_ = size_ - (12 + payload_offset);
+  payload_ = buffer_ + 12 + payload_offset;
+  if (payload) {
+    memcpy(payload, payload_, payload_size_);
+  }
+
+  return payload_size_;
+}
+
 size_t RtpPacket::DecodeData(uint8_t *payload) {
   version_ = (buffer_[0] >> 6) & 0x03;
   has_padding_ = (buffer_[0] >> 5) & 0x01;

src/rtp/rtp_packet.h

@@ -209,6 +209,7 @@ class RtpPacket {
   size_t DecodeH264Fua(uint8_t *payload = nullptr);
   size_t DecodeH264FecSource(uint8_t *payload = nullptr);
   size_t DecodeH264FecRepair(uint8_t *payload = nullptr);
+  size_t DecodeOpus(uint8_t *payload = nullptr);
 
  public:
   // Get Header

src/rtp/rtp_video_receiver.h

@@ -7,12 +7,12 @@
 #include <set>
 
 #include "fec_decoder.h"
-#include "frame.h"
 #include "ringbuffer.h"
 #include "rtcp_receiver_report.h"
 #include "rtp_codec.h"
 #include "rtp_statistics.h"
 #include "thread_base.h"
+#include "video_frame.h"
 
 class RtpVideoReceiver : public ThreadBase {
  public: