[feat] update transport module and channel module

This commit is contained in:
dijunkun
2025-01-06 17:10:56 +08:00
parent eef35ff0d4
commit 0737eee95a
17 changed files with 927 additions and 605 deletions

View File

@@ -1 +1,59 @@
#include "audio_channel_receive.h"
#include "log.h"
AudioChannelReceive::AudioChannelReceive() {}
AudioChannelReceive::AudioChannelReceive(
std::shared_ptr<IceAgent> ice_agent,
std::shared_ptr<IOStatistics> ice_io_statistics,
std::function<void(const char *, size_t)> on_receive_audio)
: ice_agent_(ice_agent),
ice_io_statistics_(ice_io_statistics),
on_receive_audio_(on_receive_audio) {}
AudioChannelReceive::~AudioChannelReceive() {}
void AudioChannelReceive::Initialize(RtpPacket::PAYLOAD_TYPE payload_type) {
audio_rtp_codec_ = std::make_unique<RtpCodec>(payload_type);
rtp_audio_receiver_ = std::make_unique<RtpAudioReceiver>(ice_io_statistics_);
rtp_audio_receiver_->SetOnReceiveData(
[this](const char *data, size_t size) -> void {
ice_io_statistics_->UpdateAudioInboundBytes((uint32_t)size);
if (on_receive_audio_) {
on_receive_audio_(data, size);
}
});
rtp_audio_receiver_->SetSendDataFunc(
[this](const char *data, size_t size) -> int {
if (!ice_agent_) {
LOG_ERROR("ice_agent_ is nullptr");
return -1;
}
auto ice_state = ice_agent_->GetIceState();
if (ice_state != NICE_COMPONENT_STATE_CONNECTED &&
ice_state != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
nice_component_state_to_string(ice_state));
return -2;
}
ice_io_statistics_->UpdateAudioOutboundBytes((uint32_t)size);
return ice_agent_->Send(data, size);
});
}
void AudioChannelReceive::Destroy() {}
int AudioChannelReceive::OnReceiveRtpPacket(const char *data, size_t size) {
if (rtp_audio_receiver_) {
rtp_audio_receiver_->InsertRtpPacket(RtpPacket((uint8_t *)data, size));
}
return 0;
}

View File

@@ -7,12 +7,30 @@
#ifndef _AUDIO_CHANNEL_RECEIVE_H_
#define _AUDIO_CHANNEL_RECEIVE_H_
#include "ice_agent.h"
#include "rtp_audio_receiver.h"
#include "rtp_codec.h"
class AudioChannelReceive {
public:
AudioChannelReceive();
AudioChannelReceive(
std::shared_ptr<IceAgent> ice_agent,
std::shared_ptr<IOStatistics> ice_io_statistics,
std::function<void(const char *, size_t)> on_receive_audio);
~AudioChannelReceive();
public:
void Initialize(RtpPacket::PAYLOAD_TYPE payload_type);
void Destroy();
int OnReceiveRtpPacket(const char *data, size_t size);
private:
std::shared_ptr<IceAgent> ice_agent_ = nullptr;
std::shared_ptr<IOStatistics> ice_io_statistics_ = nullptr;
std::unique_ptr<RtpCodec> audio_rtp_codec_ = nullptr;
std::unique_ptr<RtpAudioReceiver> rtp_audio_receiver_ = nullptr;
std::function<void(const char *, size_t)> on_receive_audio_ = nullptr;
};
#endif

View File

@@ -1 +1,55 @@
#include "audio_channel_send.h"
#include "log.h"
AudioChannelSend::AudioChannelSend() {}
AudioChannelSend::~AudioChannelSend() {}
AudioChannelSend::AudioChannelSend(
std::shared_ptr<IceAgent> ice_agent,
std::shared_ptr<IOStatistics> ice_io_statistics)
: ice_agent_(ice_agent), ice_io_statistics_(ice_io_statistics) {}
void AudioChannelSend::Initialize(RtpPacket::PAYLOAD_TYPE payload_type) {
audio_rtp_codec_ = std::make_unique<RtpCodec>(payload_type);
rtp_audio_sender_ = std::make_unique<RtpAudioSender>(ice_io_statistics_);
rtp_audio_sender_->SetSendDataFunc(
[this](const char *data, size_t size) -> int {
if (!ice_agent_) {
LOG_ERROR("ice_agent_ is nullptr");
return -1;
}
auto ice_state = ice_agent_->GetIceState();
if (ice_state != NICE_COMPONENT_STATE_CONNECTED &&
ice_state != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
nice_component_state_to_string(ice_state));
return -2;
}
ice_io_statistics_->UpdateAudioOutboundBytes((uint32_t)size);
return ice_agent_->Send(data, size);
});
rtp_audio_sender_->Start();
}
void AudioChannelSend::Destroy() {
if (rtp_audio_sender_) {
rtp_audio_sender_->Stop();
}
}
int AudioChannelSend::SendAudio(char *data, size_t size) {
if (audio_rtp_codec_) {
std::vector<RtpPacket> packets;
audio_rtp_codec_->Encode((uint8_t *)data, (uint32_t)size, packets);
rtp_audio_sender_->Enqueue(packets);
}
return 0;
}

View File

@@ -7,12 +7,27 @@
#ifndef _AUDIO_CHANNEL_SEND_H_
#define _AUDIO_CHANNEL_SEND_H_
#include "ice_agent.h"
#include "rtp_audio_sender.h"
#include "rtp_codec.h"
class AudioChannelSend {
public:
AudioChannelSend();
AudioChannelSend(std::shared_ptr<IceAgent> ice_agent,
std::shared_ptr<IOStatistics> ice_io_statistics);
~AudioChannelSend();
public:
void Initialize(RtpPacket::PAYLOAD_TYPE payload_type);
void Destroy();
int SendAudio(char *data, size_t size);
private:
std::shared_ptr<IceAgent> ice_agent_ = nullptr;
std::shared_ptr<IOStatistics> ice_io_statistics_ = nullptr;
std::unique_ptr<RtpCodec> audio_rtp_codec_ = nullptr;
std::unique_ptr<RtpAudioSender> rtp_audio_sender_ = nullptr;
};
#endif

View File

@@ -1 +1,60 @@
#include "data_channel_receive.h"
#include "log.h"
DataChannelReceive::DataChannelReceive() {}
DataChannelReceive::DataChannelReceive(
std::shared_ptr<IceAgent> ice_agent,
std::shared_ptr<IOStatistics> ice_io_statistics,
std::function<void(const char *, size_t)> on_receive_data)
: ice_agent_(ice_agent),
ice_io_statistics_(ice_io_statistics),
on_receive_data_(on_receive_data) {}
DataChannelReceive::~DataChannelReceive() {}
void DataChannelReceive::Initialize(RtpPacket::PAYLOAD_TYPE payload_type) {
data_rtp_codec_ = std::make_unique<RtpCodec>(payload_type);
rtp_data_receiver_ = std::make_unique<RtpDataReceiver>(ice_io_statistics_);
rtp_data_receiver_->SetOnReceiveData(
[this](const char *data, size_t size) -> void {
ice_io_statistics_->UpdateDataInboundBytes((uint32_t)size);
if (on_receive_data_) {
on_receive_data_(data, size);
}
});
rtp_data_receiver_->SetSendDataFunc(
[this](const char *data, size_t size) -> int {
if (!ice_agent_) {
LOG_ERROR("ice_agent_ is nullptr");
return -1;
}
auto ice_state = ice_agent_->GetIceState();
if (ice_state != NICE_COMPONENT_STATE_CONNECTED &&
ice_state != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
nice_component_state_to_string(ice_state));
return -2;
}
ice_io_statistics_->UpdateDataOutboundBytes((uint32_t)size);
return ice_agent_->Send(data, size);
});
}
void DataChannelReceive::Destroy() {}
int DataChannelReceive::OnReceiveRtpPacket(const char *data, size_t size) {
if (rtp_data_receiver_) {
rtp_data_receiver_->InsertRtpPacket(RtpPacket((uint8_t *)data, size));
}
return -1;
}

View File

@@ -7,12 +7,29 @@
#ifndef _DATA_CHANNEL_RECEIVE_H_
#define _DATA_CHANNEL_RECEIVE_H_
#include "ice_agent.h"
#include "rtp_codec.h"
#include "rtp_data_receiver.h"
class DataChannelReceive {
public:
DataChannelReceive();
DataChannelReceive(std::shared_ptr<IceAgent> ice_agent,
std::shared_ptr<IOStatistics> ice_io_statistics,
std::function<void(const char *, size_t)> on_receive_data);
~DataChannelReceive();
public:
void Initialize(RtpPacket::PAYLOAD_TYPE payload_type);
void Destroy();
int OnReceiveRtpPacket(const char *data, size_t size);
private:
std::shared_ptr<IceAgent> ice_agent_ = nullptr;
std::shared_ptr<IOStatistics> ice_io_statistics_ = nullptr;
std::unique_ptr<RtpCodec> data_rtp_codec_ = nullptr;
std::unique_ptr<RtpDataReceiver> rtp_data_receiver_ = nullptr;
std::function<void(const char *, size_t)> on_receive_data_ = nullptr;
};
#endif

View File

@@ -1 +1,58 @@
#include "data_channel_send.h"
#include "log.h"
DataChannelSend::DataChannelSend() {}
DataChannelSend::~DataChannelSend() {}
DataChannelSend::DataChannelSend(
std::shared_ptr<IceAgent> ice_agent,
std::shared_ptr<IOStatistics> ice_io_statistics)
: ice_agent_(ice_agent), ice_io_statistics_(ice_io_statistics) {}
void DataChannelSend::Initialize(RtpPacket::PAYLOAD_TYPE payload_type) {
data_rtp_codec_ = std::make_unique<RtpCodec>(payload_type);
rtp_data_sender_ = std::make_unique<RtpDataSender>(ice_io_statistics_);
rtp_data_sender_->SetSendDataFunc(
[this](const char *data, size_t size) -> int {
if (!ice_agent_) {
LOG_ERROR("ice_agent_ is nullptr");
return -1;
}
auto ice_state = ice_agent_->GetIceState();
if (ice_state != NICE_COMPONENT_STATE_CONNECTED &&
ice_state != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
nice_component_state_to_string(ice_state));
return -2;
}
ice_io_statistics_->UpdateDataOutboundBytes((uint32_t)size);
return ice_agent_->Send(data, size);
});
rtp_data_sender_->Start();
}
void DataChannelSend::Destroy() {
if (rtp_data_sender_) {
rtp_data_sender_->Stop();
}
}
int DataChannelSend::SendData(const char *data, size_t size) {
std::vector<RtpPacket> packets;
if (rtp_data_sender_) {
if (data_rtp_codec_) {
data_rtp_codec_->Encode((uint8_t *)data, (uint32_t)size, packets);
rtp_data_sender_->Enqueue(packets);
}
}
return 0;
}

View File

@@ -7,12 +7,27 @@
#ifndef _DATA_CHANNEL_SEND_H_
#define _DATA_CHANNEL_SEND_H_
#include "ice_agent.h"
#include "rtp_codec.h"
#include "rtp_data_sender.h"
class DataChannelSend {
public:
DataChannelSend();
DataChannelSend(std::shared_ptr<IceAgent> ice_agent,
std::shared_ptr<IOStatistics> ice_io_statistics);
~DataChannelSend();
public:
void Initialize(RtpPacket::PAYLOAD_TYPE payload_type);
void Destroy();
int SendData(const char *data, size_t size);
private:
std::shared_ptr<IceAgent> ice_agent_ = nullptr;
std::shared_ptr<IOStatistics> ice_io_statistics_ = nullptr;
std::unique_ptr<RtpCodec> data_rtp_codec_ = nullptr;
std::unique_ptr<RtpDataSender> rtp_data_sender_ = nullptr;
};
#endif

View File

@@ -1 +1,59 @@
#include "video_channel_receive.h"
#include "log.h"
VideoChannelReceive::VideoChannelReceive() {}
VideoChannelReceive::VideoChannelReceive(
std::shared_ptr<IceAgent> ice_agent,
std::shared_ptr<IOStatistics> ice_io_statistics,
std::function<void(VideoFrame &)> on_receive_complete_frame)
: ice_agent_(ice_agent),
ice_io_statistics_(ice_io_statistics),
on_receive_complete_frame_(on_receive_complete_frame) {}
VideoChannelReceive::~VideoChannelReceive() {}
void VideoChannelReceive::Initialize(RtpPacket::PAYLOAD_TYPE payload_type) {
video_rtp_codec_ = std::make_unique<RtpCodec>(payload_type);
rtp_video_receiver_ = std::make_unique<RtpVideoReceiver>(ice_io_statistics_);
rtp_video_receiver_->SetOnReceiveCompleteFrame(
[this](VideoFrame &video_frame) -> void {
ice_io_statistics_->UpdateVideoInboundBytes(
(uint32_t)video_frame.Size());
on_receive_complete_frame_(video_frame);
});
rtp_video_receiver_->SetSendDataFunc(
[this](const char *data, size_t size) -> int {
if (!ice_agent_) {
LOG_ERROR("ice_agent_ is nullptr");
return -1;
}
auto ice_state = ice_agent_->GetIceState();
if (ice_state != NICE_COMPONENT_STATE_CONNECTED &&
ice_state != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
nice_component_state_to_string(ice_state));
return -2;
}
ice_io_statistics_->UpdateVideoInboundBytes((uint32_t)size);
return ice_agent_->Send(data, size);
});
rtp_video_receiver_->Start();
}
void VideoChannelReceive::Destroy() {}
int VideoChannelReceive::OnReceiveRtpPacket(const char *data, size_t size) {
if (rtp_video_receiver_) {
rtp_video_receiver_->InsertRtpPacket(RtpPacket((uint8_t *)data, size));
}
return 0;
}

View File

@@ -7,12 +7,32 @@
#ifndef _VIDEO_CHANNEL_RECEIVE_H_
#define _VIDEO_CHANNEL_RECEIVE_H_
#include "ice_agent.h"
#include "rtp_codec.h"
#include "rtp_video_receiver.h"
class VideoChannelReceive {
public:
VideoChannelReceive();
VideoChannelReceive(
std::shared_ptr<IceAgent> ice_agent,
std::shared_ptr<IOStatistics> ice_io_statistics,
std::function<void(VideoFrame &)> on_receive_complete_frame);
~VideoChannelReceive();
public:
void Initialize(RtpPacket::PAYLOAD_TYPE payload_type);
void Destroy();
int OnReceiveRtpPacket(const char *data, size_t size);
private:
std::shared_ptr<IceAgent> ice_agent_ = nullptr;
std::shared_ptr<IOStatistics> ice_io_statistics_ = nullptr;
std::unique_ptr<RtpCodec> video_rtp_codec_ = nullptr;
std::unique_ptr<RtpVideoReceiver> rtp_video_receiver_ = nullptr;
std::function<void(VideoFrame &)> on_receive_complete_frame_ = nullptr;
};
#endif

View File

@@ -1,11 +1,18 @@
#include "video_channel_send.h"
#include "log.h"
VideoChannelSend::VideoChannelSend() {}
VideoChannelSend::~VideoChannelSend() {}
void VideoChannelSend::Initialize(RtpPacket::PAYLOAD_TYPE negotiated_video_pt) {
video_rtp_codec_ = std::make_unique<RtpCodec>(negotiated_video_pt);
VideoChannelSend::VideoChannelSend(
std::shared_ptr<IceAgent> ice_agent,
std::shared_ptr<IOStatistics> ice_io_statistics)
: ice_agent_(ice_agent), ice_io_statistics_(ice_io_statistics){};
void VideoChannelSend::Initialize(RtpPacket::PAYLOAD_TYPE payload_type) {
video_rtp_codec_ = std::make_unique<RtpCodec>(payload_type);
rtp_video_sender_ = std::make_unique<RtpVideoSender>(ice_io_statistics_);
rtp_video_sender_->SetSendDataFunc(
@@ -15,10 +22,12 @@ void VideoChannelSend::Initialize(RtpPacket::PAYLOAD_TYPE negotiated_video_pt) {
return -1;
}
if (state_ != NICE_COMPONENT_STATE_CONNECTED &&
state_ != NICE_COMPONENT_STATE_READY) {
auto ice_state = ice_agent_->GetIceState();
if (ice_state != NICE_COMPONENT_STATE_CONNECTED &&
ice_state != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
nice_component_state_to_string(state_));
nice_component_state_to_string(ice_state));
return -2;
}
@@ -35,13 +44,11 @@ void VideoChannelSend::Destroy() {
}
}
int VideoChannelSend::SendVideo(char *encoded_frame, size_t size) {
int VideoChannelSend::SendVideo(char *data, size_t size) {
std::vector<RtpPacket> packets;
if (rtp_video_sender_) {
if (video_rtp_codec_) {
video_rtp_codec_->Encode(
static_cast<RtpCodec::VideoFrameType>(frame_type),
(uint8_t *)encoded_frame, (uint32_t)size, packets);
video_rtp_codec_->Encode((uint8_t *)data, (uint32_t)size, packets);
}
rtp_video_sender_->Enqueue(packets);
}

View File

@@ -7,21 +7,26 @@
#ifndef _VIDEO_CHANNEL_SEND_H_
#define _VIDEO_CHANNEL_SEND_H_
#include "ice_agent.h"
#include "rtp_codec.h"
#include "rtp_video_sender.h"
class VideoChannelSend {
public:
VideoChannelSend();
VideoChannelSend(std::shared_ptr<IceAgent> ice_agent,
std::shared_ptr<IOStatistics> ice_io_statistics);
~VideoChannelSend();
public:
void Initialize(RtpPacket::PAYLOAD_TYPE negotiated_video_pt);
void Initialize(RtpPacket::PAYLOAD_TYPE payload_type);
void Destroy();
int SendVideo(char *encoded_frame, size_t size);
int SendVideo(char *data, size_t size);
private:
std::shared_ptr<IceAgent> ice_agent_ = nullptr;
std::shared_ptr<IOStatistics> ice_io_statistics_ = nullptr;
std::unique_ptr<RtpCodec> video_rtp_codec_ = nullptr;
std::unique_ptr<RtpVideoSender> rtp_video_sender_ = nullptr;
};

View File

@@ -55,304 +55,311 @@ RtpCodec::~RtpCodec() {
// }
}
void RtpCodec::Encode(uint8_t* buffer, uint32_t size,
std::vector<RtpPacket>& packets) {
if (RtpPacket::PAYLOAD_TYPE::H264 == payload_type_) {
if (fec_enable_ && IsKeyFrame((const uint8_t*)buffer, size)) {
uint8_t** fec_packets = fec_encoder_.Encode((const char*)buffer, size);
if (nullptr == fec_packets) {
LOG_ERROR("Invalid fec_packets");
return;
}
uint8_t num_of_total_packets = 0;
uint8_t num_of_source_packets = 0;
unsigned int last_packet_size = 0;
fec_encoder_.GetFecPacketsParams(size, num_of_total_packets,
num_of_source_packets, last_packet_size);
// void RtpCodec::Encode(uint8_t* buffer, uint32_t size,
// std::vector<RtpPacket>& packets) {
// if (RtpPacket::PAYLOAD_TYPE::H264 == payload_type_) {
// if (fec_enable_ && IsKeyFrame((const uint8_t*)buffer, size)) {
// uint8_t** fec_packets = fec_encoder_.Encode((const char*)buffer, size);
// if (nullptr == fec_packets) {
// LOG_ERROR("Invalid fec_packets");
// return;
// }
// uint8_t num_of_total_packets = 0;
// uint8_t num_of_source_packets = 0;
// unsigned int last_packet_size = 0;
// fec_encoder_.GetFecPacketsParams(size, num_of_total_packets,
// num_of_source_packets,
// last_packet_size);
timestamp_ = std::chrono::duration_cast<std::chrono::microseconds>(
std::chrono::system_clock::now().time_since_epoch())
.count();
for (uint8_t index = 0; index < num_of_total_packets; index++) {
RtpPacket rtp_packet;
if (index < num_of_source_packets) {
rtp_packet.SetVerion(version_);
rtp_packet.SetHasPadding(has_padding_);
rtp_packet.SetHasExtension(has_extension_);
rtp_packet.SetMarker((index == (num_of_source_packets - 1)) ? 1 : 0);
rtp_packet.SetPayloadType(RtpPacket::PAYLOAD_TYPE::H264_FEC_SOURCE);
rtp_packet.SetSequenceNumber(sequence_number_++);
rtp_packet.SetTimestamp(timestamp_);
rtp_packet.SetSsrc(ssrc_);
if (!csrcs_.empty()) {
rtp_packet.SetCsrcs(csrcs_);
}
if (has_extension_) {
rtp_packet.SetExtensionProfile(extension_profile_);
rtp_packet.SetExtensionData(extension_data_, extension_len_);
}
RtpPacket::FU_INDICATOR fu_indicator;
fu_indicator.forbidden_bit = 0;
fu_indicator.nal_reference_idc = 0;
fu_indicator.nal_unit_type = FU_A;
RtpPacket::FU_HEADER fu_header;
fu_header.start = index == 0 ? 1 : 0;
fu_header.end = index == num_of_source_packets - 1 ? 1 : 0;
fu_header.remain_bit = 0;
fu_header.nal_unit_type = FU_A;
rtp_packet.SetFuIndicator(fu_indicator);
rtp_packet.SetFuHeader(fu_header);
if (index == num_of_source_packets - 1) {
if (last_packet_size > 0) {
rtp_packet.EncodeH264FecSource(fec_packets[index],
last_packet_size, index,
num_of_source_packets);
} else {
rtp_packet.EncodeH264FecSource(fec_packets[index], MAX_NALU_LEN,
index, num_of_source_packets);
}
} else {
rtp_packet.EncodeH264FecSource(fec_packets[index], MAX_NALU_LEN,
index, num_of_source_packets);
}
} else if (index >= num_of_source_packets &&
index < num_of_total_packets) {
rtp_packet.SetVerion(version_);
rtp_packet.SetHasPadding(has_padding_);
rtp_packet.SetHasExtension(has_extension_);
rtp_packet.SetMarker(index == num_of_total_packets - 1 ? 1 : 0);
rtp_packet.SetPayloadType(RtpPacket::PAYLOAD_TYPE::H264_FEC_REPAIR);
rtp_packet.SetSequenceNumber(sequence_number_++);
rtp_packet.SetTimestamp(timestamp_);
rtp_packet.SetSsrc(ssrc_);
if (!csrcs_.empty()) {
rtp_packet.SetCsrcs(csrcs_);
}
if (has_extension_) {
rtp_packet.SetExtensionProfile(extension_profile_);
rtp_packet.SetExtensionData(extension_data_, extension_len_);
}
rtp_packet.EncodeH264FecRepair(fec_packets[index], MAX_NALU_LEN,
index, num_of_source_packets);
}
packets.emplace_back(rtp_packet);
// timestamp_ = std::chrono::duration_cast<std::chrono::microseconds>(
// std::chrono::system_clock::now().time_since_epoch())
// .count();
// for (uint8_t index = 0; index < num_of_total_packets; index++) {
// RtpPacket rtp_packet;
// if (index < num_of_source_packets) {
// rtp_packet.EncodeH264Fua(fec_packets[index], MAX_NALU_LEN);
// rtp_packet.SetVerion(version_);
// rtp_packet.SetHasPadding(has_padding_);
// rtp_packet.SetHasExtension(has_extension_);
// rtp_packet.SetMarker((index == (num_of_source_packets - 1)) ? 1 :
// 0);
// rtp_packet.SetPayloadType(RtpPacket::PAYLOAD_TYPE::H264_FEC_SOURCE);
// rtp_packet.SetSequenceNumber(sequence_number_++);
// rtp_packet.SetTimestamp(timestamp_);
// rtp_packet.SetSsrc(ssrc_);
// if (!csrcs_.empty()) {
// rtp_packet.SetCsrcs(csrcs_);
// }
// if (has_extension_) {
// rtp_packet.SetExtensionProfile(extension_profile_);
// rtp_packet.SetExtensionData(extension_data_, extension_len_);
// }
// RtpPacket::FU_INDICATOR fu_indicator;
// fu_indicator.forbidden_bit = 0;
// fu_indicator.nal_reference_idc = 0;
// fu_indicator.nal_unit_type = FU_A;
// RtpPacket::FU_HEADER fu_header;
// fu_header.start = index == 0 ? 1 : 0;
// fu_header.end = index == num_of_source_packets - 1 ? 1 : 0;
// fu_header.remain_bit = 0;
// fu_header.nal_unit_type = FU_A;
// rtp_packet.SetFuIndicator(fu_indicator);
// rtp_packet.SetFuHeader(fu_header);
// if (index == num_of_source_packets - 1) {
// if (last_packet_size > 0) {
// rtp_packet.EncodeH264FecSource(fec_packets[index],
// last_packet_size, index,
// num_of_source_packets);
// } else {
// rtp_packet.EncodeH264FecSource(fec_packets[index],
// MAX_NALU_LEN,
// index, num_of_source_packets);
// }
// } else {
// rtp_packet.EncodeH264FecSource(fec_packets[index], MAX_NALU_LEN,
// index, num_of_source_packets);
// }
// } else if (index >= num_of_source_packets &&
// index < num_of_total_packets) {
// rtp_packet.SetVerion(version_);
// rtp_packet.SetHasPadding(has_padding_);
// rtp_packet.SetHasExtension(has_extension_);
// rtp_packet.SetMarker(index == num_of_total_packets - 1 ? 1 : 0);
// rtp_packet.SetPayloadType(RtpPacket::PAYLOAD_TYPE::H264_FEC_REPAIR);
// rtp_packet.SetSequenceNumber(sequence_number_++);
// rtp_packet.SetTimestamp(timestamp_);
// rtp_packet.SetSsrc(ssrc_);
// if (!csrcs_.empty()) {
// rtp_packet.SetCsrcs(csrcs_);
// }
// if (has_extension_) {
// rtp_packet.SetExtensionProfile(extension_profile_);
// rtp_packet.SetExtensionData(extension_data_, extension_len_);
// }
// rtp_packet.EncodeH264FecRepair(fec_packets[index], MAX_NALU_LEN,
// index, num_of_source_packets);
// }
// packets.emplace_back(rtp_packet);
// // if (index < num_of_source_packets) {
// // rtp_packet.EncodeH264Fua(fec_packets[index], MAX_NALU_LEN);
// // packets.emplace_back(rtp_packet);
// // }
// }
// fec_encoder_.ReleaseFecPackets(fec_packets, size);
// return;
// }
// if (size <= MAX_NALU_LEN) {
// 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::duration_cast<std::chrono::microseconds>(
// std::chrono::system_clock::now().time_since_epoch())
// .count();
// rtp_packet.SetTimestamp(timestamp_);
// rtp_packet.SetSsrc(ssrc_);
// if (!csrcs_.empty()) {
// rtp_packet.SetCsrcs(csrcs_);
// }
// if (has_extension_) {
// rtp_packet.SetExtensionProfile(extension_profile_);
// rtp_packet.SetExtensionData(extension_data_, extension_len_);
// }
// RtpPacket::FU_INDICATOR fu_indicator;
// fu_indicator.forbidden_bit = 0;
// fu_indicator.nal_reference_idc = 1;
// fu_indicator.nal_unit_type = NALU;
// rtp_packet.SetFuIndicator(fu_indicator);
// rtp_packet.EncodeH264Nalu(buffer, size);
// packets.emplace_back(rtp_packet);
// } else {
// uint32_t last_packet_size = size % MAX_NALU_LEN;
// uint32_t packet_num = size / MAX_NALU_LEN + (last_packet_size ? 1 : 0);
// timestamp_ = std::chrono::duration_cast<std::chrono::microseconds>(
// std::chrono::system_clock::now().time_since_epoch())
// .count();
// for (uint32_t index = 0; index < packet_num; index++) {
// RtpPacket rtp_packet;
// rtp_packet.SetVerion(version_);
// rtp_packet.SetHasPadding(has_padding_);
// rtp_packet.SetHasExtension(has_extension_);
// rtp_packet.SetMarker(index == packet_num - 1 ? 1 : 0);
// rtp_packet.SetPayloadType(RtpPacket::PAYLOAD_TYPE(payload_type_));
// rtp_packet.SetSequenceNumber(sequence_number_++);
// rtp_packet.SetTimestamp(timestamp_);
// rtp_packet.SetSsrc(ssrc_);
// if (!csrcs_.empty()) {
// rtp_packet.SetCsrcs(csrcs_);
// }
// if (has_extension_) {
// rtp_packet.SetExtensionProfile(extension_profile_);
// rtp_packet.SetExtensionData(extension_data_, extension_len_);
// }
// RtpPacket::FU_INDICATOR fu_indicator;
// fu_indicator.forbidden_bit = 0;
// fu_indicator.nal_reference_idc = 0;
// fu_indicator.nal_unit_type = FU_A;
// RtpPacket::FU_HEADER fu_header;
// fu_header.start = index == 0 ? 1 : 0;
// fu_header.end = index == packet_num - 1 ? 1 : 0;
// fu_header.remain_bit = 0;
// fu_header.nal_unit_type = FU_A;
// rtp_packet.SetFuIndicator(fu_indicator);
// rtp_packet.SetFuHeader(fu_header);
// if (index == packet_num - 1 && last_packet_size > 0) {
// rtp_packet.EncodeH264Fua(buffer + index * MAX_NALU_LEN,
// last_packet_size);
// } else {
// rtp_packet.EncodeH264Fua(buffer + index * MAX_NALU_LEN,
// MAX_NALU_LEN);
// }
// packets.emplace_back(rtp_packet);
// }
}
// }
// } else if (RtpPacket::PAYLOAD_TYPE::AV1 == payload_type_) {
// std::vector<Obu> obus = ParseObus(buffer, size);
// LOG_ERROR("Total size = [{}]", size);
// for (int i = 0; i < obus.size(); i++) {
// LOG_ERROR("[{}] Obu size = [{}], Obu type [{}]", i, obus[i].size,
// ObuTypeToString((OBU_TYPE)ObuType(obus[i].header)));
// if (obus[i].size <= MAX_NALU_LEN) {
// 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_++);
fec_encoder_.ReleaseFecPackets(fec_packets, size);
return;
}
if (size <= MAX_NALU_LEN) {
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::duration_cast<std::chrono::microseconds>(
// std::chrono::system_clock::now().time_since_epoch())
// .count();
// rtp_packet.SetTimestamp(timestamp_);
// rtp_packet.SetSsrc(ssrc_);
timestamp_ = std::chrono::duration_cast<std::chrono::microseconds>(
std::chrono::system_clock::now().time_since_epoch())
.count();
rtp_packet.SetTimestamp(timestamp_);
rtp_packet.SetSsrc(ssrc_);
// if (!csrcs_.empty()) {
// rtp_packet.SetCsrcs(csrcs_);
// }
if (!csrcs_.empty()) {
rtp_packet.SetCsrcs(csrcs_);
}
// if (has_extension_) {
// rtp_packet.SetExtensionProfile(extension_profile_);
// rtp_packet.SetExtensionData(extension_data_, extension_len_);
// }
if (has_extension_) {
rtp_packet.SetExtensionProfile(extension_profile_);
rtp_packet.SetExtensionData(extension_data_, extension_len_);
}
// rtp_packet.SetAv1AggrHeader(0, 0, 1, 0);
RtpPacket::FU_INDICATOR fu_indicator;
fu_indicator.forbidden_bit = 0;
fu_indicator.nal_reference_idc = 1;
fu_indicator.nal_unit_type = NALU;
rtp_packet.SetFuIndicator(fu_indicator);
// rtp_packet.EncodeAv1(obus[i].payload.data(), obus[i].payload.size());
// packets.emplace_back(rtp_packet);
// } else {
// uint32_t last_packet_size = obus[i].payload.size() % MAX_NALU_LEN;
// size_t packet_num =
// obus[i].payload.size() / MAX_NALU_LEN + (last_packet_size ? 1 :
// 0);
// timestamp_ = std::chrono::duration_cast<std::chrono::microseconds>(
// std::chrono::system_clock::now().time_since_epoch())
// .count();
// for (uint32_t index = 0; index < packet_num; index++) {
// RtpPacket rtp_packet;
// rtp_packet.SetVerion(version_);
// rtp_packet.SetHasPadding(has_padding_);
// rtp_packet.SetHasExtension(has_extension_);
// rtp_packet.SetMarker(index == packet_num - 1 ? 1 : 0);
// rtp_packet.SetPayloadType(RtpPacket::PAYLOAD_TYPE(payload_type_));
// rtp_packet.SetSequenceNumber(sequence_number_++);
// rtp_packet.SetTimestamp(timestamp_);
// rtp_packet.SetSsrc(ssrc_);
// if (!csrcs_.empty()) {
// rtp_packet.SetCsrcs(csrcs_);
// }
// if (has_extension_) {
// rtp_packet.SetExtensionProfile(extension_profile_);
// rtp_packet.SetExtensionData(extension_data_, extension_len_);
// }
rtp_packet.EncodeH264Nalu(buffer, size);
packets.emplace_back(rtp_packet);
// int z = index != 0 ? 1 : 0;
// int y = index != packet_num - 1 ? 1 : 0;
// int w = 1;
// int n = 0;
// rtp_packet.SetAv1AggrHeader(z, y, w, n);
} else {
uint32_t last_packet_size = size % MAX_NALU_LEN;
uint32_t packet_num = size / MAX_NALU_LEN + (last_packet_size ? 1 : 0);
timestamp_ = std::chrono::duration_cast<std::chrono::microseconds>(
std::chrono::system_clock::now().time_since_epoch())
.count();
// if (index == packet_num - 1 && last_packet_size > 0) {
// rtp_packet.EncodeAv1(obus[i].payload.data() + index *
// MAX_NALU_LEN,
// last_packet_size);
// } else {
// rtp_packet.EncodeAv1(obus[i].payload.data() + index *
// MAX_NALU_LEN,
// MAX_NALU_LEN);
// }
// 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_++);
for (uint32_t index = 0; index < packet_num; index++) {
RtpPacket rtp_packet;
rtp_packet.SetVerion(version_);
rtp_packet.SetHasPadding(has_padding_);
rtp_packet.SetHasExtension(has_extension_);
rtp_packet.SetMarker(index == packet_num - 1 ? 1 : 0);
rtp_packet.SetPayloadType(RtpPacket::PAYLOAD_TYPE(payload_type_));
rtp_packet.SetSequenceNumber(sequence_number_++);
rtp_packet.SetTimestamp(timestamp_);
rtp_packet.SetSsrc(ssrc_);
// timestamp_ = std::chrono::duration_cast<std::chrono::microseconds>(
// std::chrono::system_clock::now().time_since_epoch())
// .count();
// rtp_packet.SetTimestamp(timestamp_);
// rtp_packet.SetSsrc(ssrc_);
if (!csrcs_.empty()) {
rtp_packet.SetCsrcs(csrcs_);
}
// 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_);
// 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_++);
if (has_extension_) {
rtp_packet.SetExtensionProfile(extension_profile_);
rtp_packet.SetExtensionData(extension_data_, extension_len_);
}
// timestamp_ = std::chrono::duration_cast<std::chrono::microseconds>(
// std::chrono::system_clock::now().time_since_epoch())
// .count();
// rtp_packet.SetTimestamp(timestamp_);
// rtp_packet.SetSsrc(ssrc_);
RtpPacket::FU_INDICATOR fu_indicator;
fu_indicator.forbidden_bit = 0;
fu_indicator.nal_reference_idc = 0;
fu_indicator.nal_unit_type = FU_A;
// rtp_packet.Encode(buffer, size);
// packets.emplace_back(rtp_packet);
// }
// }
RtpPacket::FU_HEADER fu_header;
fu_header.start = index == 0 ? 1 : 0;
fu_header.end = index == packet_num - 1 ? 1 : 0;
fu_header.remain_bit = 0;
fu_header.nal_unit_type = FU_A;
rtp_packet.SetFuIndicator(fu_indicator);
rtp_packet.SetFuHeader(fu_header);
if (index == packet_num - 1 && last_packet_size > 0) {
rtp_packet.EncodeH264Fua(buffer + index * MAX_NALU_LEN,
last_packet_size);
} else {
rtp_packet.EncodeH264Fua(buffer + index * MAX_NALU_LEN, MAX_NALU_LEN);
}
packets.emplace_back(rtp_packet);
}
}
} else if (RtpPacket::PAYLOAD_TYPE::AV1 == payload_type_) {
std::vector<Obu> obus = ParseObus(buffer, size);
LOG_ERROR("Total size = [{}]", size);
for (int i = 0; i < obus.size(); i++) {
LOG_ERROR("[{}] Obu size = [{}], Obu type [{}]", i, obus[i].size,
ObuTypeToString((OBU_TYPE)ObuType(obus[i].header)));
if (obus[i].size <= MAX_NALU_LEN) {
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::duration_cast<std::chrono::microseconds>(
std::chrono::system_clock::now().time_since_epoch())
.count();
rtp_packet.SetTimestamp(timestamp_);
rtp_packet.SetSsrc(ssrc_);
if (!csrcs_.empty()) {
rtp_packet.SetCsrcs(csrcs_);
}
if (has_extension_) {
rtp_packet.SetExtensionProfile(extension_profile_);
rtp_packet.SetExtensionData(extension_data_, extension_len_);
}
rtp_packet.SetAv1AggrHeader(0, 0, 1, 0);
rtp_packet.EncodeAv1(obus[i].payload.data(), obus[i].payload.size());
packets.emplace_back(rtp_packet);
} else {
uint32_t last_packet_size = obus[i].payload.size() % MAX_NALU_LEN;
size_t packet_num =
obus[i].payload.size() / MAX_NALU_LEN + (last_packet_size ? 1 : 0);
timestamp_ = std::chrono::duration_cast<std::chrono::microseconds>(
std::chrono::system_clock::now().time_since_epoch())
.count();
for (uint32_t index = 0; index < packet_num; index++) {
RtpPacket rtp_packet;
rtp_packet.SetVerion(version_);
rtp_packet.SetHasPadding(has_padding_);
rtp_packet.SetHasExtension(has_extension_);
rtp_packet.SetMarker(index == packet_num - 1 ? 1 : 0);
rtp_packet.SetPayloadType(RtpPacket::PAYLOAD_TYPE(payload_type_));
rtp_packet.SetSequenceNumber(sequence_number_++);
rtp_packet.SetTimestamp(timestamp_);
rtp_packet.SetSsrc(ssrc_);
if (!csrcs_.empty()) {
rtp_packet.SetCsrcs(csrcs_);
}
if (has_extension_) {
rtp_packet.SetExtensionProfile(extension_profile_);
rtp_packet.SetExtensionData(extension_data_, extension_len_);
}
int z = index != 0 ? 1 : 0;
int y = index != packet_num - 1 ? 1 : 0;
int w = 1;
int n = 0;
rtp_packet.SetAv1AggrHeader(z, y, w, n);
if (index == packet_num - 1 && last_packet_size > 0) {
rtp_packet.EncodeAv1(obus[i].payload.data() + index * MAX_NALU_LEN,
last_packet_size);
} else {
rtp_packet.EncodeAv1(obus[i].payload.data() + index * MAX_NALU_LEN,
MAX_NALU_LEN);
}
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::duration_cast<std::chrono::microseconds>(
std::chrono::system_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_);
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::duration_cast<std::chrono::microseconds>(
std::chrono::system_clock::now().time_since_epoch())
.count();
rtp_packet.SetTimestamp(timestamp_);
rtp_packet.SetSsrc(ssrc_);
rtp_packet.Encode(buffer, size);
packets.emplace_back(rtp_packet);
}
}
void RtpCodec::Encode(VideoFrameType frame_type, uint8_t* buffer, uint32_t size,
void RtpCodec::Encode(uint8_t* buffer, uint32_t size,
std::vector<RtpPacket>& packets) {
if (RtpPacket::PAYLOAD_TYPE::H264 == payload_type_) {
if (fec_enable_ && IsKeyFrame((const uint8_t*)buffer, size)) {
@@ -595,10 +602,7 @@ void RtpCodec::Encode(VideoFrameType frame_type, uint8_t* buffer, uint32_t size,
int z = index != 0 ? 1 : 0;
int y = index != packet_num - 1 ? 1 : 0;
int w = 1;
int n = (frame_type == VideoFrameType::kVideoFrameKey) &&
(ObuType(obus[i].header) == kObuTypeSequenceHeader)
? 1
: 0;
int n = (ObuType(obus[i].header) == kObuTypeSequenceHeader) ? 1 : 0;
rtp_packet.SetAv1AggrHeader(z, y, w, n);
if (index == packet_num - 1 && last_packet_size > 0) {
rtp_packet.EncodeAv1(obus[i].payload.data() + index * MAX_NALU_LEN,

View File

@@ -23,8 +23,6 @@ class RtpCodec {
public:
void Encode(uint8_t* buffer, uint32_t size, std::vector<RtpPacket>& packets);
void Encode(VideoFrameType frame_type, uint8_t* buffer, uint32_t size,
std::vector<RtpPacket>& packets);
size_t Decode(RtpPacket& packet, uint8_t* payload);
// protected:

View File

@@ -62,11 +62,49 @@ int IceTransport::InitIceTransmission(
std::string &stun_ip, int stun_port, std::string &turn_ip, int turn_port,
std::string &turn_username, std::string &turn_password,
RtpPacket::PAYLOAD_TYPE video_codec_payload_type) {
ice_agent_ = std::make_unique<IceAgent>(
offer_peer_, use_trickle_ice_, use_reliable_ice_, enable_turn_,
force_turn_, stun_ip, stun_port, turn_ip, turn_port, turn_username,
turn_password);
InitializeIOStatistics();
InitializeChannels(video_codec_payload_type);
ice_agent_->CreateIceAgent(
[](NiceAgent *agent, guint stream_id, guint component_id,
NiceComponentState state, gpointer user_ptr) {
static_cast<IceTransport *>(user_ptr)->OnIceStateChange(
agent, stream_id, component_id, state, user_ptr);
},
[](NiceAgent *agent, guint stream_id, guint component_id,
gchar *foundation, gpointer user_ptr) {
static_cast<IceTransport *>(user_ptr)->OnNewLocalCandidate(
agent, stream_id, component_id, foundation, user_ptr);
},
[](NiceAgent *agent, guint stream_id, gpointer user_ptr) {
static_cast<IceTransport *>(user_ptr)->OnGatheringDone(agent, stream_id,
user_ptr);
},
[](NiceAgent *agent, guint stream_id, guint component_id,
const char *lfoundation, const char *rfoundation, gpointer user_ptr) {
static_cast<IceTransport *>(user_ptr)->OnNewSelectedPair(
agent, stream_id, component_id, lfoundation, rfoundation, user_ptr);
},
[](NiceAgent *agent, guint stream_id, guint component_id, guint size,
gchar *buffer, gpointer user_ptr) {
static_cast<IceTransport *>(user_ptr)->OnReceiveBuffer(
agent, stream_id, component_id, size, buffer, user_ptr);
},
this);
return 0;
}
void IceTransport::InitializeIOStatistics() {
ice_io_statistics_ = std::make_unique<IOStatistics>(
[this](const IOStatistics::NetTrafficStats &net_traffic_stats) {
if (on_receive_net_status_report_) {
XNetTrafficStats xnet_traffic_stats;
memcpy(&xnet_traffic_stats, &net_traffic_stats,
sizeof(XNetTrafficStats));
on_receive_net_status_report_(
@@ -75,32 +113,151 @@ int IceTransport::InitIceTransmission(
remote_user_id_.size(), user_data_);
}
});
}
void IceTransport::InitializeChannels(
RtpPacket::PAYLOAD_TYPE video_codec_payload_type) {
video_codec_payload_type_ = video_codec_payload_type;
rtp_video_receiver_ = std::make_unique<RtpVideoReceiver>(ice_io_statistics_);
// rr sender
rtp_video_receiver_->SetSendDataFunc(
[this](const char *data, size_t size) -> int {
if (!ice_agent_) {
LOG_ERROR("ice_agent_ is nullptr");
return -1;
}
video_channel_send_ =
std::make_unique<VideoChannelSend>(ice_agent_, ice_io_statistics_);
audio_channel_send_ =
std::make_unique<AudioChannelSend>(ice_agent_, ice_io_statistics_);
data_channel_send_ =
std::make_unique<DataChannelSend>(ice_agent_, ice_io_statistics_);
if (state_ != NICE_COMPONENT_STATE_CONNECTED &&
state_ != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
nice_component_state_to_string(state_));
return -2;
}
video_channel_send_->Initialize(video_codec_payload_type_);
audio_channel_send_->Initialize(RtpPacket::PAYLOAD_TYPE::OPUS);
data_channel_send_->Initialize(RtpPacket::PAYLOAD_TYPE::DATA);
return ice_agent_->Send(data, size);
video_channel_receive_ = std::make_unique<VideoChannelReceive>(
ice_agent_, ice_io_statistics_,
[this](VideoFrame &video_frame) { OnReceiveCompleteFrame(video_frame); });
audio_channel_receive_ = std::make_unique<AudioChannelReceive>(
ice_agent_, ice_io_statistics_, [this](const char *data, size_t size) {
OnReceiveCompleteAudio(data, size);
});
rtp_video_receiver_->SetOnReceiveCompleteFrame(
[this](VideoFrame &video_frame) -> void {
ice_io_statistics_->UpdateVideoInboundBytes(
(uint32_t)video_frame.Size());
[[maybe_unused]] int num_frame_returned = video_decoder_->Decode(
data_channel_receive_ = std::make_unique<DataChannelReceive>(
ice_agent_, ice_io_statistics_, [this](const char *data, size_t size) {
OnReceiveCompleteData(data, size);
});
video_channel_receive_->Initialize(video_codec_payload_type_);
audio_channel_receive_->Initialize(RtpPacket::PAYLOAD_TYPE::OPUS);
data_channel_receive_->Initialize(RtpPacket::PAYLOAD_TYPE::DATA);
}
void IceTransport::OnIceStateChange(NiceAgent *agent, guint stream_id,
guint component_id,
NiceComponentState state,
gpointer user_ptr) {
if (!is_closed_) {
LOG_INFO("[{}->{}] state_change: {}", user_id_, remote_user_id_,
nice_component_state_to_string(state));
state_ = state;
if (state == NICE_COMPONENT_STATE_READY ||
state == NICE_COMPONENT_STATE_CONNECTED) {
ice_io_statistics_->Start();
}
on_ice_status_change_(nice_component_state_to_string(state),
remote_user_id_);
}
}
void IceTransport::OnNewLocalCandidate(NiceAgent *agent, guint stream_id,
guint component_id, gchar *foundation,
gpointer user_ptr) {
if (use_trickle_ice_) {
GSList *cands =
nice_agent_get_local_candidates(agent, stream_id, component_id);
NiceCandidate *cand;
for (GSList *i = cands; i; i = i->next) {
cand = (NiceCandidate *)i->data;
if (g_strcmp0(cand->foundation, foundation) == 0) {
new_local_candidate_ =
nice_agent_generate_local_candidate_sdp(agent, cand);
json message = {{"type", "new_candidate"},
{"transmission_id", transmission_id_},
{"user_id", user_id_},
{"remote_user_id", remote_user_id_},
{"sdp", new_local_candidate_}};
if (ice_ws_transport_) {
ice_ws_transport_->Send(message.dump());
}
}
}
g_slist_free_full(cands, (GDestroyNotify)nice_candidate_free);
}
}
void IceTransport::OnGatheringDone(NiceAgent *agent, guint stream_id,
gpointer user_ptr) {
LOG_INFO("[{}->{}] gather_done", user_id_, remote_user_id_);
if (!use_trickle_ice_) {
if (offer_peer_) {
SendOffer();
} else {
SendAnswer();
}
}
}
void IceTransport::OnNewSelectedPair(NiceAgent *agent, guint stream_id,
guint component_id,
const char *lfoundation,
const char *rfoundation,
gpointer user_ptr) {
LOG_INFO("new selected pair: [{}] [{}]", lfoundation, rfoundation);
NiceCandidate *local = nullptr;
NiceCandidate *remote = nullptr;
nice_agent_get_selected_pair(agent, stream_id, component_id, &local, &remote);
if (local->type == NICE_CANDIDATE_TYPE_RELAYED &&
remote->type == NICE_CANDIDATE_TYPE_RELAYED) {
LOG_INFO("Traversal using relay server");
traversal_type_ = TraversalType::TRelay;
} else {
LOG_INFO("Traversal using p2p");
traversal_type_ = TraversalType::TP2P;
}
XNetTrafficStats net_traffic_stats;
memset(&net_traffic_stats, 0, sizeof(net_traffic_stats));
on_receive_net_status_report_(user_id_.data(), user_id_.size(),
TraversalMode(traversal_type_),
&net_traffic_stats, remote_user_id_.data(),
remote_user_id_.size(), user_data_);
}
void IceTransport::OnReceiveBuffer(NiceAgent *agent, guint stream_id,
guint component_id, guint size,
gchar *buffer, gpointer user_ptr) {
if (!is_closed_) {
if (CheckIsRtpPacket(buffer, size)) {
if (CheckIsVideoPacket(buffer, size)) {
video_channel_receive_->OnReceiveRtpPacket(buffer, size);
} else if (CheckIsAudioPacket(buffer, size)) {
audio_channel_receive_->OnReceiveRtpPacket(buffer, size);
} else if (CheckIsDataPacket(buffer, size)) {
data_channel_receive_->OnReceiveRtpPacket(buffer, size);
}
} else if (CheckIsRtcpPacket(buffer, size)) {
LOG_ERROR("Rtcp packet [{}]", (uint8_t)(buffer[1]));
} else {
LOG_ERROR("Unknown packet");
}
}
}
void IceTransport::OnReceiveCompleteFrame(VideoFrame &video_frame) {
int num_frame_returned = video_decoder_->Decode(
(uint8_t *)video_frame.Buffer(), video_frame.Size(),
[this](VideoFrame video_frame) {
if (on_receive_video_) {
@@ -113,261 +270,23 @@ int IceTransport::InitIceTransmission(
remote_user_id_.size(), user_data_);
}
});
});
rtp_video_receiver_->Start();
rtp_video_sender_ = std::make_unique<RtpVideoSender>(ice_io_statistics_);
rtp_video_sender_->SetSendDataFunc(
[this](const char *data, size_t size) -> int {
if (!ice_agent_) {
LOG_ERROR("ice_agent_ is nullptr");
return -1;
}
if (state_ != NICE_COMPONENT_STATE_CONNECTED &&
state_ != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
nice_component_state_to_string(state_));
return -2;
}
ice_io_statistics_->UpdateVideoOutboundBytes((uint32_t)size);
return ice_agent_->Send(data, size);
});
rtp_video_sender_->Start();
rtp_audio_receiver_ = std::make_unique<RtpAudioReceiver>(ice_io_statistics_);
// rr sender
rtp_audio_receiver_->SetSendDataFunc(
[this](const char *data, size_t size) -> int {
if (!ice_agent_) {
LOG_ERROR("ice_agent_ is nullptr");
return -1;
}
if (state_ != NICE_COMPONENT_STATE_CONNECTED &&
state_ != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
nice_component_state_to_string(state_));
return -2;
}
return ice_agent_->Send(data, size);
});
rtp_audio_receiver_->SetOnReceiveData([this](const char *data,
size_t size) -> void {
ice_io_statistics_->UpdateAudioInboundBytes((uint32_t)size);
[[maybe_unused]] int num_frame_returned = audio_decoder_->Decode(
void IceTransport::OnReceiveCompleteAudio(const char *data, size_t size) {
int num_frame_returned = audio_decoder_->Decode(
(uint8_t *)data, size, [this](uint8_t *data, int size) {
if (on_receive_audio_) {
on_receive_audio_((const char *)data, size, remote_user_id_.data(),
remote_user_id_.size(), user_data_);
}
});
});
rtp_audio_sender_ = std::make_unique<RtpAudioSender>(ice_io_statistics_);
rtp_audio_sender_->SetSendDataFunc(
[this](const char *data, size_t size) -> int {
if (!ice_agent_) {
LOG_ERROR("ice_agent_ is nullptr");
return -1;
}
if (state_ != NICE_COMPONENT_STATE_CONNECTED &&
state_ != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
nice_component_state_to_string(state_));
return -2;
}
ice_io_statistics_->UpdateAudioOutboundBytes((uint32_t)size);
return ice_agent_->Send(data, size);
});
rtp_audio_sender_->Start();
rtp_data_receiver_ = std::make_unique<RtpDataReceiver>(ice_io_statistics_);
// rr sender
rtp_data_receiver_->SetSendDataFunc(
[this](const char *data, size_t size) -> int {
if (!ice_agent_) {
LOG_ERROR("ice_agent_ is nullptr");
return -1;
}
if (state_ != NICE_COMPONENT_STATE_CONNECTED &&
state_ != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
nice_component_state_to_string(state_));
return -2;
}
return ice_agent_->Send(data, size);
});
rtp_data_receiver_->SetOnReceiveData(
[this](const char *data, size_t size) -> void {
ice_io_statistics_->UpdateDataInboundBytes((uint32_t)size);
void IceTransport::OnReceiveCompleteData(const char *data, size_t size) {
if (on_receive_data_) {
on_receive_data_(data, size, remote_user_id_.data(),
remote_user_id_.size(), user_data_);
on_receive_data_(data, size, remote_user_id_.data(), remote_user_id_.size(),
user_data_);
}
});
rtp_data_sender_ = std::make_unique<RtpDataSender>(ice_io_statistics_);
rtp_data_sender_->SetSendDataFunc(
[this](const char *data, size_t size) -> int {
if (!ice_agent_) {
LOG_ERROR("ice_agent_ is nullptr");
return -1;
}
if (state_ != NICE_COMPONENT_STATE_CONNECTED &&
state_ != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
nice_component_state_to_string(state_));
return -2;
}
ice_io_statistics_->UpdateDataOutboundBytes((uint32_t)size);
return ice_agent_->Send(data, size);
});
rtp_data_sender_->Start();
ice_agent_ = std::make_unique<IceAgent>(
offer_peer_, use_trickle_ice_, use_reliable_ice_, enable_turn_,
force_turn_, stun_ip, stun_port, turn_ip, turn_port, turn_username,
turn_password);
ice_agent_->CreateIceAgent(
[]([[maybe_unused]] NiceAgent *agent, [[maybe_unused]] guint stream_id,
[[maybe_unused]] guint component_id, NiceComponentState state,
gpointer user_ptr) {
if (auto ice_transport = static_cast<IceTransport *>(user_ptr)) {
if (!ice_transport->is_closed_) {
LOG_INFO("[{}->{}] state_change: {}", ice_transport->user_id_,
ice_transport->remote_user_id_,
nice_component_state_to_string(state));
ice_transport->state_ = state;
if (state == NICE_COMPONENT_STATE_READY ||
state == NICE_COMPONENT_STATE_CONNECTED) {
ice_transport->ice_io_statistics_->Start();
}
ice_transport->on_ice_status_change_(
nice_component_state_to_string(state),
ice_transport->remote_user_id_);
}
}
},
[](NiceAgent *agent, guint stream_id, guint component_id,
gchar *foundation, gpointer user_ptr) {
if (auto ice_transport = static_cast<IceTransport *>(user_ptr)) {
if (ice_transport->use_trickle_ice_) {
GSList *cands =
nice_agent_get_local_candidates(agent, stream_id, component_id);
NiceCandidate *cand;
for (GSList *i = cands; i; i = i->next) {
cand = (NiceCandidate *)i->data;
if (g_strcmp0(cand->foundation, foundation) == 0) {
ice_transport->new_local_candidate_ =
nice_agent_generate_local_candidate_sdp(agent, cand);
json message = {
{"type", "new_candidate"},
{"transmission_id", ice_transport->transmission_id_},
{"user_id", ice_transport->user_id_},
{"remote_user_id", ice_transport->remote_user_id_},
{"sdp", ice_transport->new_local_candidate_}};
// LOG_INFO("[{}] Send new candidate to [{}]]:[{}]",
// ice_transport->user_id_,
// ice_transport->remote_user_id_,
// ice_transport->new_local_candidate_);
if (ice_transport->ice_ws_transport_) {
ice_transport->ice_ws_transport_->Send(message.dump());
}
}
}
g_slist_free_full(cands, (GDestroyNotify)nice_candidate_free);
}
}
},
[]([[maybe_unused]] NiceAgent *agent, [[maybe_unused]] guint stream_id,
gpointer user_ptr) {
// non-trickle
if (auto ice_transport = static_cast<IceTransport *>(user_ptr)) {
LOG_INFO("[{}->{}] gather_done", ice_transport->user_id_,
ice_transport->remote_user_id_);
if (!ice_transport->use_trickle_ice_) {
if (ice_transport->offer_peer_) {
ice_transport->SendOffer();
} else {
ice_transport->SendAnswer();
}
}
}
},
[](NiceAgent *agent, guint stream_id, guint component_id,
const char *lfoundation, const char *rfoundation, gpointer user_ptr) {
LOG_INFO("new selected pair: [{}] [{}]", lfoundation, rfoundation);
NiceCandidate *local = nullptr;
NiceCandidate *remote = nullptr;
nice_agent_get_selected_pair(agent, stream_id, component_id, &local,
&remote);
if (auto ice_transport = static_cast<IceTransport *>(user_ptr)) {
if (local->type == NICE_CANDIDATE_TYPE_RELAYED &&
remote->type == NICE_CANDIDATE_TYPE_RELAYED) {
LOG_INFO("Traversal using relay server");
ice_transport->traversal_type_ = TraversalType::TRelay;
} else {
LOG_INFO("Traversal using p2p");
ice_transport->traversal_type_ = TraversalType::TP2P;
}
XNetTrafficStats net_traffic_stats;
memset(&net_traffic_stats, 0, sizeof(net_traffic_stats));
ice_transport->on_receive_net_status_report_(
ice_transport->user_id_.data(), ice_transport->user_id_.size(),
TraversalMode(ice_transport->traversal_type_), &net_traffic_stats,
ice_transport->remote_user_id_.data(),
ice_transport->remote_user_id_.size(), ice_transport->user_data_);
}
},
[]([[maybe_unused]] NiceAgent *agent, [[maybe_unused]] guint stream_id,
[[maybe_unused]] guint component_id, guint size, gchar *buffer,
gpointer user_ptr) {
if (auto ice_transport = static_cast<IceTransport *>(user_ptr)) {
if (ice_transport && !ice_transport->is_closed_) {
if (ice_transport->CheckIsRtpPacket(buffer, size)) {
if (ice_transport->CheckIsVideoPacket(buffer, size)) {
RtpPacket packet((uint8_t *)buffer, size);
ice_transport->rtp_video_receiver_->InsertRtpPacket(packet);
} else if (ice_transport->CheckIsAudioPacket(buffer, size)) {
RtpPacket packet((uint8_t *)buffer, size);
ice_transport->rtp_audio_receiver_->InsertRtpPacket(packet);
} else if (ice_transport->CheckIsDataPacket(buffer, size)) {
RtpPacket packet((uint8_t *)buffer, size);
ice_transport->rtp_data_receiver_->InsertRtpPacket(packet);
}
} else if (ice_transport->CheckIsRtcpPacket(buffer, size)) {
LOG_ERROR("Rtcp packet [{}]", (uint8_t)(buffer[1]));
} else {
LOG_ERROR("Unknown packet");
}
}
}
},
this);
return 0;
}
int IceTransport::DestroyIceTransmission() {
@@ -972,14 +891,10 @@ int IceTransport::SendAudioFrame(const char *data, size_t size) {
int ret = audio_encoder_->Encode(
(uint8_t *)data, size,
[this](char *encoded_audio_buffer, size_t size) -> int {
if (rtp_audio_sender_) {
if (audio_rtp_codec_) {
std::vector<RtpPacket> packets;
audio_rtp_codec_->Encode((uint8_t *)encoded_audio_buffer,
(uint32_t)size, packets);
rtp_audio_sender_->Enqueue(packets);
}
if (audio_channel_send_) {
audio_channel_send_->SendAudio(encoded_audio_buffer, size);
}
return 0;
});
@@ -994,13 +909,8 @@ int IceTransport::SendDataFrame(const char *data, size_t size) {
return -2;
}
std::vector<RtpPacket> packets;
if (rtp_data_sender_) {
if (data_rtp_codec_) {
data_rtp_codec_->Encode((uint8_t *)data, (uint32_t)size, packets);
rtp_data_sender_->Enqueue(packets);
}
if (data_channel_send_) {
data_channel_send_->SendData(data, size);
}
return 0;

View File

@@ -143,6 +143,33 @@ class IceTransport {
uint8_t CheckIsAudioPacket(const char *buffer, size_t size);
uint8_t CheckIsDataPacket(const char *buffer, size_t size);
private:
void InitializeIOStatistics();
void InitializeChannels(RtpPacket::PAYLOAD_TYPE video_codec_payload_type);
void OnIceStateChange(NiceAgent *agent, guint stream_id, guint component_id,
NiceComponentState state, gpointer user_ptr);
void OnNewLocalCandidate(NiceAgent *agent, guint stream_id,
guint component_id, gchar *foundation,
gpointer user_ptr);
void OnGatheringDone(NiceAgent *agent, guint stream_id, gpointer user_ptr);
void OnNewSelectedPair(NiceAgent *agent, guint stream_id, guint component_id,
const char *lfoundation, const char *rfoundation,
gpointer user_ptr);
void OnReceiveBuffer(NiceAgent *agent, guint stream_id, guint component_id,
guint size, gchar *buffer, gpointer user_ptr);
void OnReceiveCompleteFrame(VideoFrame &video_frame);
void OnReceiveCompleteAudio(const char *data, size_t size);
void OnReceiveCompleteData(const char *data, size_t size);
private:
bool use_trickle_ice_ = true;
bool enable_turn_ = false;
@@ -168,7 +195,7 @@ class IceTransport {
void *user_data_ = nullptr;
private:
std::unique_ptr<IceAgent> ice_agent_ = nullptr;
std::shared_ptr<IceAgent> ice_agent_ = nullptr;
bool is_closed_ = false;
std::shared_ptr<WsClient> ice_ws_transport_ = nullptr;
// CongestionControl *congestion_control_ = nullptr;

View File

@@ -206,7 +206,7 @@ target("transport")
target("channel")
set_kind("object")
add_deps("log", "rtp", "rtcp")
add_deps("log", "rtp", "rtcp", "ice")
add_files("src/channel/*.cpp")
add_includedirs("src/rtp", "src/rtcp", {public = true})