YangWX/crossdesk
1
0
Fork 0
mirror of https://github.com/kunkundi/crossdesk.git synced 2025-10-26 12:15:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

[feat] separate rtp send/receive module from ice transport module

Browse Source
This commit is contained in:
dijunkun
2025-01-03 17:34:46 +08:00
parent 7ddcca53e4
commit eef35ff0d4
18 changed files with 384 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/channel/audio_channel_receive.cpp Normal file
View File

@@ -0,0 +1 @@
#include "audio_channel_receive.h"

18
src/channel/audio_channel_receive.h Normal file
View File

@@ -0,0 +1,18 @@
/*
* @Author: DI JUNKUN
* @Date: 2025-01-03
* Copyright (c) 2025 by DI JUNKUN, All Rights Reserved.
*/
#ifndef _AUDIO_CHANNEL_RECEIVE_H_
#define _AUDIO_CHANNEL_RECEIVE_H_
class AudioChannelReceive {
public:
AudioChannelReceive();
~AudioChannelReceive();
public:
};
#endif

1
src/channel/audio_channel_send.cpp Normal file
View File

@@ -0,0 +1 @@
#include "audio_channel_send.h"

18
src/channel/audio_channel_send.h Normal file
View File

@@ -0,0 +1,18 @@
/*
* @Author: DI JUNKUN
* @Date: 2025-01-03
* Copyright (c) 2025 by DI JUNKUN, All Rights Reserved.
*/
#ifndef _AUDIO_CHANNEL_SEND_H_
#define _AUDIO_CHANNEL_SEND_H_
class AudioChannelSend {
public:
AudioChannelSend();
~AudioChannelSend();
public:
};
#endif

1
src/channel/data_channel_receive.cpp Normal file
View File

@@ -0,0 +1 @@
#include "data_channel_receive.h"

18
src/channel/data_channel_receive.h Normal file
View File

@@ -0,0 +1,18 @@
/*
* @Author: DI JUNKUN
* @Date: 2025-01-03
* Copyright (c) 2025 by DI JUNKUN, All Rights Reserved.
*/
#ifndef _DATA_CHANNEL_RECEIVE_H_
#define _DATA_CHANNEL_RECEIVE_H_
class DataChannelReceive {
public:
DataChannelReceive();
~DataChannelReceive();
public:
};
#endif

1
src/channel/data_channel_send.cpp Normal file
View File

@@ -0,0 +1 @@
#include "data_channel_send.h"

18
src/channel/data_channel_send.h Normal file
View File

@@ -0,0 +1,18 @@
/*
* @Author: DI JUNKUN
* @Date: 2025-01-03
* Copyright (c) 2025 by DI JUNKUN, All Rights Reserved.
*/
#ifndef _DATA_CHANNEL_SEND_H_
#define _DATA_CHANNEL_SEND_H_
class DataChannelSend {
public:
DataChannelSend();
~DataChannelSend();
public:
};
#endif

1
src/channel/video_channel_receive.cpp Normal file
View File

@@ -0,0 +1 @@
#include "video_channel_receive.h"

18
src/channel/video_channel_receive.h Normal file
View File

@@ -0,0 +1,18 @@
/*
* @Author: DI JUNKUN
* @Date: 2025-01-03
* Copyright (c) 2025 by DI JUNKUN, All Rights Reserved.
*/
#ifndef _VIDEO_CHANNEL_RECEIVE_H_
#define _VIDEO_CHANNEL_RECEIVE_H_
class VideoChannelReceive {
public:
VideoChannelReceive();
~VideoChannelReceive();
public:
};
#endif

50
src/channel/video_channel_send.cpp Normal file
View File

@@ -0,0 +1,50 @@
#include "video_channel_send.h"
VideoChannelSend::VideoChannelSend() {}
VideoChannelSend::~VideoChannelSend() {}
void VideoChannelSend::Initialize(RtpPacket::PAYLOAD_TYPE negotiated_video_pt) {
video_rtp_codec_ = std::make_unique<RtpCodec>(negotiated_video_pt);
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();
}
void VideoChannelSend::Destroy() {
if (rtp_video_sender_) {
rtp_video_sender_->Stop();
}
}
int VideoChannelSend::SendVideo(char *encoded_frame, 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);
}
rtp_video_sender_->Enqueue(packets);
}
return 0;
}

29
src/channel/video_channel_send.h Normal file
View File

@@ -0,0 +1,29 @@
/*
* @Author: DI JUNKUN
* @Date: 2025-01-03
* Copyright (c) 2025 by DI JUNKUN, All Rights Reserved.
*/
#ifndef _VIDEO_CHANNEL_SEND_H_
#define _VIDEO_CHANNEL_SEND_H_
#include "rtp_codec.h"
#include "rtp_video_sender.h"
class VideoChannelSend {
public:
VideoChannelSend();
~VideoChannelSend();
public:
void Initialize(RtpPacket::PAYLOAD_TYPE negotiated_video_pt);
void Destroy();
int SendVideo(char *encoded_frame, size_t size);
private:
std::unique_ptr<RtpCodec> video_rtp_codec_ = nullptr;
std::unique_ptr<RtpVideoSender> rtp_video_sender_ = nullptr;
};
#endif

113
src/pc/peer_connection.cpp
View File

@@ -140,24 +140,24 @@ int PeerConnection::Init(PeerConnectionParams params,
user_id.size(), user_data_);
} else if ("connected" == ice_status) {
// std::string transmission_id = std::string(user_id, user_id_size);
// is_ice_transmission_ready_[user_id] = true;
// is_ice_transport_ready_[user_id] = true;
// on_connection_status_(ConnectionStatus::Connected, user_id.data(),
// user_id.size(), user_data_);
// b_force_i_frame_ = true;
LOG_INFO("Ice connected");
} else if ("ready" == ice_status) {
is_ice_transmission_ready_[user_id] = true;
is_ice_transport_ready_[user_id] = true;
b_force_i_frame_ = true;
LOG_INFO("Ice ready");
on_connection_status_(ConnectionStatus::Connected, user_id.data(),
user_id.size(), user_data_);
} else if ("closed" == ice_status) {
is_ice_transmission_ready_[user_id] = false;
is_ice_transport_ready_[user_id] = false;
LOG_INFO("Ice closed");
on_connection_status_(ConnectionStatus::Closed, user_id.data(),
user_id.size(), user_data_);
} else if ("failed" == ice_status) {
is_ice_transmission_ready_[user_id] = false;
is_ice_transport_ready_[user_id] = false;
if (offer_peer_ && try_rejoin_with_turn_) {
enable_turn_ = true;
reliable_ice_ = false;
@@ -179,7 +179,7 @@ int PeerConnection::Init(PeerConnectionParams params,
user_id.size(), user_data_);
}
} else {
is_ice_transmission_ready_[user_id] = false;
is_ice_transport_ready_[user_id] = false;
LOG_INFO("Unknown ice state [{}]", ice_status);
}
};
@@ -298,7 +298,7 @@ int PeerConnection::Leave(const std::string &transmission_id) {
transmission_id);
}
is_ice_transmission_ready_[user_id_] = false;
is_ice_transport_ready_[user_id_] = false;
leave_ = true;
ReleaseAllIceTransmission();
@@ -306,11 +306,11 @@ int PeerConnection::Leave(const std::string &transmission_id) {
}
int PeerConnection::ReleaseAllIceTransmission() {
for (auto &user_id_it : ice_transmission_list_) {
for (auto &user_id_it : ice_transport_list_) {
user_id_it.second->DestroyIceTransmission();
}
ice_transmission_list_.clear();
is_ice_transmission_ready_.clear();
ice_transport_list_.clear();
is_ice_transport_ready_.clear();
return 0;
}
@@ -349,12 +349,12 @@ SignalStatus PeerConnection::GetSignalStatus() {
}
int PeerConnection::SendVideoFrame(const XVideoFrame *video_frame) {
if (ice_transmission_list_.empty()) {
if (ice_transport_list_.empty()) {
return -1;
}
for (auto &ice_trans : ice_transmission_list_) {
if (!is_ice_transmission_ready_[ice_trans.first]) {
for (auto &ice_trans : ice_transport_list_) {
if (!is_ice_transport_ready_[ice_trans.first]) {
continue;
}
@@ -365,12 +365,12 @@ int PeerConnection::SendVideoFrame(const XVideoFrame *video_frame) {
}
int PeerConnection::SendAudioFrame(const char *data, size_t size) {
if (ice_transmission_list_.empty()) {
if (ice_transport_list_.empty()) {
return -1;
}
for (auto &ice_trans : ice_transmission_list_) {
if (!is_ice_transmission_ready_[ice_trans.first]) {
for (auto &ice_trans : ice_transport_list_) {
if (!is_ice_transport_ready_[ice_trans.first]) {
continue;
}
ice_trans.second->SendAudioFrame(data, size);
@@ -380,8 +380,8 @@ int PeerConnection::SendAudioFrame(const char *data, size_t size) {
}
int PeerConnection::SendDataFrame(const char *data, size_t size) {
for (auto &ice_trans : ice_transmission_list_) {
if (!is_ice_transmission_ready_[ice_trans.first]) {
for (auto &ice_trans : ice_transport_list_) {
if (!is_ice_transport_ready_[ice_trans.first]) {
continue;
}
ice_trans.second->SendDataFrame(data, size);
@@ -594,31 +594,30 @@ void PeerConnection::ProcessIceWorkMsg(const IceWorkMsg &msg) {
LOG_INFO("]");
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_ice_status_change_, user_data_);
ice_transport_list_[remote_user_id] = std::make_unique<IceTransport>(
true, transmission_id, user_id_, remote_user_id, ws_transport_,
on_ice_status_change_, user_data_);
ice_transmission_list_[remote_user_id]->SetLocalCapabilities(
ice_transport_list_[remote_user_id]->SetLocalCapabilities(
hardware_acceleration_, trickle_ice_, reliable_ice_, enable_turn_,
false, video_payload_types_, audio_payload_types_);
ice_transmission_list_[remote_user_id]->SetOnReceiveVideoFunc(
ice_transport_list_[remote_user_id]->SetOnReceiveVideoFunc(
on_receive_video_frame_);
ice_transmission_list_[remote_user_id]->SetOnReceiveAudioFunc(
ice_transport_list_[remote_user_id]->SetOnReceiveAudioFunc(
on_receive_audio_buffer_);
ice_transmission_list_[remote_user_id]->SetOnReceiveDataFunc(
ice_transport_list_[remote_user_id]->SetOnReceiveDataFunc(
on_receive_data_buffer_);
ice_transmission_list_[remote_user_id]->SetOnReceiveNetStatusReportFunc(
ice_transport_list_[remote_user_id]->SetOnReceiveNetStatusReportFunc(
net_status_report_);
ice_transmission_list_[remote_user_id]->InitIceTransmission(
ice_transport_list_[remote_user_id]->InitIceTransmission(
cfg_stun_server_ip_, stun_server_port_, cfg_turn_server_ip_,
turn_server_port_, cfg_turn_server_username_,
cfg_turn_server_password_,
av1_encoding_ ? RtpPacket::AV1 : RtpPacket::H264);
ice_transmission_list_[remote_user_id]->JoinTransmission();
ice_transport_list_[remote_user_id]->JoinTransmission();
}
break;
@@ -627,11 +626,11 @@ void PeerConnection::ProcessIceWorkMsg(const IceWorkMsg &msg) {
std::string user_id = msg.user_id;
LOG_INFO("Receive notification: user id [{}] leave transmission",
user_id);
auto user_id_it = ice_transmission_list_.find(user_id);
if (user_id_it != ice_transmission_list_.end()) {
auto user_id_it = ice_transport_list_.find(user_id);
if (user_id_it != ice_transport_list_.end()) {
user_id_it->second->DestroyIceTransmission();
ice_transmission_list_.erase(user_id_it);
is_ice_transmission_ready_[user_id] = false;
ice_transport_list_.erase(user_id_it);
is_ice_transport_ready_[user_id] = false;
LOG_INFO("Terminate transmission to user [{}]", user_id);
}
break;
@@ -639,40 +638,37 @@ void PeerConnection::ProcessIceWorkMsg(const IceWorkMsg &msg) {
case IceWorkMsg::Type::Offer: {
std::string transmission_id = msg.transmission_id;
std::string remote_user_id = msg.remote_user_id;
if (ice_transmission_list_.end() ==
ice_transmission_list_.find(remote_user_id)) {
if (ice_transport_list_.end() ==
ice_transport_list_.find(remote_user_id)) {
// Enable TURN for answer peer by default
ice_transmission_list_[remote_user_id] =
std::make_unique<IceTransmission>(
false, transmission_id, user_id_, remote_user_id, ws_transport_,
on_ice_status_change_, user_data_);
ice_transport_list_[remote_user_id] = std::make_unique<IceTransport>(
false, transmission_id, user_id_, remote_user_id, ws_transport_,
on_ice_status_change_, user_data_);
ice_transmission_list_[remote_user_id]->SetLocalCapabilities(
ice_transport_list_[remote_user_id]->SetLocalCapabilities(
hardware_acceleration_, trickle_ice_, reliable_ice_, enable_turn_,
false, video_payload_types_, audio_payload_types_);
ice_transmission_list_[remote_user_id]->SetOnReceiveVideoFunc(
ice_transport_list_[remote_user_id]->SetOnReceiveVideoFunc(
on_receive_video_frame_);
ice_transmission_list_[remote_user_id]->SetOnReceiveAudioFunc(
ice_transport_list_[remote_user_id]->SetOnReceiveAudioFunc(
on_receive_audio_buffer_);
ice_transmission_list_[remote_user_id]->SetOnReceiveDataFunc(
ice_transport_list_[remote_user_id]->SetOnReceiveDataFunc(
on_receive_data_buffer_);
ice_transmission_list_[remote_user_id]->SetOnReceiveNetStatusReportFunc(
ice_transport_list_[remote_user_id]->SetOnReceiveNetStatusReportFunc(
net_status_report_);
ice_transmission_list_[remote_user_id]->InitIceTransmission(
ice_transport_list_[remote_user_id]->InitIceTransmission(
cfg_stun_server_ip_, stun_server_port_, cfg_turn_server_ip_,
turn_server_port_, cfg_turn_server_username_,
cfg_turn_server_password_,
av1_encoding_ ? RtpPacket::AV1 : RtpPacket::H264);
ice_transmission_list_[remote_user_id]->SetTransmissionId(
transmission_id);
ice_transport_list_[remote_user_id]->SetTransmissionId(transmission_id);
}
std::string remote_sdp = msg.remote_sdp;
int ret =
ice_transmission_list_[remote_user_id]->SetRemoteSdp(remote_sdp);
int ret = ice_transport_list_[remote_user_id]->SetRemoteSdp(remote_sdp);
if (0 != ret) {
NegotiationFailed();
break;
@@ -680,19 +676,18 @@ void PeerConnection::ProcessIceWorkMsg(const IceWorkMsg &msg) {
if (trickle_ice_) {
sdp_without_cands_ = remote_sdp;
ice_transmission_list_[remote_user_id]->SendAnswer();
ice_transport_list_[remote_user_id]->SendAnswer();
}
ice_transmission_list_[remote_user_id]->GatherCandidates();
ice_transport_list_[remote_user_id]->GatherCandidates();
break;
}
case IceWorkMsg::Type::Answer: {
std::string remote_user_id = msg.remote_user_id;
std::string remote_sdp = msg.remote_sdp;
if (ice_transmission_list_.find(remote_user_id) !=
ice_transmission_list_.end()) {
int ret =
ice_transmission_list_[remote_user_id]->SetRemoteSdp(remote_sdp);
if (ice_transport_list_.find(remote_user_id) !=
ice_transport_list_.end()) {
int ret = ice_transport_list_[remote_user_id]->SetRemoteSdp(remote_sdp);
if (0 != ret) {
Leave(remote_transmission_id_);
break;
@@ -700,7 +695,7 @@ void PeerConnection::ProcessIceWorkMsg(const IceWorkMsg &msg) {
if (trickle_ice_) {
sdp_without_cands_ = remote_sdp;
ice_transmission_list_[remote_user_id]->GatherCandidates();
ice_transport_list_[remote_user_id]->GatherCandidates();
}
}
@@ -714,10 +709,10 @@ void PeerConnection::ProcessIceWorkMsg(const IceWorkMsg &msg) {
// LOG_INFO("[{}] receive new candidate from [{}]:[{}]", user_id_,
// remote_user_id, new_candidate);
if (ice_transmission_list_.find(remote_user_id) !=
ice_transmission_list_.end()) {
ice_transmission_list_[remote_user_id]->SetRemoteSdp(
sdp_without_cands_ + new_candidate);
if (ice_transport_list_.find(remote_user_id) !=
ice_transport_list_.end()) {
ice_transport_list_[remote_user_id]->SetRemoteSdp(sdp_without_cands_ +
new_candidate);
}
break;
}

7
src/pc/peer_connection.h
View File

@@ -13,7 +13,7 @@
#include "audio_decoder.h"
#include "audio_encoder.h"
#include "ice_transmission.h"
#include "ice_transport.h"
#include "video_decoder_factory.h"
#include "video_encoder_factory.h"
#include "ws_client.h"
@@ -169,9 +169,8 @@ class PeerConnection {
std::string sdp_without_cands_ = "";
private:
std::map<std::string, std::unique_ptr<IceTransmission>>
ice_transmission_list_;
std::map<std::string, bool> is_ice_transmission_ready_;
std::map<std::string, std::unique_ptr<IceTransport>> ice_transport_list_;
std::map<std::string, bool> is_ice_transport_ready_;
std::function<void(std::string, const std::string &)> on_ice_status_change_ =
nullptr;

4
src/rtp/rtp_packet/rtp_codec.cpp
View File

@@ -31,7 +31,7 @@ uint32_t GenerateUniqueSsrc() {
return new_ssrc;
}
RtpCodec ::RtpCodec(RtpPacket::PAYLOAD_TYPE payload_type)
RtpCodec::RtpCodec(RtpPacket::PAYLOAD_TYPE payload_type)
: version_(RTP_VERSION),
has_padding_(false),
has_extension_(false),
@@ -41,7 +41,7 @@ RtpCodec ::RtpCodec(RtpPacket::PAYLOAD_TYPE payload_type)
fec_encoder_.Init();
}
RtpCodec ::~RtpCodec() {
RtpCodec::~RtpCodec() {
SSRCManager::Instance().DeleteSsrc(ssrc_);
if (extension_data_) {

239
src/transmission/ice_transmission.cpp → src/transport/ice_transport.cpp
View File

@@ -1,4 +1,4 @@
#include "ice_transmission.h"
#include "ice_transport.h"
#include <chrono>
#include <map>
@@ -14,7 +14,7 @@
using nlohmann::json;
IceTransmission::IceTransmission(
IceTransport::IceTransport(
bool offer_peer, std::string &transmission_id, std::string &user_id,
std::string &remote_user_id, std::shared_ptr<WsClient> ice_ws_transmission,
std::function<void(std::string, const std::string &)> on_ice_status_change,
@@ -27,7 +27,7 @@ IceTransmission::IceTransmission(
on_ice_status_change_(on_ice_status_change),
user_data_(user_data) {}
IceTransmission::~IceTransmission() {
IceTransport::~IceTransport() {
user_data_ = nullptr;
video_codec_inited_ = false;
audio_codec_inited_ = false;
@@ -41,10 +41,12 @@ IceTransmission::~IceTransmission() {
#endif
}
int IceTransmission::SetLocalCapabilities(
bool hardware_acceleration, bool use_trickle_ice, bool use_reliable_ice,
bool enable_turn, bool force_turn, std::vector<int> &video_payload_types,
std::vector<int> &audio_payload_types) {
int IceTransport::SetLocalCapabilities(bool hardware_acceleration,
bool use_trickle_ice,
bool use_reliable_ice, bool enable_turn,
bool force_turn,
std::vector<int> &video_payload_types,
std::vector<int> &audio_payload_types) {
hardware_acceleration_ = hardware_acceleration;
use_trickle_ice_ = use_trickle_ice;
use_reliable_ice_ = use_reliable_ice;
@@ -56,7 +58,7 @@ int IceTransmission::SetLocalCapabilities(
return 0;
}
int IceTransmission::InitIceTransmission(
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) {
@@ -246,56 +248,50 @@ int IceTransmission::InitIceTransmission(
[]([[maybe_unused]] NiceAgent *agent, [[maybe_unused]] guint stream_id,
[[maybe_unused]] guint component_id, NiceComponentState state,
gpointer user_ptr) {
if (user_ptr) {
IceTransmission *ice_transmission_obj =
static_cast<IceTransmission *>(user_ptr);
if (!ice_transmission_obj->is_closed_) {
LOG_INFO("[{}->{}] state_change: {}",
ice_transmission_obj->user_id_,
ice_transmission_obj->remote_user_id_,
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_transmission_obj->state_ = state;
ice_transport->state_ = state;
if (state == NICE_COMPONENT_STATE_READY ||
state == NICE_COMPONENT_STATE_CONNECTED) {
ice_transmission_obj->ice_io_statistics_->Start();
ice_transport->ice_io_statistics_->Start();
}
ice_transmission_obj->on_ice_status_change_(
ice_transport->on_ice_status_change_(
nice_component_state_to_string(state),
ice_transmission_obj->remote_user_id_);
ice_transport->remote_user_id_);
}
}
},
[](NiceAgent *agent, guint stream_id, guint component_id,
gchar *foundation, gpointer user_ptr) {
if (user_ptr) {
IceTransmission *ice_transmission_obj =
static_cast<IceTransmission *>(user_ptr);
if (ice_transmission_obj->use_trickle_ice_) {
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_transmission_obj->new_local_candidate_ =
ice_transport->new_local_candidate_ =
nice_agent_generate_local_candidate_sdp(agent, cand);
json message = {
{"type", "new_candidate"},
{"transmission_id", ice_transmission_obj->transmission_id_},
{"user_id", ice_transmission_obj->user_id_},
{"remote_user_id", ice_transmission_obj->remote_user_id_},
{"sdp", ice_transmission_obj->new_local_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_transmission_obj->user_id_,
// ice_transmission_obj->remote_user_id_,
// ice_transmission_obj->new_local_candidate_);
// ice_transport->user_id_,
// ice_transport->remote_user_id_,
// ice_transport->new_local_candidate_);
if (ice_transmission_obj->ice_ws_transport_) {
ice_transmission_obj->ice_ws_transport_->Send(message.dump());
if (ice_transport->ice_ws_transport_) {
ice_transport->ice_ws_transport_->Send(message.dump());
}
}
}
@@ -307,17 +303,15 @@ int IceTransmission::InitIceTransmission(
[]([[maybe_unused]] NiceAgent *agent, [[maybe_unused]] guint stream_id,
gpointer user_ptr) {
// non-trickle
if (user_ptr) {
IceTransmission *ice_transmission_obj =
static_cast<IceTransmission *>(user_ptr);
LOG_INFO("[{}->{}] gather_done", ice_transmission_obj->user_id_,
ice_transmission_obj->remote_user_id_);
if (auto ice_transport = static_cast<IceTransport *>(user_ptr)) {
LOG_INFO("[{}->{}] gather_done", ice_transport->user_id_,
ice_transport->remote_user_id_);
if (!ice_transmission_obj->use_trickle_ice_) {
if (ice_transmission_obj->offer_peer_) {
ice_transmission_obj->SendOffer();
if (!ice_transport->use_trickle_ice_) {
if (ice_transport->offer_peer_) {
ice_transport->SendOffer();
} else {
ice_transmission_obj->SendAnswer();
ice_transport->SendAnswer();
}
}
}
@@ -329,49 +323,43 @@ int IceTransmission::InitIceTransmission(
NiceCandidate *remote = nullptr;
nice_agent_get_selected_pair(agent, stream_id, component_id, &local,
&remote);
if (user_ptr) {
IceTransmission *ice_transmission_obj =
static_cast<IceTransmission *>(user_ptr);
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_transmission_obj->traversal_type_ = TraversalType::TRelay;
ice_transport->traversal_type_ = TraversalType::TRelay;
} else {
LOG_INFO("Traversal using p2p");
ice_transmission_obj->traversal_type_ = TraversalType::TP2P;
ice_transport->traversal_type_ = TraversalType::TP2P;
}
XNetTrafficStats net_traffic_stats;
memset(&net_traffic_stats, 0, sizeof(net_traffic_stats));
ice_transmission_obj->on_receive_net_status_report_(
ice_transmission_obj->user_id_.data(),
ice_transmission_obj->user_id_.size(),
TraversalMode(ice_transmission_obj->traversal_type_),
&net_traffic_stats, ice_transmission_obj->remote_user_id_.data(),
ice_transmission_obj->remote_user_id_.size(),
ice_transmission_obj->user_data_);
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 (user_ptr) {
IceTransmission *ice_transmission_obj =
static_cast<IceTransmission *>(user_ptr);
if (ice_transmission_obj && !ice_transmission_obj->is_closed_) {
if (ice_transmission_obj->CheckIsVideoPacket(buffer, size)) {
RtpPacket packet((uint8_t *)buffer, size);
ice_transmission_obj->rtp_video_receiver_->InsertRtpPacket(
packet);
} else if (ice_transmission_obj->CheckIsAudioPacket(buffer, size)) {
RtpPacket packet((uint8_t *)buffer, size);
ice_transmission_obj->rtp_audio_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]));
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");
}
@@ -382,7 +370,7 @@ int IceTransmission::InitIceTransmission(
return 0;
}
int IceTransmission::DestroyIceTransmission() {
int IceTransport::DestroyIceTransmission() {
LOG_INFO("[{}->{}] Destroy ice transmission", user_id_, remote_user_id_);
is_closed_ = true;
@@ -413,15 +401,15 @@ int IceTransmission::DestroyIceTransmission() {
return ice_agent_->DestroyIceAgent();
}
int IceTransmission::CreateMediaCodec() {
int IceTransport::CreateMediaCodec() {
video_rtp_codec_ = std::make_unique<RtpCodec>(negotiated_video_pt_);
audio_rtp_codec_ = std::make_unique<RtpCodec>(negotiated_audio_pt_);
data_rtp_codec_ = std::make_unique<RtpCodec>(negotiated_data_pt_);
return 0;
}
int IceTransmission::CreateVideoCodec(RtpPacket::PAYLOAD_TYPE video_pt,
bool hardware_acceleration) {
int IceTransport::CreateVideoCodec(RtpPacket::PAYLOAD_TYPE video_pt,
bool hardware_acceleration) {
if (video_codec_inited_) {
return 0;
}
@@ -493,7 +481,7 @@ int IceTransmission::CreateVideoCodec(RtpPacket::PAYLOAD_TYPE video_pt,
return 0;
}
int IceTransmission::CreateAudioCodec() {
int IceTransport::CreateAudioCodec() {
if (audio_codec_inited_) {
return 0;
}
@@ -517,13 +505,13 @@ int IceTransmission::CreateAudioCodec() {
return 0;
}
int IceTransmission::SetTransmissionId(const std::string &transmission_id) {
int IceTransport::SetTransmissionId(const std::string &transmission_id) {
transmission_id_ = transmission_id;
return 0;
}
int IceTransmission::JoinTransmission() {
int IceTransport::JoinTransmission() {
LOG_INFO("[{}] Join transmission", user_id_);
if (use_trickle_ice_) {
@@ -534,7 +522,7 @@ int IceTransmission::JoinTransmission() {
return 0;
}
int IceTransmission::GatherCandidates() {
int IceTransport::GatherCandidates() {
int ret = ice_agent_->GatherCandidates();
if (ret < 0) {
LOG_ERROR("Gather candidates failed");
@@ -542,7 +530,7 @@ int IceTransmission::GatherCandidates() {
return 0;
}
int IceTransmission::SetRemoteSdp(const std::string &remote_sdp) {
int IceTransport::SetRemoteSdp(const std::string &remote_sdp) {
std::string media_stream_sdp = GetRemoteCapabilities(remote_sdp);
if (media_stream_sdp.empty()) {
LOG_ERROR("Set remote sdp failed due to negotiation failed");
@@ -556,7 +544,7 @@ int IceTransmission::SetRemoteSdp(const std::string &remote_sdp) {
return 0;
}
int IceTransmission::SendOffer() {
int IceTransport::SendOffer() {
local_sdp_ = ice_agent_->GenerateLocalSdp();
AppendLocalCapabilitiesToOffer(local_sdp_);
json message = {{"type", "offer"},
@@ -572,7 +560,7 @@ int IceTransmission::SendOffer() {
return 0;
}
int IceTransmission::SendAnswer() {
int IceTransport::SendAnswer() {
local_sdp_ = ice_agent_->GenerateLocalSdp();
AppendLocalCapabilitiesToAnswer(local_sdp_);
json message = {{"type", "answer"},
@@ -589,7 +577,7 @@ int IceTransmission::SendAnswer() {
return 0;
}
int IceTransmission::AppendLocalCapabilitiesToOffer(
int IceTransport::AppendLocalCapabilitiesToOffer(
const std::string &remote_sdp) {
std::string preferred_video_pt;
std::string to_replace = "ICE/SDP";
@@ -647,7 +635,7 @@ int IceTransmission::AppendLocalCapabilitiesToOffer(
return 0;
}
int IceTransmission::AppendLocalCapabilitiesToAnswer(
int IceTransport::AppendLocalCapabilitiesToAnswer(
const std::string &remote_sdp) {
std::string negotiated_video_pt;
std::string negotiated_audio_pt;
@@ -692,8 +680,7 @@ int IceTransmission::AppendLocalCapabilitiesToAnswer(
return 0;
}
std::string IceTransmission::GetRemoteCapabilities(
const std::string &remote_sdp) {
std::string IceTransport::GetRemoteCapabilities(const std::string &remote_sdp) {
std::string media_stream_sdp;
std::size_t video_start = remote_sdp.find("m=video");
std::size_t video_end = remote_sdp.find("m=audio");
@@ -745,7 +732,7 @@ std::string IceTransmission::GetRemoteCapabilities(
return media_stream_sdp;
}
bool IceTransmission::NegotiateVideoPayloadType(const std::string &remote_sdp) {
bool IceTransport::NegotiateVideoPayloadType(const std::string &remote_sdp) {
std::string remote_video_capabilities;
std::string local_video_capabilities;
std::string remote_prefered_video_pt;
@@ -817,7 +804,7 @@ bool IceTransmission::NegotiateVideoPayloadType(const std::string &remote_sdp) {
}
}
bool IceTransmission::NegotiateAudioPayloadType(const std::string &remote_sdp) {
bool IceTransport::NegotiateAudioPayloadType(const std::string &remote_sdp) {
std::string remote_audio_capabilities;
std::string remote_prefered_audio_pt;
@@ -877,7 +864,7 @@ bool IceTransmission::NegotiateAudioPayloadType(const std::string &remote_sdp) {
}
}
bool IceTransmission::NegotiateDataPayloadType(const std::string &remote_sdp) {
bool IceTransport::NegotiateDataPayloadType(const std::string &remote_sdp) {
std::string remote_data_capabilities;
std::string remote_prefered_data_pt;
@@ -937,12 +924,11 @@ bool IceTransmission::NegotiateDataPayloadType(const std::string &remote_sdp) {
}
}
std::vector<RtpPacket::PAYLOAD_TYPE>
IceTransmission::GetNegotiatedCapabilities() {
std::vector<RtpPacket::PAYLOAD_TYPE> IceTransport::GetNegotiatedCapabilities() {
return {negotiated_video_pt_, negotiated_audio_pt_, negotiated_data_pt_};
}
int IceTransmission::SendVideoFrame(const XVideoFrame *video_frame) {
int IceTransport::SendVideoFrame(const XVideoFrame *video_frame) {
if (state_ != NICE_COMPONENT_STATE_CONNECTED &&
state_ != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
@@ -960,14 +946,8 @@ int IceTransmission::SendVideoFrame(const XVideoFrame *video_frame) {
video_frame,
[this](char *encoded_frame, size_t size,
VideoEncoder::VideoFrameType frame_type) -> int {
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);
}
rtp_video_sender_->Enqueue(packets);
if (video_channel_send_) {
video_channel_send_->SendVideo(encoded_frame, size);
}
return 0;
@@ -981,7 +961,7 @@ int IceTransmission::SendVideoFrame(const XVideoFrame *video_frame) {
return 0;
}
int IceTransmission::SendAudioFrame(const char *data, size_t size) {
int IceTransport::SendAudioFrame(const char *data, size_t size) {
if (state_ != NICE_COMPONENT_STATE_CONNECTED &&
state_ != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
@@ -1006,7 +986,7 @@ int IceTransmission::SendAudioFrame(const char *data, size_t size) {
return ret;
}
int IceTransmission::SendDataFrame(const char *data, size_t size) {
int IceTransport::SendDataFrame(const char *data, size_t size) {
if (state_ != NICE_COMPONENT_STATE_CONNECTED &&
state_ != NICE_COMPONENT_STATE_READY) {
LOG_ERROR("Ice is not connected, state = [{}]",
@@ -1026,41 +1006,34 @@ int IceTransmission::SendDataFrame(const char *data, size_t size) {
return 0;
}
uint8_t IceTransmission::CheckIsRtcpPacket(const char *buffer, size_t size) {
if (size < 4) {
uint8_t IceTransport::CheckIsRtpPacket(const char *buffer, size_t size) {
if (size < 2) {
return 0;
}
uint8_t v = (buffer[0] >> 6) & 0x03;
if (2 != v) {
uint8_t payload_type = buffer[1] & 0x7F;
if (payload_type >= 96 && payload_type <= 127) {
return payload_type;
} else {
return 0;
}
uint8_t pt = buffer[1];
switch (pt) {
case RtcpHeader::PAYLOAD_TYPE::SR: {
return pt;
}
case RtcpHeader::PAYLOAD_TYPE::RR: {
return pt;
}
case RtcpHeader::PAYLOAD_TYPE::SDES: {
return pt;
}
case RtcpHeader::PAYLOAD_TYPE::BYE: {
return pt;
}
case RtcpHeader::PAYLOAD_TYPE::APP: {
return pt;
}
default: {
return 0;
}
}
}
uint8_t IceTransmission::CheckIsVideoPacket(const char *buffer, size_t size) {
uint8_t IceTransport::CheckIsRtcpPacket(const char *buffer, size_t size) {
if (size < 2) {
return 0;
}
uint8_t payload_type = buffer[1];
if (payload_type >= 192 && payload_type <= 223) {
return payload_type;
;
}
return 0;
}
uint8_t IceTransport::CheckIsVideoPacket(const char *buffer, size_t size) {
if (size < 4) {
return 0;
}
@@ -1081,7 +1054,7 @@ uint8_t IceTransmission::CheckIsVideoPacket(const char *buffer, size_t size) {
}
}
uint8_t IceTransmission::CheckIsAudioPacket(const char *buffer, size_t size) {
uint8_t IceTransport::CheckIsAudioPacket(const char *buffer, size_t size) {
if (size < 4) {
return 0;
}
@@ -1099,7 +1072,7 @@ uint8_t IceTransmission::CheckIsAudioPacket(const char *buffer, size_t size) {
}
}
uint8_t IceTransmission::CheckIsDataPacket(const char *buffer, size_t size) {
uint8_t IceTransport::CheckIsDataPacket(const char *buffer, size_t size) {
if (size < 4) {
return 0;
}

42
src/transmission/ice_transmission.h → src/transport/ice_transport.h
View File

@@ -1,17 +1,21 @@
/*
* @Author: DI JUNKUN
* @Date: 2023-11-24
* Copyright (c) 2023 by DI JUNKUN, All Rights Reserved.
* @Date: 2025-01-03
* Copyright (c) 2025 by DI JUNKUN, All Rights Reserved.
*/
#ifndef _ICE_TRANSMISSION_H_
#define _ICE_TRANSMISSION_H_
#ifndef _ICE_TRANSPORT_H_
#define _ICE_TRANSPORT_H_
#include <iostream>
#include "audio_decoder.h"
#include "audio_encoder.h"
#include "congestion_control.h"
// #include "congestion_control.h"
#include "audio_channel_receive.h"
#include "audio_channel_send.h"
#include "data_channel_receive.h"
#include "data_channel_send.h"
#include "ice_agent.h"
#include "io_statistics.h"
#include "ringbuffer.h"
@@ -23,11 +27,13 @@
#include "rtp_packet.h"
#include "rtp_video_receiver.h"
#include "rtp_video_sender.h"
#include "video_channel_receive.h"
#include "video_channel_send.h"
#include "video_decoder_factory.h"
#include "video_encoder_factory.h"
#include "ws_client.h"
class IceTransmission {
class IceTransport {
public:
typedef enum { VIDEO = 96, AUDIO = 97, DATA = 127 } DATA_TYPE;
typedef enum { H264 = 96, AV1 = 99 } VIDEO_TYPE;
@@ -40,13 +46,13 @@ class IceTransmission {
enum TraversalType { TP2P = 0, TRelay = 1, TUnknown = 2 };
public:
IceTransmission(bool offer_peer, std::string &transmission_id,
std::string &user_id, std::string &remote_user_id,
std::shared_ptr<WsClient> ice_ws_transmission,
std::function<void(std::string, const std::string &)>
on_ice_status_change,
void *user_data);
~IceTransmission();
IceTransport(bool offer_peer, std::string &transmission_id,
std::string &user_id, std::string &remote_user_id,
std::shared_ptr<WsClient> ice_ws_transmission,
std::function<void(std::string, const std::string &)>
on_ice_status_change,
void *user_data);
~IceTransport();
public:
int SetLocalCapabilities(bool hardware_acceleration, bool use_trickle_ice,
@@ -131,6 +137,7 @@ class IceTransmission {
int CreateAudioCodec();
private:
uint8_t CheckIsRtpPacket(const char *buffer, size_t size);
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);
@@ -164,7 +171,7 @@ class IceTransmission {
std::unique_ptr<IceAgent> ice_agent_ = nullptr;
bool is_closed_ = false;
std::shared_ptr<WsClient> ice_ws_transport_ = nullptr;
CongestionControl *congestion_control_ = nullptr;
// CongestionControl *congestion_control_ = nullptr;
std::function<void(const XVideoFrame *, const char *, const size_t, void *)>
on_receive_video_ = nullptr;
std::function<void(const char *, size_t, const char *, const size_t, void *)>
@@ -181,6 +188,13 @@ class IceTransmission {
on_receive_net_status_report_ = nullptr;
private:
std::unique_ptr<VideoChannelSend> video_channel_send_ = nullptr;
std::unique_ptr<VideoChannelReceive> video_channel_receive_ = nullptr;
std::unique_ptr<AudioChannelSend> audio_channel_send_ = nullptr;
std::unique_ptr<AudioChannelReceive> audio_channel_receive_ = nullptr;
std::unique_ptr<DataChannelSend> data_channel_send_ = nullptr;
std::unique_ptr<DataChannelReceive> data_channel_receive_ = nullptr;
std::unique_ptr<RtpCodec> video_rtp_codec_ = nullptr;
std::unique_ptr<RtpCodec> audio_rtp_codec_ = nullptr;
std::unique_ptr<RtpCodec> data_rtp_codec_ = nullptr;

23
xmake.lua
View File

@@ -192,18 +192,29 @@ target("qos")
add_files("src/qos/*.cpp")
add_includedirs("src/qos", {public = true})
-- target("transport")
-- set_kind("object")
-- add_deps("log", "ws", "ice", "qos", "rtp", "rtcp", "statistics", "media")
-- add_files("src/transport/*.cpp")
-- add_includedirs("src/ws", "src/ice", "src/qos", {public = true})
target("transmission")
target("transport")
set_kind("object")
add_deps("log", "ws", "ice", "qos", "rtp", "rtcp", "statistics", "media")
add_files("src/transmission/*.cpp")
add_includedirs("src/ws", "src/ice", "src/qos", {public = true})
add_deps("log", "ws", "ice", "channel", "rtp", "rtcp", "statistics", "media")
add_files("src/transport/*.cpp")
add_includedirs("src/ws", "src/ice", "src/channel", {public = true})
target("channel")
set_kind("object")
add_deps("log", "rtp", "rtcp")
add_files("src/channel/*.cpp")
add_includedirs("src/rtp", "src/rtcp", {public = true})
target("pc")
set_kind("object")
add_deps("log", "ws", "ice", "transmission", "inih", "common")
add_deps("log", "ws", "ice", "transport", "inih", "common")
add_files("src/pc/*.cpp")
add_includedirs("src/transmission", "src/interface", {public = true})
add_includedirs("src/transport", "src/interface", {public = true})
target("projectx")
set_kind("static")