THENG
/
ZLMediaKit
mirror of https://gitcode.com/gh_mirrors/zlme/ZLMediaKit.git
6
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:ZLMediaKit/ZLMediaKit into feature/transcode

This commit is contained in:
cqm 2023-03-07 17:20:34 +08:00
parent 8e55e5dee4 23296ae5fa
commit 06ddad0512
64 changed files with 3981 additions and 3917 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
3rdpart/ZLToolKit

@ -1 +1 @@
Subproject commit 57901f9d341478378b1526f7efe99ebc79b2ddb5
Subproject commit 7d52e11ae4e8d6d2c20aa0349dbfea8a1d82a968

1
AUTHORS
View File

@ -66,3 +66,4 @@ WuPeng <wp@zafu.edu.cn>
[KevinZang](https://github.com/ZSC714725)
[gongluck](https://github.com/gongluck)
[a-ucontrol](https://github.com/a-ucontrol)
[TalusL](https://github.com/TalusL)

1
README.md
View File

@ -291,6 +291,7 @@ bash build_docker_images.sh
[KevinZang](https://github.com/ZSC714725)
[gongluck](https://github.com/gongluck)
[a-ucontrol](https://github.com/a-ucontrol)
[TalusL](https://github.com/TalusL)
## 使用案例

3
api/CMakeLists.txt
View File

@ -81,4 +81,5 @@ install(FILES ${API_HEADER_LIST}
DESTINATION ${INSTALL_PATH_INCLUDE})
install(TARGETS mk_api
ARCHIVE DESTINATION ${INSTALL_PATH_LIB}
LIBRARY DESTINATION ${INSTALL_PATH_LIB})
LIBRARY DESTINATION ${INSTALL_PATH_LIB}
RUNTIME DESTINATION ${INSTALL_PATH_RUNTIME})

23
api/include/mk_events_objects.h
View File

@ -18,7 +18,7 @@ extern "C" {
///////////////////////////////////////////MP4Info/////////////////////////////////////////////
//MP4Info对象的C映射
typedef void* mk_mp4_info;
typedef struct mk_mp4_info_t *mk_mp4_info;
// GMT 标准时间，单位秒
API_EXPORT uint64_t API_CALL mk_mp4_info_get_start_time(const mk_mp4_info ctx);
// 录像长度，单位秒
@ -42,7 +42,7 @@ API_EXPORT const char* API_CALL mk_mp4_info_get_stream(const mk_mp4_info ctx);
///////////////////////////////////////////Parser/////////////////////////////////////////////
//Parser对象的C映射
typedef void* mk_parser;
typedef struct mk_parser_t *mk_parser;
//Parser::Method(),获取命令字，譬如GET/POST
API_EXPORT const char* API_CALL mk_parser_get_method(const mk_parser ctx);
//Parser::Url(),获取HTTP的访问url(不包括?后面的参数)
@ -60,7 +60,7 @@ API_EXPORT const char* API_CALL mk_parser_get_content(const mk_parser ctx, size_
///////////////////////////////////////////MediaInfo/////////////////////////////////////////////
//MediaInfo对象的C映射
typedef void* mk_media_info;
typedef struct mk_media_info_t *mk_media_info;
//MediaInfo::_param_strs
API_EXPORT const char* API_CALL mk_media_info_get_params(const mk_media_info ctx);
//MediaInfo::_schema
@ -79,7 +79,7 @@ API_EXPORT uint16_t API_CALL mk_media_info_get_port(const mk_media_info ctx);
///////////////////////////////////////////MediaSource/////////////////////////////////////////////
//MediaSource对象的C映射
typedef void* mk_media_source;
typedef struct mk_media_source_t *mk_media_source;
//查找MediaSource的回调函数
typedef void(API_CALL *on_mk_media_source_find_cb)(void *user_data, const mk_media_source ctx);
@ -138,7 +138,7 @@ API_EXPORT void API_CALL mk_media_source_for_each(void *user_data, on_mk_media_s
///////////////////////////////////////////HttpBody/////////////////////////////////////////////
//HttpBody对象的C映射
typedef void* mk_http_body;
typedef struct mk_http_body_t *mk_http_body;
/**
* 生成HttpStringBody
* @param str 字符串指针
@ -173,7 +173,7 @@ API_EXPORT void API_CALL mk_http_body_release(mk_http_body ctx);
///////////////////////////////////////////HttpResponseInvoker/////////////////////////////////////////////
//HttpSession::HttpResponseInvoker对象的C映射
typedef void* mk_http_response_invoker;
typedef struct mk_http_response_invoker_t *mk_http_response_invoker;
/**
* HttpSession::HttpResponseInvoker(const string &codeOut, const StrCaseMap &headerOut, const HttpBody::Ptr &body);
@ -219,7 +219,8 @@ API_EXPORT void API_CALL mk_http_response_invoker_clone_release(const mk_http_re
///////////////////////////////////////////HttpAccessPathInvoker/////////////////////////////////////////////
//HttpSession::HttpAccessPathInvoker对象的C映射
typedef void* mk_http_access_path_invoker;
typedef struct mk_http_access_path_invoker_t *mk_http_access_path_invoker;
/**
* HttpSession::HttpAccessPathInvoker(const string &errMsg,const string &accessPath, int cookieLifeSecond);
* @param err_msg 如果为空，则代表鉴权通过，否则为错误提示,可以为null
@ -244,7 +245,7 @@ API_EXPORT void API_CALL mk_http_access_path_invoker_clone_release(const mk_http
///////////////////////////////////////////RtspSession::onGetRealm/////////////////////////////////////////////
//RtspSession::onGetRealm对象的C映射
typedef void* mk_rtsp_get_realm_invoker;
typedef struct mk_rtsp_get_realm_invoker_t *mk_rtsp_get_realm_invoker;
/**
* 执行RtspSession::onGetRealm
* @param realm 该rtsp流是否需要开启rtsp专属鉴权，至null或空字符串则不鉴权
@ -265,7 +266,7 @@ API_EXPORT void API_CALL mk_rtsp_get_realm_invoker_clone_release(const mk_rtsp_g
///////////////////////////////////////////RtspSession::onAuth/////////////////////////////////////////////
//RtspSession::onAuth对象的C映射
typedef void* mk_rtsp_auth_invoker;
typedef struct mk_rtsp_auth_invoker_t *mk_rtsp_auth_invoker;
/**
* 执行RtspSession::onAuth
@ -289,7 +290,7 @@ API_EXPORT void API_CALL mk_rtsp_auth_invoker_clone_release(const mk_rtsp_auth_i
///////////////////////////////////////////Broadcast::PublishAuthInvoker/////////////////////////////////////////////
//Broadcast::PublishAuthInvoker对象的C映射
typedef void* mk_publish_auth_invoker;
typedef struct mk_publish_auth_invoker_t *mk_publish_auth_invoker;
/**
* 执行Broadcast::PublishAuthInvoker
@ -315,7 +316,7 @@ API_EXPORT void API_CALL mk_publish_auth_invoker_clone_release(const mk_publish_
///////////////////////////////////////////Broadcast::AuthInvoker/////////////////////////////////////////////
//Broadcast::AuthInvoker对象的C映射
typedef void* mk_auth_invoker;
typedef struct mk_auth_invoker_t *mk_auth_invoker;
/**
* 执行Broadcast::AuthInvoker

2
api/include/mk_frame.h
View File

@ -39,7 +39,7 @@ API_EXPORT extern const int MKCodecVP9;
API_EXPORT extern const int MKCodecAV1;
API_EXPORT extern const int MKCodecJPEG;
typedef void *mk_frame;
typedef struct mk_frame_t *mk_frame;
// 用户自定义free回调函数
typedef void(API_CALL *on_mk_frame_data_release)(void *user_data, char *ptr);

2
api/include/mk_h264_splitter.h
View File

@ -17,7 +17,7 @@
extern "C" {
#endif
typedef void *mk_h264_splitter;
typedef struct mk_h264_splitter_t *mk_h264_splitter;
/**
* h264 分帧器输出回调函数

4
api/include/mk_httpclient.h
View File

@ -20,7 +20,7 @@ extern "C" {
///////////////////////////////////////////HttpDownloader/////////////////////////////////////////////
typedef void *mk_http_downloader;
typedef struct mk_http_downloader_t *mk_http_downloader;
/**
* @param user_data 用户数据指针
@ -54,7 +54,7 @@ API_EXPORT void API_CALL mk_http_downloader_start(mk_http_downloader ctx, const
API_EXPORT void API_CALL mk_http_downloader_start2(mk_http_downloader ctx, const char *url, const char *file, on_mk_download_complete cb, void *user_data, on_user_data_free user_data_free);
///////////////////////////////////////////HttpRequester/////////////////////////////////////////////
typedef void *mk_http_requester;
typedef struct mk_http_requester_t *mk_http_requester;
/**
* http请求结果回调

2
api/include/mk_media.h
View File

@ -22,7 +22,7 @@
extern "C" {
#endif
typedef void *mk_media;
typedef struct mk_media_t *mk_media;
/**
* 创建一个媒体源

2
api/include/mk_player.h
View File

@ -19,7 +19,7 @@
extern "C" {
#endif
typedef void* mk_player;
typedef struct mk_player_t *mk_player;
/**
* 播放结果或播放中断事件的回调

2
api/include/mk_proxyplayer.h
View File

@ -17,7 +17,7 @@
extern "C" {
#endif
typedef void *mk_proxy_player;
typedef struct mk_proxy_player_t *mk_proxy_player;
/**
* 创建一个代理播放器

2
api/include/mk_pusher.h
View File

@ -18,7 +18,7 @@
extern "C" {
#endif
typedef void* mk_pusher;
typedef struct mk_pusher_t *mk_pusher;
/**
* 推流结果或推流中断事件的回调

2
api/include/mk_recorder.h
View File

@ -19,7 +19,7 @@ extern "C" {
///////////////////////////////////////////flv录制/////////////////////////////////////////////
typedef void* mk_flv_recorder;
typedef struct mk_flv_recorder_t *mk_flv_recorder;
/**
* 创建flv录制器

2
api/include/mk_rtp_server.h
View File

@ -14,7 +14,7 @@
extern "C" {
#endif
typedef void* mk_rtp_server;
typedef struct mk_rtp_server_t *mk_rtp_server;
/**
* 创建GB28181 RTP 服务器

10
api/include/mk_tcp.h
View File

@ -19,7 +19,7 @@ extern "C" {
///////////////////////////////////////////Buffer::Ptr/////////////////////////////////////////////
typedef void *mk_buffer;
typedef struct mk_buffer_t *mk_buffer;
typedef void(API_CALL *on_mk_buffer_free)(void *user_data, void *data);
/**
@ -39,7 +39,7 @@ API_EXPORT size_t API_CALL mk_buffer_get_size(mk_buffer buffer);
///////////////////////////////////////////SockInfo/////////////////////////////////////////////
//SockInfo对象的C映射
typedef void* mk_sock_info;
typedef struct mk_sock_info_t *mk_sock_info;
//SockInfo::get_peer_ip()
API_EXPORT const char* API_CALL mk_sock_info_peer_ip(const mk_sock_info ctx, char *buf);
@ -66,8 +66,8 @@ API_EXPORT uint16_t API_CALL mk_sock_info_local_port(const mk_sock_info ctx);
#endif
///////////////////////////////////////////TcpSession/////////////////////////////////////////////
//TcpSession对象的C映射
typedef void* mk_tcp_session;
typedef void* mk_tcp_session_ref;
typedef struct mk_tcp_session_t *mk_tcp_session;
typedef struct mk_tcp_session_ref_t *mk_tcp_session_ref;
//获取基类指针以便获取其网络相关信息
API_EXPORT mk_sock_info API_CALL mk_tcp_session_get_sock_info(const mk_tcp_session ctx);
@ -168,7 +168,7 @@ API_EXPORT void API_CALL mk_tcp_server_events_listen(const mk_tcp_session_events
///////////////////////////////////////////自定义tcp客户端/////////////////////////////////////////////
typedef void* mk_tcp_client;
typedef struct mk_tcp_client_t *mk_tcp_client;
//获取基类指针以便获取其网络相关信息
API_EXPORT mk_sock_info API_CALL mk_tcp_client_get_sock_info(const mk_tcp_client ctx);

8
api/include/mk_thread.h
View File

@ -20,7 +20,7 @@ extern "C" {
#endif
///////////////////////////////////////////事件线程/////////////////////////////////////////////
typedef void* mk_thread;
typedef struct mk_thread_t *mk_thread;
/**
* 获取tcp会话对象所在事件线程
@ -52,7 +52,7 @@ API_EXPORT mk_thread API_CALL mk_thread_from_pool();
*/
API_EXPORT mk_thread API_CALL mk_thread_from_pool_work();
typedef void* mk_thread_pool;
typedef struct mk_thread_pool_t *mk_thread_pool;
/**
* 创建线程池
@ -108,7 +108,7 @@ API_EXPORT void API_CALL mk_async_do_delay2(mk_thread ctx, size_t ms, on_mk_asyn
API_EXPORT void API_CALL mk_sync_do(mk_thread ctx, on_mk_async cb, void *user_data);
///////////////////////////////////////////定时器/////////////////////////////////////////////
typedef void* mk_timer;
typedef struct mk_timer_t *mk_timer;
/**
* 定时器触发事件
@ -135,7 +135,7 @@ API_EXPORT void API_CALL mk_timer_release(mk_timer ctx);
///////////////////////////////////////////信号量/////////////////////////////////////////////
typedef void* mk_sem;
typedef struct mk_sem_t *mk_sem;
/**
* 创建信号量

2
api/include/mk_track.h
View File

@ -19,7 +19,7 @@ extern "C" {
#endif
//音视频轨道
typedef void* mk_track;
typedef struct mk_track_t *mk_track;
//输出frame回调
typedef void(API_CALL *on_mk_frame_out)(void *user_data, mk_frame frame);

6
api/include/mk_transcode.h
View File

@ -20,11 +20,11 @@ extern "C" {
#endif
//解码器对象
typedef void *mk_decoder;
typedef struct mk_decoder_t *mk_decoder;
//解码后的frame
typedef void *mk_frame_pix;
typedef struct mk_frame_pix_t *mk_frame_pix;
//SwsContext的包装
typedef void *mk_swscale;
typedef struct mk_swscale_t *mk_swscale;
//FFmpeg原始解码帧对象
typedef struct AVFrame AVFrame;
//FFmpeg编解码器对象

2
api/include/mk_util.h
View File

@ -58,7 +58,7 @@ API_EXPORT char* API_CALL mk_util_get_current_time_string(const char *fmt);
API_EXPORT char* API_CALL mk_util_hex_dump(const void *buf, int len);
///////////////////////////////////////////mk ini/////////////////////////////////////////////
typedef void* mk_ini;
typedef struct mk_ini_t *mk_ini;
/**
* 创建ini配置对象

24
api/source/mk_events_objects.cpp
View File

@ -245,14 +245,14 @@ API_EXPORT void API_CALL mk_media_source_find(const char *schema,
on_mk_media_source_find_cb cb) {
assert(schema && vhost && app && stream && cb);
auto src = MediaSource::find(schema, vhost, app, stream, from_mp4);
cb(user_data, src.get());
cb(user_data, (mk_media_source)src.get());
}
API_EXPORT void API_CALL mk_media_source_for_each(void *user_data, on_mk_media_source_find_cb cb, const char *schema,
const char *vhost, const char *app, const char *stream) {
assert(cb);
MediaSource::for_each_media([&](const MediaSource::Ptr &src) {
cb(user_data, src.get());
cb(user_data, (mk_media_source)src.get());
}, schema ? schema : "", vhost ? vhost : "", app ? app : "", stream ? stream : "");
}
@ -263,17 +263,17 @@ API_EXPORT mk_http_body API_CALL mk_http_body_from_string(const char *str, size_
if(!len){
len = strlen(str);
}
return new HttpBody::Ptr(new HttpStringBody(std::string(str, len)));
return (mk_http_body)new HttpBody::Ptr(new HttpStringBody(std::string(str, len)));
}
API_EXPORT mk_http_body API_CALL mk_http_body_from_buffer(mk_buffer buffer) {
assert(buffer);
return new HttpBody::Ptr(new HttpBufferBody(*((Buffer::Ptr *) buffer)));
return (mk_http_body)new HttpBody::Ptr(new HttpBufferBody(*((Buffer::Ptr *) buffer)));
}
API_EXPORT mk_http_body API_CALL mk_http_body_from_file(const char *file_path){
assert(file_path);
return new HttpBody::Ptr(new HttpFileBody(file_path));
return (mk_http_body)new HttpBody::Ptr(new HttpFileBody(file_path));
}
template <typename C = StrCaseMap>
@ -294,7 +294,7 @@ static C get_http_header( const char *response_header[]){
API_EXPORT mk_http_body API_CALL mk_http_body_from_multi_form(const char *key_val[],const char *file_path){
assert(key_val && file_path);
return new HttpBody::Ptr(new HttpMultiFormBody(get_http_header<HttpArgs>(key_val),file_path));
return (mk_http_body)new HttpBody::Ptr(new HttpMultiFormBody(get_http_header<HttpArgs>(key_val),file_path));
}
API_EXPORT void API_CALL mk_http_body_release(mk_http_body ctx){
@ -338,7 +338,7 @@ API_EXPORT void API_CALL mk_http_response_invoker_do(const mk_http_response_invo
API_EXPORT mk_http_response_invoker API_CALL mk_http_response_invoker_clone(const mk_http_response_invoker ctx){
assert(ctx);
HttpSession::HttpResponseInvoker *invoker = (HttpSession::HttpResponseInvoker *)ctx;
return new HttpSession::HttpResponseInvoker (*invoker);
return (mk_http_response_invoker)new HttpSession::HttpResponseInvoker (*invoker);
}
API_EXPORT void API_CALL mk_http_response_invoker_clone_release(const mk_http_response_invoker ctx){
@ -362,7 +362,7 @@ API_EXPORT void API_CALL mk_http_access_path_invoker_do(const mk_http_access_pat
API_EXPORT mk_http_access_path_invoker API_CALL mk_http_access_path_invoker_clone(const mk_http_access_path_invoker ctx){
assert(ctx);
HttpSession::HttpAccessPathInvoker *invoker = (HttpSession::HttpAccessPathInvoker *)ctx;
return new HttpSession::HttpAccessPathInvoker(*invoker);
return (mk_http_access_path_invoker)new HttpSession::HttpAccessPathInvoker(*invoker);
}
API_EXPORT void API_CALL mk_http_access_path_invoker_clone_release(const mk_http_access_path_invoker ctx){
@ -382,7 +382,7 @@ API_EXPORT void API_CALL mk_rtsp_get_realm_invoker_do(const mk_rtsp_get_realm_in
API_EXPORT mk_rtsp_get_realm_invoker API_CALL mk_rtsp_get_realm_invoker_clone(const mk_rtsp_get_realm_invoker ctx){
assert(ctx);
RtspSession::onGetRealm *invoker = (RtspSession::onGetRealm *)ctx;
return new RtspSession::onGetRealm (*invoker);
return (mk_rtsp_get_realm_invoker)new RtspSession::onGetRealm (*invoker);
}
API_EXPORT void API_CALL mk_rtsp_get_realm_invoker_clone_release(const mk_rtsp_get_realm_invoker ctx){
@ -403,7 +403,7 @@ API_EXPORT void API_CALL mk_rtsp_auth_invoker_do(const mk_rtsp_auth_invoker ctx,
API_EXPORT mk_rtsp_auth_invoker API_CALL mk_rtsp_auth_invoker_clone(const mk_rtsp_auth_invoker ctx){
assert(ctx);
RtspSession::onAuth *invoker = (RtspSession::onAuth *)ctx;
return new RtspSession::onAuth(*invoker);
return (mk_rtsp_auth_invoker)new RtspSession::onAuth(*invoker);
}
API_EXPORT void API_CALL mk_rtsp_auth_invoker_clone_release(const mk_rtsp_auth_invoker ctx){
@ -428,7 +428,7 @@ API_EXPORT void API_CALL mk_publish_auth_invoker_do(const mk_publish_auth_invoke
API_EXPORT mk_publish_auth_invoker API_CALL mk_publish_auth_invoker_clone(const mk_publish_auth_invoker ctx){
assert(ctx);
Broadcast::PublishAuthInvoker *invoker = (Broadcast::PublishAuthInvoker *)ctx;
return new Broadcast::PublishAuthInvoker(*invoker);
return (mk_publish_auth_invoker)new Broadcast::PublishAuthInvoker(*invoker);
}
API_EXPORT void API_CALL mk_publish_auth_invoker_clone_release(const mk_publish_auth_invoker ctx){
@ -447,7 +447,7 @@ API_EXPORT void API_CALL mk_auth_invoker_do(const mk_auth_invoker ctx, const cha
API_EXPORT mk_auth_invoker API_CALL mk_auth_invoker_clone(const mk_auth_invoker ctx){
assert(ctx);
Broadcast::AuthInvoker *invoker = (Broadcast::AuthInvoker *)ctx;
return new Broadcast::AuthInvoker(*invoker);
return (mk_auth_invoker)new Broadcast::AuthInvoker(*invoker);
}
API_EXPORT void API_CALL mk_auth_invoker_clone_release(const mk_auth_invoker ctx){

8
api/source/mk_frame.cpp
View File

@ -74,13 +74,13 @@ static mk_frame mk_frame_create_complex(int codec_id, uint64_t dts, uint64_t pts
char *data, size_t size, on_mk_frame_data_release cb, std::shared_ptr<void> user_data) {
switch (codec_id) {
case CodecH264:
return new Frame::Ptr(new H264FrameHelper<FrameFromPtrForC>(
return (mk_frame)new Frame::Ptr(new H264FrameHelper<FrameFromPtrForC>(
cb, frame_flags, cb, std::move(user_data), (CodecId)codec_id, data, size, dts, pts, prefix_size));
case CodecH265:
return new Frame::Ptr(new H265FrameHelper<FrameFromPtrForC>(
return (mk_frame)new Frame::Ptr(new H265FrameHelper<FrameFromPtrForC>(
cb, frame_flags, cb, std::move(user_data), (CodecId)codec_id, data, size, dts, pts, prefix_size));
default:
return new Frame::Ptr(new FrameFromPtrForC(
return (mk_frame)new Frame::Ptr(new FrameFromPtrForC(
cb, frame_flags, cb, std::move(user_data), (CodecId)codec_id, data, size, dts, pts, prefix_size));
}
}
@ -117,7 +117,7 @@ API_EXPORT void API_CALL mk_frame_unref(mk_frame frame) {
API_EXPORT mk_frame API_CALL mk_frame_ref(mk_frame frame) {
assert(frame);
return new Frame::Ptr(Frame::getCacheAbleFrame(*((Frame::Ptr *) frame)));
return (mk_frame)new Frame::Ptr(Frame::getCacheAbleFrame(*((Frame::Ptr *) frame)));
}
API_EXPORT int API_CALL mk_frame_codec_id(mk_frame frame) {

2
api/source/mk_httpclient.cpp
View File

@ -48,7 +48,7 @@ API_EXPORT void API_CALL mk_http_downloader_start2(mk_http_downloader ctx, const
///////////////////////////////////////////HttpRequester/////////////////////////////////////////////
API_EXPORT mk_http_requester API_CALL mk_http_requester_create(){
HttpRequester::Ptr *ret = new HttpRequester::Ptr(new HttpRequester);
return ret;
return (mk_http_requester)ret;
}
API_EXPORT void API_CALL mk_http_requester_clear(mk_http_requester ctx){

4
api/source/mk_media.cpp
View File

@ -99,7 +99,7 @@ protected:
void onRegist(MediaSource &sender, bool regist) override{
if (_on_regist) {
_on_regist(_on_regist_data.get(), &sender, regist);
_on_regist(_on_regist_data.get(), (mk_media_source)&sender, regist);
}
}
@ -320,5 +320,5 @@ API_EXPORT void API_CALL mk_media_stop_send_rtp(mk_media ctx, const char *ssrc){
API_EXPORT mk_thread API_CALL mk_media_get_owner_thread(mk_media ctx) {
MediaHelper::Ptr *obj = (MediaHelper::Ptr *)ctx;
return (*obj)->getChannel()->getOwnerPoller(MediaSource::NullMediaSource()).get();
return (mk_thread)(*obj)->getChannel()->getOwnerPoller(MediaSource::NullMediaSource()).get();
}

2
api/source/mk_player.cpp
View File

@ -105,7 +105,7 @@ private:
API_EXPORT mk_player API_CALL mk_player_create() {
MediaPlayerForC::Ptr *obj = new MediaPlayerForC::Ptr(new MediaPlayerForC());
(*obj)->setup();
return obj;
return (mk_player)obj;
}
API_EXPORT void API_CALL mk_player_release(mk_player ctx) {
assert(ctx);

4
api/source/mk_pusher.cpp
View File

@ -18,14 +18,14 @@ using namespace mediakit;
API_EXPORT mk_pusher API_CALL mk_pusher_create(const char *schema,const char *vhost,const char *app, const char *stream){
assert(schema && vhost && app && schema);
MediaPusher::Ptr *obj = new MediaPusher::Ptr(new MediaPusher(schema,vhost,app,stream));
return obj;
return (mk_pusher)obj;
}
API_EXPORT mk_pusher API_CALL mk_pusher_create_src(mk_media_source ctx){
assert(ctx);
MediaSource *src = (MediaSource *)ctx;
MediaPusher::Ptr *obj = new MediaPusher::Ptr(new MediaPusher(src->shared_from_this()));
return obj;
return (mk_pusher)obj;
}
API_EXPORT void API_CALL mk_pusher_release(mk_pusher ctx){

2
api/source/mk_recorder.cpp
View File

@ -18,7 +18,7 @@ using namespace mediakit;
API_EXPORT mk_flv_recorder API_CALL mk_flv_recorder_create(){
FlvRecorder::Ptr *ret = new FlvRecorder::Ptr(new FlvRecorder);
return ret;
return (mk_flv_recorder)ret;
}
API_EXPORT void API_CALL mk_flv_recorder_release(mk_flv_recorder ctx){
assert(ctx);

2
api/source/mk_rtp_server.cpp
View File

@ -19,7 +19,7 @@ using namespace mediakit;
API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create(uint16_t port, int tcp_mode, const char *stream_id) {
RtpServer::Ptr *server = new RtpServer::Ptr(new RtpServer);
(*server)->start(port, stream_id, (RtpServer::TcpMode)tcp_mode);
return server;
return (mk_rtp_server)server;
}
API_EXPORT void API_CALL mk_rtp_server_connect(mk_rtp_server ctx, const char *dst_url, uint16_t dst_port, on_mk_rtp_server_connected cb, void *user_data) {

24
api/source/mk_tcp.cpp
View File

@ -65,12 +65,12 @@ API_EXPORT mk_buffer API_CALL mk_buffer_from_char(const char *data, size_t len,
API_EXPORT mk_buffer API_CALL mk_buffer_from_char2(const char *data, size_t len, on_mk_buffer_free cb, void *user_data, on_user_data_free user_data_free) {
assert(data);
std::shared_ptr<void> ptr(user_data, user_data_free ? user_data_free : [](void *) {});
return new Buffer::Ptr(std::make_shared<BufferForC>(data, len, cb, std::move(ptr)));
return (mk_buffer)new Buffer::Ptr(std::make_shared<BufferForC>(data, len, cb, std::move(ptr)));
}
API_EXPORT mk_buffer API_CALL mk_buffer_ref(mk_buffer buffer) {
assert(buffer);
return new Buffer::Ptr(*((Buffer::Ptr *) buffer));
return (mk_buffer)new Buffer::Ptr(*((Buffer::Ptr *) buffer));
}
API_EXPORT void API_CALL mk_buffer_unref(mk_buffer buffer) {
@ -115,7 +115,7 @@ API_EXPORT uint16_t API_CALL mk_sock_info_local_port(const mk_sock_info ctx){
API_EXPORT mk_sock_info API_CALL mk_tcp_session_get_sock_info(const mk_tcp_session ctx){
assert(ctx);
SessionForC *session = (SessionForC *)ctx;
return (SockInfo *)session;
return (mk_sock_info)session;
}
API_EXPORT void API_CALL mk_tcp_session_shutdown(const mk_tcp_session ctx,int err,const char *err_msg){
@ -155,7 +155,7 @@ API_EXPORT void API_CALL mk_tcp_session_send_buffer_safe(const mk_tcp_session ct
API_EXPORT mk_tcp_session_ref API_CALL mk_tcp_session_ref_from(const mk_tcp_session ctx) {
auto ref = ((SessionForC *) ctx)->shared_from_this();
return new std::shared_ptr<SessionForC>(std::dynamic_pointer_cast<SessionForC>(ref));
return (mk_tcp_session_ref)new std::shared_ptr<SessionForC>(std::dynamic_pointer_cast<SessionForC>(ref));
}
API_EXPORT void mk_tcp_session_ref_release(const mk_tcp_session_ref ref) {
@ -163,7 +163,7 @@ API_EXPORT void mk_tcp_session_ref_release(const mk_tcp_session_ref ref) {
}
API_EXPORT mk_tcp_session mk_tcp_session_from_ref(const mk_tcp_session_ref ref) {
return ((std::shared_ptr<SessionForC> *) ref)->get();
return (mk_tcp_session)((std::shared_ptr<SessionForC> *) ref)->get();
}
API_EXPORT void API_CALL mk_tcp_session_send_safe(const mk_tcp_session ctx, const char *data, size_t len) {
@ -179,25 +179,25 @@ static mk_tcp_session_events s_events_server = {0};
SessionForC::SessionForC(const Socket::Ptr &pSock) : Session(pSock) {
_local_port = get_local_port();
if (s_events_server.on_mk_tcp_session_create) {
s_events_server.on_mk_tcp_session_create(_local_port,this);
s_events_server.on_mk_tcp_session_create(_local_port, (mk_tcp_session) this);
}
}
void SessionForC::onRecv(const Buffer::Ptr &buffer) {
if (s_events_server.on_mk_tcp_session_data) {
s_events_server.on_mk_tcp_session_data(_local_port, this, (mk_buffer)&buffer);
s_events_server.on_mk_tcp_session_data(_local_port, (mk_tcp_session)this, (mk_buffer)&buffer);
}
}
void SessionForC::onError(const SockException &err) {
if (s_events_server.on_mk_tcp_session_disconnect) {
s_events_server.on_mk_tcp_session_disconnect(_local_port,this, err.getErrCode(), err.what());
s_events_server.on_mk_tcp_session_disconnect(_local_port, (mk_tcp_session)this, err.getErrCode(), err.what());
}
}
void SessionForC::onManager() {
if (s_events_server.on_mk_tcp_session_manager) {
s_events_server.on_mk_tcp_session_manager(_local_port,this);
s_events_server.on_mk_tcp_session_manager(_local_port, (mk_tcp_session)this);
}
}
@ -320,13 +320,13 @@ TcpClientForC::Ptr *mk_tcp_client_create_l(mk_tcp_client_events *events, mk_tcp_
API_EXPORT mk_sock_info API_CALL mk_tcp_client_get_sock_info(const mk_tcp_client ctx){
assert(ctx);
TcpClientForC::Ptr *client = (TcpClientForC::Ptr *)ctx;
return (SockInfo *)client->get();
return (mk_sock_info)(SockInfo *)client->get();
}
API_EXPORT mk_tcp_client API_CALL mk_tcp_client_create(mk_tcp_client_events *events, mk_tcp_type type){
auto ret = mk_tcp_client_create_l(events,type);
(*ret)->setClient(ret);
return ret;
(*ret)->setClient((mk_tcp_client)ret);
return (mk_tcp_client)ret;
}
API_EXPORT void API_CALL mk_tcp_client_release(mk_tcp_client ctx){

16
api/source/mk_thread.cpp
View File

@ -19,21 +19,21 @@ using namespace toolkit;
API_EXPORT mk_thread API_CALL mk_thread_from_tcp_session(mk_tcp_session ctx){
assert(ctx);
SessionForC *obj = (SessionForC *)ctx;
return obj->getPoller().get();
return (mk_thread)(obj->getPoller().get());
}
API_EXPORT mk_thread API_CALL mk_thread_from_tcp_client(mk_tcp_client ctx){
assert(ctx);
TcpClientForC::Ptr *client = (TcpClientForC::Ptr *)ctx;
return (*client)->getPoller().get();
return (mk_thread)((*client)->getPoller().get());
}
API_EXPORT mk_thread API_CALL mk_thread_from_pool(){
return EventPollerPool::Instance().getPoller().get();
return (mk_thread)(EventPollerPool::Instance().getPoller().get());
}
API_EXPORT mk_thread API_CALL mk_thread_from_pool_work(){
return WorkThreadPool::Instance().getPoller().get();
return (mk_thread)(WorkThreadPool::Instance().getPoller().get());
}
API_EXPORT void API_CALL mk_async_do(mk_thread ctx,on_mk_async cb, void *user_data){
@ -123,7 +123,7 @@ API_EXPORT mk_timer API_CALL mk_timer_create2(mk_thread ctx, uint64_t delay_ms,
std::shared_ptr<void> ptr(user_data, user_data_free ? user_data_free : [](void *) {});
TimerForC::Ptr *ret = new TimerForC::Ptr(new TimerForC(cb, ptr));
(*ret)->start(delay_ms,*poller);
return ret;
return (mk_timer)ret;
}
API_EXPORT void API_CALL mk_timer_release(mk_timer ctx){
@ -148,7 +148,7 @@ public:
};
API_EXPORT mk_thread_pool API_CALL mk_thread_pool_create(const char *name, size_t n_thread, int priority) {
return new WorkThreadPoolForC(name, n_thread, priority);
return (mk_thread_pool)new WorkThreadPoolForC(name, n_thread, priority);
}
API_EXPORT int API_CALL mk_thread_pool_release(mk_thread_pool pool) {
@ -159,11 +159,11 @@ API_EXPORT int API_CALL mk_thread_pool_release(mk_thread_pool pool) {
API_EXPORT mk_thread API_CALL mk_thread_from_thread_pool(mk_thread_pool pool) {
assert(pool);
return ((WorkThreadPoolForC *) pool)->getPoller().get();
return (mk_thread)(((WorkThreadPoolForC *) pool)->getPoller().get());
}
API_EXPORT mk_sem API_CALL mk_sem_create() {
return new toolkit::semaphore;
return (mk_sem)new toolkit::semaphore;
}
API_EXPORT void API_CALL mk_sem_release(mk_sem sem) {

6
api/source/mk_track.cpp
View File

@ -82,8 +82,8 @@ public:
API_EXPORT mk_track API_CALL mk_track_create(int codec_id, codec_args *args) {
switch (getTrackType((CodecId) codec_id)) {
case TrackVideo: return new Track::Ptr(std::make_shared<VideoTrackForC>(codec_id, args));
case TrackAudio: return new Track::Ptr(std::make_shared<AudioTrackForC>(codec_id, args));
case TrackVideo: return (mk_track)new Track::Ptr(std::make_shared<VideoTrackForC>(codec_id, args));
case TrackAudio: return (mk_track)new Track::Ptr(std::make_shared<AudioTrackForC>(codec_id, args));
default: WarnL << "unrecognized codec:" << codec_id; return nullptr;
}
}
@ -95,7 +95,7 @@ API_EXPORT void API_CALL mk_track_unref(mk_track track) {
API_EXPORT mk_track API_CALL mk_track_ref(mk_track track) {
assert(track);
return new Track::Ptr(*( (Track::Ptr *)track));
return (mk_track)new Track::Ptr(*( (Track::Ptr *)track));
}
API_EXPORT int API_CALL mk_track_codec_id(mk_track track) {

14
api/source/mk_transcode.cpp
View File

@ -29,12 +29,12 @@ std::vector<std::string> toCodecList(const char *codec_name_list[]) {
API_EXPORT mk_decoder API_CALL mk_decoder_create(mk_track track, int thread_num) {
assert(track);
return new FFmpegDecoder(*((Track::Ptr *) track), thread_num);
return (mk_decoder)new FFmpegDecoder(*((Track::Ptr *) track), thread_num);
}
API_EXPORT mk_decoder API_CALL mk_decoder_create2(mk_track track, int thread_num, const char *codec_name_list[]) {
assert(track && codec_name_list);
return new FFmpegDecoder(*((Track::Ptr *) track), thread_num, toCodecList(codec_name_list));
return (mk_decoder)new FFmpegDecoder(*((Track::Ptr *) track), thread_num, toCodecList(codec_name_list));
}
API_EXPORT void API_CALL mk_decoder_release(mk_decoder ctx, int flush_frame) {
@ -77,12 +77,12 @@ API_EXPORT const AVCodecContext *API_CALL mk_decoder_get_context(mk_decoder ctx)
API_EXPORT mk_frame_pix API_CALL mk_frame_pix_ref(mk_frame_pix frame) {
assert(frame);
return new FFmpegFrame::Ptr(*(FFmpegFrame::Ptr *) frame);
return (mk_frame_pix)new FFmpegFrame::Ptr(*(FFmpegFrame::Ptr *) frame);
}
API_EXPORT mk_frame_pix API_CALL mk_frame_pix_from_av_frame(AVFrame *frame) {
assert(frame);
return new FFmpegFrame::Ptr(std::make_shared<FFmpegFrame>(std::shared_ptr<AVFrame>(av_frame_clone(frame), [](AVFrame *frame){
return (mk_frame_pix)new FFmpegFrame::Ptr(std::make_shared<FFmpegFrame>(std::shared_ptr<AVFrame>(av_frame_clone(frame), [](AVFrame *frame){
av_frame_free(&frame);
})));
}
@ -99,7 +99,7 @@ API_EXPORT mk_frame_pix API_CALL mk_frame_pix_from_buffer(mk_buffer plane_data[]
frame->linesize[i] = line_size[i];
buffer_array.emplace_back(buffer);
}
return new FFmpegFrame::Ptr(new FFmpegFrame(std::move(frame)), [buffer_array](FFmpegFrame *frame) {
return (mk_frame_pix)new FFmpegFrame::Ptr(new FFmpegFrame(std::move(frame)), [buffer_array](FFmpegFrame *frame) {
for (auto &buffer : buffer_array) {
mk_buffer_unref(buffer);
}
@ -120,7 +120,7 @@ API_EXPORT AVFrame *API_CALL mk_frame_pix_get_av_frame(mk_frame_pix frame) {
/////////////////////////////////////////////////////////////////////////////////////////////
API_EXPORT mk_swscale mk_swscale_create(int output, int width, int height) {
return new FFmpegSws((AVPixelFormat) output, width, height);
return (mk_swscale)new FFmpegSws((AVPixelFormat) output, width, height);
}
API_EXPORT void mk_swscale_release(mk_swscale ctx) {
@ -132,7 +132,7 @@ API_EXPORT int mk_swscale_input_frame(mk_swscale ctx, mk_frame_pix frame, uint8_
}
API_EXPORT mk_frame_pix mk_swscale_input_frame2(mk_swscale ctx, mk_frame_pix frame){
return new FFmpegFrame::Ptr(((FFmpegSws *) ctx)->inputFrame(*(FFmpegFrame::Ptr *) frame));
return (mk_frame_pix)new FFmpegFrame::Ptr(((FFmpegSws *) ctx)->inputFrame(*(FFmpegFrame::Ptr *) frame));
}
API_EXPORT uint8_t **API_CALL mk_get_av_frame_data(AVFrame *frame) {

4
api/source/mk_util.cpp
View File

@ -55,11 +55,11 @@ API_EXPORT char* API_CALL mk_util_hex_dump(const void *buf, int len){
}
API_EXPORT mk_ini API_CALL mk_ini_create() {
return new mINI;
return (mk_ini)new mINI;
}
API_EXPORT mk_ini API_CALL mk_ini_default() {
return &(mINI::Instance());
return (mk_ini)&(mINI::Instance());
}
static void emit_ini_file_reload(mk_ini ini) {

2
api/tests/pusher.c
View File

@ -32,7 +32,7 @@ void release_player(mk_player *ptr) {
}
}
void release_pusher(mk_media *ptr) {
void release_pusher(mk_pusher *ptr) {
if (ptr && *ptr) {
mk_pusher_release(*ptr);
*ptr = NULL;

31
src/Extension/SPSParser.c
View File

@ -252,9 +252,9 @@ static inline int getBitsLeft(void *pvHandle)
*functions
********************************************/
/**
* @brief Function getOneBit() Á1¸öbit
* @brief Function getOneBit() get next bit
* @param[in] h T_GetBitContext structrue
* @retval 0: success, -1 : failure
* @retval other : success, -1 : failure
* @pre
* @post
*/
@ -291,10 +291,10 @@ exit:
/**
* @brief Function getBits() Án¸öbits£¬n²»Äܳ¬¹ý32
* @brief Function getBits() get next bits
* @param[in] h T_GetBitContext structrue
* @param[in] n how many bits you want?
* @retval 0: success, -1 : failure
* @retval other : success, -1 : failure
* @pre
* @post
*/
@ -446,7 +446,7 @@ static inline unsigned int showBitsLong(void *pvHandle, int iN)
/**
* @brief Function parseCodenum() Ö¸Êý¸çÂײ¼±àÂë½âÎö£¬²Î¿¼h264±ê×¼µÚ9½Ú
* @brief Function parseCodenum()
* @param[in] buf
* @retval u32CodeNum
* @pre
@ -469,7 +469,7 @@ static int parseCodenum(void *pvBuf)
}
/**
* @brief Function parseUe() Ö¸Êý¸çÂײ¼±àÂë½âÎö ue(),²Î¿¼h264±ê×¼µÚ9½Ú
* @brief Function parseUe()
* @param[in] buf sps_pps parse buf
* @retval u32CodeNum
* @pre
@ -482,7 +482,7 @@ static int parseUe(void *pvBuf)
/**
* @brief Function parseSe() Ö¸Êý¸çÂײ¼±àÂë½âÎö se(), ²Î¿¼h264±ê×¼µÚ9½Ú
* @brief Function parseSe()
* @param[in] buf sps_pps parse buf
* @retval u32CodeNum
* @pre
@ -502,7 +502,7 @@ static int parseSe(void *pvBuf)
/**
* @brief Function getBitContextFree() ÉêÇëµÄget_bit_context½á¹¹ÄÚ´æÊÍ·Å
* @brief Function getBitContextFree()
* @param[in] buf T_GetBitContext buf
* @retval none
* @pre
@ -527,18 +527,13 @@ static void getBitContextFree(void *pvBuf)
/**
* @brief Function deEmulationPrevention() ½â¾ºÕù´úÂë
* @brief Function deEmulationPrevention()
* @param[in] buf T_GetBitContext buf
* @retval none
* @pre
* @post
* @note:
* µ÷ÊÔʱ×ÜÊÇ·¢ÏÖvui.time_scaleÖµÌرðÆæ¹Ö£¬×ÜÊÇ16777216£¬ºóÀ´²éѯԭÒòÈçÏÂ:
* http://www.cnblogs.com/eustoma/archive/2012/02/13/2415764.html
* H.264±àÂëʱ£¬ÔÚÿ¸öNALÇ°Ìí¼ÓÆðʼÂë 0x000001£¬½âÂëÆ÷ÔÚÂëÁ÷Öмì²âµ½ÆðʼÂ룬µ±Ç°NAL½áÊø¡£
* ΪÁË·ÀÖ¹NALÄÚ²¿³öÏÖ0x000001µÄÊý¾Ý£¬h.264ÓÖÌá³ö'·ÀÖ¹¾ºÕù emulation prevention"»úÖÆ£¬
* ÔÚ±àÂëÍêÒ»¸öNALʱ£¬Èç¹û¼ì²â³öÓÐÁ¬ÐøÁ½¸ö0x00×Ö½Ú£¬¾ÍÔÚºóÃæ²åÈëÒ»¸ö0x03¡£
* µ±½âÂëÆ÷ÔÚNALÄÚ²¿¼ì²âµ½0x000003µÄÊý¾Ý£¬¾Í°Ñ0x03Å×Æú£¬»Ö¸´Ô­Ê¼Êý¾Ý¡£
* 0x000000 >>>>>> 0x00000300
* 0x000001 >>>>>> 0x00000301
* 0x000002 >>>>>> 0x00000302
@ -581,22 +576,20 @@ static void *deEmulationPrevention(void *pvBuf)
tmp_buf_size = ptPtr->iBufSize;
for(i=0; i<(tmp_buf_size-2); i++)
{
/*¼ì²â0x000003*/
iVal = (pu8TmpPtr[i]^0x00) + (pu8TmpPtr[i+1]^0x00) + (pu8TmpPtr[i+2]^0x03);
if(iVal == 0)
{
/*ÌÞ³ý0x03*/
for(j=i+2; j<tmp_buf_size-1; j++)
{
pu8TmpPtr[j] = pu8TmpPtr[j+1];
}
/*ÏàÓ¦µÄbufsizeÒª¼õС*/
ptPtr->iBufSize--;
}
}
/*ÖØмÆËãtotal_bit*/
ptPtr->iTotalBit = ptPtr->iBufSize << 3;
return (void *)ptPtr;

10
src/Extension/SPSParser.h
View File

@ -439,11 +439,11 @@ typedef struct T_HEVCSPS {
typedef struct T_GetBitContext{
uint8_t *pu8Buf; /*Ö¸ÏòSPS start*/
int iBufSize; /*SPS ³¤¶È*/
int iBitPos; /*bitÒѶÁȡλÖÃ*/
int iTotalBit; /*bit×ܳ¤¶È*/
int iCurBitPos; /*µ±Ç°¶ÁȡλÖÃ*/
uint8_t *pu8Buf; // buf
int iBufSize; // buf size
int iBitPos; // bit position
int iTotalBit; // bit number
int iCurBitPos; // current bit position
}T_GetBitContext;

4
src/Http/HttpFileManager.cpp
View File

@ -496,6 +496,10 @@ void HttpFileManager::onAccessPath(Session &sender, Parser &parser, const HttpFi
auto fullUrl = string(HTTP_SCHEMA) + "://" + parser["Host"] + parser.FullUrl();
MediaInfo media_info(fullUrl);
auto file_path = getFilePath(parser, media_info, sender);
if (file_path.size() == 0) {
sendNotFound(cb);
return;
}
//访问的是文件夹
if (File::is_dir(file_path.data())) {
auto indexFile = searchIndexFile(file_path);

4
src/Rtp/RtpProcess.cpp
View File

@ -71,6 +71,10 @@ RtpProcess::~RtpProcess() {
}
bool RtpProcess::inputRtp(bool is_udp, const Socket::Ptr &sock, const char *data, size_t len, const struct sockaddr *addr, uint64_t *dts_out) {
if (!isRtp(data, len)) {
WarnP(this) << "Not rtp packet";
return false;
}
if (_sock != sock) {
// 第一次运行本函数
bool first = !_sock;

2
src/Rtp/RtpSelector.cpp
View File

@ -42,7 +42,7 @@ RtpProcess::Ptr RtpSelector::getProcess(const string &stream_id,bool makeNew) {
}
if (it != _map_rtp_process.end() && makeNew) {
//已经被其他线程持有了，不得再被持有，否则会存在线程安全的问题
throw std::runtime_error(StrPrinter << "RtpProcess(" << stream_id << ") already existed");
throw ProcessExisted(StrPrinter << "RtpProcess(" << stream_id << ") already existed");
}
RtpProcessHelper::Ptr &ref = _map_rtp_process[stream_id];
if (!ref) {

7
src/Rtp/RtpSelector.h
View File

@ -44,6 +44,13 @@ public:
RtpSelector() = default;
~RtpSelector() = default;
class ProcessExisted : public std::runtime_error {
public:
template<typename ...T>
ProcessExisted(T && ...args) : std::runtime_error(std::forward<T>(args)...) {}
~ProcessExisted() override = default;
};
static bool getSSRC(const char *data,size_t data_len, uint32_t &ssrc);
static RtpSelector &Instance();

22
src/Rtp/RtpSession.cpp
View File

@ -12,6 +12,7 @@
#include "RtpSession.h"
#include "RtpSelector.h"
#include "Network/TcpServer.h"
#include "Rtsp/Rtsp.h"
#include "Rtsp/RtpReceiver.h"
#include "Common/config.h"
@ -75,6 +76,15 @@ void RtpSession::onManager() {
}
void RtpSession::onRtpPacket(const char *data, size_t len) {
if (_delay_close) {
// 正在延时关闭中，忽略所有数据
return;
}
if (!isRtp(data, len)) {
// 忽略非rtp数据
WarnP(this) << "Not rtp packet";
return;
}
if (!_is_udp) {
if (_search_rtp) {
//搜索上下文期间，数据丢弃
@ -101,8 +111,18 @@ void RtpSession::onRtpPacket(const char *data, size_t len) {
//未指定流id就使用ssrc为流id
_stream_id = printSSRC(_ssrc);
}
//tcp情况下，一个tcp链接只可能是一路流，不需要通过多个ssrc来区分，所以不需要频繁getProcess
try {
_process = RtpSelector::Instance().getProcess(_stream_id, true);
} catch (RtpSelector::ProcessExisted &ex) {
if (!_is_udp) {
// tcp情况下立即断开连接
throw;
}
// udp情况下延时断开连接(等待超时自动关闭)，防止频繁创建销毁RtpSession对象
WarnP(this) << ex.what();
_delay_close = true;
return;
}
_process->setOnlyAudio(_only_audio);
_process->setDelegate(dynamic_pointer_cast<RtpSession>(shared_from_this()));
}

1
src/Rtp/RtpSession.h
View File

@ -44,6 +44,7 @@ protected:
const char *onSearchPacketTail(const char *data, size_t len) override;
private:
bool _delay_close = false;
bool _is_udp = false;
bool _search_rtp = false;
bool _search_rtp_finished = false;

4
src/Rtsp/RtpReceiver.cpp
View File

@ -14,7 +14,7 @@
namespace mediakit {
RtpTrack::RtpTrack() {
setOnSort([this](uint16_t seq, RtpPacket::Ptr &packet) {
setOnSort([this](uint16_t seq, RtpPacket::Ptr packet) {
onRtpSorted(std::move(packet));
});
}
@ -114,7 +114,7 @@ void RtpTrack::setNtpStamp(uint32_t rtp_stamp, uint64_t ntp_stamp_ms) {
}
}
void RtpTrack::setPT(uint8_t pt){
void RtpTrack::setPayloadType(uint8_t pt) {
_pt = pt;
}

165
src/Rtsp/RtpReceiver.h
View File

@ -18,42 +18,37 @@
#include "Extension/Frame.h"
// for NtpStamp
#include "Common/Stamp.h"
#include "Util/TimeTicker.h"
namespace mediakit {
template<typename T, typename SEQ = uint16_t, size_t kMax = 1024, size_t kMin = 32>
template<typename T, typename SEQ = uint16_t>
class PacketSortor {
public:
static constexpr SEQ SEQ_MAX = (std::numeric_limits<SEQ>::max)();
PacketSortor() = default;
~PacketSortor() = default;
void setOnSort(std::function<void(SEQ seq, T &packet)> cb) {
_cb = std::move(cb);
}
void setOnSort(std::function<void(SEQ seq, T packet)> cb) { _cb = std::move(cb); }
/**
* 清空状态
*/
void clear() {
_started = false;
_seq_cycle_count = 0;
_pkt_sort_cache_map.clear();
_next_seq_out = 0;
_max_sort_size = kMin;
}
/**
* 获取排序缓存长度
*/
size_t getJitterSize() const{
return _pkt_sort_cache_map.size();
}
size_t getJitterSize() const { return _pkt_sort_cache_map.size(); }
/**
* 获取seq回环次数
*/
size_t getCycleCount() const{
return _seq_cycle_count;
}
size_t getCycleCount() const { return _seq_cycle_count; }
/**
* 输入并排序
@ -61,110 +56,106 @@ public:
* @param packet 包负载
*/
void sortPacket(SEQ seq, T packet) {
if(!_is_inited && _next_seq_out == 0){
_next_seq_out = seq;
_is_inited = true;
if (!_started) {
// 记录第一个seq
_started = true;
_last_seq_out = seq - 1;
}
if (seq < _next_seq_out) {
if (_next_seq_out < seq + kMax) {
//过滤seq回退包(回环包除外)
return;
}
} else if (_next_seq_out && seq - _next_seq_out > ((std::numeric_limits<SEQ>::max)() >> 1)) {
//过滤seq跳变非常大的包(防止回环时乱序时收到非常大的seq)
if (seq == static_cast<SEQ>(_last_seq_out + 1)) {
// 收到下一个seq
output(seq, std::move(packet));
return;
}
if (seq < _last_seq_out && _last_seq_out != SEQ_MAX && seq < 1024 && _last_seq_out > SEQ_MAX - 1024) {
// seq回环,清空回环前缓存
flush();
_last_seq_out = SEQ_MAX;
++_seq_cycle_count;
sortPacket(seq, std::move(packet));
return;
}
if (seq <= _last_seq_out && _last_seq_out != SEQ_MAX) {
// 这个回退包已经不再等待
return;
}
//放入排序缓存
_pkt_sort_cache_map.emplace(seq, std::move(packet));
//尝试输出排序后的包
tryPopPacket();
auto it_min = _pkt_sort_cache_map.begin();
auto it_max = _pkt_sort_cache_map.rbegin();
if (it_max->first - it_min->first > (SEQ_MAX >> 1)) {
// 回环后，收到回环前的大值seq, 忽略掉
_pkt_sort_cache_map.erase((++it_max).base());
return;
}
void flush(){
//清空缓存
tryFlushFrontPacket();
if (_pkt_sort_cache_map.size() > _max_buffer_size || (_ticker.elapsedTime() > _max_buffer_ms && !_pkt_sort_cache_map.empty())) {
// buffer太长，强行减小
WarnL << "packet dropped: " << static_cast<SEQ>(_last_seq_out + 1) << " -> "
<< static_cast<SEQ>(_pkt_sort_cache_map.begin()->first - 1)
<< ", jitter buffer size: " << _pkt_sort_cache_map.size()
<< ", jitter buffer ms: " << _ticker.elapsedTime();
popIterator(_pkt_sort_cache_map.begin());
}
}
void flush() {
// 清空缓存
while (!_pkt_sort_cache_map.empty()) {
popIterator(_pkt_sort_cache_map.begin());
}
}
private:
void popPacket() {
void tryFlushFrontPacket() {
while (!_pkt_sort_cache_map.empty()) {
auto it = _pkt_sort_cache_map.begin();
if (it->first >= _next_seq_out) {
//过滤回跳包
popIterator(it);
return;
}
if (_next_seq_out - it->first > (0xFFFF >> 1)) {
//产生回环了
if (_pkt_sort_cache_map.size() < 2 * kMin) {
//等足够多的数据后才处理回环, 因为后面还可能出现大的SEQ
return;
}
++_seq_cycle_count;
//找到大的SEQ并清空掉，然后从小的SEQ重新开始排序
auto hit = _pkt_sort_cache_map.upper_bound((SEQ) (_next_seq_out - _pkt_sort_cache_map.size()));
while (hit != _pkt_sort_cache_map.end()) {
//回环前，清空剩余的大的SEQ的数据
_cb(hit->first, hit->second);
hit = _pkt_sort_cache_map.erase(hit);
}
//下一个回环的数据
popIterator(_pkt_sort_cache_map.begin());
return;
}
//删除回跳的数据包
auto next_seq = static_cast<SEQ>(_last_seq_out + 1);
if (it->first < next_seq) {
_pkt_sort_cache_map.erase(it);
continue;
}
if (it->first == next_seq) {
// 连续的seq
popIterator(it);
continue;
}
break;
}
}
void popIterator(typename std::map<SEQ, T>::iterator it) {
auto seq = it->first;
auto data = std::move(it->second);
_pkt_sort_cache_map.erase(it);
_next_seq_out = seq + 1;
_cb(seq, data);
output(seq, std::move(data));
}
void tryPopPacket() {
int count = 0;
while ((!_pkt_sort_cache_map.empty() && _pkt_sort_cache_map.begin()->first == _next_seq_out)) {
//找到下个包，直接输出
popPacket();
++count;
}
if (count) {
setSortSize();
} else if (_pkt_sort_cache_map.size() > _max_sort_size) {
//排序缓存溢出，不再继续排序
popPacket();
setSortSize();
}
}
void setSortSize() {
_max_sort_size = kMin + _pkt_sort_cache_map.size();
if (_max_sort_size > kMax) {
_max_sort_size = kMax;
}
void output(SEQ seq, T packet) {
_last_seq_out = seq;
_cb(seq, std::move(packet));
_ticker.resetTime();
}
private:
//第一个包是已经进入
bool _is_inited = false;
bool _started = false;
//排序缓存最大保存数据长度，单位毫秒
size_t _max_buffer_ms = 3000;
//排序缓存最大保存数据个数
size_t _max_buffer_size = 1024;
//记录上次output至今的时间
toolkit::Ticker _ticker;
//下次应该输出的SEQ
SEQ _next_seq_out = 0;
SEQ _last_seq_out = 0;
//seq回环次数计数
size_t _seq_cycle_count = 0;
//排序缓存长度
size_t _max_sort_size = kMin;
//pkt排序缓存，根据seq排序
std::map<SEQ, T> _pkt_sort_cache_map;
//回调
std::function<void(SEQ seq, T &packet)> _cb;
std::function<void(SEQ seq, T packet)> _cb;
};
class RtpTrack : private PacketSortor<RtpPacket::Ptr> {
@ -183,7 +174,7 @@ public:
uint32_t getSSRC() const;
RtpPacket::Ptr inputRtp(TrackType type, int sample_rate, uint8_t *ptr, size_t len);
void setNtpStamp(uint32_t rtp_stamp, uint64_t ntp_stamp_ms);
void setPT(uint8_t pt);
void setPayloadType(uint8_t pt);
protected:
virtual void onRtpSorted(RtpPacket::Ptr rtp) {}
@ -261,9 +252,9 @@ public:
_track[index].setNtpStamp(rtp_stamp, ntp_stamp_ms);
}
void setPT(int index, uint8_t pt){
void setPayloadType(int index, uint8_t pt){
assert(index < kCount && index >= 0);
_track[index].setPT(pt);
_track[index].setPayloadType(pt);
}
void clear() {

16
src/Rtsp/Rtsp.cpp
View File

@ -442,6 +442,22 @@ string printSSRC(uint32_t ui32Ssrc) {
return tmp;
}
bool isRtp(const char *buf, size_t size) {
if (size < 2) {
return false;
}
RtpHeader *header = (RtpHeader *)buf;
return ((header->pt < 64) || (header->pt >= 96));
}
bool isRtcp(const char *buf, size_t size) {
if (size < 2) {
return false;
}
RtpHeader *header = (RtpHeader *)buf;
return ((header->pt >= 64) && (header->pt < 96));
}
Buffer::Ptr makeRtpOverTcpPrefix(uint16_t size, uint8_t interleaved) {
auto rtp_tcp = BufferRaw::create();
rtp_tcp->setCapacity(RtpPacket::kRtpTcpHeaderSize);

3
src/Rtsp/Rtsp.h
View File

@ -337,5 +337,8 @@ void makeSockPair(std::pair<toolkit::Socket::Ptr, toolkit::Socket::Ptr> &pair, c
//十六进制方式打印ssrc
std::string printSSRC(uint32_t ui32Ssrc);
bool isRtp(const char *buf, size_t size);
bool isRtcp(const char *buf, size_t size);
} //namespace mediakit
#endif //RTSP_RTSP_H_

2
src/Rtsp/RtspPlayer.cpp
View File

@ -225,7 +225,7 @@ void RtspPlayer::handleResDESCRIBE(const Parser& parser) {
_rtcp_context.clear();
for (auto &track : _sdp_track) {
if(track->_pt != 0xff){
setPT(_rtcp_context.size(),track->_pt);
setPayloadType(_rtcp_context.size(),track->_pt);
}
_rtcp_context.emplace_back(std::make_shared<RtcpContextForRecv>());
}

36
tests/test_rtcp_nack.cpp
View File

@ -15,20 +15,38 @@ using namespace std;
using namespace toolkit;
using namespace mediakit;
extern void testFCI();
int main() {
Logger::Instance().add(std::make_shared<ConsoleChannel>());
Logger::Instance().setWriter(std::make_shared<AsyncLogWriter>());
srand((unsigned) time(NULL));
srand((unsigned)time(NULL));
NackContext ctx;
for (int i = 1; i < 1000; ++i) {
if (i % (1 + (rand() % 30)) == 0) {
DebugL << "drop:" << i;
ctx.setOnNack([](const FCI_NACK &nack){
InfoL << nack.dumpString();
});
auto drop_start = 0;
auto drop_len = 0;
uint16_t offset = 0xFFFF - 200 - 50;
for (int i = 1; i < 10000; ++i) {
if (i % 100 == 0) {
drop_start = i + rand() % 16;
drop_len = 4 + rand() % 16;
InfoL << "start drop:" << (uint16_t)(drop_start + offset) << " -> "
<< (uint16_t)(drop_start + offset + drop_len);
}
uint16_t seq = i + offset;
if ((i >= drop_start && i <= drop_start + drop_len) || seq == 65535 || seq == 0 || seq == 1) {
TraceL << "drop:" << (uint16_t)(i + offset);
} else {
ctx.received(i);
static auto last_seq = seq;
if (seq - last_seq > 16) {
ctx.received(last_seq);
ctx.received(seq);
DebugL << "seq reduce:" << last_seq;
last_seq = seq;
} else {
ctx.received(seq);
}
}
}
sleep(1);

2
tests/test_sortor.cpp
View File

@ -102,7 +102,7 @@ void test_real() {
PacketSortor<uint16_t, uint16_t> sortor;
list<uint16_t> sorted_list;
sortor.setOnSort([&](uint16_t seq, const uint16_t &packet) {
sortor.setOnSort([&](uint16_t seq, uint16_t packet) {
sorted_list.push_back(seq);
});

3
webrtc/DtlsTransport.cpp
View File

@ -1469,8 +1469,7 @@ namespace RTC
// Workaround for https://github.com/openssl/openssl/issues/7998.
if (this->handshakeDone)
{
MS_DEBUG_DEV("handshake is done so return");
// MS_DEBUG_DEV("handshake is done so return");
return;
}

28
webrtc/IceServer.cpp
View File

@ -198,8 +198,12 @@ namespace RTC
// Create a success response.
RTC::StunPacket* response = packet->CreateSuccessResponse();
sockaddr_storage peerAddr;
socklen_t addr_len = sizeof(peerAddr);
getpeername(tuple->getSock()->rawFD(), (struct sockaddr *)&peerAddr, &addr_len);
// Add XOR-MAPPED-ADDRESS.
response->SetXorMappedAddress(tuple);
response->SetXorMappedAddress((struct sockaddr *)&peerAddr);
// Authenticate the response.
if (this->oldPassword.empty())
@ -260,9 +264,9 @@ namespace RTC
for (; it != this->tuples.end(); ++it)
{
RTC::TransportTuple* storedTuple = std::addressof(*it);
RTC::TransportTuple* storedTuple = *it;
if (memcmp(storedTuple, tuple, sizeof (RTC::TransportTuple)) == 0)
if (storedTuple == tuple)
{
removedTuple = storedTuple;
@ -285,9 +289,9 @@ namespace RTC
this->selectedTuple = nullptr;
// Mark the first tuple as selected tuple (if any).
if (this->tuples.begin() != this->tuples.end())
if (!this->tuples.empty())
{
SetSelectedTuple(std::addressof(*this->tuples.begin()));
SetSelectedTuple(this->tuples.front());
}
// Or just emit 'disconnected'.
else
@ -477,12 +481,10 @@ namespace RTC
MS_TRACE();
// Add the new tuple at the beginning of the list.
this->tuples.push_front(*tuple);
auto* storedTuple = std::addressof(*this->tuples.begin());
this->tuples.push_front(tuple);
// Return the address of the inserted tuple.
return storedTuple;
return tuple;
}
inline RTC::TransportTuple* IceServer::HasTuple(const RTC::TransportTuple* tuple) const
@ -495,15 +497,14 @@ namespace RTC
return nullptr;
// Check the current selected tuple.
if (memcmp(selectedTuple, tuple, sizeof (RTC::TransportTuple)) == 0)
if (selectedTuple == tuple)
return this->selectedTuple;
// Otherwise check other stored tuples.
for (const auto& it : this->tuples)
{
auto* storedTuple = const_cast<RTC::TransportTuple*>(std::addressof(it));
if (memcmp(storedTuple, tuple, sizeof (RTC::TransportTuple)) == 0)
auto& storedTuple = it;
if (storedTuple == tuple)
return storedTuple;
}
@ -519,6 +520,7 @@ namespace RTC
return;
this->selectedTuple = storedTuple;
this->lastSelectedTuple = storedTuple->shared_from_this();
// Notify the listener.
this->listener->OnIceServerSelectedTuple(this, this->selectedTuple);

16
webrtc/IceServer.hpp
View File

@ -20,6 +20,7 @@ OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#define MS_RTC_ICE_SERVER_HPP
#include "StunPacket.hpp"
#include "Network/Session.h"
#include "logger.h"
#include "Utils.hpp"
#include <list>
@ -27,11 +28,9 @@ OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
#include <functional>
#include <memory>
using _TransportTuple = struct sockaddr;
namespace RTC
{
using TransportTuple = _TransportTuple;
using TransportTuple = toolkit::Session;
class IceServer
{
public:
@ -80,9 +79,9 @@ namespace RTC
{
return this->state;
}
RTC::TransportTuple* GetSelectedTuple() const
RTC::TransportTuple* GetSelectedTuple(bool try_last_tuple = false) const
{
return this->selectedTuple;
return try_last_tuple ? this->lastSelectedTuple.lock().get() : this->selectedTuple;
}
void SetUsernameFragment(const std::string& usernameFragment)
{
@ -100,6 +99,8 @@ namespace RTC
// and the given tuple must be an already valid tuple.
void ForceSelectedTuple(const RTC::TransportTuple* tuple);
const std::list<RTC::TransportTuple *>& GetTuples() const { return tuples; }
private:
void HandleTuple(RTC::TransportTuple* tuple, bool hasUseCandidate);
/**
@ -125,8 +126,9 @@ namespace RTC
std::string oldUsernameFragment;
std::string oldPassword;
IceState state{ IceState::NEW };
std::list<RTC::TransportTuple> tuples;
RTC::TransportTuple* selectedTuple{ nullptr };
std::list<RTC::TransportTuple *> tuples;
RTC::TransportTuple *selectedTuple { nullptr };
std::weak_ptr<RTC::TransportTuple> lastSelectedTuple;
//最大不超过mtu
static constexpr size_t StunSerializeBufferSize{ 1600 };
uint8_t StunSerializeBuffer[StunSerializeBufferSize];

139
webrtc/Nack.cpp
View File

@ -9,7 +9,7 @@
*/
#include "Nack.h"
#include <algorithm>
using namespace std;
using namespace toolkit;
@ -27,7 +27,7 @@ void NackList::pushBack(RtpPacket::Ptr rtp) {
}
_cache_ms_check = 0;
while (getCacheMS() >= kMaxNackMS) {
//需要清除部分nack缓存
// 需要清除部分nack缓存
popFront();
}
}
@ -36,7 +36,7 @@ void NackList::forEach(const FCI_NACK &nack, const function<void(const RtpPacket
auto seq = nack.getPid();
for (auto bit : nack.getBitArray()) {
if (bit) {
//丢包
// 丢包
RtpPacket::Ptr *ptr = getRtp(seq);
if (ptr) {
func(*ptr);
@ -79,7 +79,7 @@ uint32_t NackList::getCacheMS() {
if (back_stamp >= front_stamp) {
return back_stamp - front_stamp;
}
//很有可能回环了
// 很有可能回环了
return back_stamp + (UINT32_MAX - front_stamp);
}
return 0;
@ -95,101 +95,119 @@ int64_t NackList::getRtpStamp(uint16_t seq) {
////////////////////////////////////////////////////////////////////////////////////////////////
NackContext::NackContext() {
setOnNack(nullptr);
}
void NackContext::received(uint16_t seq, bool is_rtx) {
if (!_last_max_seq && _seq.empty()) {
_last_max_seq = seq - 1;
if (!_started) {
// 记录第一个seq
_started = true;
_nack_seq = seq - 1;
}
if (is_rtx || (seq < _last_max_seq && !(seq < 1024 && _last_max_seq > UINT16_MAX - 1024))) {
//重传包或
// seq回退，且非回环，那么这个应该是重传包
onRtx(seq);
if (seq < _nack_seq && _nack_seq != UINT16_MAX && seq < 1024 && _nack_seq > UINT16_MAX - 1024) {
// seq回环,清空回环前状态
makeNack(UINT16_MAX, true);
_seq.emplace(seq);
return;
}
if (is_rtx || (seq < _nack_seq && _nack_seq != UINT16_MAX)) {
// seq非回环回退包，猜测其为重传包，清空其nack状态
clearNackStatus(seq);
return;
}
auto pr = _seq.emplace(seq);
if (!pr.second) {
// seq重复, 忽略
return;
}
_seq.emplace(seq);
auto max_seq = *_seq.rbegin();
auto min_seq = *_seq.begin();
auto diff = max_seq - min_seq;
if (!diff) {
if (diff > (UINT16_MAX >> 1)) {
// 回环后，收到回环前的大值seq, 忽略掉
_seq.erase(max_seq);
return;
}
if (diff > UINT16_MAX / 2) {
//回环
if (min_seq == (uint16_t)(_nack_seq + 1) && _seq.size() == (size_t)diff + 1) {
// 都是连续的seq，未丢包
_seq.clear();
_last_max_seq = min_seq;
_nack_send_status.clear();
return;
}
if (_seq.size() == (size_t)diff + 1 && _last_max_seq + 1 == min_seq) {
//都是连续的seq，未丢包
_seq.clear();
_last_max_seq = max_seq;
_nack_seq = max_seq;
} else {
// seq不连续，有丢包
if (min_seq == _last_max_seq + 1) {
//前面部分seq是连续的，未丢包，移除之
eraseFrontSeq();
makeNack(max_seq, false);
}
}
//有丢包，丢包从_last_max_seq开始
auto nack_rtp_count = FCI_NACK::kBitSize;
if (max_seq > nack_rtp_count + _last_max_seq) {
void NackContext::makeNack(uint16_t max_seq, bool flush) {
// 尝试移除前面部分连续的seq
eraseFrontSeq();
// 最多生成5个nack包，防止seq大幅跳跃导致一直循环
auto max_nack = 5u;
while (_nack_seq != max_seq && max_nack--) {
// 一次不能发送超过16+1个rtp的状态
uint16_t nack_rtp_count = std::min<uint16_t>(FCI_NACK::kBitSize, max_seq - (uint16_t)(_nack_seq + 1));
if (!flush && nack_rtp_count < kNackRtpSize) {
// 非flush状态下，seq个数不足以发送一次nack
break;
}
vector<bool> vec;
vec.resize(FCI_NACK::kBitSize, false);
vec.resize(nack_rtp_count, false);
for (size_t i = 0; i < nack_rtp_count; ++i) {
vec[i] = _seq.find(_last_max_seq + i + 2) == _seq.end();
vec[i] = _seq.find((uint16_t)(_nack_seq + i + 2)) == _seq.end();
}
doNack(FCI_NACK(_last_max_seq + 1, vec), true);
_last_max_seq += nack_rtp_count + 1;
if (_last_max_seq >= max_seq) {
_seq.clear();
} else {
auto it = _seq.emplace_hint(_seq.begin(), _last_max_seq + 1);
doNack(FCI_NACK(_nack_seq + 1, vec), true);
_nack_seq += nack_rtp_count + 1;
// 返回第一个比_last_max_seq大的元素
auto it = _seq.upper_bound(_nack_seq);
// 移除 <=_last_max_seq 的seq
_seq.erase(_seq.begin(), it);
}
}
}
}
void NackContext::setOnNack(onNack cb) {
if (cb) {
_cb = std::move(cb);
} else {
_cb = [](const FCI_NACK &nack) {};
}
}
void NackContext::doNack(const FCI_NACK &nack, bool record_nack) {
if (record_nack) {
recordNack(nack);
}
if (_cb) {
_cb(nack);
}
}
void NackContext::eraseFrontSeq() {
//前面部分seq是连续的，未丢包，移除之
// 前面部分seq是连续的，未丢包，移除之
for (auto it = _seq.begin(); it != _seq.end();) {
if (*it != _last_max_seq + 1) {
if (*it != (uint16_t)(_nack_seq + 1)) {
// seq不连续，丢包了
break;
}
_last_max_seq = *it;
_nack_seq = *it;
it = _seq.erase(it);
}
}
void NackContext::onRtx(uint16_t seq) {
void NackContext::clearNackStatus(uint16_t seq) {
auto it = _nack_send_status.find(seq);
if (it == _nack_send_status.end()) {
return;
}
auto rtt = getCurrentMillisecond() - it->second.update_stamp;
//收到重传包与第一个nack包间的时间约等于rtt时间
auto rtt = getCurrentMillisecond() - it->second.first_stamp;
_nack_send_status.erase(it);
if (rtt >= 0) {
// rtt不肯小于0
// rtt不能小于0
_rtt = rtt;
// InfoL << "rtt:" << rtt;
}
}
@ -205,7 +223,7 @@ void NackContext::recordNack(const FCI_NACK &nack) {
}
++i;
}
//记录太多了，移除一部分早期的记录
// 记录太多了，移除一部分早期的记录
while (_nack_send_status.size() > kNackMaxSize) {
_nack_send_status.erase(_nack_send_status.begin());
}
@ -216,18 +234,18 @@ uint64_t NackContext::reSendNack() {
auto now = getCurrentMillisecond();
for (auto it = _nack_send_status.begin(); it != _nack_send_status.end();) {
if (now - it->second.first_stamp > kNackMaxMS) {
//该rtp丢失太久了，不再要求重传
// 该rtp丢失太久了，不再要求重传
it = _nack_send_status.erase(it);
continue;
}
if (now - it->second.update_stamp < kNackIntervalRatio * _rtt) {
//距离上次nack不足2倍的rtt，不用再发送nack
// 距离上次nack不足2倍的rtt，不用再发送nack
++it;
continue;
}
//此rtp需要请求重传
// 此rtp需要请求重传
nack_rtp.emplace(it->first);
//更新nack发送时间戳
// 更新nack发送时间戳
it->second.update_stamp = now;
if (++(it->second.nack_count) == kNackMaxCount) {
// nack次数太多，移除之
@ -237,11 +255,6 @@ uint64_t NackContext::reSendNack() {
++it;
}
if (_nack_send_status.empty()) {
//不需要再发送nack
return 0;
}
int pid = -1;
vector<bool> vec;
for (auto it = nack_rtp.begin(); it != nack_rtp.end();) {
@ -253,12 +266,12 @@ uint64_t NackContext::reSendNack() {
}
auto inc = *it - pid;
if (inc > (ssize_t)FCI_NACK::kBitSize) {
//新的nack包
// 新的nack包
doNack(FCI_NACK(pid, vec), false);
pid = -1;
continue;
}
//这个包丢了
// 这个包丢了
vec[inc - 1] = true;
++it;
}
@ -266,8 +279,8 @@ uint64_t NackContext::reSendNack() {
doNack(FCI_NACK(pid, vec), false);
}
//重传间隔不得低于5ms
return max(_rtt, 5);
// 没有任何包需要重传时返回0，否则返回下次重传间隔(不得低于5ms)
return _nack_send_status.empty() ? 0 : max(_rtt, 5);
}
} // namespace mediakit

15
webrtc/Nack.h
View File

@ -49,11 +49,15 @@ public:
// rtp丢包状态最长保留时间
static constexpr auto kNackMaxMS = 3 * 1000;
// nack最多请求重传10次
static constexpr auto kNackMaxCount = 10;
static constexpr auto kNackMaxCount = 15;
// nack重传频率，rtt的倍数
static constexpr auto kNackIntervalRatio = 1.0f;
// nack包中rtp个数，减小此值可以让nack包响应更灵敏
static constexpr auto kNackRtpSize = 8;
NackContext() = default;
static_assert(kNackRtpSize >=0 && kNackRtpSize <= FCI_NACK::kBitSize, "NackContext::kNackRtpSize must between 0 and 16");
NackContext();
~NackContext() = default;
void received(uint16_t seq, bool is_rtx = false);
@ -64,13 +68,16 @@ private:
void eraseFrontSeq();
void doNack(const FCI_NACK &nack, bool record_nack);
void recordNack(const FCI_NACK &nack);
void onRtx(uint16_t seq);
void clearNackStatus(uint16_t seq);
void makeNack(uint16_t max, bool flush = false);
private:
bool _started = false;
int _rtt = 50;
onNack _cb;
std::set<uint16_t> _seq;
uint16_t _last_max_seq = 0;
// 最新nack包中的rtp seq值
uint16_t _nack_seq = 0;
struct NackStatus {
uint64_t first_stamp;

10
webrtc/WebRtcPlayer.cpp
View File

@ -70,21 +70,17 @@ void WebRtcPlayer::onStartWebRTC() {
}
}
void WebRtcPlayer::onDestory() {
WebRtcTransportImp::onDestory();
auto duration = getDuration();
auto bytes_usage = getBytesUsage();
//流量统计事件广播
GET_CONFIG(uint32_t, iFlowThreshold, General::kFlowThreshold);
if (_reader && getSession()) {
WarnL << "RTC播放器("
<< _media_info.shortUrl()
<< ")结束播放,耗时(s):" << duration;
WarnL << "RTC播放器(" << _media_info.shortUrl() << ")结束播放,耗时(s):" << duration;
if (bytes_usage >= iFlowThreshold * 1024) {
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _media_info, bytes_usage, duration,
true, static_cast<SockInfo &>(*getSession()));
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _media_info, bytes_usage, duration, true, static_cast<SockInfo &>(*getSession()));
}
}
WebRtcTransportImp::onDestory();
}
void WebRtcPlayer::onRtcConfigure(RtcConfigure &configure) const {

10
webrtc/WebRtcPusher.cpp
View File

@ -118,20 +118,15 @@ void WebRtcPusher::onStartWebRTC() {
}
void WebRtcPusher::onDestory() {
WebRtcTransportImp::onDestory();
auto duration = getDuration();
auto bytes_usage = getBytesUsage();
//流量统计事件广播
GET_CONFIG(uint32_t, iFlowThreshold, General::kFlowThreshold);
if (getSession()) {
WarnL << "RTC推流器("
<< _media_info.shortUrl()
<< ")结束推流,耗时(s):" << duration;
WarnL << "RTC推流器(" << _media_info.shortUrl() << ")结束推流,耗时(s):" << duration;
if (bytes_usage >= iFlowThreshold * 1024) {
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _media_info, bytes_usage, duration,
false, static_cast<SockInfo &>(*getSession()));
NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastFlowReport, _media_info, bytes_usage, duration, false, static_cast<SockInfo &>(*getSession()));
}
}
@ -142,6 +137,7 @@ void WebRtcPusher::onDestory() {
auto push_src = std::move(_push_src);
getPoller()->doDelayTask(_continue_push_ms, [push_src]() { return 0; });
}
WebRtcTransportImp::onDestory();
}
void WebRtcPusher::onRtcConfigure(RtcConfigure &configure) const {

12
webrtc/WebRtcSession.cpp
View File

@ -48,8 +48,6 @@ EventPoller::Ptr WebRtcSession::queryPoller(const Buffer::Ptr &buffer) {
////////////////////////////////////////////////////////////////////////////////
WebRtcSession::WebRtcSession(const Socket::Ptr &sock) : Session(sock) {
socklen_t addr_len = sizeof(_peer_addr);
getpeername(sock->rawFD(), (struct sockaddr *)&_peer_addr, &addr_len);
_over_tcp = sock->sockType() == SockNum::Sock_TCP;
}
@ -87,14 +85,12 @@ void WebRtcSession::onRecv_l(const char *data, size_t len) {
//3、销毁原先的socket和WebRtcSession(原先的对象跟WebRtcTransport不在同一条线程)
throw std::runtime_error("webrtc over tcp change poller: " + getPoller()->getThreadName() + " -> " + sock->getPoller()->getThreadName());
}
transport->setSession(shared_from_this());
_transport = std::move(transport);
InfoP(this);
}
_ticker.resetTime();
CHECK(_transport);
_transport->inputSockData((char *)data, len, (struct sockaddr *)&_peer_addr);
_transport->inputSockData((char *)data, len, this);
}
void WebRtcSession::onRecv(const Buffer::Ptr &buffer) {
@ -114,9 +110,13 @@ void WebRtcSession::onError(const SockException &err) {
if (!_transport) {
return;
}
auto self = shared_from_this();
auto transport = std::move(_transport);
getPoller()->async([transport] {
getPoller()->async([transport, self]() mutable {
//延时减引用，防止使用transport对象时，销毁对象
transport->removeTuple(self.get());
//确保transport在Session对象前销毁，防止WebRtcTransport::onDestory()时获取不到Session对象
transport = nullptr;
}, false);
}

1
webrtc/WebRtcSession.h
View File

@ -46,7 +46,6 @@ private:
bool _over_tcp = false;
bool _find_transport = true;
Ticker _ticker;
struct sockaddr_storage _peer_addr;
std::weak_ptr<toolkit::TcpServer> _server;
WebRtcTransportImp::Ptr _transport;
};

96
webrtc/WebRtcTransport.cpp
View File

@ -15,6 +15,7 @@
#include "Rtcp/Rtcp.h"
#include "Rtcp/RtcpFCI.h"
#include "Rtcp/RtcpContext.h"
#include "Rtsp/Rtsp.h"
#include "Rtsp/RtpReceiver.h"
#include "WebRtcTransport.h"
@ -74,6 +75,17 @@ static void translateIPFromEnv(std::vector<std::string> &v) {
}
}
const char* sockTypeStr(Session* session) {
if (session) {
switch (session->getSock()->sockType()) {
case SockNum::Sock_TCP: return "tcp";
case SockNum::Sock_UDP: return "udp";
default: break;
}
}
return "unknown";
}
WebRtcTransport::WebRtcTransport(const EventPoller::Ptr &poller) {
_poller = poller;
_identifier = "zlm_" + to_string(++s_key);
@ -108,16 +120,18 @@ void WebRtcTransport::OnIceServerSendStunPacket(
sendSockData((char *)packet->GetData(), packet->GetSize(), tuple);
}
void WebRtcTransport::OnIceServerSelectedTuple(const RTC::IceServer *iceServer, RTC::TransportTuple *tuple) {
InfoL;
void WebRtcTransportImp::OnIceServerSelectedTuple(const RTC::IceServer *iceServer, RTC::TransportTuple *tuple) {
InfoL << getIdentifier() << " select tuple " << sockTypeStr(tuple) << " " << tuple->get_peer_ip() << ":" << tuple->get_peer_port();
tuple->setSendFlushFlag(false);
unrefSelf();
}
void WebRtcTransport::OnIceServerConnected(const RTC::IceServer *iceServer) {
InfoL;
InfoL << getIdentifier();
}
void WebRtcTransport::OnIceServerCompleted(const RTC::IceServer *iceServer) {
InfoL;
InfoL << getIdentifier();
if (_answer_sdp->media[0].role == DtlsRole::passive) {
_dtls_transport->Run(RTC::DtlsTransport::Role::SERVER);
} else {
@ -126,7 +140,7 @@ void WebRtcTransport::OnIceServerCompleted(const RTC::IceServer *iceServer) {
}
void WebRtcTransport::OnIceServerDisconnected(const RTC::IceServer *iceServer) {
InfoL;
InfoL << getIdentifier();
}
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
@ -134,7 +148,7 @@ void WebRtcTransport::OnIceServerDisconnected(const RTC::IceServer *iceServer) {
void WebRtcTransport::OnDtlsTransportConnected(
const RTC::DtlsTransport *dtlsTransport, RTC::SrtpSession::CryptoSuite srtpCryptoSuite, uint8_t *srtpLocalKey,
size_t srtpLocalKeyLen, uint8_t *srtpRemoteKey, size_t srtpRemoteKeyLen, std::string &remoteCert) {
InfoL;
InfoL << getIdentifier();
_srtp_session_send = std::make_shared<RTC::SrtpSession>(
RTC::SrtpSession::Type::OUTBOUND, srtpCryptoSuite, srtpLocalKey, srtpLocalKeyLen);
_srtp_session_recv = std::make_shared<RTC::SrtpSession>(
@ -152,16 +166,16 @@ void WebRtcTransport::OnDtlsTransportSendData(
}
void WebRtcTransport::OnDtlsTransportConnecting(const RTC::DtlsTransport *dtlsTransport) {
InfoL;
InfoL << getIdentifier();
}
void WebRtcTransport::OnDtlsTransportFailed(const RTC::DtlsTransport *dtlsTransport) {
InfoL;
InfoL << getIdentifier();
onShutdown(SockException(Err_shutdown, "dtls transport failed"));
}
void WebRtcTransport::OnDtlsTransportClosed(const RTC::DtlsTransport *dtlsTransport) {
InfoL;
InfoL << getIdentifier();
onShutdown(SockException(Err_shutdown, "dtls close notify received"));
}
@ -177,7 +191,7 @@ void WebRtcTransport::OnDtlsTransportApplicationDataReceived(
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
#ifdef ENABLE_SCTP
void WebRtcTransport::OnSctpAssociationConnecting(RTC::SctpAssociation *sctpAssociation) {
TraceL;
TraceL << getIdentifier();
}
void WebRtcTransport::OnSctpAssociationConnected(RTC::SctpAssociation *sctpAssociation) {
@ -214,8 +228,9 @@ void WebRtcTransport::sendSockData(const char *buf, size_t len, RTC::TransportTu
onSendSockData(std::move(pkt), true, tuple ? tuple : _ice_server->GetSelectedTuple());
}
RTC::TransportTuple *WebRtcTransport::getSelectedTuple() const {
return _ice_server->GetSelectedTuple();
Session::Ptr WebRtcTransport::getSession() const {
auto tuple = _ice_server->GetSelectedTuple(true);
return tuple ? tuple->shared_from_this() : nullptr;
}
void WebRtcTransport::sendRtcpRemb(uint32_t ssrc, size_t bit_rate) {
@ -287,22 +302,12 @@ std::string WebRtcTransport::getAnswerSdp(const string &offer) {
}
}
static bool is_dtls(char *buf) {
static bool isDtls(char *buf) {
return ((*buf > 19) && (*buf < 64));
}
static bool is_rtp(char *buf) {
RtpHeader *header = (RtpHeader *)buf;
return ((header->pt < 64) || (header->pt >= 96));
}
static bool is_rtcp(char *buf) {
RtpHeader *header = (RtpHeader *)buf;
return ((header->pt >= 64) && (header->pt < 96));
}
static string getPeerAddress(RTC::TransportTuple *tuple) {
return SockUtil::inet_ntoa(tuple);
return tuple->get_peer_ip();
}
void WebRtcTransport::inputSockData(char *buf, int len, RTC::TransportTuple *tuple) {
@ -315,11 +320,11 @@ void WebRtcTransport::inputSockData(char *buf, int len, RTC::TransportTuple *tup
_ice_server->ProcessStunPacket(packet.get(), tuple);
return;
}
if (is_dtls(buf)) {
if (isDtls(buf)) {
_dtls_transport->ProcessDtlsData((uint8_t *)buf, len);
return;
}
if (is_rtp(buf)) {
if (isRtp(buf, len)) {
if (!_srtp_session_recv) {
WarnL << "received rtp packet when dtls not completed from:" << getPeerAddress(tuple);
return;
@ -329,7 +334,7 @@ void WebRtcTransport::inputSockData(char *buf, int len, RTC::TransportTuple *tup
}
return;
}
if (is_rtcp(buf)) {
if (isRtcp(buf, len)) {
if (!_srtp_session_recv) {
WarnL << "received rtcp packet when dtls not completed from:" << getPeerAddress(tuple);
return;
@ -418,24 +423,27 @@ void WebRtcTransportImp::onDestory() {
}
void WebRtcTransportImp::onSendSockData(Buffer::Ptr buf, bool flush, RTC::TransportTuple *tuple) {
if (!_selected_session) {
if (tuple == nullptr) {
tuple = _ice_server->GetSelectedTuple();
if (!tuple) {
WarnL << "send data failed:" << buf->size();
return;
}
}
// 一次性发送一帧的rtp数据，提高网络io性能
if (_selected_session->getSock()->sockType() == SockNum::Sock_TCP) {
if (tuple->getSock()->sockType() == SockNum::Sock_TCP) {
// 增加tcp两字节头
auto len = buf->size();
char tcp_len[2] = { 0 };
tcp_len[0] = (len >> 8) & 0xff;
tcp_len[1] = len & 0xff;
_selected_session->SockSender::send(tcp_len, 2);
tuple->SockSender::send(tcp_len, 2);
}
_selected_session->send(std::move(buf));
tuple->send(std::move(buf));
if (flush) {
_selected_session->flushAll();
tuple->flushAll();
}
}
@ -1049,28 +1057,14 @@ void WebRtcTransportImp::onBeforeEncryptRtp(const char *buf, int &len, void *ctx
void WebRtcTransportImp::onShutdown(const SockException &ex) {
WarnL << ex.what();
unrefSelf();
for (auto &pr : _history_sessions) {
auto session = pr.second.lock();
if (session) {
session->shutdown(ex);
}
for (auto &tuple : _ice_server->GetTuples()) {
tuple->shutdown(ex);
}
}
void WebRtcTransportImp::setSession(Session::Ptr session) {
_history_sessions.emplace(session.get(), session);
if (_selected_session) {
InfoL << "rtc network changed: " << _selected_session->get_peer_ip() << ":"
<< _selected_session->get_peer_port() << " -> " << session->get_peer_ip() << ":"
<< session->get_peer_port() << ", id:" << getIdentifier();
}
_selected_session = std::move(session);
_selected_session->setSendFlushFlag(false);
unrefSelf();
}
const Session::Ptr &WebRtcTransportImp::getSession() const {
return _selected_session;
void WebRtcTransportImp::removeTuple(RTC::TransportTuple *tuple) {
InfoL << getIdentifier() << " remove tuple " << tuple->get_peer_ip() << ":" << tuple->get_peer_port();
this->_ice_server->RemoveTuple(tuple);
}
uint64_t WebRtcTransportImp::getBytesUsage() const {

13
webrtc/WebRtcTransport.h
View File

@ -110,6 +110,7 @@ public:
void sendRtcpPacket(const char *buf, int len, bool flush, void *ctx = nullptr);
const EventPoller::Ptr& getPoller() const;
Session::Ptr getSession() const;
protected:
//// dtls相关的回调 ////
@ -130,7 +131,6 @@ protected:
protected:
//// ice相关的回调 ///
void OnIceServerSendStunPacket(const RTC::IceServer *iceServer, const RTC::StunPacket *packet, RTC::TransportTuple *tuple) override;
void OnIceServerSelectedTuple(const RTC::IceServer *iceServer, RTC::TransportTuple *tuple) override;
void OnIceServerConnected(const RTC::IceServer *iceServer) override;
void OnIceServerCompleted(const RTC::IceServer *iceServer) override;
void OnIceServerDisconnected(const RTC::IceServer *iceServer) override;
@ -159,7 +159,6 @@ protected:
virtual void onRtcpBye() = 0;
protected:
RTC::TransportTuple* getSelectedTuple() const;
void sendRtcpRemb(uint32_t ssrc, size_t bit_rate);
void sendRtcpPli(uint32_t ssrc);
@ -170,11 +169,11 @@ private:
protected:
RtcSession::Ptr _offer_sdp;
RtcSession::Ptr _answer_sdp;
std::shared_ptr<RTC::IceServer> _ice_server;
private:
std::string _identifier;
EventPoller::Ptr _poller;
std::shared_ptr<RTC::IceServer> _ice_server;
std::shared_ptr<RTC::DtlsTransport> _dtls_transport;
std::shared_ptr<RTC::SrtpSession> _srtp_session_send;
std::shared_ptr<RTC::SrtpSession> _srtp_session_recv;
@ -239,8 +238,6 @@ public:
using Ptr = std::shared_ptr<WebRtcTransportImp>;
~WebRtcTransportImp() override;
void setSession(Session::Ptr session);
const Session::Ptr& getSession() const;
uint64_t getBytesUsage() const;
uint64_t getDuration() const;
bool canSendRtp() const;
@ -248,8 +245,10 @@ public:
void onSendRtp(const RtpPacket::Ptr &rtp, bool flush, bool rtx = false);
void createRtpChannel(const std::string &rid, uint32_t ssrc, MediaTrack &track);
void removeTuple(RTC::TransportTuple* tuple);
protected:
void OnIceServerSelectedTuple(const RTC::IceServer *iceServer, RTC::TransportTuple *tuple) override;
WebRtcTransportImp(const EventPoller::Ptr &poller,bool preferred_tcp = false);
void OnDtlsTransportApplicationDataReceived(const RTC::DtlsTransport *dtlsTransport, const uint8_t *data, size_t len) override;
void onStartWebRTC() override;
@ -292,10 +291,6 @@ private:
Ticker _alive_ticker;
//pli rtcp计时器
Ticker _pli_ticker;
//当前选中的udp链接
Session::Ptr _selected_session;
//链接迁移前后使用过的udp链接
std::unordered_map<Session *, std::weak_ptr<Session> > _history_sessions;
//twcc rtcp发送上下文对象
TwccContext _twcc_ctx;
//根据发送rtp的track类型获取相关信息