增加多路RTP视频流输出

50927548 · hewenyuan · f7433b0f · 50927548 · 50927548 · 50927548
Commit 50927548 authored Nov 27, 2020 by hewenyuan
--- a/server/WebApi.cpp
+++ b/server/WebApi.cpp
 /*
 * Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
 *
 * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
 *
 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
 */
 #include <sys/stat.h>
 #include <math.h>
 #include <signal.h>
 #include <functional>
 #include <sstream>
 #include <unordered_map>
 #include "jsoncpp/json.h"
 #include "Util/util.h"
 #include "Util/logger.h"
 #include "Util/onceToken.h"
 #include "Util/NoticeCenter.h"
 #ifdef ENABLE_MYSQL
 #include "Util/SqlPool.h"
 #endif //ENABLE_MYSQL
 #include "Common/config.h"
 #include "Common/MediaSource.h"
 #include "Http/HttpRequester.h"
 #include "Http/HttpSession.h"
 #include "Network/TcpServer.h"
 #include "Player/PlayerProxy.h"
 #include "Util/MD5.h"
 #include "WebApi.h"
 #include "WebHook.h"
 #include "Thread/WorkThreadPool.h"
 #include "Rtp/RtpSelector.h"
 #include "FFmpegSource.h"
 #if defined(ENABLE_RTPPROXY)
 #include "Rtp/RtpServer.h"
 #endif
 using namespace Json;
 using namespace toolkit;
 using namespace mediakit;
 namespace API {
 typedef enum {
    Exception = -400,//代码抛异常
    InvalidArgs = -300,//参数不合法
    SqlFailed = -200,//sql执行失败
    AuthFailed = -100,//鉴权失败
    OtherFailed = -1,//业务代码执行失败，
    Success = 0//执行成功
 } ApiErr;
 #define API_FIELD "api."
 const string kApiDebug = API_FIELD"apiDebug";
 const string kSecret = API_FIELD"secret";
 const string kSnapRoot = API_FIELD"snapRoot";
 const string kDefaultSnap = API_FIELD"defaultSnap";
 static onceToken token([]() {
    mINI::Instance()[kApiDebug] = "1";
    mINI::Instance()[kSecret] = "035c73f7-bb6b-4889-a715-d9eb2d1925cc";
    mINI::Instance()[kSnapRoot] = "./www/snap/";
    mINI::Instance()[kDefaultSnap] = "./www/logo.png";
 });
 }//namespace API
 class ApiRetException: public std::runtime_error {
 public:
    ApiRetException(const char *str = "success" ,int code = API::Success):runtime_error(str){
        _code = code;
    }
    ~ApiRetException() = default;
    int code(){ return _code; }
 private:
    int _code;
 };
 class AuthException : public ApiRetException {
 public:
    AuthException(const char *str):ApiRetException(str,API::AuthFailed){}
    ~AuthException() = default;
 };
 class InvalidArgsException: public ApiRetException {
 public:
    InvalidArgsException(const char *str):ApiRetException(str,API::InvalidArgs){}
    ~InvalidArgsException() = default;
 };
 class SuccessException: public ApiRetException {
 public:
    SuccessException():ApiRetException("success",API::Success){}
    ~SuccessException() = default;
 };
 #define API_ARGS1 SockInfo &sender,HttpSession::KeyValue &headerIn, HttpSession::KeyValue &headerOut, ApiArgsType &allArgs, Json::Value &val
 #define API_ARGS2 API_ARGS1, const HttpSession::HttpResponseInvoker &invoker
 #define API_ARGS_VALUE1 sender,headerIn,headerOut,allArgs,val
 #define API_ARGS_VALUE2 API_ARGS_VALUE1, invoker
 typedef map<string, variant, StrCaseCompare> ApiArgsType;
 //http api列表
 static map<string, std::function<void(API_ARGS2)> > s_map_api;
 template<typename FUNC>
 static void api_regist1(const string &api_path, FUNC &&func) {
    s_map_api.emplace(api_path, [func](API_ARGS2) {
        func(API_ARGS_VALUE1);
        invoker("200 OK", headerOut, val.toStyledString());
    });
 }
 template<typename FUNC>
 static void api_regist2(const string &api_path, FUNC &&func) {
    s_map_api.emplace(api_path, std::forward<FUNC>(func));
 }
 //获取HTTP请求中url参数、content参数
 static ApiArgsType getAllArgs(const Parser &parser) {
    ApiArgsType allArgs;
    if (parser["Content-Type"].find("application/x-www-form-urlencoded") == 0) {
        auto contentArgs = parser.parseArgs(parser.Content());
        for (auto &pr : contentArgs) {
            allArgs[pr.first] = HttpSession::urlDecode(pr.second);
        }
    } else if (parser["Content-Type"].find("application/json") == 0) {
        try {
            stringstream ss(parser.Content());
            Value jsonArgs;
            ss >> jsonArgs;
            auto keys = jsonArgs.getMemberNames();
            for (auto key = keys.begin(); key != keys.end(); ++key) {
                allArgs[*key] = jsonArgs[*key].asString();
            }
        } catch (std::exception &ex) {
            WarnL << ex.what();
        }
    } else if (!parser["Content-Type"].empty()) {
        WarnL << "invalid Content-Type:" << parser["Content-Type"];
    }
    for (auto &pr :  parser.getUrlArgs()) {
        allArgs[pr.first] = pr.second;
    }
    return allArgs;
 }
 static inline void addHttpListener(){
    GET_CONFIG(bool, api_debug, API::kApiDebug);
    //注册监听kBroadcastHttpRequest事件
    NoticeCenter::Instance().addListener(nullptr, Broadcast::kBroadcastHttpRequest, [](BroadcastHttpRequestArgs) {
        auto it = s_map_api.find(parser.Url());
        if (it == s_map_api.end()) {
            return;
        }
        //该api已被消费
        consumed = true;
        //执行API
        Json::Value val;
        val["code"] = API::Success;
        HttpSession::KeyValue headerOut;
        auto allArgs = getAllArgs(parser);
        HttpSession::KeyValue &headerIn = parser.getHeader();
        GET_CONFIG(string,charSet,Http::kCharSet);
        headerOut["Content-Type"] = StrPrinter << "application/json; charset=" << charSet;
        if(api_debug){
            auto newInvoker = [invoker,parser,allArgs](const string &codeOut,
                                                       const HttpSession::KeyValue &headerOut,
                                                       const HttpBody::Ptr &body){
                stringstream ss;
                for(auto &pr : allArgs ){
                    ss << pr.first << " : " << pr.second << "\r\n";
                }
                //body默认为空
                int64_t size = 0;
                if (body && body->remainSize()) {
                    //有body，获取body大小
                    size = body->remainSize();
                }
                if(size && size < 4 * 1024){
                    string contentOut = body->readData(size)->toString();
                    DebugL << "\r\n# request:\r\n" << parser.Method() << " " << parser.FullUrl() << "\r\n"
                           << "# content:\r\n" << parser.Content() << "\r\n"
                           << "# args:\r\n" << ss.str()
                           << "# response:\r\n"
                           << contentOut << "\r\n";
                    invoker(codeOut,headerOut,contentOut);
                } else{
                    DebugL << "\r\n# request:\r\n" << parser.Method() << " " << parser.FullUrl() << "\r\n"
                           << "# content:\r\n" << parser.Content() << "\r\n"
                           << "# args:\r\n" << ss.str()
                           << "# response size:"
                           << size <<"\r\n";
                    invoker(codeOut,headerOut,body);
                }
            };
            ((HttpSession::HttpResponseInvoker &)invoker) = newInvoker;
        }
        try {
            it->second(sender,headerIn, headerOut, allArgs, val, invoker);
        }  catch(ApiRetException &ex){
            val["code"] = ex.code();
            val["msg"] = ex.what();
            invoker("200 OK", headerOut, val.toStyledString());
        }
 #ifdef ENABLE_MYSQL
        catch(SqlException &ex){
            val["code"] = API::SqlFailed;
            val["msg"] = StrPrinter << "操作数据库失败:" << ex.what() << ":" << ex.getSql();
            WarnL << ex.what() << ":" << ex.getSql();
            invoker("200 OK", headerOut, val.toStyledString());
        }
 #endif// ENABLE_MYSQL
        catch (std::exception &ex) {
            val["code"] = API::Exception;
            val["msg"] = ex.what();
            invoker("200 OK", headerOut, val.toStyledString());
        }
    });
 }
 template <typename Args,typename First>
 bool checkArgs(Args &&args,First &&first){
    return !args[first].empty();
 }
 template <typename Args,typename First,typename ...KeyTypes>
 bool checkArgs(Args &&args,First &&first,KeyTypes && ...keys){
    return !args[first].empty() && checkArgs(std::forward<Args>(args),std::forward<KeyTypes>(keys)...);
 }
 #define CHECK_ARGS(...)  \
    if(!checkArgs(allArgs,##__VA_ARGS__)){ \
        throw InvalidArgsException("缺少必要参数:" #__VA_ARGS__); \
    }
 #define CHECK_SECRET() \
    if(sender.get_peer_ip() != "127.0.0.1"){ \
        CHECK_ARGS("secret"); \
        if(api_secret != allArgs["secret"]){ \
            throw AuthException("secret错误"); \
        } \
    }
 //拉流代理器列表
 static unordered_map<string ,PlayerProxy::Ptr> s_proxyMap;
 static recursive_mutex s_proxyMapMtx;
 //FFmpeg拉流代理器列表
 static unordered_map<string ,FFmpegSource::Ptr> s_ffmpegMap;
 static recursive_mutex s_ffmpegMapMtx;
 #if defined(ENABLE_RTPPROXY)
 //rtp服务器列表
 static unordered_map<string, RtpServer::Ptr> s_rtpServerMap;
 static recursive_mutex s_rtpServerMapMtx;
 #endif
 static inline string getProxyKey(const string &vhost,const string &app,const string &stream){
    return vhost + "/" + app + "/" + stream;
 }
 /**
 * 安装api接口
 * 所有api都支持GET和POST两种方式
 * POST方式参数支持application/json和application/x-www-form-urlencoded方式
 */
 void installWebApi() {
    addHttpListener();
    GET_CONFIG(string,api_secret,API::kSecret);
    //获取线程负载
    //测试url http://127.0.0.1/index/api/getThreadsLoad
    api_regist2("/index/api/getThreadsLoad",[](API_ARGS2){
        EventPollerPool::Instance().getExecutorDelay([invoker, headerOut](const vector<int> &vecDelay) {
            Value val;
            auto vec = EventPollerPool::Instance().getExecutorLoad();
            int i = API::Success;
            for (auto load : vec) {
                Value obj(objectValue);
                obj["load"] = load;
                obj["delay"] = vecDelay[i++];
                val["data"].append(obj);
            }
            val["code"] = API::Success;
            invoker("200 OK", headerOut, val.toStyledString());
        });
    });
    //获取后台工作线程负载
    //测试url http://127.0.0.1/index/api/getWorkThreadsLoad
    api_regist2("/index/api/getWorkThreadsLoad", [](API_ARGS2){
        WorkThreadPool::Instance().getExecutorDelay([invoker, headerOut](const vector<int> &vecDelay) {
            Value val;
            auto vec = WorkThreadPool::Instance().getExecutorLoad();
            int i = 0;
            for (auto load : vec) {
                Value obj(objectValue);
                obj["load"] = load;
                obj["delay"] = vecDelay[i++];
                val["data"].append(obj);
            }
            val["code"] = API::Success;
            invoker("200 OK", headerOut, val.toStyledString());
        });
    });
    //获取服务器配置
    //测试url http://127.0.0.1/index/api/getServerConfig
    api_regist1("/index/api/getServerConfig",[](API_ARGS1){
        CHECK_SECRET();
        Value obj;
        for (auto &pr : mINI::Instance()) {
            obj[pr.first] = (string &) pr.second;
        }
        val["data"].append(obj);
    });
    //设置服务器配置
    //测试url(比如关闭http api调试) http://127.0.0.1/index/api/setServerConfig?api.apiDebug=0
    //你也可以通过http post方式传参，可以通过application/x-www-form-urlencoded或application/json方式传参
    api_regist1("/index/api/setServerConfig",[](API_ARGS1){
        CHECK_SECRET();
        auto &ini = mINI::Instance();
        int changed = API::Success;
        for (auto &pr : allArgs) {
            if (ini.find(pr.first) == ini.end()) {
                //没有这个key
                continue;
            }
            if (ini[pr.first] == pr.second) {
                continue;
            }
            ini[pr.first] = pr.second;
            //替换成功
            ++changed;
        }
        if (changed > 0) {
            NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastReloadConfig);
            ini.dumpFile(g_ini_file);
        }
        val["changed"] = changed;
    });
    static auto s_get_api_list = [](API_ARGS1){
        CHECK_SECRET();
        for(auto &pr : s_map_api){
            val["data"].append(pr.first);
        }
    };
    //获取服务器api列表
    //测试url http://127.0.0.1/index/api/getApiList
    api_regist1("/index/api/getApiList",[](API_ARGS1){
        s_get_api_list(API_ARGS_VALUE1);
    });
    //获取服务器api列表
    //测试url http://127.0.0.1/index/
    api_regist1("/index/",[](API_ARGS1){
        s_get_api_list(API_ARGS_VALUE1);
    });
 #if !defined(_WIN32)
    //重启服务器,只有Daemon方式才能重启，否则是直接关闭！
    //测试url http://127.0.0.1/index/api/restartServer
    api_regist1("/index/api/restartServer",[](API_ARGS1){
        CHECK_SECRET();
        EventPollerPool::Instance().getPoller()->doDelayTask(1000,[](){
            //尝试正常退出
            ::kill(getpid(), SIGINT);
            //3秒后强制退出
            EventPollerPool::Instance().getPoller()->doDelayTask(3000,[](){
                exit(0);
                return 0;
            });
            return 0;
        });
        val["msg"] = "服务器将在一秒后自动重启";
    });
 #endif//#if !defined(_WIN32)
    static auto makeMediaSourceJson = [](const MediaSource::Ptr &media){
        Value item;
        item["schema"] = media->getSchema();
        item["vhost"] = media->getVhost();
        item["app"] = media->getApp();
        item["stream"] = media->getId();
        item["createStamp"] = (Json::UInt64) media->getCreateStamp();
        item["aliveSecond"] = (Json::UInt64) media->getAliveSecond();
        item["bytesSpeed"] = media->getBytesSpeed();
        item["readerCount"] = media->readerCount();
        item["totalReaderCount"] = media->totalReaderCount();
        item["originType"] = (int) media->getOriginType();
        item["originTypeStr"] = getOriginTypeString(media->getOriginType());
        item["originUrl"] = media->getOriginUrl();
        auto originSock = media->getOriginSock();
        if (originSock) {
            item["originSock"]["local_ip"] = originSock->get_local_ip();
            item["originSock"]["local_port"] = originSock->get_local_port();
            item["originSock"]["peer_ip"] = originSock->get_peer_ip();
            item["originSock"]["peer_port"] = originSock->get_peer_port();
            item["originSock"]["identifier"] = originSock->getIdentifier();
        } else {
            item["originSock"] = Json::nullValue;
        }
        for(auto &track : media->getTracks()){
            Value obj;
            auto codec_type = track->getTrackType();
            obj["codec_id"] = track->getCodecId();
            obj["codec_id_name"] = track->getCodecName();
            obj["ready"] = track->ready();
            obj["codec_type"] = codec_type;
            switch(codec_type){
                case TrackAudio : {
                    auto audio_track = dynamic_pointer_cast<AudioTrack>(track);
                    obj["sample_rate"] = audio_track->getAudioSampleRate();
                    obj["channels"] = audio_track->getAudioChannel();
                    obj["sample_bit"] = audio_track->getAudioSampleBit();
                    break;
                }
                case TrackVideo : {
                    auto video_track = dynamic_pointer_cast<VideoTrack>(track);
                    obj["width"] = video_track->getVideoWidth();
                    obj["height"] = video_track->getVideoHeight();
                    obj["fps"] = round(video_track->getVideoFps());
                    break;
                }
                default:
                    break;
            }
            item["tracks"].append(obj);
        }
        return item;
    };
    //获取流列表，可选筛选参数
    //测试url0(获取所有流) http://127.0.0.1/index/api/getMediaList
    //测试url1(获取虚拟主机为"__defaultVost__"的流) http://127.0.0.1/index/api/getMediaList?vhost=__defaultVost__
    //测试url2(获取rtsp类型的流) http://127.0.0.1/index/api/getMediaList?schema=rtsp
    api_regist1("/index/api/getMediaList",[](API_ARGS1){
        CHECK_SECRET();
        //获取所有MediaSource列表
        MediaSource::for_each_media([&](const MediaSource::Ptr &media){
            if(!allArgs["schema"].empty() && allArgs["schema"] != media->getSchema()){
                return;
            }
            if(!allArgs["vhost"].empty() && allArgs["vhost"] != media->getVhost()){
                return;
            }
            if(!allArgs["app"].empty() && allArgs["app"] != media->getApp()){
                return;
            }
            val["data"].append(makeMediaSourceJson(media));
        });
    });
    //测试url http://127.0.0.1/index/api/isMediaOnline?schema=rtsp&vhost=__defaultVhost__&app=live&stream=obs
    api_regist1("/index/api/isMediaOnline",[](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("schema","vhost","app","stream");
        val["online"] = (bool) (MediaSource::find(allArgs["schema"],allArgs["vhost"],allArgs["app"],allArgs["stream"]));
    });
    //测试url http://127.0.0.1/index/api/getMediaInfo?schema=rtsp&vhost=__defaultVhost__&app=live&stream=obs
    api_regist1("/index/api/getMediaInfo",[](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("schema","vhost","app","stream");
        auto src = MediaSource::find(allArgs["schema"],allArgs["vhost"],allArgs["app"],allArgs["stream"]);
        if(!src){
            val["online"] = false;
            return;
        }
        val = makeMediaSourceJson(src);
        val["online"] = true;
        val["code"] = API::Success;
    });
    //主动关断流，包括关断拉流、推流
    //测试url http://127.0.0.1/index/api/close_stream?schema=rtsp&vhost=__defaultVhost__&app=live&stream=obs&force=1
    api_regist1("/index/api/close_stream",[](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("schema","vhost","app","stream");
        //踢掉推流器
        auto src = MediaSource::find(allArgs["schema"],
                                     allArgs["vhost"],
                                     allArgs["app"],
                                     allArgs["stream"]);
        if (src) {
            bool flag = src->close(allArgs["force"].as<bool>());
            val["result"] = flag ? 0 : -1;
            val["msg"] = flag ? "success" : "close failed";
            val["code"] = flag ? API::Success : API::OtherFailed;
        } else {
            val["result"] = -2;
            val["msg"] = "can not find the stream";
            val["code"] = API::OtherFailed;
        }
    });
    //批量主动关断流，包括关断拉流、推流
    //测试url http://127.0.0.1/index/api/close_streams?schema=rtsp&vhost=__defaultVhost__&app=live&stream=obs&force=1
    api_regist1("/index/api/close_streams",[](API_ARGS1){
        CHECK_SECRET();
        //筛选命中个数
        int count_hit = 0;
        int count_closed = 0;
        list<MediaSource::Ptr> media_list;
        MediaSource::for_each_media([&](const MediaSource::Ptr &media){
            if(!allArgs["schema"].empty() && allArgs["schema"] != media->getSchema()){
                return;
            }
            if(!allArgs["vhost"].empty() && allArgs["vhost"] != media->getVhost()){
                return;
            }
            if(!allArgs["app"].empty() && allArgs["app"] != media->getApp()){
                return;
            }
            if(!allArgs["stream"].empty() && allArgs["stream"] != media->getId()){
                return;
            }
            ++count_hit;
            media_list.emplace_back(media);
        });
        bool force = allArgs["force"].as<bool>();
        for(auto &media : media_list){
            if(media->close(force)){
                ++count_closed;
            }
        }
        val["count_hit"] = count_hit;
        val["count_closed"] = count_closed;
    });
    //获取所有TcpSession列表信息
    //可以根据本地端口和远端ip来筛选
    //测试url(筛选某端口下的tcp会话) http://127.0.0.1/index/api/getAllSession?local_port=1935
    api_regist1("/index/api/getAllSession",[](API_ARGS1){
        CHECK_SECRET();
        Value jsession;
        uint16_t local_port = allArgs["local_port"].as<uint16_t>();
        string &peer_ip = allArgs["peer_ip"];
        SessionMap::Instance().for_each_session([&](const string &id,const TcpSession::Ptr &session){
            if(local_port != 0 && local_port != session->get_local_port()){
                return;
            }
            if(!peer_ip.empty() && peer_ip != session->get_peer_ip()){
                return;
            }
            jsession["peer_ip"] = session->get_peer_ip();
            jsession["peer_port"] = session->get_peer_port();
            jsession["local_ip"] = session->get_local_ip();
            jsession["local_port"] = session->get_local_port();
            jsession["id"] = id;
            jsession["typeid"] = typeid(*session).name();
            val["data"].append(jsession);
        });
    });
    //断开tcp连接，比如说可以断开rtsp、rtmp播放器等
    //测试url http://127.0.0.1/index/api/kick_session?id=123456
    api_regist1("/index/api/kick_session",[](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("id");
        //踢掉tcp会话
        auto session = SessionMap::Instance().get(allArgs["id"]);
        if(!session){
            throw ApiRetException("can not find the target",API::OtherFailed);
        }
        session->safeShutdown();
    });
    //批量断开tcp连接，比如说可以断开rtsp、rtmp播放器等
    //测试url http://127.0.0.1/index/api/kick_sessions?local_port=1935
    api_regist1("/index/api/kick_sessions",[](API_ARGS1){
        CHECK_SECRET();
        uint16_t local_port = allArgs["local_port"].as<uint16_t>();
        string &peer_ip = allArgs["peer_ip"];
        uint64_t count_hit = 0;
        list<TcpSession::Ptr> session_list;
        SessionMap::Instance().for_each_session([&](const string &id,const TcpSession::Ptr &session){
            if(local_port != 0 && local_port != session->get_local_port()){
                return;
            }
            if(!peer_ip.empty() && peer_ip != session->get_peer_ip()){
                return;
            }
            session_list.emplace_back(session);
            ++count_hit;
        });
        for(auto &session : session_list){
            session->safeShutdown();
        }
        val["count_hit"] = (Json::UInt64)count_hit;
    });
    static auto addStreamProxy = [](const string &vhost,
                                    const string &app,
                                    const string &stream,
                                    const string &url,
                                    bool enable_hls,
                                    bool enable_mp4,
                                    int rtp_type,
                                    const function<void(const SockException &ex,const string &key)> &cb){
        auto key = getProxyKey(vhost,app,stream);
        lock_guard<recursive_mutex> lck(s_proxyMapMtx);
        if(s_proxyMap.find(key) != s_proxyMap.end()){
            //已经在拉流了
            cb(SockException(Err_success),key);
            return;
        }
        //添加拉流代理
        PlayerProxy::Ptr player(new PlayerProxy(vhost, app, stream, enable_hls, enable_mp4));
        s_proxyMap[key] = player;
        //指定RTP over TCP(播放rtsp时有效)
        (*player)[kRtpType] = rtp_type;
        //开始播放，如果播放失败或者播放中止，将会自动重试若干次，默认一直重试
        player->setPlayCallbackOnce([cb,key](const SockException &ex){
            if(ex){
                lock_guard<recursive_mutex> lck(s_proxyMapMtx);
                s_proxyMap.erase(key);
            }
            cb(ex,key);
        });
        //被主动关闭拉流
        player->setOnClose([key](){
            lock_guard<recursive_mutex> lck(s_proxyMapMtx);
            s_proxyMap.erase(key);
        });
        player->play(url);
    };
    //动态添加rtsp/rtmp拉流代理
    //测试url http://127.0.0.1/index/api/addStreamProxy?vhost=__defaultVhost__&app=proxy&enable_rtsp=1&enable_rtmp=1&stream=0&url=rtmp://127.0.0.1/live/obs
    api_regist2("/index/api/addStreamProxy",[](API_ARGS2){
        CHECK_SECRET();
        CHECK_ARGS("vhost","app","stream","url");
        addStreamProxy(allArgs["vhost"],
                       allArgs["app"],
                       allArgs["stream"],
                       allArgs["url"],
                       allArgs["enable_hls"],/* 是否hls转发 */
                       allArgs["enable_mp4"],/* 是否MP4录制 */
                       allArgs["rtp_type"],
                       [invoker,val,headerOut](const SockException &ex,const string &key){
                           if(ex){
                               const_cast<Value &>(val)["code"] = API::OtherFailed;
                               const_cast<Value &>(val)["msg"] = ex.what();
                           }else{
                               const_cast<Value &>(val)["data"]["key"] = key;
                           }
                           invoker("200 OK", headerOut, val.toStyledString());
                       });
    });
    //关闭拉流代理
    //测试url http://127.0.0.1/index/api/delStreamProxy?key=__defaultVhost__/proxy/0
    api_regist1("/index/api/delStreamProxy",[](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("key");
        lock_guard<recursive_mutex> lck(s_proxyMapMtx);
        val["data"]["flag"] = s_proxyMap.erase(allArgs["key"]) == 1;
    });
    static auto addFFmpegSource = [](const string &src_url,
                                     const string &dst_url,
                                     int timeout_ms,
                                     const function<void(const SockException &ex,const string &key)> &cb){
        auto key = MD5(dst_url).hexdigest();
        lock_guard<decltype(s_ffmpegMapMtx)> lck(s_ffmpegMapMtx);
        if(s_ffmpegMap.find(key) != s_ffmpegMap.end()){
            //已经在拉流了
            cb(SockException(Err_success),key);
            return;
        }
        FFmpegSource::Ptr ffmpeg = std::make_shared<FFmpegSource>();
        s_ffmpegMap[key] = ffmpeg;
        ffmpeg->setOnClose([key](){
            lock_guard<decltype(s_ffmpegMapMtx)> lck(s_ffmpegMapMtx);
            s_ffmpegMap.erase(key);
        });
        ffmpeg->play(src_url, dst_url,timeout_ms,[cb , key](const SockException &ex){
            if(ex){
                lock_guard<decltype(s_ffmpegMapMtx)> lck(s_ffmpegMapMtx);
                s_ffmpegMap.erase(key);
            }
            cb(ex,key);
        });
    };
    //动态添加rtsp/rtmp拉流代理
    //测试url http://127.0.0.1/index/api/addFFmpegSource?src_url=http://live.hkstv.hk.lxdns.com/live/hks2/playlist.m3u8&dst_url=rtmp://127.0.0.1/live/hks2&timeout_ms=10000
    api_regist2("/index/api/addFFmpegSource",[](API_ARGS2){
        CHECK_SECRET();
        CHECK_ARGS("src_url","dst_url","timeout_ms");
        auto src_url = allArgs["src_url"];
        auto dst_url = allArgs["dst_url"];
        int timeout_ms = allArgs["timeout_ms"];
        addFFmpegSource(src_url,dst_url,timeout_ms,[invoker,val,headerOut](const SockException &ex,const string &key){
            if(ex){
                const_cast<Value &>(val)["code"] = API::OtherFailed;
                const_cast<Value &>(val)["msg"] = ex.what();
            }else{
                const_cast<Value &>(val)["data"]["key"] = key;
            }
            invoker("200 OK", headerOut, val.toStyledString());
        });
    });
    static auto api_delFFmpegSource = [](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("key");
        lock_guard<decltype(s_ffmpegMapMtx)> lck(s_ffmpegMapMtx);
        val["data"]["flag"] = s_ffmpegMap.erase(allArgs["key"]) == 1;
    };
    //关闭拉流代理
    //测试url http://127.0.0.1/index/api/delFFmepgSource?key=key
    api_regist1("/index/api/delFFmpegSource",[](API_ARGS1){
        api_delFFmpegSource(API_ARGS_VALUE1);
    });
    //此处为了兼容之前的拼写错误
    api_regist1("/index/api/delFFmepgSource",[](API_ARGS1){
        api_delFFmpegSource(API_ARGS_VALUE1);
    });
    //新增http api下载可执行程序文件接口
    //测试url http://127.0.0.1/index/api/downloadBin
    api_regist2("/index/api/downloadBin",[](API_ARGS2){
        CHECK_SECRET();
        invoker.responseFile(headerIn,StrCaseMap(),exePath());
    });
 #if defined(ENABLE_RTPPROXY)
    api_regist1("/index/api/getRtpInfo",[](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("stream_id");
        auto process = RtpSelector::Instance().getProcess(allArgs["stream_id"], false);
        if (!process) {
            val["exist"] = false;
            return;
        }
        val["exist"] = true;
        val["peer_ip"] = process->get_peer_ip();
        val["peer_port"] = process->get_peer_port();
        val["local_port"] = process->get_local_port();
        val["local_ip"] = process->get_local_ip();
    });
    api_regist1("/index/api/openRtpServer",[](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("port", "enable_tcp", "stream_id");
        auto stream_id = allArgs["stream_id"];
        lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
        if(s_rtpServerMap.find(stream_id) != s_rtpServerMap.end()) {
            //为了防止RtpProcess所有权限混乱的问题，不允许重复添加相同的stream_id
            throw InvalidArgsException("该stream_id已存在");
        }
        RtpServer::Ptr server = std::make_shared<RtpServer>();
        server->start(allArgs["port"], stream_id, allArgs["enable_tcp"].as<bool>());
        server->setOnDetach([stream_id]() {
            //设置rtp超时移除事件
            lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
            s_rtpServerMap.erase(stream_id);
        });
        //保存对象
        s_rtpServerMap.emplace(stream_id, server);
        //回复json
        val["port"] = server->getPort();
    });
    api_regist1("/index/api/closeRtpServer",[](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("stream_id");
        lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
        auto it = s_rtpServerMap.find(allArgs["stream_id"]);
        if(it == s_rtpServerMap.end()){
            val["hit"] = 0;
            return;
        }
        auto server = it->second;
        s_rtpServerMap.erase(it);
        val["hit"] = 1;
    });
    api_regist1("/index/api/listRtpServer",[](API_ARGS1){
        CHECK_SECRET();
        lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
        for (auto &pr : s_rtpServerMap) {
            Value obj;
            obj["stream_id"] = pr.first;
            obj["port"] = pr.second->getPort();
            val["data"].append(obj);
        }
    });
    api_regist2("/index/api/startSendRtp",[](API_ARGS2){
        CHECK_SECRET();
-        CHECK_ARGS("vhost", "app", "stream", "ssrc", "dst_url", "dst_port", "is_udp");
+        CHECK_ARGS("vhost", "app", "stream", "ssrc", "dst_url", "dst_port", "is_udp", "src_port");
        auto src = MediaSource::find(allArgs["vhost"], allArgs["app"], allArgs["stream"]);
        if (!src) {
            throw ApiRetException("该媒体流不存在", API::OtherFailed);
        }
-        src->startSendRtp(allArgs["dst_url"], allArgs["dst_port"], allArgs["ssrc"], allArgs["is_udp"], [val, headerOut, invoker](const SockException &ex){
+        src->startSendRtp(allArgs["dst_url"], allArgs["dst_port"], allArgs["ssrc"], allArgs["is_udp"], allArgs["src_port"], [val, headerOut, invoker](const SockException &ex){
            if (ex) {
                const_cast<Value &>(val)["code"] = API::OtherFailed;
                const_cast<Value &>(val)["msg"] = ex.what();
            }
            invoker("200 OK", headerOut, val.toStyledString());
        });
    });
    api_regist1("/index/api/stopSendRtp",[](API_ARGS1){
        CHECK_SECRET();
-        CHECK_ARGS("vhost", "app", "stream");
+        CHECK_ARGS("vhost", "app", "stream", "ssrc");
        auto src = MediaSource::find(allArgs["vhost"], allArgs["app"], allArgs["stream"]);
        if (!src) {
            throw ApiRetException("该媒体流不存在", API::OtherFailed);
        }
-        if (!src->stopSendRtp()) {
+        if (!src->stopSendRtp(allArgs["ssrc"])) {
            throw ApiRetException("尚未开始推流,停止失败", API::OtherFailed);
        }
    });
 #endif//ENABLE_RTPPROXY
    // 开始录制hls或MP4
    api_regist1("/index/api/startRecord",[](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("type","vhost","app","stream");
        auto result = Recorder::startRecord((Recorder::type) allArgs["type"].as<int>(),
                                              allArgs["vhost"],
                                              allArgs["app"],
                                              allArgs["stream"],
                                              allArgs["customized_path"]);
        val["result"] = result;
        val["code"] = result ? API::Success : API::OtherFailed;
        val["msg"] = result ? "success" :  "start record failed";
    });
    // 停止录制hls或MP4
    api_regist1("/index/api/stopRecord",[](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("type","vhost","app","stream");
        auto result = Recorder::stopRecord((Recorder::type) allArgs["type"].as<int>(),
                                             allArgs["vhost"],
                                             allArgs["app"],
                                             allArgs["stream"]);
        val["result"] = result;
        val["code"] = result ? API::Success : API::OtherFailed;
        val["msg"] = result ? "success" :  "stop record failed";
    });
    // 获取hls或MP4录制状态
    api_regist1("/index/api/isRecording",[](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("type","vhost","app","stream");
        val["status"] = Recorder::isRecording((Recorder::type) allArgs["type"].as<int>(),
                                              allArgs["vhost"],
                                              allArgs["app"],
                                              allArgs["stream"]);
    });
    //获取录像文件夹列表或mp4文件列表
    //http://127.0.0.1/index/api/getMp4RecordFile?vhost=__defaultVhost__&app=live&stream=ss&period=2020-01
    api_regist1("/index/api/getMp4RecordFile", [](API_ARGS1){
        CHECK_SECRET();
        CHECK_ARGS("vhost", "app", "stream");
        auto record_path = Recorder::getRecordPath(Recorder::type_mp4, allArgs["vhost"], allArgs["app"],allArgs["stream"]);
        auto period = allArgs["period"];
        //判断是获取mp4文件列表还是获取文件夹列表
        bool search_mp4 = period.size() == sizeof("2020-02-01") - 1;
        if (search_mp4) {
            record_path = record_path + period + "/";
        }
        Json::Value paths(arrayValue);
        //这是筛选日期，获取文件夹列表
        File::scanDir(record_path, [&](const string &path, bool isDir) {
            int pos = path.rfind('/');
            if (pos != string::npos) {
                string relative_path = path.substr(pos + 1);
                if (search_mp4) {
                    if (!isDir) {
                        //我们只收集mp4文件，对文件夹不感兴趣
                        paths.append(relative_path);
                    }
                } else if (isDir && relative_path.find(period) == 0) {
                    //匹配到对应日期的文件夹
                    paths.append(relative_path);
                }
            }
            return true;
        }, false);
        val["data"]["rootPath"] = record_path;
        val["data"]["paths"] = paths;
    });
    static auto responseSnap = [](const string &snap_path,
                                  const HttpSession::KeyValue &headerIn,
                                  const HttpSession::HttpResponseInvoker &invoker) {
        StrCaseMap headerOut;
        struct stat statbuf = {0};
        GET_CONFIG(string, defaultSnap, API::kDefaultSnap);
        if (!(stat(snap_path.data(), &statbuf) == 0 && statbuf.st_size != 0) && !defaultSnap.empty()) {
            //空文件且设置了预设图，则返回预设图片(也就是FFmpeg生成截图中空档期的默认图片)
            const_cast<string&>(snap_path) = File::absolutePath(defaultSnap, "");
            headerOut["Content-Type"] = HttpFileManager::getContentType(snap_path.data());
        } else {
            //之前生成的截图文件，我们默认为jpeg格式
            headerOut["Content-Type"] = HttpFileManager::getContentType(".jpeg");
        }
        //返回图片给http客户端
        invoker.responseFile(headerIn, headerOut, snap_path);
    };
    //获取截图缓存或者实时截图
    //http://127.0.0.1/index/api/getSnap?url=rtmp://127.0.0.1/record/robot.mp4&timeout_sec=10&expire_sec=3
    api_regist2("/index/api/getSnap", [](API_ARGS2){
        CHECK_SECRET();
        CHECK_ARGS("url", "timeout_sec", "expire_sec");
        GET_CONFIG(string, snap_root, API::kSnapRoot);
        bool have_old_snap = false, res_old_snap = false;
        int expire_sec = allArgs["expire_sec"];
        auto scan_path = File::absolutePath(MD5(allArgs["url"]).hexdigest(), snap_root) + "/";
        string new_snap = StrPrinter << scan_path << time(NULL) << ".jpeg";
        File::scanDir(scan_path, [&](const string &path, bool isDir) {
            if (isDir || !end_with(path, ".jpeg")) {
                //忽略文件夹或其他类型的文件
                return true;
            }
            //找到截图
            auto tm = FindField(path.data() + scan_path.size(), nullptr, ".jpeg");
            if (atoll(tm.data()) + expire_sec < time(NULL)) {
                //截图已经过期，改名，以便再次请求时，可以返回老截图
                rename(path.data(), new_snap.data());
                have_old_snap = true;
                return true;
            }
            //截图存在，且未过期，那么返回之
            res_old_snap = true;
            responseSnap(path, headerIn, invoker);
            //中断遍历
            return false;
        });
        if (res_old_snap) {
            //已经回复了旧的截图
            return;
        }
        //无截图或者截图已经过期
        if (!have_old_snap) {
            //无过期截图，生成一个空文件，目的是顺便创建文件夹路径
            //同时防止在FFmpeg生成截图途中不停的尝试调用该api多次启动FFmpeg进程
            auto file = File::create_file(new_snap.data(), "wb");
            if (file) {
                fclose(file);
            }
        }
        //启动FFmpeg进程，开始截图，生成临时文件，截图成功后替换为正式文件
        auto new_snap_tmp = new_snap + ".tmp";
        FFmpegSnap::makeSnap(allArgs["url"], new_snap_tmp, allArgs["timeout_sec"], [invoker, headerIn, new_snap, new_snap_tmp](bool success) {
            if (!success) {
                //生成截图失败，可能残留空文件
                File::delete_file(new_snap_tmp.data());
            } else {
                //临时文件改成正式文件
                File::delete_file(new_snap.data());
                rename(new_snap_tmp.data(), new_snap.data());
            }
            responseSnap(new_snap, headerIn, invoker);
        });
    });
    ////////////以下是注册的Hook API////////////
    api_regist1("/index/hook/on_publish",[](API_ARGS1){
        //开始推流事件
        //转换成rtsp或rtmp
        val["enableRtxp"] = true;
        //转换hls
        val["enableHls"] = true;
        //不录制mp4
        val["enableMP4"] = false;
    });
    api_regist1("/index/hook/on_play",[](API_ARGS1){
        //开始播放事件
    });
    api_regist1("/index/hook/on_flow_report",[](API_ARGS1){
        //流量统计hook api
    });
    api_regist1("/index/hook/on_rtsp_realm",[](API_ARGS1){
        //rtsp是否需要鉴权，默认需要鉴权
        val["code"] = API::Success;
        val["realm"] = "zlmediakit_reaml";
    });
    api_regist1("/index/hook/on_rtsp_auth",[](API_ARGS1){
        //rtsp鉴权密码，密码等于用户名
        //rtsp可以有双重鉴权！后面还会触发on_play事件
        CHECK_ARGS("user_name");
        val["code"] = API::Success;
        val["encrypted"] = false;
        val["passwd"] = allArgs["user_name"].data();
    });
    api_regist1("/index/hook/on_stream_changed",[](API_ARGS1){
        //媒体注册或反注册事件
    });
 #if !defined(_WIN32)
    api_regist2("/index/hook/on_stream_not_found_ffmpeg",[](API_ARGS2){
        //媒体未找到事件,我们都及时拉流hks作为替代品，目的是为了测试按需拉流
        CHECK_SECRET();
        CHECK_ARGS("vhost","app","stream");
        //通过FFmpeg按需拉流
        GET_CONFIG(int,rtmp_port,Rtmp::kPort);
        GET_CONFIG(int,timeout_sec,Hook::kTimeoutSec);
        string dst_url = StrPrinter
                << "rtmp://127.0.0.1:"
                << rtmp_port << "/"
                << allArgs["app"] << "/"
                << allArgs["stream"] << "?vhost="
                << allArgs["vhost"];
        addFFmpegSource("http://hls-ott-zhibo.wasu.tv/live/272/index.m3u8",/** ffmpeg拉流支持任意编码格式任意协议 **/
                        dst_url,
                        (1000 * timeout_sec) - 500,
                        [invoker,val,headerOut](const SockException &ex,const string &key){
                            if(ex){
                                const_cast<Value &>(val)["code"] = API::OtherFailed;
                                const_cast<Value &>(val)["msg"] = ex.what();
                            }else{
                                const_cast<Value &>(val)["data"]["key"] = key;
                            }
                            invoker("200 OK", headerOut, val.toStyledString());
                        });
    });
 #endif//!defined(_WIN32)
    api_regist2("/index/hook/on_stream_not_found",[](API_ARGS2){
        //媒体未找到事件,我们都及时拉流hks作为替代品，目的是为了测试按需拉流
        CHECK_SECRET();
        CHECK_ARGS("vhost","app","stream");
        //通过内置支持的rtsp/rtmp按需拉流
        addStreamProxy(allArgs["vhost"],
                       allArgs["app"],
                       allArgs["stream"],
                       /** 支持rtsp和rtmp方式拉流 ，rtsp支持h265/h264/aac,rtmp仅支持h264/aac **/
                       "rtsp://184.72.239.149/vod/mp4:BigBuckBunny_115k.mov",
                       true,/* 开启hls转发 */
                       false,/* 禁用MP4录制 */
                       0,//rtp over tcp方式拉流
                       [invoker,val,headerOut](const SockException &ex,const string &key){
                           if(ex){
                               const_cast<Value &>(val)["code"] = API::OtherFailed;
                               const_cast<Value &>(val)["msg"] = ex.what();
                           }else{
                               const_cast<Value &>(val)["data"]["key"] = key;
                           }
                           invoker("200 OK", headerOut, val.toStyledString());
                       });
    });
    api_regist1("/index/hook/on_record_mp4",[](API_ARGS1){
        //录制mp4分片完毕事件
    });
    api_regist1("/index/hook/on_shell_login",[](API_ARGS1){
        //shell登录调试事件
    });
    api_regist1("/index/hook/on_stream_none_reader",[](API_ARGS1){
        //无人观看流默认关闭
        val["close"] = true;
    });
    static auto checkAccess = [](const string &params){
        //我们假定大家都要权限访问
        return true;
    };
    api_regist1("/index/hook/on_http_access",[](API_ARGS1){
        //在这里根据allArgs["params"](url参数)来判断该http客户端是否有权限访问该文件
        if(!checkAccess(allArgs["params"])){
            //无访问权限
            val["err"] = "无访问权限";
            //仅限制访问当前目录
            val["path"] = "";
            //标记该客户端无权限1分钟
            val["second"] = 60;
            return;
        }
        //可以访问
        val["err"] = "";
        //只能访问当前目录
        val["path"] = "";
        //该http客户端用户被授予10分钟的访问权限，该权限仅限访问当前目录
        val["second"] = 10 * 60;
    });
    api_regist1("/index/hook/on_server_started",[](API_ARGS1){
        //服务器重启报告
    });
 }
 void unInstallWebApi(){
    {
        lock_guard<recursive_mutex> lck(s_proxyMapMtx);
        s_proxyMap.clear();
    }
    {
        lock_guard<recursive_mutex> lck(s_ffmpegMapMtx);
        s_ffmpegMap.clear();
    }
    {
 #if defined(ENABLE_RTPPROXY)
        RtpSelector::Instance().clear();
        lock_guard<recursive_mutex> lck(s_rtpServerMapMtx);
        s_rtpServerMap.clear();
 #endif
    }
 }
\ No newline at end of file
--- a/src/Common/MediaSink.cpp
+++ b/src/Common/MediaSink.cpp
 /*
 * Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
 *
 * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
 *
 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
 */
 #include "MediaSink.h"
 //最多等待未初始化的Track 10秒，超时之后会忽略未初始化的Track
 #define MAX_WAIT_MS_READY 10000
 //如果添加Track，最多等待5秒
-#define MAX_WAIT_MS_ADD_TRACK 5000
+#define MAX_WAIT_MS_ADD_TRACK 1000
 namespace mediakit{
 void MediaSink::addTrack(const Track::Ptr &track_in) {
    lock_guard<recursive_mutex> lck(_mtx);
    if (_all_track_ready) {
        WarnL << "all track is ready, add this track too late!";
        return;
    }
    //克隆Track，只拷贝其数据，不拷贝其数据转发关系
    auto track = track_in->clone();
    auto codec_id = track->getCodecId();
    _track_map[codec_id] = track;
    _track_ready_callback[codec_id] = [this, track]() {
        onTrackReady(track);
    };
    _ticker.resetTime();
    track->addDelegate(std::make_shared<FrameWriterInterfaceHelper>([this](const Frame::Ptr &frame) {
        if (_all_track_ready) {
            onTrackFrame(frame);
        } else {
            //还有Track未就绪，先缓存之
            _frame_unread[frame->getCodecId()].emplace_back(Frame::getCacheAbleFrame(frame));
        }
    }));
 }
 void MediaSink::resetTracks() {
    lock_guard<recursive_mutex> lck(_mtx);
    _all_track_ready = false;
    _track_map.clear();
    _track_ready_callback.clear();
    _ticker.resetTime();
    _max_track_size = 2;
    _frame_unread.clear();
 }
 void MediaSink::inputFrame(const Frame::Ptr &frame) {
    lock_guard<recursive_mutex> lck(_mtx);
    auto it = _track_map.find(frame->getCodecId());
    if (it == _track_map.end()) {
        return;
    }
    it->second->inputFrame(frame);
    checkTrackIfReady(nullptr);
 }
 void MediaSink::checkTrackIfReady_l(const Track::Ptr &track){
    //Track由未就绪状态转换成就绪状态，我们就触发onTrackReady回调
    auto it_callback = _track_ready_callback.find(track->getCodecId());
    if (it_callback != _track_ready_callback.end() && track->ready()) {
        it_callback->second();
        _track_ready_callback.erase(it_callback);
    }
 }
 void MediaSink::checkTrackIfReady(const Track::Ptr &track){
    if (!_all_track_ready && !_track_ready_callback.empty()) {
        if (track) {
            checkTrackIfReady_l(track);
        } else {
            for (auto &pr : _track_map) {
                checkTrackIfReady_l(pr.second);
            }
        }
    }
    if(!_all_track_ready){
        if(_ticker.elapsedTime() > MAX_WAIT_MS_READY){
            //如果超过规定时间，那么不再等待并忽略未准备好的Track
            emitAllTrackReady();
            return;
        }
        if(!_track_ready_callback.empty()){
            //在超时时间内，如果存在未准备好的Track，那么继续等待
            return;
        }
        if(_track_map.size() == _max_track_size){
            //如果已经添加了音视频Track，并且不存在未准备好的Track，那么说明所有Track都准备好了
            emitAllTrackReady();
            return;
        }
        if(_track_map.size() == 1 && _ticker.elapsedTime() > MAX_WAIT_MS_ADD_TRACK){
            //如果只有一个Track，那么在该Track添加后，我们最多还等待若干时间(可能后面还会添加Track)
            emitAllTrackReady();
            return;
        }
    }
 }
 void MediaSink::addTrackCompleted(){
    lock_guard<recursive_mutex> lck(_mtx);
    _max_track_size = _track_map.size();
    checkTrackIfReady(nullptr);
 }
 void MediaSink::emitAllTrackReady() {
    if (_all_track_ready) {
        return;
    }
    DebugL << "all track ready use " << _ticker.elapsedTime() << "ms";
    if (!_track_ready_callback.empty()) {
        //这是超时强制忽略未准备好的Track
        _track_ready_callback.clear();
        //移除未准备好的Track
        for (auto it = _track_map.begin(); it != _track_map.end();) {
            if (!it->second->ready()) {
                WarnL << "track not ready for a long time, ignored: " << it->second->getCodecName();
                it = _track_map.erase(it);
                continue;
            }
            ++it;
        }
    }
    if (!_track_map.empty()) {
        //最少有一个有效的Track
        _all_track_ready = true;
        onAllTrackReady();
        //全部Track就绪，我们一次性把之前的帧输出
        for(auto &pr : _frame_unread){
            if (_track_map.find(pr.first) == _track_map.end()) {
                //该Track已经被移除
                continue;
            }
            pr.second.for_each([&](const Frame::Ptr &frame) {
                onTrackFrame(frame);
            });
        }
        _frame_unread.clear();
    }
 }
 vector<Track::Ptr> MediaSink::getTracks(bool trackReady) const{
    vector<Track::Ptr> ret;
    lock_guard<recursive_mutex> lck(_mtx);
    for (auto &pr : _track_map){
        if(trackReady && !pr.second->ready()){
            continue;
        }
        ret.emplace_back(pr.second);
    }
    return ret;
 }
 }//namespace mediakit
--- a/src/Common/MediaSource.cpp
+++ b/src/Common/MediaSource.cpp
 /*
 * Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
 *
 * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
 *
 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
 */
 #include "MediaSource.h"
 #include "Record/MP4Reader.h"
 #include "Util/util.h"
 #include "Network/sockutil.h"
 #include "Network/TcpSession.h"
 using namespace toolkit;
 namespace mediakit {
 recursive_mutex s_media_source_mtx;
 MediaSource::SchemaVhostAppStreamMap s_media_source_map;
 string getOriginTypeString(MediaOriginType type){
 #define SWITCH_CASE(type) case MediaOriginType::type : return #type
    switch (type) {
        SWITCH_CASE(unknown);
        SWITCH_CASE(rtmp_push);
        SWITCH_CASE(rtsp_push);
        SWITCH_CASE(rtp_push);
        SWITCH_CASE(pull);
        SWITCH_CASE(ffmpeg_pull);
        SWITCH_CASE(mp4_vod);
        SWITCH_CASE(device_chn);
    }
 }
 MediaSource::MediaSource(const string &schema, const string &vhost, const string &app, const string &stream_id){
    GET_CONFIG(bool, enableVhost, General::kEnableVhost);
    if (!enableVhost) {
        _vhost = DEFAULT_VHOST;
    } else {
        _vhost = vhost.empty() ? DEFAULT_VHOST : vhost;
    }
    _schema = schema;
    _app = app;
    _stream_id = stream_id;
    _create_stamp = time(NULL);
 }
 MediaSource::~MediaSource() {
    unregist();
 }
 const string& MediaSource::getSchema() const {
    return _schema;
 }
 const string& MediaSource::getVhost() const {
    return _vhost;
 }
 const string& MediaSource::getApp() const {
    //获取该源的id
    return _app;
 }
 const string& MediaSource::getId() const {
    return _stream_id;
 }
 int MediaSource::getBytesSpeed(){
    return _speed.getSpeed();
 }
 uint64_t MediaSource::getCreateStamp() const {
    return _create_stamp;
 }
 uint64_t MediaSource::getAliveSecond() const {
    //使用Ticker对象获取存活时间的目的是防止修改系统时间导致回退
    return _ticker.createdTime() / 1000;
 }
 vector<Track::Ptr> MediaSource::getTracks(bool ready) const {
    auto listener = _listener.lock();
    if(!listener){
        return vector<Track::Ptr>();
    }
    return listener->getTracks(const_cast<MediaSource &>(*this), ready);
 }
 void MediaSource::setListener(const std::weak_ptr<MediaSourceEvent> &listener){
    _listener = listener;
 }
 std::weak_ptr<MediaSourceEvent> MediaSource::getListener(bool next) const{
    if (!next) {
        return _listener;
    }
    auto listener = dynamic_pointer_cast<MediaSourceEventInterceptor>(_listener.lock());
    if (!listener) {
        //不是MediaSourceEventInterceptor对象或者对象已经销毁
        return _listener;
    }
    //获取被拦截的对象
    auto next_obj = listener->getDelegate();
    //有则返回之
    return next_obj ? next_obj : _listener;
 }
 int MediaSource::totalReaderCount(){
    auto listener = _listener.lock();
    if(!listener){
        return readerCount();
    }
    return listener->totalReaderCount(*this);
 }
 MediaOriginType MediaSource::getOriginType() const {
    auto listener = _listener.lock();
    if (!listener) {
        return MediaOriginType::unknown;
    }
    return listener->getOriginType(const_cast<MediaSource &>(*this));
 }
 string MediaSource::getOriginUrl() const {
    auto listener = _listener.lock();
    if (!listener) {
        return "";
    }
    return listener->getOriginUrl(const_cast<MediaSource &>(*this));
 }
 std::shared_ptr<SockInfo> MediaSource::getOriginSock() const {
    auto listener = _listener.lock();
    if (!listener) {
        return nullptr;
    }
    return listener->getOriginSock(const_cast<MediaSource &>(*this));
 }
 bool MediaSource::seekTo(uint32_t stamp) {
    auto listener = _listener.lock();
    if(!listener){
        return false;
    }
    return listener->seekTo(*this, stamp);
 }
 bool MediaSource::close(bool force) {
    auto listener = _listener.lock();
    if(!listener){
        return false;
    }
    return listener->close(*this,force);
 }
 void MediaSource::onReaderChanged(int size) {
    auto listener = _listener.lock();
    if (listener) {
        listener->onReaderChanged(*this, size);
    }
 }
 bool MediaSource::setupRecord(Recorder::type type, bool start, const string &custom_path){
    auto listener = _listener.lock();
    if (!listener) {
        WarnL << "未设置MediaSource的事件监听者，setupRecord失败:" << getSchema() << "/" << getVhost() << "/" << getApp() << "/" << getId();
        return false;
    }
    return listener->setupRecord(*this, type, start, custom_path);
 }
 bool MediaSource::isRecording(Recorder::type type){
    auto listener = _listener.lock();
    if(!listener){
        return false;
    }
    return listener->isRecording(*this, type);
 }
-void MediaSource::startSendRtp(const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
+void MediaSource::startSendRtp(const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, uint16_t src_port, const function<void(const SockException &ex)> &cb){
    auto listener = _listener.lock();
    if (!listener) {
        cb(SockException(Err_other, "尚未设置事件监听器"));
        return;
    }
-    return listener->startSendRtp(*this, dst_url, dst_port, ssrc, is_udp, cb);
+    return listener->startSendRtp(*this, dst_url, dst_port, ssrc, is_udp, src_port, cb);
 }
-bool MediaSource::stopSendRtp() {
+bool MediaSource::stopSendRtp(const string &ssrc) {
    auto listener = _listener.lock();
    if (!listener) {
        return false;
    }
-    return listener->stopSendRtp(*this);
+    return listener->stopSendRtp(*this, ssrc);
 }
 void MediaSource::for_each_media(const function<void(const MediaSource::Ptr &src)> &cb) {
    decltype(s_media_source_map) copy;
    {
        //拷贝s_media_source_map后再遍历，考虑到是高频使用的全局单例锁，并且在上锁时会执行回调代码
        //很容易导致多个锁交叉死锁的情况，而且该函数使用频率不高，拷贝开销相对来说是可以接受的
        lock_guard<recursive_mutex> lock(s_media_source_mtx);
        copy = s_media_source_map;
    }
    for (auto &pr0 : copy) {
        for (auto &pr1 : pr0.second) {
            for (auto &pr2 : pr1.second) {
                for (auto &pr3 : pr2.second) {
                    auto src = pr3.second.lock();
                    if(src){
                        cb(src);
                    }
                }
            }
        }
    }
 }
 template<typename MAP, typename FUNC>
 static bool searchMedia(MAP &map, const string &schema, const string &vhost, const string &app, const string &id, FUNC &&func) {
    auto it0 = map.find(schema);
    if (it0 == map.end()) {
        //未找到协议
        return false;
    }
    auto it1 = it0->second.find(vhost);
    if (it1 == it0->second.end()) {
        //未找到vhost
        return false;
    }
    auto it2 = it1->second.find(app);
    if (it2 == it1->second.end()) {
        //未找到app
        return false;
    }
    auto it3 = it2->second.find(id);
    if (it3 == it2->second.end()) {
        //未找到streamId
        return false;
    }
    return func(it0, it1, it2, it3);
 }
 template<typename MAP, typename IT0, typename IT1, typename IT2>
 static void eraseIfEmpty(MAP &map, IT0 it0, IT1 it1, IT2 it2) {
    if (it2->second.empty()) {
        it1->second.erase(it2);
        if (it1->second.empty()) {
            it0->second.erase(it1);
            if (it0->second.empty()) {
                map.erase(it0);
            }
        }
    }
 }
 static MediaSource::Ptr find_l(const string &schema, const string &vhost_in, const string &app, const string &id, bool create_new) {
    string vhost = vhost_in;
    GET_CONFIG(bool,enableVhost,General::kEnableVhost);
    if(vhost.empty() || !enableVhost){
        vhost = DEFAULT_VHOST;
    }
    MediaSource::Ptr ret;
    {
        lock_guard<recursive_mutex> lock(s_media_source_mtx);
        //查找某一媒体源，找到后返回
        searchMedia(s_media_source_map, schema, vhost, app, id,
                    [&](MediaSource::SchemaVhostAppStreamMap::iterator &it0, MediaSource::VhostAppStreamMap::iterator &it1,
                        MediaSource::AppStreamMap::iterator &it2, MediaSource::StreamMap::iterator &it3) {
            ret = it3->second.lock();
            if (!ret) {
                //该对象已经销毁
                it2->second.erase(it3);
                eraseIfEmpty(s_media_source_map, it0, it1, it2);
                return false;
            }
            return true;
        });
    }
    if(!ret && create_new && schema != HLS_SCHEMA){
        //未查找媒体源，则读取mp4创建一个
        //播放hls不触发mp4点播(因为HLS也可以用于录像，不是纯粹的直播)
        ret = MediaSource::createFromMP4(schema, vhost, app, id);
    }
    return ret;
 }
 static void findAsync_l(const MediaInfo &info, const std::shared_ptr<TcpSession> &session, bool retry,
                        const function<void(const MediaSource::Ptr &src)> &cb){
    auto src = find_l(info._schema, info._vhost, info._app, info._streamid, true);
    if (src || !retry) {
        cb(src);
        return;
    }
    void *listener_tag = session.get();
    weak_ptr<TcpSession> weak_session = session;
    GET_CONFIG(int, maxWaitMS, General::kMaxStreamWaitTimeMS);
    auto on_timeout = session->getPoller()->doDelayTask(maxWaitMS, [cb, listener_tag]() {
        //最多等待一定时间，如果这个时间内，流未注册上，那么返回未找到流
        NoticeCenter::Instance().delListener(listener_tag, Broadcast::kBroadcastMediaChanged);
        cb(nullptr);
        return 0;
    });
    auto cancel_all = [on_timeout, listener_tag]() {
        //取消延时任务，防止多次回调
        on_timeout->cancel();
        //取消媒体注册事件监听
        NoticeCenter::Instance().delListener(listener_tag, Broadcast::kBroadcastMediaChanged);
    };
    function<void()> close_player = [cb, cancel_all]() {
        cancel_all();
        //告诉播放器，流不存在，这样会立即断开播放器
        cb(nullptr);
    };
    auto on_regist = [weak_session, info, cb, cancel_all](BroadcastMediaChangedArgs) {
        auto strong_session = weak_session.lock();
        if (!strong_session) {
            //自己已经销毁
            cancel_all();
            return;
        }
        if (!bRegist ||
            sender.getSchema() != info._schema ||
            sender.getVhost() != info._vhost ||
            sender.getApp() != info._app ||
            sender.getId() != info._streamid) {
            //不是自己感兴趣的事件，忽略之
            return;
        }
        cancel_all();
        //播发器请求的流终于注册上了，切换到自己的线程再回复
        strong_session->async([weak_session, info, cb]() {
            auto strongSession = weak_session.lock();
            if (!strongSession) {
                return;
            }
            DebugL << "收到媒体注册事件,回复播放器:" << info._schema << "/" << info._vhost << "/" << info._app << "/" << info._streamid;
            //再找一遍媒体源，一般能找到
            findAsync_l(info, strongSession, false, cb);
        }, false);
    };
    //监听媒体注册事件
    NoticeCenter::Instance().addListener(listener_tag, Broadcast::kBroadcastMediaChanged, on_regist);
    //广播未找到流,此时可以立即去拉流，这样还来得及
    NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastNotFoundStream, info, static_cast<SockInfo &>(*session), close_player);
 }
 void MediaSource::findAsync(const MediaInfo &info, const std::shared_ptr<TcpSession> &session,const function<void(const Ptr &src)> &cb){
    return findAsync_l(info, session, true, cb);
 }
 MediaSource::Ptr MediaSource::find(const string &schema, const string &vhost, const string &app, const string &id) {
    return find_l(schema, vhost, app, id, false);
 }
 MediaSource::Ptr MediaSource::find(const string &vhost, const string &app, const string &stream_id){
    auto src = MediaSource::find(RTMP_SCHEMA, vhost, app, stream_id);
    if (src) {
        return src;
    }
    src = MediaSource::find(RTSP_SCHEMA, vhost, app, stream_id);
    if (src) {
        return src;
    }
    return MediaSource::find(HLS_SCHEMA, vhost, app, stream_id);
 }
 void MediaSource::emitEvent(bool regist){
    auto listener = _listener.lock();
    if (listener) {
        //触发回调
        listener->onRegist(*this, regist);
    }
    //触发广播
    NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastMediaChanged, regist, *this);
    InfoL << (regist ? "媒体注册:" : "媒体注销:") << _schema << " " << _vhost << " " << _app << " " << _stream_id;
 }
 void MediaSource::regist() {
    {
        //减小互斥锁临界区
        lock_guard<recursive_mutex> lock(s_media_source_mtx);
        s_media_source_map[_schema][_vhost][_app][_stream_id] = shared_from_this();
    }
    emitEvent(true);
 }
 //反注册该源
 bool MediaSource::unregist() {
    bool ret;
    {
        //减小互斥锁临界区
        lock_guard<recursive_mutex> lock(s_media_source_mtx);
        ret = searchMedia(s_media_source_map, _schema, _vhost, _app, _stream_id,
                          [&](SchemaVhostAppStreamMap::iterator &it0, VhostAppStreamMap::iterator &it1,
                              AppStreamMap::iterator &it2, StreamMap::iterator &it3) {
          auto strong_self = it3->second.lock();
          if (strong_self && this != strong_self.get()) {
              //不是自己,不允许反注册
              return false;
          }
          it2->second.erase(it3);
          eraseIfEmpty(s_media_source_map, it0, it1, it2);
          return true;
      });
    }
    if (ret) {
        emitEvent(false);
    }
    return ret;
 }
 /////////////////////////////////////MediaInfo//////////////////////////////////////
 void MediaInfo::parse(const string &url_in){
    _full_url = url_in;
    string url = url_in;
    auto pos = url.find("?");
    if (pos != string::npos) {
        _param_strs = url.substr(pos + 1);
        url.erase(pos);
    }
    auto schema_pos = url.find("://");
    if (schema_pos != string::npos) {
        _schema = url.substr(0, schema_pos);
    } else {
        schema_pos = -3;
    }
    auto split_vec = split(url.substr(schema_pos + 3), "/");
    if (split_vec.size() > 0) {
        auto vhost = split_vec[0];
        auto pos = vhost.find(":");
        if (pos != string::npos) {
            _host = _vhost = vhost.substr(0, pos);
            _port = vhost.substr(pos + 1);
        } else {
            _host = _vhost = vhost;
        }
        if (_vhost == "localhost" || INADDR_NONE != inet_addr(_vhost.data())) {
            //如果访问的是localhost或ip，那么则为默认虚拟主机
            _vhost = DEFAULT_VHOST;
        }
    }
    if (split_vec.size() > 1) {
        _app = split_vec[1];
    }
    if (split_vec.size() > 2) {
        string stream_id;
        for (int i = 2; i < split_vec.size(); ++i) {
            stream_id.append(split_vec[i] + "/");
        }
        if (stream_id.back() == '/') {
            stream_id.pop_back();
        }
        _streamid = stream_id;
    }
    auto params = Parser::parseArgs(_param_strs);
    if (params.find(VHOST_KEY) != params.end()) {
        _vhost = params[VHOST_KEY];
    }
    GET_CONFIG(bool, enableVhost, General::kEnableVhost);
    if (!enableVhost || _vhost.empty()) {
        //如果关闭虚拟主机或者虚拟主机为空，则设置虚拟主机为默认
        _vhost = DEFAULT_VHOST;
    }
 }
 MediaSource::Ptr MediaSource::createFromMP4(const string &schema, const string &vhost, const string &app, const string &stream, const string &file_path , bool check_app){
    GET_CONFIG(string, appName, Record::kAppName);
    if (check_app && app != appName) {
        return nullptr;
    }
 #ifdef ENABLE_MP4
    try {
        MP4Reader::Ptr pReader(new MP4Reader(vhost, app, stream, file_path));
        pReader->startReadMP4();
        return MediaSource::find(schema, vhost, app, stream);
    } catch (std::exception &ex) {
        WarnL << ex.what();
        return nullptr;
    }
 #else
    WarnL << "创建MP4点播失败，请编译时打开\"ENABLE_MP4\"选项";
    return nullptr;
 #endif //ENABLE_MP4
 }
 /////////////////////////////////////MediaSourceEvent//////////////////////////////////////
 void MediaSourceEvent::onReaderChanged(MediaSource &sender, int size){
    if (size || totalReaderCount(sender)) {
        //还有人观看该视频，不触发关闭事件
        return;
    }
    //没有任何人观看该视频源，表明该源可以关闭了
    GET_CONFIG(string, record_app, Record::kAppName);
    GET_CONFIG(int, stream_none_reader_delay, General::kStreamNoneReaderDelayMS);
    //如果mp4点播, 无人观看时我们强制关闭点播
    bool is_mp4_vod = sender.getApp() == record_app;
    weak_ptr<MediaSource> weak_sender = sender.shared_from_this();
    _async_close_timer = std::make_shared<Timer>(stream_none_reader_delay / 1000.0, [weak_sender, is_mp4_vod]() {
        auto strong_sender = weak_sender.lock();
        if (!strong_sender) {
            //对象已经销毁
            return false;
        }
        if (strong_sender->totalReaderCount()) {
            //还有人观看该视频，不触发关闭事件
            return false;
        }
        if (!is_mp4_vod) {
            //直播时触发无人观看事件，让开发者自行选择是否关闭
            WarnL << "无人观看事件:"
                  << strong_sender->getSchema() << "/"
                  << strong_sender->getVhost() << "/"
                  << strong_sender->getApp() << "/"
                  << strong_sender->getId();
            NoticeCenter::Instance().emitEvent(Broadcast::kBroadcastStreamNoneReader, *strong_sender);
        } else {
            //这个是mp4点播，我们自动关闭
            WarnL << "MP4点播无人观看,自动关闭:"
                  << strong_sender->getSchema() << "/"
                  << strong_sender->getVhost() << "/"
                  << strong_sender->getApp() << "/"
                  << strong_sender->getId();
            strong_sender->close(false);
        }
        return false;
    }, nullptr);
 }
 MediaOriginType MediaSourceEventInterceptor::getOriginType(MediaSource &sender) const {
    auto listener = _listener.lock();
    if (!listener) {
        return MediaOriginType::unknown;
    }
    return listener->getOriginType(sender);
 }
 string MediaSourceEventInterceptor::getOriginUrl(MediaSource &sender) const {
    auto listener = _listener.lock();
    if (!listener) {
        return "";
    }
    return listener->getOriginUrl(sender);
 }
 std::shared_ptr<SockInfo> MediaSourceEventInterceptor::getOriginSock(MediaSource &sender) const {
    auto listener = _listener.lock();
    if (!listener) {
        return nullptr;
    }
    return listener->getOriginSock(sender);
 }
 bool MediaSourceEventInterceptor::seekTo(MediaSource &sender, uint32_t stamp) {
    auto listener = _listener.lock();
    if (!listener) {
        return false;
    }
    return listener->seekTo(sender, stamp);
 }
 bool MediaSourceEventInterceptor::close(MediaSource &sender, bool force) {
    auto listener = _listener.lock();
    if (!listener) {
        return false;
    }
    return listener->close(sender, force);
 }
 int MediaSourceEventInterceptor::totalReaderCount(MediaSource &sender) {
    auto listener = _listener.lock();
    if (!listener) {
        return sender.readerCount();
    }
    return listener->totalReaderCount(sender);
 }
 void MediaSourceEventInterceptor::onReaderChanged(MediaSource &sender, int size) {
    auto listener = _listener.lock();
    if (!listener) {
        MediaSourceEvent::onReaderChanged(sender, size);
    } else {
        listener->onReaderChanged(sender, size);
    }
 }
 void MediaSourceEventInterceptor::onRegist(MediaSource &sender, bool regist) {
    auto listener = _listener.lock();
    if (listener) {
        listener->onRegist(sender, regist);
    }
 }
 bool MediaSourceEventInterceptor::setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) {
    auto listener = _listener.lock();
    if (!listener) {
        return false;
    }
    return listener->setupRecord(sender, type, start, custom_path);
 }
 bool MediaSourceEventInterceptor::isRecording(MediaSource &sender, Recorder::type type) {
    auto listener = _listener.lock();
    if (!listener) {
        return false;
    }
    return listener->isRecording(sender, type);
 }
 vector<Track::Ptr> MediaSourceEventInterceptor::getTracks(MediaSource &sender, bool trackReady) const {
    auto listener = _listener.lock();
    if (!listener) {
        return vector<Track::Ptr>();
    }
    return listener->getTracks(sender, trackReady);
 }
-void MediaSourceEventInterceptor::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
+void MediaSourceEventInterceptor::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, uint16_t src_port, const function<void(const SockException &ex)> &cb){
    auto listener = _listener.lock();
    if (listener) {
-        listener->startSendRtp(sender, dst_url, dst_port, ssrc, is_udp, cb);
+        listener->startSendRtp(sender, dst_url, dst_port, ssrc, is_udp, src_port, cb);
    } else {
-        MediaSourceEvent::startSendRtp(sender, dst_url, dst_port, ssrc, is_udp, cb);
+        MediaSourceEvent::startSendRtp(sender, dst_url, dst_port, ssrc, is_udp, src_port, cb);
    }
 }
-bool MediaSourceEventInterceptor::stopSendRtp(MediaSource &sender){
+bool MediaSourceEventInterceptor::stopSendRtp(MediaSource &sender, const string &ssrc){
    auto listener = _listener.lock();
    if (listener) {
-        return listener->stopSendRtp(sender);
+        return listener->stopSendRtp(sender, ssrc);
    }
    return false;
 }
 void MediaSourceEventInterceptor::setDelegate(const std::weak_ptr<MediaSourceEvent> &listener) {
    if (listener.lock().get() == this) {
        throw std::invalid_argument("can not set self as a delegate");
    }
    _listener = listener;
 }
 std::shared_ptr<MediaSourceEvent> MediaSourceEventInterceptor::getDelegate() const{
    return _listener.lock();
 }
 /////////////////////////////////////FlushPolicy//////////////////////////////////////
 static bool isFlushAble_default(bool is_video, uint64_t last_stamp, uint64_t new_stamp, int cache_size) {
    if (new_stamp + 500 < last_stamp) {
        //时间戳回退比较大(可能seek中)，由于rtp中时间戳是pts，是可能存在一定程度的回退的
        return true;
    }
    //时间戳发送变化或者缓存超过1024个,sendmsg接口一般最多只能发送1024个数据包
    return last_stamp != new_stamp || cache_size >= 1024;
 }
 static bool isFlushAble_merge(bool is_video, uint64_t last_stamp, uint64_t new_stamp, int cache_size, int merge_ms) {
    if (new_stamp + 500 < last_stamp) {
        //时间戳回退比较大(可能seek中)，由于rtp中时间戳是pts，是可能存在一定程度的回退的
        return true;
    }
    if (new_stamp > last_stamp + merge_ms) {
        //时间戳增量超过合并写阈值
        return true;
    }
    //缓存数超过1024个,这个逻辑用于避免时间戳异常的流导致的内存暴增问题
    //而且sendmsg接口一般最多只能发送1024个数据包
    return cache_size >= 1024;
 }
 bool FlushPolicy::isFlushAble(bool is_video, bool is_key, uint64_t new_stamp, int cache_size) {
    bool flush_flag = false;
    if (is_key && is_video) {
        //遇到关键帧flush掉前面的数据，确保关键帧为该组数据的第一帧，确保GOP缓存有效
        flush_flag = true;
    } else {
        GET_CONFIG(int, mergeWriteMS, General::kMergeWriteMS);
        if (mergeWriteMS <= 0) {
            //关闭了合并写或者合并写阈值小于等于0
            flush_flag = isFlushAble_default(is_video, _last_stamp[is_video], new_stamp, cache_size);
        } else {
            flush_flag = isFlushAble_merge(is_video, _last_stamp[is_video], new_stamp, cache_size, mergeWriteMS);
        }
    }
    if (flush_flag) {
        _last_stamp[is_video] = new_stamp;
    }
    return flush_flag;
 }
 } /* namespace mediakit */
\ No newline at end of file
--- a/src/Common/MediaSource.h
+++ b/src/Common/MediaSource.h
 /*
 * Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
 *
 * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
 *
 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
 */
 #ifndef ZLMEDIAKIT_MEDIASOURCE_H
 #define ZLMEDIAKIT_MEDIASOURCE_H
 #include <mutex>
 #include <string>
 #include <memory>
 #include <functional>
 #include <unordered_map>
 #include "Common/config.h"
 #include "Common/Parser.h"
 #include "Util/logger.h"
 #include "Util/TimeTicker.h"
 #include "Util/NoticeCenter.h"
 #include "Util/List.h"
 #include "Network/Socket.h"
 #include "Rtsp/Rtsp.h"
 #include "Rtmp/Rtmp.h"
 #include "Extension/Track.h"
 #include "Record/Recorder.h"
 using namespace std;
 using namespace toolkit;
 namespace toolkit{
    class TcpSession;
 }// namespace toolkit
 namespace mediakit {
 enum class MediaOriginType : uint8_t {
    unknown = 0,
    rtmp_push ,
    rtsp_push,
    rtp_push,
    pull,
    ffmpeg_pull,
    mp4_vod,
    device_chn
 };
 string getOriginTypeString(MediaOriginType type);
 class MediaSource;
 class MediaSourceEvent{
 public:
    friend class MediaSource;
    MediaSourceEvent(){};
    virtual ~MediaSourceEvent(){};
    // 获取媒体源类型
    virtual MediaOriginType getOriginType(MediaSource &sender) const { return MediaOriginType::unknown; }
    // 获取媒体源url或者文件路径
    virtual string getOriginUrl(MediaSource &sender) const { return ""; }
    // 获取媒体源客户端相关信息
    virtual std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const { return nullptr; }
    // 通知拖动进度条
    virtual bool seekTo(MediaSource &sender, uint32_t stamp) { return false; }
    // 通知其停止产生流
    virtual bool close(MediaSource &sender, bool force) { return false; }
    // 获取观看总人数
    virtual int totalReaderCount(MediaSource &sender) = 0;
    // 通知观看人数变化
    virtual void onReaderChanged(MediaSource &sender, int size);
    //流注册或注销事件
    virtual void onRegist(MediaSource &sender, bool regist) {};
    ////////////////////////仅供MultiMediaSourceMuxer对象继承////////////////////////
    // 开启或关闭录制
    virtual bool setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) { return false; };
    // 获取录制状态
    virtual bool isRecording(MediaSource &sender, Recorder::type type) { return false; };
    // 获取所有track相关信息
    virtual vector<Track::Ptr> getTracks(MediaSource &sender, bool trackReady = true) const { return vector<Track::Ptr>(); };
    // 开始发送ps-rtp
-    virtual void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) { cb(SockException(Err_other, "not implemented"));};
+    virtual void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, uint16_t src_port, const function<void(const SockException &ex)> &cb) { cb(SockException(Err_other, "not implemented"));};
    // 停止发送ps-rtp
-    virtual bool stopSendRtp(MediaSource &sender) {return false; }
+    virtual bool stopSendRtp(MediaSource &sender, const string &ssrc) {return false; }
 private:
    Timer::Ptr _async_close_timer;
 };
 //该对象用于拦截感兴趣的MediaSourceEvent事件
 class MediaSourceEventInterceptor : public MediaSourceEvent{
 public:
    MediaSourceEventInterceptor(){}
    ~MediaSourceEventInterceptor() override {}
    void setDelegate(const std::weak_ptr<MediaSourceEvent> &listener);
    std::shared_ptr<MediaSourceEvent> getDelegate() const;
    MediaOriginType getOriginType(MediaSource &sender) const override;
    string getOriginUrl(MediaSource &sender) const override;
    std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
    bool seekTo(MediaSource &sender, uint32_t stamp) override;
    bool close(MediaSource &sender, bool force) override;
    int totalReaderCount(MediaSource &sender) override;
    void onReaderChanged(MediaSource &sender, int size) override;
    void onRegist(MediaSource &sender, bool regist) override;
    bool setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) override;
    bool isRecording(MediaSource &sender, Recorder::type type) override;
    vector<Track::Ptr> getTracks(MediaSource &sender, bool trackReady = true) const override;
-    void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) override;
+    void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, uint16_t src_port, const function<void(const SockException &ex)> &cb) override;
-    bool stopSendRtp(MediaSource &sender) override;
+    bool stopSendRtp(MediaSource &sender, const string &ssrc) override;
 private:
    std::weak_ptr<MediaSourceEvent> _listener;
 };
 /**
 * 解析url获取媒体相关信息
 */
 class MediaInfo{
 public:
    ~MediaInfo() {}
    MediaInfo() {}
    MediaInfo(const string &url) { parse(url); }
    void parse(const string &url);
 public:
    string _full_url;
    string _schema;
    string _host;
    string _port;
    string _vhost;
    string _app;
    string _streamid;
    string _param_strs;
 };
 class BytesSpeed {
 public:
    BytesSpeed() = default;
    ~BytesSpeed() = default;
    /**
     * 添加统计字节
     */
    BytesSpeed& operator += (uint64_t bytes) {
        _bytes += bytes;
        if (_bytes > 1024 * 1024) {
            //数据大于1MB就计算一次网速
            computeSpeed();
        }
        return *this;
    }
    /**
     * 获取速度，单位bytes/s
     */
    int getSpeed() {
        if (_ticker.elapsedTime() < 1000) {
            //获取频率小于1秒，那么返回上次计算结果
            return _speed;
        }
        return computeSpeed();
    }
 private:
    uint64_t computeSpeed() {
        auto elapsed = _ticker.elapsedTime();
        if (!elapsed) {
            return _speed;
        }
        _speed = _bytes * 1000 / elapsed;
        _ticker.resetTime();
        _bytes = 0;
        return _speed;
    }
 private:
    int _speed = 0;
    uint64_t _bytes = 0;
    Ticker _ticker;
 };
 /**
 * 媒体源，任何rtsp/rtmp的直播流都源自该对象
 */
 class MediaSource: public TrackSource, public enable_shared_from_this<MediaSource> {
 public:
    typedef std::shared_ptr<MediaSource> Ptr;
    typedef unordered_map<string, weak_ptr<MediaSource> > StreamMap;
    typedef unordered_map<string, StreamMap > AppStreamMap;
    typedef unordered_map<string, AppStreamMap > VhostAppStreamMap;
    typedef unordered_map<string, VhostAppStreamMap > SchemaVhostAppStreamMap;
    MediaSource(const string &schema, const string &vhost, const string &app, const string &stream_id) ;
    virtual ~MediaSource() ;
    ////////////////获取MediaSource相关信息////////////////
    // 获取协议类型
    const string& getSchema() const;
    // 虚拟主机
    const string& getVhost() const;
    // 应用名
    const string& getApp() const;
    // 流id
    const string& getId() const;
    // 获取所有Track
    vector<Track::Ptr> getTracks(bool ready = true) const override;
    // 获取流当前时间戳
    virtual uint32_t getTimeStamp(TrackType type) { return 0; };
    // 设置时间戳
    virtual void setTimeStamp(uint32_t stamp) {};
    // 获取数据速率，单位bytes/s
    int getBytesSpeed();
    // 获取流创建GMT unix时间戳，单位秒
    uint64_t getCreateStamp() const;
    // 获取流上线时间，单位秒
    uint64_t getAliveSecond() const;
    ////////////////MediaSourceEvent相关接口实现////////////////
    // 设置监听者
    virtual void setListener(const std::weak_ptr<MediaSourceEvent> &listener);
    // 获取监听者
    std::weak_ptr<MediaSourceEvent> getListener(bool next = false) const;
    // 本协议获取观看者个数，可能返回本协议的观看人数，也可能返回总人数
    virtual int readerCount() = 0;
    // 观看者个数，包括(hls/rtsp/rtmp)
    virtual int totalReaderCount();
    // 获取媒体源类型
    MediaOriginType getOriginType() const;
    // 获取媒体源url或者文件路径
    string getOriginUrl() const;
    // 获取媒体源客户端相关信息
    std::shared_ptr<SockInfo> getOriginSock() const;
    // 拖动进度条
    bool seekTo(uint32_t stamp);
    // 关闭该流
    bool close(bool force);
    // 该流观看人数变化
    void onReaderChanged(int size);
    // 开启或关闭录制
    bool setupRecord(Recorder::type type, bool start, const string &custom_path);
    // 获取录制状态
    bool isRecording(Recorder::type type);
    // 开始发送ps-rtp
-    void startSendRtp(const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb);
+    void startSendRtp(const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, uint16_t src_port, const function<void(const SockException &ex)> &cb);
    // 停止发送ps-rtp
-    bool stopSendRtp();
+    bool stopSendRtp(const string &ssrc);
    ////////////////static方法，查找或生成MediaSource////////////////
    // 同步查找流
    static Ptr find(const string &schema, const string &vhost, const string &app, const string &id);
    // 忽略类型，同步查找流，可能返回rtmp/rtsp/hls类型
    static Ptr find(const string &vhost, const string &app, const string &stream_id);
    // 异步查找流
    static void findAsync(const MediaInfo &info, const std::shared_ptr<TcpSession> &session, const function<void(const Ptr &src)> &cb);
    // 遍历所有流
    static void for_each_media(const function<void(const Ptr &src)> &cb);
    // 从mp4文件生成MediaSource
    static MediaSource::Ptr createFromMP4(const string &schema, const string &vhost, const string &app, const string &stream, const string &file_path = "", bool check_app = true);
 protected:
    //媒体注册
    void regist();
 private:
    //媒体注销
    bool unregist();
    //触发媒体事件
    void emitEvent(bool regist);
 protected:
    BytesSpeed _speed;
 private:
    time_t _create_stamp;
    Ticker _ticker;
    string _schema;
    string _vhost;
    string _app;
    string _stream_id;
    std::weak_ptr<MediaSourceEvent> _listener;
 };
 ///缓存刷新策略类
 class FlushPolicy {
 public:
    FlushPolicy() = default;
    ~FlushPolicy() = default;
    bool isFlushAble(bool is_video, bool is_key, uint64_t new_stamp, int cache_size);
 private:
    uint64_t _last_stamp[2] = {0, 0};
 };
 /// 合并写缓存模板
 /// \tparam packet 包类型
 /// \tparam policy 刷新缓存策略
 /// \tparam packet_list 包缓存类型
 template<typename packet, typename policy = FlushPolicy, typename packet_list = List<std::shared_ptr<packet> > >
 class PacketCache {
 public:
    PacketCache(){
        _cache = std::make_shared<packet_list>();
    }
    virtual ~PacketCache() = default;
    void inputPacket(uint64_t stamp, bool is_video, std::shared_ptr<packet> pkt, bool key_pos) {
        if (_policy.isFlushAble(is_video, key_pos, stamp, _cache->size())) {
            flushAll();
        }
        //追加数据到最后
        _cache->emplace_back(std::move(pkt));
        if (key_pos) {
            _key_pos = key_pos;
        }
    }
    virtual void clearCache() {
        _cache->clear();
    }
    virtual void onFlush(std::shared_ptr<packet_list>, bool key_pos) = 0;
 private:
    void flushAll() {
        if (_cache->empty()) {
            return;
        }
        onFlush(std::move(_cache), _key_pos);
        _cache = std::make_shared<packet_list>();
        _key_pos = false;
    }
 private:
    bool _key_pos = false;
    policy _policy;
    std::shared_ptr<packet_list> _cache;
 };
 } /* namespace mediakit */
 #endif //ZLMEDIAKIT_MEDIASOURCE_H
\ No newline at end of file
--- a/src/Common/MultiMediaSourceMuxer.cpp
+++ b/src/Common/MultiMediaSourceMuxer.cpp
 /*
 * Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
 *
 * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
 *
 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
 */
 #include <math.h>
 #include "Common/config.h"
 #include "MultiMediaSourceMuxer.h"
 namespace mediakit {
 ///////////////////////////////MultiMuxerPrivate//////////////////////////////////
 MultiMuxerPrivate::~MultiMuxerPrivate() {}
 MultiMuxerPrivate::MultiMuxerPrivate(const string &vhost, const string &app, const string &stream, float dur_sec,
                                     bool enable_rtsp, bool enable_rtmp, bool enable_hls, bool enable_mp4) {
    _stream_url = vhost + " " + app + " " + stream;
    if (enable_rtmp) {
        _rtmp = std::make_shared<RtmpMediaSourceMuxer>(vhost, app, stream, std::make_shared<TitleMeta>(dur_sec));
    }
    if (enable_rtsp) {
        _rtsp = std::make_shared<RtspMediaSourceMuxer>(vhost, app, stream, std::make_shared<TitleSdp>(dur_sec));
    }
    if (enable_hls) {
        _hls = dynamic_pointer_cast<HlsRecorder>(Recorder::createRecorder(Recorder::type_hls, vhost, app, stream));
    }
    if (enable_mp4) {
        _mp4 = Recorder::createRecorder(Recorder::type_mp4, vhost, app, stream);
    }
    _ts = std::make_shared<TSMediaSourceMuxer>(vhost, app, stream);
 #if defined(ENABLE_MP4)
    _fmp4 = std::make_shared<FMP4MediaSourceMuxer>(vhost, app, stream);
 #endif
 }
 void MultiMuxerPrivate::resetTracks() {
    if (_rtmp) {
        _rtmp->resetTracks();
    }
    if (_rtsp) {
        _rtsp->resetTracks();
    }
    if (_ts) {
        _ts->resetTracks();
    }
 #if defined(ENABLE_MP4)
    if (_fmp4) {
        _fmp4->resetTracks();
    }
 #endif
    //拷贝智能指针，目的是为了防止跨线程调用设置录像相关api导致的线程竞争问题
    auto hls = _hls;
    if (hls) {
        hls->resetTracks();
    }
    auto mp4 = _mp4;
    if (mp4) {
        mp4->resetTracks();
    }
 }
 void MultiMuxerPrivate::setMediaListener(const std::weak_ptr<MediaSourceEvent> &listener) {
    _listener = listener;
    if (_rtmp) {
        _rtmp->setListener(listener);
    }
    if (_rtsp) {
        _rtsp->setListener(listener);
    }
    if (_ts) {
        _ts->setListener(listener);
    }
 #if defined(ENABLE_MP4)
    if (_fmp4) {
        _fmp4->setListener(listener);
    }
 #endif
    auto hls = _hls;
    if (hls) {
        hls->setListener(listener);
    }
 }
 int MultiMuxerPrivate::totalReaderCount() const {
    auto hls = _hls;
    return (_rtsp ? _rtsp->readerCount() : 0) +
           (_rtmp ? _rtmp->readerCount() : 0) +
           (_ts ? _ts->readerCount() : 0) +
 #if defined(ENABLE_MP4)
           (_fmp4 ? _fmp4->readerCount() : 0) +
 #endif
           (hls ? hls->readerCount() : 0);
 }
 static std::shared_ptr<MediaSinkInterface> makeRecorder(const vector<Track::Ptr> &tracks, Recorder::type type, const string &custom_path, MediaSource &sender){
    auto recorder = Recorder::createRecorder(type, sender.getVhost(), sender.getApp(), sender.getId(), custom_path);
    for (auto &track : tracks) {
        recorder->addTrack(track);
    }
    return recorder;
 }
 //此函数可能跨线程调用
 bool MultiMuxerPrivate::setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path){
    switch (type) {
        case Recorder::type_hls : {
            if (start && !_hls) {
                //开始录制
                auto hls = dynamic_pointer_cast<HlsRecorder>(makeRecorder(getTracks(true), type, custom_path, sender));
                if (hls) {
                    //设置HlsMediaSource的事件监听器
                    hls->setListener(_listener);
                }
                _hls = hls;
            } else if (!start && _hls) {
                //停止录制
                _hls = nullptr;
            }
            return true;
        }
        case Recorder::type_mp4 : {
            if (start && !_mp4) {
                //开始录制
                _mp4 = makeRecorder(getTracks(true), type, custom_path, sender);
            } else if (!start && _mp4) {
                //停止录制
                _mp4 = nullptr;
            }
            return true;
        }
        default : return false;
    }
 }
 //此函数可能跨线程调用
 bool MultiMuxerPrivate::isRecording(MediaSource &sender, Recorder::type type){
    switch (type){
        case Recorder::type_hls :
            return _hls ? true : false;
        case Recorder::type_mp4 :
            return _mp4 ? true : false;
        default:
            return false;
    }
 }
 void MultiMuxerPrivate::setTimeStamp(uint32_t stamp) {
    if (_rtmp) {
        _rtmp->setTimeStamp(stamp);
    }
    if (_rtsp) {
        _rtsp->setTimeStamp(stamp);
    }
 }
 void MultiMuxerPrivate::setTrackListener(Listener *listener) {
    _track_listener = listener;
 }
 void MultiMuxerPrivate::onTrackReady(const Track::Ptr &track) {
    if (_rtmp) {
        _rtmp->addTrack(track);
    }
    if (_rtsp) {
        _rtsp->addTrack(track);
    }
    if (_ts) {
        _ts->addTrack(track);
    }
 #if defined(ENABLE_MP4)
    if (_fmp4) {
        _fmp4->addTrack(track);
    }
 #endif
    //拷贝智能指针，目的是为了防止跨线程调用设置录像相关api导致的线程竞争问题
    auto hls = _hls;
    if (hls) {
        hls->addTrack(track);
    }
    auto mp4 = _mp4;
    if (mp4) {
        mp4->addTrack(track);
    }
 }
 bool MultiMuxerPrivate::isEnabled(){
    auto hls = _hls;
    return (_rtmp ? _rtmp->isEnabled() : false) ||
           (_rtsp ? _rtsp->isEnabled() : false) ||
           (_ts ? _ts->isEnabled() : false) ||
 #if defined(ENABLE_MP4)
           (_fmp4 ? _fmp4->isEnabled() : false) ||
 #endif
           (hls ? hls->isEnabled() : false) || _mp4;
 }
 void MultiMuxerPrivate::onTrackFrame(const Frame::Ptr &frame) {
    if (_rtmp) {
        _rtmp->inputFrame(frame);
    }
    if (_rtsp) {
        _rtsp->inputFrame(frame);
    }
    if (_ts) {
        _ts->inputFrame(frame);
    }
 #if defined(ENABLE_MP4)
    if (_fmp4) {
        _fmp4->inputFrame(frame);
    }
 #endif
    //拷贝智能指针，目的是为了防止跨线程调用设置录像相关api导致的线程竞争问题
    //此处使用智能指针拷贝来确保线程安全，比互斥锁性能更优
    auto hls = _hls;
    if (hls) {
        hls->inputFrame(frame);
    }
    auto mp4 = _mp4;
    if (mp4) {
        mp4->inputFrame(frame);
    }
 }
 static string getTrackInfoStr(const TrackSource *track_src){
    _StrPrinter codec_info;
    auto tracks = track_src->getTracks(true);
    for (auto &track : tracks) {
        auto codec_type = track->getTrackType();
        codec_info << track->getCodecName();
        switch (codec_type) {
            case TrackAudio : {
                auto audio_track = dynamic_pointer_cast<AudioTrack>(track);
                codec_info << "["
                           << audio_track->getAudioSampleRate() << "/"
                           << audio_track->getAudioChannel() << "/"
                           << audio_track->getAudioSampleBit() << "] ";
                break;
            }
            case TrackVideo : {
                auto video_track = dynamic_pointer_cast<VideoTrack>(track);
                codec_info << "["
                           << video_track->getVideoWidth() << "/"
                           << video_track->getVideoHeight() << "/"
                           << round(video_track->getVideoFps()) << "] ";
                break;
            }
            default:
                break;
        }
    }
    return codec_info;
 }
 void MultiMuxerPrivate::onAllTrackReady() {
    if (_rtmp) {
        _rtmp->onAllTrackReady();
    }
    if (_rtsp) {
        _rtsp->onAllTrackReady();
    }
 #if defined(ENABLE_MP4)
    if (_fmp4) {
        _fmp4->onAllTrackReady();
    }
 #endif
    if (_track_listener) {
        _track_listener->onAllTrackReady();
    }
    InfoL << "stream: " << _stream_url << " , codec info: " << getTrackInfoStr(this);
 }
 ///////////////////////////////MultiMediaSourceMuxer//////////////////////////////////
 MultiMediaSourceMuxer::~MultiMediaSourceMuxer() {}
 MultiMediaSourceMuxer::MultiMediaSourceMuxer(const string &vhost, const string &app, const string &stream, float dur_sec,
                                             bool enable_rtsp, bool enable_rtmp, bool enable_hls, bool enable_mp4) {
    _muxer.reset(new MultiMuxerPrivate(vhost, app, stream, dur_sec, enable_rtsp, enable_rtmp, enable_hls, enable_mp4));
    _muxer->setTrackListener(this);
 }
 void MultiMediaSourceMuxer::setMediaListener(const std::weak_ptr<MediaSourceEvent> &listener) {
    setDelegate(listener);
    //拦截事件
    _muxer->setMediaListener(shared_from_this());
 }
 void MultiMediaSourceMuxer::setTrackListener(const std::weak_ptr<MultiMuxerPrivate::Listener> &listener) {
    _track_listener = listener;
 }
 int MultiMediaSourceMuxer::totalReaderCount() const {
    return _muxer->totalReaderCount();
 }
 void MultiMediaSourceMuxer::setTimeStamp(uint32_t stamp) {
    _muxer->setTimeStamp(stamp);
 }
 vector<Track::Ptr> MultiMediaSourceMuxer::getTracks(MediaSource &sender, bool trackReady) const {
    return _muxer->getTracks(trackReady);
 }
 int MultiMediaSourceMuxer::totalReaderCount(MediaSource &sender) {
    auto listener = getDelegate();
    if (!listener) {
        return totalReaderCount();
    }
    return listener->totalReaderCount(sender);
 }
 bool MultiMediaSourceMuxer::setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) {
    return _muxer->setupRecord(sender, type, start, custom_path);
 }
 bool MultiMediaSourceMuxer::isRecording(MediaSource &sender, Recorder::type type) {
    return _muxer->isRecording(sender,type);
 }
-void MultiMediaSourceMuxer::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb){
+void MultiMediaSourceMuxer::startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, uint16_t src_port, const function<void(const SockException &ex)> &cb){
 #if defined(ENABLE_RTPPROXY)
    RtpSender::Ptr rtp_sender = std::make_shared<RtpSender>(atoi(ssrc.data()));
    weak_ptr<MultiMediaSourceMuxer> weak_self = shared_from_this();
-    rtp_sender->startSend(dst_url, dst_port, is_udp, [weak_self, rtp_sender, cb](const SockException &ex) {
+    rtp_sender->startSend(dst_url, dst_port, is_udp, src_port, [weak_self, rtp_sender, cb, ssrc](const SockException &ex) {
        cb(ex);
        auto strong_self = weak_self.lock();
        if (!strong_self || ex) {
            return;
        }
        for (auto &track : strong_self->_muxer->getTracks(false)) {
            rtp_sender->addTrack(track);
        }
        rtp_sender->addTrackCompleted();
-        strong_self->_rtp_sender = rtp_sender;
+        strong_self->_rtp_sender[ssrc] = rtp_sender;
    });
 #else
    cb(SockException(Err_other, "该功能未启用，编译时请打开ENABLE_RTPPROXY宏"));
 #endif//ENABLE_RTPPROXY
 }
-bool MultiMediaSourceMuxer::stopSendRtp(MediaSource &sender){
+bool MultiMediaSourceMuxer::stopSendRtp(MediaSource &sender, const string& ssrc){
 #if defined(ENABLE_RTPPROXY)
-    if (_rtp_sender) {
+	map<string, RtpSender::Ptr>::iterator ite = _rtp_sender.find(ssrc);
-        _rtp_sender = nullptr;
+	if (ite != _rtp_sender.end())
-        return true;
+	{
-    }
+		ite->second = nullptr;
-#endif//ENABLE_RTPPROXY
+		_rtp_sender.erase(ite);
-    return false;
+		return true;
-}
+	}
+#endif//ENABLE_RTPPROXY
-void MultiMediaSourceMuxer::addTrack(const Track::Ptr &track) {
+    return false;
-    _muxer->addTrack(track);
+}
-}
+void MultiMediaSourceMuxer::addTrack(const Track::Ptr &track) {
-void MultiMediaSourceMuxer::addTrackCompleted() {
+    _muxer->addTrack(track);
-    _muxer->addTrackCompleted();
+}
-}
+void MultiMediaSourceMuxer::addTrackCompleted() {
-void MultiMediaSourceMuxer::onAllTrackReady(){
+    _muxer->addTrackCompleted();
-    _muxer->setMediaListener(shared_from_this());
+}
-    auto listener = _track_listener.lock();
-    if(listener){
+void MultiMediaSourceMuxer::onAllTrackReady(){
-        listener->onAllTrackReady();
+    _muxer->setMediaListener(shared_from_this());
-    }
+    auto listener = _track_listener.lock();
-}
+    if(listener){
+        listener->onAllTrackReady();
-void MultiMediaSourceMuxer::resetTracks() {
+    }
-    _muxer->resetTracks();
+}
-}
+void MultiMediaSourceMuxer::resetTracks() {
-//该类实现frame级别的时间戳覆盖
+    _muxer->resetTracks();
-class FrameModifyStamp : public Frame{
+}
-public:
-    typedef std::shared_ptr<FrameModifyStamp> Ptr;
+//该类实现frame级别的时间戳覆盖
-    FrameModifyStamp(const Frame::Ptr &frame, Stamp &stamp){
+class FrameModifyStamp : public Frame{
-        _frame = frame;
+public:
-        //覆盖时间戳
+    typedef std::shared_ptr<FrameModifyStamp> Ptr;
-        stamp.revise(frame->dts(), frame->pts(), _dts, _pts, true);
+    FrameModifyStamp(const Frame::Ptr &frame, Stamp &stamp){
-    }
+        _frame = frame;
-    ~FrameModifyStamp() override {}
+        //覆盖时间戳
+        stamp.revise(frame->dts(), frame->pts(), _dts, _pts, true);
-    uint32_t dts() const override{
+    }
-        return _dts;
+    ~FrameModifyStamp() override {}
-    }
+    uint32_t dts() const override{
-    uint32_t pts() const override{
+        return _dts;
-        return _pts;
+    }
-    }
+    uint32_t pts() const override{
-    uint32_t prefixSize() const override {
+        return _pts;
-        return _frame->prefixSize();
+    }
-    }
+    uint32_t prefixSize() const override {
-    bool keyFrame() const override {
+        return _frame->prefixSize();
-        return _frame->keyFrame();
+    }
-    }
+    bool keyFrame() const override {
-    bool configFrame() const override {
+        return _frame->keyFrame();
-        return _frame->configFrame();
+    }
-    }
+    bool configFrame() const override {
-    bool cacheAble() const override {
+        return _frame->configFrame();
-        return _frame->cacheAble();
+    }
-    }
+    bool cacheAble() const override {
-    char *data() const override {
+        return _frame->cacheAble();
-        return _frame->data();
+    }
-    }
+    char *data() const override {
-    uint32_t size() const override {
+        return _frame->data();
-        return _frame->size();
+    }
-    }
+    uint32_t size() const override {
-    CodecId getCodecId() const override {
+        return _frame->size();
-        return _frame->getCodecId();
+    }
-    }
-private:
+    CodecId getCodecId() const override {
-    int64_t _dts;
+        return _frame->getCodecId();
-    int64_t _pts;
+    }
-    Frame::Ptr _frame;
+private:
-};
+    int64_t _dts;
+    int64_t _pts;
-void MultiMediaSourceMuxer::inputFrame(const Frame::Ptr &frame_in) {
+    Frame::Ptr _frame;
-    GET_CONFIG(bool, modify_stamp, General::kModifyStamp);
+};
-    auto frame = frame_in;
-    if (modify_stamp) {
+void MultiMediaSourceMuxer::inputFrame(const Frame::Ptr &frame_in) {
-        //开启了时间戳覆盖
+    GET_CONFIG(bool, modify_stamp, General::kModifyStamp);
-        frame = std::make_shared<FrameModifyStamp>(frame, _stamp[frame->getTrackType()]);
+    auto frame = frame_in;
-    }
+    if (modify_stamp) {
-    _muxer->inputFrame(frame);
+        //开启了时间戳覆盖
+        frame = std::make_shared<FrameModifyStamp>(frame, _stamp[frame->getTrackType()]);
-#if defined(ENABLE_RTPPROXY)
+    }
-    auto rtp_sender = _rtp_sender;
+    _muxer->inputFrame(frame);
-    if (rtp_sender) {
-        rtp_sender->inputFrame(frame);
+#if defined(ENABLE_RTPPROXY)
-    }
+	map<string, RtpSender::Ptr>::iterator ite = _rtp_sender.begin();
-#endif //ENABLE_RTPPROXY
+	while (ite != _rtp_sender.end())
+	{
-}
+		if (ite->second)
+		{
-bool MultiMediaSourceMuxer::isEnabled(){
+			ite->second->inputFrame(frame);
-    GET_CONFIG(uint32_t, stream_none_reader_delay_ms, General::kStreamNoneReaderDelayMS);
+		}
-    if (!_is_enable || _last_check.elapsedTime() > stream_none_reader_delay_ms) {
+		ite++;
-        //无人观看时，每次检查是否真的无人观看
+	}
-        //有人观看时，则延迟一定时间检查一遍是否无人观看了(节省性能)
+#endif //ENABLE_RTPPROXY
-#if defined(ENABLE_RTPPROXY)
-        _is_enable = (_muxer->isEnabled() || _rtp_sender);
+}
-#else
-        _is_enable = _muxer->isEnabled();
+bool MultiMediaSourceMuxer::isEnabled(){
-#endif //ENABLE_RTPPROXY
+    GET_CONFIG(uint32_t, stream_none_reader_delay_ms, General::kStreamNoneReaderDelayMS);
-        if (_is_enable) {
+    if (!_is_enable || _last_check.elapsedTime() > stream_none_reader_delay_ms) {
-            //无人观看时，不刷新计时器,因为无人观看时每次都会检查一遍，所以刷新计数器无意义且浪费cpu
+        //无人观看时，每次检查是否真的无人观看
-            _last_check.resetTime();
+        //有人观看时，则延迟一定时间检查一遍是否无人观看了(节省性能)
-        }
+#if defined(ENABLE_RTPPROXY)
-    }
+        _is_enable = (_muxer->isEnabled() || _rtp_sender.size());
-    return _is_enable;
+#else
-}
+        _is_enable = _muxer->isEnabled();
+#endif //ENABLE_RTPPROXY
+        if (_is_enable) {
-}//namespace mediakit
+            //无人观看时，不刷新计时器,因为无人观看时每次都会检查一遍，所以刷新计数器无意义且浪费cpu
+            _last_check.resetTime();
+        }
+    }
+    return _is_enable;
+}
+}//namespace mediakit
--- a/src/Common/MultiMediaSourceMuxer.h
+++ b/src/Common/MultiMediaSourceMuxer.h
 /*
 * Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
 *
 * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
 *
 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
 */
 #ifndef ZLMEDIAKIT_MULTIMEDIASOURCEMUXER_H
 #define ZLMEDIAKIT_MULTIMEDIASOURCEMUXER_H
 #include "Common/Stamp.h"
 #include "Rtp/RtpSender.h"
 #include "Record/Recorder.h"
 #include "Record/HlsRecorder.h"
 #include "Record/HlsMediaSource.h"
 #include "Rtsp/RtspMediaSourceMuxer.h"
 #include "Rtmp/RtmpMediaSourceMuxer.h"
 #include "TS/TSMediaSourceMuxer.h"
 #include "FMP4/FMP4MediaSourceMuxer.h"
 namespace mediakit{
 class MultiMuxerPrivate : public MediaSink, public std::enable_shared_from_this<MultiMuxerPrivate>{
 public:
    friend class MultiMediaSourceMuxer;
    typedef std::shared_ptr<MultiMuxerPrivate> Ptr;
    class Listener{
    public:
        Listener() = default;
        virtual ~Listener() = default;
        virtual void onAllTrackReady() = 0;
    };
    ~MultiMuxerPrivate() override;
 private:
    MultiMuxerPrivate(const string &vhost,const string &app, const string &stream,float dur_sec,
                      bool enable_rtsp, bool enable_rtmp, bool enable_hls, bool enable_mp4);
    void resetTracks() override;
    void setMediaListener(const std::weak_ptr<MediaSourceEvent> &listener);
    int totalReaderCount() const;
    void setTimeStamp(uint32_t stamp);
    void setTrackListener(Listener *listener);
    bool setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path);
    bool isRecording(MediaSource &sender, Recorder::type type);
    bool isEnabled();
    void onTrackReady(const Track::Ptr & track) override;
    void onTrackFrame(const Frame::Ptr &frame) override;
    void onAllTrackReady() override;
 private:
    string _stream_url;
    Listener *_track_listener = nullptr;
    RtmpMediaSourceMuxer::Ptr _rtmp;
    RtspMediaSourceMuxer::Ptr _rtsp;
    HlsRecorder::Ptr _hls;
    MediaSinkInterface::Ptr _mp4;
    TSMediaSourceMuxer::Ptr _ts;
 #if defined(ENABLE_MP4)
    FMP4MediaSourceMuxer::Ptr _fmp4;
 #endif
    std::weak_ptr<MediaSourceEvent> _listener;
 };
 class MultiMediaSourceMuxer : public MediaSourceEventInterceptor, public MediaSinkInterface, public MultiMuxerPrivate::Listener, public std::enable_shared_from_this<MultiMediaSourceMuxer>{
 public:
    typedef MultiMuxerPrivate::Listener Listener;
    typedef std::shared_ptr<MultiMediaSourceMuxer> Ptr;
    ~MultiMediaSourceMuxer() override;
    MultiMediaSourceMuxer(const string &vhost, const string &app, const string &stream, float dur_sec = 0.0,
                          bool enable_rtsp = true, bool enable_rtmp = true, bool enable_hls = true, bool enable_mp4 = false);
    /**
     * 设置事件监听器
     * @param listener 监听器
     */
    void setMediaListener(const std::weak_ptr<MediaSourceEvent> &listener);
     /**
      * 随着Track就绪事件监听器
      * @param listener 事件监听器
     */
    void setTrackListener(const std::weak_ptr<MultiMuxerPrivate::Listener> &listener);
    /**
     * 返回总的消费者个数
     */
    int totalReaderCount() const;
    /**
     * 判断是否生效(是否正在转其他协议)
     */
    bool isEnabled();
    /**
     * 设置MediaSource时间戳
     * @param stamp 时间戳
     */
    void setTimeStamp(uint32_t stamp);
    /////////////////////////////////MediaSourceEvent override/////////////////////////////////
    /**
     * 获取所有Track
     * @param trackReady 是否筛选过滤未就绪的track
     * @return 所有Track
     */
    vector<Track::Ptr> getTracks(MediaSource &sender, bool trackReady = true) const override;
    /**
     * 观看总人数
     * @param sender 事件发送者
     * @return 观看总人数
     */
    int totalReaderCount(MediaSource &sender) override;
    /**
     * 设置录制状态
     * @param type 录制类型
     * @param start 开始或停止
     * @param custom_path 开启录制时，指定自定义路径
     * @return 是否设置成功
     */
    bool setupRecord(MediaSource &sender, Recorder::type type, bool start, const string &custom_path) override;
    /**
     * 获取录制状态
     * @param type 录制类型
     * @return 录制状态
     */
    bool isRecording(MediaSource &sender, Recorder::type type) override;
    /**
     * 开始发送ps-rtp流
     * @param dst_url 目标ip或域名
     * @param dst_port 目标端口
     * @param ssrc rtp的ssrc
     * @param is_udp 是否为udp
     * @param cb 启动成功或失败回调
     */
-    void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, const function<void(const SockException &ex)> &cb) override;
+    void startSendRtp(MediaSource &sender, const string &dst_url, uint16_t dst_port, const string &ssrc, bool is_udp, uint16_t src_port, const function<void(const SockException &ex)> &cb) override;
    /**
     * 停止ps-rtp发送
     * @return 是否成功
     */
-    bool stopSendRtp(MediaSource &sender) override;
+    bool stopSendRtp(MediaSource &sender, const string &ssrc) override;
    /////////////////////////////////MediaSinkInterface override/////////////////////////////////
    /**
    * 添加track，内部会调用Track的clone方法
    * 只会克隆sps pps这些信息 ，而不会克隆Delegate相关关系
    * @param track 添加音频或视频轨道
    */
    void addTrack(const Track::Ptr &track) override;
    /**
     * 添加track完毕
     */
    void addTrackCompleted() override;
    /**
     * 重置track
     */
    void resetTracks() override;
    /**
     * 写入帧数据
     * @param frame 帧
     */
    void inputFrame(const Frame::Ptr &frame) override;
    /////////////////////////////////MultiMuxerPrivate::Listener override/////////////////////////////////
    /**
     * 所有track全部就绪
     */
    void onAllTrackReady() override;
 private:
    bool _is_enable = false;
    Ticker _last_check;
    Stamp _stamp[2];
    MultiMuxerPrivate::Ptr _muxer;
    std::weak_ptr<MultiMuxerPrivate::Listener> _track_listener;
 #if defined(ENABLE_RTPPROXY)
-    RtpSender::Ptr _rtp_sender;
+	map<string, RtpSender::Ptr> _rtp_sender;
 #endif //ENABLE_RTPPROXY
 };
 }//namespace mediakit
 #endif //ZLMEDIAKIT_MULTIMEDIASOURCEMUXER_H
--- a/src/Rtp/RtpSender.cpp
+++ b/src/Rtp/RtpSender.cpp
 /*
 * Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
 *
 * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
 *
 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
 */
 #if defined(ENABLE_RTPPROXY)
 #include "RtpSender.h"
 #include "Rtsp/RtspSession.h"
 #include "Thread/WorkThreadPool.h"
 #include "RtpCache.h"
 namespace mediakit{
 RtpSender::RtpSender(uint32_t ssrc, uint8_t payload_type) {
    _poller = EventPollerPool::Instance().getPoller();
    _interface = std::make_shared<RtpCachePS>([this](std::shared_ptr<List<Buffer::Ptr> > list) {
        onFlushRtpList(std::move(list));
    }, ssrc, payload_type);
 }
 RtpSender::~RtpSender() {
 }
-void RtpSender::startSend(const string &dst_url, uint16_t dst_port, bool is_udp, const function<void(const SockException &ex)> &cb){
+void RtpSender::startSend(const string &dst_url, uint16_t dst_port, bool is_udp, uint16_t src_port, const function<void(const SockException &ex)> &cb){
    _is_udp = is_udp;
    _socket = Socket::createSocket(_poller, false);
    _dst_url = dst_url;
    _dst_port = dst_port;
-    weak_ptr<RtpSender> weak_self = shared_from_this();
+	_src_port = src_port;
-    if (is_udp) {
+    weak_ptr<RtpSender> weak_self = shared_from_this();
-        _socket->bindUdpSock(0);
+    if (is_udp) {
-        auto poller = _poller;
+        _socket->bindUdpSock(src_port);
-        WorkThreadPool::Instance().getPoller()->async([cb, dst_url, dst_port, weak_self, poller]() {
+        auto poller = _poller;
-            struct sockaddr addr;
+        WorkThreadPool::Instance().getPoller()->async([cb, dst_url, dst_port, weak_self, poller]() {
-            //切换线程目的是为了dns解析放在后台线程执行
+            struct sockaddr addr;
-            if (!SockUtil::getDomainIP(dst_url.data(), dst_port, addr)) {
+            //切换线程目的是为了dns解析放在后台线程执行
-                poller->async([dst_url, cb]() {
+            if (!SockUtil::getDomainIP(dst_url.data(), dst_port, addr)) {
-                    //切回自己的线程
+                poller->async([dst_url, cb]() {
-                    cb(SockException(Err_dns, StrPrinter << "dns解析域名失败:" << dst_url));
+                    //切回自己的线程
-                });
+                    cb(SockException(Err_dns, StrPrinter << "dns解析域名失败:" << dst_url));
-                return;
+                });
-            }
+                return;
+            }
-            //dns解析成功
-            poller->async([addr, weak_self, cb]() {
+            //dns解析成功
-                //切回自己的线程
+            poller->async([addr, weak_self, cb]() {
-                cb(SockException());
+                //切回自己的线程
-                auto strong_self = weak_self.lock();
+                cb(SockException());
-                if (strong_self) {
+                auto strong_self = weak_self.lock();
-                    strong_self->_socket->setSendPeerAddr(&addr);
+                if (strong_self) {
-                    strong_self->onConnect();
+                    strong_self->_socket->setSendPeerAddr(&addr);
-                }
+                    strong_self->onConnect();
-            });
+                }
-        });
+            });
-    } else {
+        });
-        _socket->connect(dst_url, dst_port, [cb, weak_self](const SockException &err) {
+    } else {
-            cb(err);
+        _socket->connect(dst_url, dst_port, [cb, weak_self](const SockException &err) {
-            auto strong_self = weak_self.lock();
+            cb(err);
-            if (strong_self && !err) {
+            auto strong_self = weak_self.lock();
-                //tcp连接成功
+            if (strong_self && !err) {
-                strong_self->onConnect();
+                //tcp连接成功
-            }
+                strong_self->onConnect();
-        });
+            }
-    }
+        }, 5.0F, "0.0.0.0", src_port);
-}
+    }
+}
-void RtpSender::onConnect(){
-    _is_connect = true;
+void RtpSender::onConnect(){
-    //加大发送缓存,防止udp丢包之类的问题
+    _is_connect = true;
-    SockUtil::setSendBuf(_socket->rawFD(), 4 * 1024 * 1024);
+    //加大发送缓存,防止udp丢包之类的问题
-    if (!_is_udp) {
+    SockUtil::setSendBuf(_socket->rawFD(), 4 * 1024 * 1024);
-        //关闭tcp no_delay并开启MSG_MORE, 提高发送性能
+    if (!_is_udp) {
-        SockUtil::setNoDelay(_socket->rawFD(), false);
+        //关闭tcp no_delay并开启MSG_MORE, 提高发送性能
-        _socket->setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE);
+        SockUtil::setNoDelay(_socket->rawFD(), false);
-    }
+        _socket->setSendFlags(SOCKET_DEFAULE_FLAGS | FLAG_MORE);
-    //连接建立成功事件
+    }
-    weak_ptr<RtpSender> weak_self = shared_from_this();
+    //连接建立成功事件
-    _socket->setOnErr([weak_self](const SockException &err) {
+    weak_ptr<RtpSender> weak_self = shared_from_this();
-        auto strong_self = weak_self.lock();
+    _socket->setOnErr([weak_self](const SockException &err) {
-        if (strong_self) {
+        auto strong_self = weak_self.lock();
-            strong_self->onErr(err);
+        if (strong_self) {
-        }
+            strong_self->onErr(err);
-    });
+        }
-    InfoL << "开始发送 rtp:" << _socket->get_peer_ip() << ":" << _socket->get_peer_port() << ", 是否为udp方式:" << _is_udp;
+    });
-}
+    InfoL << "开始发送 rtp:" << _socket->get_peer_ip() << ":" << _socket->get_peer_port() << ", 是否为udp方式:" << _is_udp;
+}
-void RtpSender::addTrack(const Track::Ptr &track){
-    _interface->addTrack(track);
+void RtpSender::addTrack(const Track::Ptr &track){
-}
+    _interface->addTrack(track);
+}
-void RtpSender::addTrackCompleted(){
-    _interface->addTrackCompleted();
+void RtpSender::addTrackCompleted(){
-}
+    _interface->addTrackCompleted();
+}
-void RtpSender::resetTracks(){
-    _interface->resetTracks();
+void RtpSender::resetTracks(){
-}
+    _interface->resetTracks();
+}
-//此函数在其他线程执行
-void RtpSender::inputFrame(const Frame::Ptr &frame) {
+//此函数在其他线程执行
-    if (_is_connect) {
+void RtpSender::inputFrame(const Frame::Ptr &frame) {
-        //连接成功后才做实质操作(节省cpu资源)
+    if (_is_connect) {
-        _interface->inputFrame(frame);
+        //连接成功后才做实质操作(节省cpu资源)
-    }
+        _interface->inputFrame(frame);
-}
+    }
+}
-//此函数在其他线程执行
-void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr> > rtp_list) {
+//此函数在其他线程执行
-    if(!_is_connect){
+void RtpSender::onFlushRtpList(shared_ptr<List<Buffer::Ptr> > rtp_list) {
-        //连接成功后才能发送数据
+    if(!_is_connect){
-        return;
+        //连接成功后才能发送数据
-    }
+        return;
+    }
-    auto is_udp = _is_udp;
-    auto socket = _socket;
+    auto is_udp = _is_udp;
-    _poller->async([rtp_list, is_udp, socket]() {
+    auto socket = _socket;
-        int i = 0;
+    _poller->async([rtp_list, is_udp, socket]() {
-        int size = rtp_list->size();
+        int i = 0;
-        rtp_list->for_each([&](Buffer::Ptr &packet) {
+        int size = rtp_list->size();
-            if (is_udp) {
+        rtp_list->for_each([&](Buffer::Ptr &packet) {
-                //udp模式，rtp over tcp前4个字节可以忽略
+            if (is_udp) {
-                socket->send(std::make_shared<BufferRtp>(std::move(packet), 4), nullptr, 0, ++i == size);
+                //udp模式，rtp over tcp前4个字节可以忽略
-            } else {
+                socket->send(std::make_shared<BufferRtp>(std::move(packet), 4), nullptr, 0, ++i == size);
-                //tcp模式, rtp over tcp前2个字节可以忽略,只保留后续rtp长度的2个字节
+            } else {
-                socket->send(std::make_shared<BufferRtp>(std::move(packet), 2), nullptr, 0, ++i == size);
+                //tcp模式, rtp over tcp前2个字节可以忽略,只保留后续rtp长度的2个字节
-            }
+                socket->send(std::make_shared<BufferRtp>(std::move(packet), 2), nullptr, 0, ++i == size);
-        });
+            }
-    });
+        });
-}
+    });
+}
-void RtpSender::onErr(const SockException &ex, bool is_connect) {
-    _is_connect = false;
+void RtpSender::onErr(const SockException &ex, bool is_connect) {
+    _is_connect = false;
-    //监听socket断开事件，方便重连
-    if (is_connect) {
+    //监听socket断开事件，方便重连
-        WarnL << "重连" << _dst_url << ":" << _dst_port << "失败, 原因为:" << ex.what();
+    if (is_connect) {
-    } else {
+        WarnL << "重连" << _dst_url << ":" << _dst_port << "失败, 原因为:" << ex.what();
-        WarnL << "停止发送 rtp:" <<  _dst_url << ":" << _dst_port << ", 原因为:" << ex.what();
+    } else {
-    }
+        WarnL << "停止发送 rtp:" <<  _dst_url << ":" << _dst_port << ", 原因为:" << ex.what();
+    }
-    weak_ptr<RtpSender> weak_self = shared_from_this();
-    _connect_timer = std::make_shared<Timer>(10.0, [weak_self]() {
+    weak_ptr<RtpSender> weak_self = shared_from_this();
-        auto strong_self = weak_self.lock();
+    _connect_timer = std::make_shared<Timer>(10.0, [weak_self]() {
-        if (!strong_self) {
+        auto strong_self = weak_self.lock();
-            return false;
+        if (!strong_self) {
-        }
+            return false;
-        strong_self->startSend(strong_self->_dst_url, strong_self->_dst_port, strong_self->_is_udp, [weak_self](const SockException &ex){
+        }
-            auto strong_self = weak_self.lock();
+        strong_self->startSend(strong_self->_dst_url, strong_self->_dst_port, strong_self->_is_udp, strong_self->_src_port, [weak_self](const SockException &ex){
-            if (strong_self && ex) {
+            auto strong_self = weak_self.lock();
-                //连接失败且本对象未销毁，那么重试连接
+            if (strong_self && ex) {
-                strong_self->onErr(ex, true);
+                //连接失败且本对象未销毁，那么重试连接
-            }
+                strong_self->onErr(ex, true);
-        });
+            }
-        return false;
+        });
-    }, _poller);
+        return false;
-}
+    }, _poller);
+}
-}//namespace mediakit
+}//namespace mediakit
 #endif// defined(ENABLE_RTPPROXY)
\ No newline at end of file
--- a/src/Rtp/RtpSender.h
+++ b/src/Rtp/RtpSender.h
 /*
 * Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
 *
 * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
 *
 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
 */
 #ifndef ZLMEDIAKIT_RTPSENDER_H
 #define ZLMEDIAKIT_RTPSENDER_H
 #if defined(ENABLE_RTPPROXY)
 #include "PSEncoder.h"
 #include "Extension/CommonRtp.h"
 namespace mediakit{
 //rtp发送客户端，支持发送GB28181协议
 class RtpSender : public MediaSinkInterface, public std::enable_shared_from_this<RtpSender>{
 public:
    typedef std::shared_ptr<RtpSender> Ptr;
    ~RtpSender() override;
    /**
     * 构造函数，创建GB28181 RTP发送客户端
     * @param ssrc rtp的ssrc
     * @param payload_type 国标中ps-rtp的pt一般为96
     */
    RtpSender(uint32_t ssrc, uint8_t payload_type = 96);
    /**
     * 开始发送ps-rtp包
     * @param dst_url 目标ip或域名
     * @param dst_port 目标端口
     * @param is_udp 是否采用udp方式发送rtp
     * @param cb 连接目标端口是否成功的回调
     */
-    void startSend(const string &dst_url, uint16_t dst_port, bool is_udp, const function<void(const SockException &ex)> &cb);
+    void startSend(const string &dst_url, uint16_t dst_port, bool is_udp, uint16_t src_port, const function<void(const SockException &ex)> &cb);
    /**
     * 输入帧数据
     */
    void inputFrame(const Frame::Ptr &frame) override;
    /**
     * 添加track，内部会调用Track的clone方法
     * 只会克隆sps pps这些信息 ，而不会克隆Delegate相关关系
     * @param track
     */
    virtual void addTrack(const Track::Ptr & track) override;
    /**
     * 添加所有Track完毕
     */
    virtual void addTrackCompleted() override;
    /**
     * 重置track
     */
    virtual void resetTracks() override;
 private:
    //合并写输出
    void onFlushRtpList(std::shared_ptr<List<Buffer::Ptr> > rtp_list);
    //udp/tcp连接成功回调
    void onConnect();
    //异常断开socket事件
    void onErr(const SockException &ex, bool is_connect = false);
 private:
    bool _is_udp;
    bool _is_connect = false;
    string _dst_url;
    uint16_t _dst_port;
-    Socket::Ptr _socket;
+	uint16_t _src_port;
-    EventPoller::Ptr _poller;
+    Socket::Ptr _socket;
-    Timer::Ptr _connect_timer;
+    EventPoller::Ptr _poller;
-    MediaSinkInterface::Ptr _interface;
+    Timer::Ptr _connect_timer;
-};
+    MediaSinkInterface::Ptr _interface;
+};
-}//namespace mediakit
-#endif// defined(ENABLE_RTPPROXY)
+}//namespace mediakit
-#endif //ZLMEDIAKIT_RTPSENDER_H
+#endif// defined(ENABLE_RTPPROXY)
+#endif //ZLMEDIAKIT_RTPSENDER_H
--- a/src/Rtp/RtpServer.cpp
+++ b/src/Rtp/RtpServer.cpp
 /*
 * Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
 *
 * This file is part of ZLMediaKit(https://github.com/xiongziliang/ZLMediaKit).
 *
 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
 */
 #if defined(ENABLE_RTPPROXY)
 #include "RtpServer.h"
 #include "RtpSelector.h"
 namespace mediakit{
 RtpServer::RtpServer() {
 }
 RtpServer::~RtpServer() {
    if(_on_clearup){
        _on_clearup();
    }
 }
 void RtpServer::start(uint16_t local_port, const string &stream_id,  bool enable_tcp, const char *local_ip) {
    //创建udp服务器
    Socket::Ptr udp_server = Socket::createSocket(nullptr, false);
    if (local_port == 0) {
        //随机端口，rtp端口采用偶数
        Socket::Ptr rtcp_server = Socket::createSocket(nullptr, false);
        auto pair = std::make_pair(udp_server, rtcp_server);
        makeSockPair(pair, local_ip);
        //取偶数端口
        udp_server = pair.first;
    } else if (!udp_server->bindUdpSock(local_port, local_ip)) {
        //用户指定端口
        throw std::runtime_error(StrPrinter << "bindUdpSock on " << local_ip << ":" << local_port << " failed:" << get_uv_errmsg(true));
    }
    //设置udp socket读缓存
    SockUtil::setRecvBuf(udp_server->rawFD(), 4 * 1024 * 1024);
    TcpServer::Ptr tcp_server;
    if (enable_tcp) {
        //创建tcp服务器
        tcp_server = std::make_shared<TcpServer>(udp_server->getPoller());
        (*tcp_server)[RtpSession::kStreamID] = stream_id;
        tcp_server->start<RtpSession>(udp_server->get_local_port(), local_ip);
    }
    RtpProcess::Ptr process;
    if (!stream_id.empty()) {
        //指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流)
        process = RtpSelector::Instance().getProcess(stream_id, true);
-        udp_server->setOnRead([udp_server, process](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
+		//udp_server->setOnRead([udp_server, process](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
-            process->inputRtp(true, udp_server, buf->data(), buf->size(), addr);
+		//    process->inputRtp(true, udp_server, buf->data(), buf->size(), addr);
-        });
+		//});
-    } else {
+		weak_ptr<Socket> weak_sock = udp_server;
-        //未指定流id，一个端口多个流，通过ssrc来分流
+		udp_server->setOnRead([weak_sock, process](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
-        auto &ref = RtpSelector::Instance();
+			process->inputRtp(true, weak_sock.lock(), buf->data(), buf->size(), addr);
-        udp_server->setOnRead([&ref, udp_server](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
+        });
-            ref.inputRtp(udp_server, buf->data(), buf->size(), addr);
+    } else {
-        });
+        //未指定流id，一个端口多个流，通过ssrc来分流
-    }
+        auto &ref = RtpSelector::Instance();
+		//udp_server->setOnRead([&ref, udp_server](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
-    _on_clearup = [udp_server, process, stream_id]() {
+		//    ref.inputRtp(udp_server, buf->data(), buf->size(), addr);
-        //去除循环引用
+		//});
-        udp_server->setOnRead(nullptr);
+		weak_ptr<Socket> weak_sock = udp_server;
-        if (process) {
+		udp_server->setOnRead([&ref, weak_sock](const Buffer::Ptr &buf, struct sockaddr *addr, int) {
-            //删除rtp处理器
+			ref.inputRtp(weak_sock.lock(), buf->data(), buf->size(), addr);
-            RtpSelector::Instance().delProcess(stream_id, process.get());
+        });
-        }
+    }
-    };
+    _on_clearup = [udp_server, process, stream_id]() {
-    _tcp_server = tcp_server;
+        //去除循环引用
-    _udp_server = udp_server;
+        //udp_server->setOnRead(nullptr);
-    _rtp_process = process;
+        if (process) {
-}
+            //删除rtp处理器
+            RtpSelector::Instance().delProcess(stream_id, process.get());
-void RtpServer::setOnDetach(const function<void()> &cb){
+        }
-    if(_rtp_process){
+    };
-        _rtp_process->setOnDetach(cb);
-    }
+    _tcp_server = tcp_server;
-}
+    _udp_server = udp_server;
+    _rtp_process = process;
-EventPoller::Ptr RtpServer::getPoller() {
+}
-    return _udp_server->getPoller();
-}
+void RtpServer::setOnDetach(const function<void()> &cb){
+    if(_rtp_process){
-uint16_t RtpServer::getPort() {
+        _rtp_process->setOnDetach(cb);
-    return _udp_server ? _udp_server->get_local_port() : 0;
+    }
 }
-}//namespace mediakit
+EventPoller::Ptr RtpServer::getPoller() {
+    return _udp_server->getPoller();
+}
+uint16_t RtpServer::getPort() {
+    return _udp_server ? _udp_server->get_local_port() : 0;
+}
+}//namespace mediakit
 #endif//defined(ENABLE_RTPPROXY)
\ No newline at end of file