move srt src like webrtc (#4471)

srt/SrtCaller.h

@@ -0,0 +1,202 @@
+/*
+ * Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
+ *
+ * This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
+ *
+ * Use of this source code is governed by MIT-like 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_SRTCALLER_H
+#define ZLMEDIAKIT_SRTCALLER_H
+
+//srt
+#include "srt/Packet.hpp"
+#include "srt/Crypto.hpp"
+#include "srt/PacketQueue.hpp"
+#include "srt/PacketSendQueue.hpp"
+#include "srt/Statistic.hpp"
+
+#include "Poller/EventPoller.h"
+#include "Network/Socket.h"
+#include "Poller/Timer.h"
+#include "Util/TimeTicker.h"
+#include "Common/MultiMediaSourceMuxer.h"
+#include "Rtp/Decoder.h"
+#include "TS/TSMediaSource.h"
+#include <memory>
+#include <string>
+
+
+namespace mediakit {
+
+// 解析srt 信令url的工具类
+class SrtUrl {
+public:
+    std::string _full_url;
+    std::string _params;
+    std::string _host;
+    uint16_t _port;
+    std::string _streamid;
+
+public:
+    void parse(const std::string &url);
+};
+
+// 实现了webrtc代理拉流功能
+class SrtCaller : public std::enable_shared_from_this<SrtCaller>{
+public:
+    using Ptr = std::shared_ptr<SrtCaller>;
+
+    using SteadyClock = std::chrono::steady_clock;
+    using TimePoint = std::chrono::time_point<SteadyClock>;
+
+    SrtCaller(const toolkit::EventPoller::Ptr &poller);
+    virtual ~SrtCaller();
+
+    const toolkit::EventPoller::Ptr &getPoller() const {return _poller;}
+
+    virtual void inputSockData(uint8_t *buf, int len, struct sockaddr *addr);
+    virtual void onSendTSData(const SRT::Buffer::Ptr &buffer, bool flush);
+
+    size_t getRecvSpeed() const;
+    size_t getRecvTotalBytes() const;
+    size_t getSendSpeed() const;
+    size_t getSendTotalBytes() const;
+
+protected:
+
+    virtual void onConnect();
+    virtual void onHandShakeFinished();
+    virtual void onResult(const toolkit::SockException &ex);
+
+    virtual void onSRTData(SRT::DataPacket::Ptr pkt);
+
+    virtual uint16_t getLatency() = 0;
+    virtual int getLatencyMul();
+    virtual int getPktBufSize();
+    virtual float getTimeOutSec();
+
+    virtual bool isPlayer() = 0;
+
+private:
+    void doHandshake();
+
+    void sendHandshakeInduction();
+    void sendHandshakeConclusion();
+    void sendACKPacket();
+    void sendLightACKPacket();
+    void sendNAKPacket(std::list<SRT::PacketQueue::LostPair> &lost_list);
+    void sendMsgDropReq(uint32_t first, uint32_t last);
+    void sendKeepLivePacket();
+    void sendShutDown();
+    void tryAnnounceKeyMaterial();
+    void sendControlPacket(SRT::ControlPacket::Ptr pkt, bool flush = true);
+    void sendDataPacket(SRT::DataPacket::Ptr pkt, char *buf, int len, bool flush = false);
+    void sendPacket(toolkit::Buffer::Ptr pkt, bool flush);
+
+    void handleHandshake(uint8_t *buf, int len, struct sockaddr *addr);
+    void handleHandshakeInduction(SRT::HandshakePacket &pkt, struct sockaddr *addr);
+    void handleHandshakeConclusion(SRT::HandshakePacket &pkt, struct sockaddr *addr);
+    void handleACK(uint8_t *buf, int len, struct sockaddr *addr);
+    void handleACKACK(uint8_t *buf, int len, struct sockaddr *addr);
+    void handleNAK(uint8_t *buf, int len, struct sockaddr *addr);
+    void handleDropReq(uint8_t *buf, int len, struct sockaddr *addr);
+    void handleKeeplive(uint8_t *buf, int len, struct sockaddr *addr);
+    void handleShutDown(uint8_t *buf, int len, struct sockaddr *addr);
+    void handlePeerError(uint8_t *buf, int len, struct sockaddr *addr);
+    void handleCongestionWarning(uint8_t *buf, int len, struct sockaddr *addr);
+    void handleUserDefinedType(uint8_t *buf, int len, struct sockaddr *addr);
+    void handleDataPacket(uint8_t *buf, int len, struct sockaddr *addr);
+    void handleKeyMaterialReqPacket(uint8_t *buf, int len, struct sockaddr *addr);
+    void handleKeyMaterialRspPacket(uint8_t *buf, int len, struct sockaddr *addr);
+
+    void checkAndSendAckNak();
+    void createTimerForCheckAlive();
+
+    std::string generateStreamId();
+    uint32_t generateSocketId();
+    int32_t generateInitSeq();
+    size_t  getPayloadSize();
+
+    virtual std::string getPassphrase() = 0;
+
+protected:
+    SrtUrl _url;
+    toolkit::EventPoller::Ptr _poller;
+
+    bool _is_handleshake_finished = false;
+
+private:
+    toolkit::Socket::Ptr _socket;
+
+    TimePoint _now;
+    TimePoint _start_timestamp;
+    // for calculate rtt for delay
+    TimePoint _induction_ts;
+
+    //the initial value of RTT is 100 milliseconds
+    //RTTVar is 50 milliseconds
+    uint32_t _rtt          = 100 * 1000;
+    uint32_t _rtt_variance = 50 * 1000;
+
+    //local
+    uint32_t _socket_id            = 0;
+    uint32_t _init_seq_number       = 0;
+    uint32_t _mtu                  = 1500;
+    uint32_t _max_flow_window_size = 8192;
+    uint16_t _delay                = 120;
+
+    //peer
+    uint32_t _sync_cookie          = 0;
+    uint32_t _peer_socket_id;
+
+    // for handshake
+    SRT::Timer::Ptr _handleshake_timer;
+    SRT::HandshakePacket::Ptr _handleshake_req;
+
+    // for keeplive 
+    SRT::Ticker _send_ticker;
+    SRT::Timer::Ptr _keeplive_timer;
+
+    // for alive
+    SRT::Ticker _alive_ticker;
+    SRT::Timer::Ptr _alive_timer;
+
+    // for recv
+    SRT::PacketQueueInterface::Ptr _recv_buf;
+    uint32_t _last_pkt_seq = 0;
+
+    // Ack
+    SRT::UTicker _ack_ticker;
+    uint32_t _last_ack_pkt_seq    = 0;
+    uint32_t _light_ack_pkt_count = 0;
+    uint32_t _ack_number_count    = 0;
+    std::map<uint32_t, TimePoint> _ack_send_timestamp;
+    // Full Ack
+    // Link Capacity and Receiving Rate Estimation
+    std::shared_ptr<SRT::PacketRecvRateContext> _pkt_recv_rate_context;
+    std::shared_ptr<SRT::EstimatedLinkCapacityContext> _estimated_link_capacity_context;
+
+    // Nak
+    SRT::UTicker _nak_ticker;
+
+    //for Send
+    SRT::PacketSendQueue::Ptr _send_buf;
+    SRT::ResourcePool<SRT::BufferRaw> _packet_pool;
+    uint32_t _send_packet_seq_number = 0;
+    uint32_t _send_msg_number        = 1;
+
+    //AckAck
+    uint32_t _last_recv_ackack_seq_num = 0;
+
+    // for encryption
+    SRT::Crypto::Ptr _crypto;
+    SRT::Timer::Ptr _announce_timer;
+    SRT::KeyMaterialPacket::Ptr _announce_req;
+};
+
+} /* namespace mediakit */
+#endif /* ZLMEDIAKIT_SRTCALLER_H */
+

srt/SrtPlayer.cpp

@@ -0,0 +1,184 @@
+/*
+ * Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
+ *
+ * This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
+ *
+ * Use of this source code is governed by MIT-like 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 "SrtPlayer.h"
+#include "SrtPlayerImp.h"
+#include "Common/config.h"
+#include "Http/HlsPlayer.h"
+
+using namespace toolkit;
+using namespace std;
+
+namespace mediakit {
+
+
+SrtPlayer::SrtPlayer(const EventPoller::Ptr &poller) 
+    : SrtCaller(poller) {
+    DebugL;
+}
+
+SrtPlayer::~SrtPlayer(void) {
+    DebugL;
+}
+
+void SrtPlayer::play(const string &strUrl) {
+    DebugL;
+    try {
+        _url.parse(strUrl);
+    } catch (std::exception &ex) {
+        onResult(SockException(Err_other, StrPrinter << "illegal srt url:" << ex.what()));
+        return;
+    }
+
+    weak_ptr<SrtPlayer> weak_self = static_pointer_cast<SrtPlayer>(shared_from_this());
+    getPoller()->async([weak_self]() {
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            return;
+        }
+        strong_self->onConnect();
+    });
+    return;
+}
+
+void SrtPlayer::teardown() {
+    SrtCaller::onResult(SockException(Err_other, StrPrinter << "teardown: " << _url._full_url));
+}
+
+void SrtPlayer::pause(bool bPause) {
+    DebugL;
+}
+
+void SrtPlayer::speed(float speed) {
+    DebugL;
+}
+
+void SrtPlayer::onHandShakeFinished() {
+    SrtCaller::onHandShakeFinished();
+    onResult(SockException(Err_success, "srt play success"));
+}
+
+void SrtPlayer::onResult(const SockException &ex) {
+    SrtCaller::onResult(ex);
+
+     if (!ex) {
+        // 播放成功
+        onPlayResult(ex);
+        _benchmark_mode = (*this)[Client::kBenchmarkMode].as<int>();
+
+        // 播放成功，恢复数据包接收超时定时器
+        _recv_ticker.resetTime();
+        auto timeout = getTimeOutSec();
+        //读取配置文件
+        weak_ptr<SrtPlayer> weakSelf = static_pointer_cast<SrtPlayer>(shared_from_this());
+        // 创建rtp数据接收超时检测定时器
+        _check_timer = std::make_shared<Timer>(timeout /2,
+            [weakSelf, timeout]() {
+                auto strongSelf = weakSelf.lock();
+                if (!strongSelf) {
+                    return false;
+                }
+                if (strongSelf->_recv_ticker.elapsedTime() > timeout * 1000) {
+                    // 接收媒体数据包超时
+                    strongSelf->onResult(SockException(Err_timeout, "receive srt media data timeout:" + strongSelf->_url._full_url));
+                    return false;
+                }
+
+                return true;
+            }, getPoller());
+    } else {
+        WarnL << ex.getErrCode() << " " << ex.what();
+        if (ex.getErrCode() == Err_shutdown) {
+            // 主动shutdown的，不触发回调
+            return;
+        }
+        if (!_is_handleshake_finished) {
+            onPlayResult(ex);
+        } else {
+            onShutdown(ex);
+        }
+    }
+    return;
+}
+
+
+void SrtPlayer::onSRTData(SRT::DataPacket::Ptr pkt) {
+    _recv_ticker.resetTime();
+}
+
+uint16_t SrtPlayer::getLatency() {
+    auto latency = (*this)[Client::kLatency].as<uint16_t>();
+    return (uint16_t)latency ;
+}
+
+float SrtPlayer::getTimeOutSec() {
+    auto timeoutMS = (*this)[Client::kTimeoutMS].as<uint64_t>();
+    return (float)timeoutMS / (float)1000;
+}
+
+std::string SrtPlayer::getPassphrase() {
+    auto passPhrase = (*this)[Client::kPassPhrase].as<string>();
+    return passPhrase;
+}
+
+size_t SrtPlayer::getRecvSpeed() {
+    return SrtCaller::getRecvSpeed();
+}
+
+size_t SrtPlayer::getRecvTotalBytes() {
+    return SrtCaller::getRecvTotalBytes();
+}
+
+///////////////////////////////////////////////////
+// SrtPlayerImp
+
+void SrtPlayerImp::onPlayResult(const toolkit::SockException &ex) {
+    if (ex) {
+        Super::onPlayResult(ex);
+    }
+    //success result only occur when addTrackCompleted
+    return;
+}
+
+std::vector<Track::Ptr> SrtPlayerImp::getTracks(bool ready /*= true*/) const {
+    return _demuxer ? static_pointer_cast<HlsDemuxer>(_demuxer)->getTracks(ready) : Super::getTracks(ready);
+}
+
+void SrtPlayerImp::addTrackCompleted() {
+    Super::onPlayResult(toolkit::SockException(toolkit::Err_success, "play success"));
+}
+
+void SrtPlayerImp::onSRTData(SRT::DataPacket::Ptr pkt) {
+    SrtPlayer::onSRTData(pkt);
+
+    if (_benchmark_mode) {
+        return;
+    }
+
+    auto strong_self = shared_from_this();
+    if (!_demuxer) {
+        auto demuxer = std::make_shared<HlsDemuxer>();
+        demuxer->start(getPoller(), this);
+        _demuxer = std::move(demuxer);
+    }
+
+    if (!_decoder && _demuxer) {
+        _decoder = DecoderImp::createDecoder(DecoderImp::decoder_ts, _demuxer.get());
+    }
+
+    if (_decoder && _demuxer) {
+        _decoder->input(reinterpret_cast<const uint8_t *>(pkt->payloadData()), pkt->payloadSize());
+    }
+
+    return;
+}
+
+} /* namespace mediakit */
+

srt/SrtPlayer.h

@@ -0,0 +1,67 @@
+/*
+ * Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
+ *
+ * This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
+ *
+ * Use of this source code is governed by MIT-like 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_SRTPLAYER_H
+#define ZLMEDIAKIT_SRTPLAYER_H
+
+#include "Network/Socket.h"
+#include "Player/PlayerBase.h"
+#include "Poller/Timer.h"
+#include "Util/TimeTicker.h"
+#include "srt/SrtTransport.hpp"
+#include "Http/HttpRequester.h"
+#include <memory>
+#include <string>
+#include "SrtCaller.h"
+
+namespace mediakit {
+
+
+// 实现了srt代理拉流功能
+class SrtPlayer
+    : public PlayerBase , public SrtCaller {
+public:
+    using Ptr = std::shared_ptr<SrtPlayer>;
+
+    SrtPlayer(const toolkit::EventPoller::Ptr &poller);
+    ~SrtPlayer() override;
+
+    //// PlayerBase override////
+    void play(const std::string &strUrl) override;
+    void teardown() override;
+    void pause(bool pause) override;
+    void speed(float speed) override;
+    size_t getRecvSpeed() override;
+    size_t getRecvTotalBytes() override;
+
+protected:
+
+    //// SrtCaller override////
+    void onHandShakeFinished() override;
+    void onSRTData(SRT::DataPacket::Ptr pkt) override;
+    void onResult(const toolkit::SockException &ex) override;
+
+    bool isPlayer() override {return true;}
+
+    uint16_t getLatency() override;
+    float getTimeOutSec() override;
+    std::string getPassphrase() override;
+
+protected:
+    //是否为性能测试模式
+    bool _benchmark_mode = false;
+
+    //超时功能实现
+    toolkit::Ticker _recv_ticker;
+    std::shared_ptr<toolkit::Timer> _check_timer;
+};
+
+} /* namespace mediakit */
+#endif /* ZLMEDIAKIT_SRTPLAYER_H */

srt/SrtPlayerImp.h

@@ -0,0 +1,51 @@
+/*
+ * Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
+ *
+ * This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
+ *
+ * Use of this source code is governed by MIT-like 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_SRtPLAYERIMP_H
+#define ZLMEDIAKIT_SRtPLAYERIMP_H
+
+#include "SrtPlayer.h"
+
+namespace mediakit {
+
+class SrtPlayerImp
+    : public PlayerImp<SrtPlayer, PlayerBase>
+    , private TrackListener {
+public:
+    using Ptr = std::shared_ptr<SrtPlayerImp>;
+    using Super = PlayerImp<SrtPlayer, PlayerBase>;
+
+    SrtPlayerImp(const toolkit::EventPoller::Ptr &poller) : Super(poller) {}
+    ~SrtPlayerImp() override { DebugL; }
+
+private:
+    //// SrtPlayer override////
+    void onSRTData(SRT::DataPacket::Ptr pkt) override;
+
+    //// PlayerBase override////
+    void onPlayResult(const toolkit::SockException &ex) override;
+    std::vector<Track::Ptr> getTracks(bool ready = true) const override;
+
+private:
+    //// TrackListener override////
+    bool addTrack(const Track::Ptr &track) override { return true; }
+    void addTrackCompleted() override;
+
+private:
+    // for player
+    DecoderImp::Ptr _decoder;
+    MediaSinkInterface::Ptr _demuxer;
+
+    // for pusher
+    TSMediaSource::RingType::RingReader::Ptr _ts_reader;
+};
+
+} /* namespace mediakit */
+#endif /* ZLMEDIAKIT_SRtPLAYERIMP_H */

srt/SrtPusher.cpp

@@ -0,0 +1,132 @@
+/*
+ * Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
+ *
+ * This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
+ *
+ * Use of this source code is governed by MIT-like 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 "SrtPusher.h"
+#include "Common/config.h"
+
+using namespace toolkit;
+using namespace std;
+namespace mediakit {
+
+SrtPusher::SrtPusher(const EventPoller::Ptr &poller, const TSMediaSource::Ptr &src) : SrtCaller(poller) {
+    _push_src = src;
+    DebugL;
+}
+
+SrtPusher::~SrtPusher(void) {
+    DebugL;
+}
+
+void SrtPusher::publish(const string &strUrl) {
+    DebugL;
+    try {
+        _url.parse(strUrl);
+    } catch (std::exception &ex) {
+        onResult(SockException(Err_other, StrPrinter << "illegal srt url:" << ex.what()));
+        return;
+    }
+
+    weak_ptr<SrtPusher> weak_self = static_pointer_cast<SrtPusher>(shared_from_this());
+    getPoller()->async([weak_self]() {
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            return;
+        }
+        strong_self->onConnect();
+    });
+    return;
+}
+
+void SrtPusher::teardown() {
+    SrtCaller::onResult(SockException(Err_other, StrPrinter << "teardown: " << _url._full_url));
+}
+
+void SrtPusher::onHandShakeFinished() {
+    SrtCaller::onHandShakeFinished();
+    onResult(SockException(Err_success, "srt push success"));
+    doPublish();
+}
+
+void SrtPusher::onResult(const SockException &ex) {
+    SrtCaller::onResult(ex);
+
+    if (!ex) {
+        onPublishResult(ex);
+    } else {
+        WarnL << ex.getErrCode() << " " << ex.what();
+        if (ex.getErrCode() == Err_shutdown) {
+            // 主动shutdown的，不触发回调
+            return;
+        }
+        if (!_is_handleshake_finished) {
+            onPublishResult(ex);
+        } else {
+            onShutdown(ex);
+        }
+    }
+    return;
+}
+
+uint16_t SrtPusher::getLatency() {
+    auto latency = (*this)[Client::kLatency].as<uint16_t>();
+    return (uint16_t)latency ;
+}
+
+float SrtPusher::getTimeOutSec() {
+    auto timeoutMS = (*this)[Client::kTimeoutMS].as<uint64_t>();
+    return (float)timeoutMS / (float)1000;
+}
+
+std::string SrtPusher::getPassphrase() {
+    auto passPhrase = (*this)[Client::kPassPhrase].as<string>();
+    return passPhrase;
+}
+
+void SrtPusher::doPublish() {
+    auto src = _push_src.lock();
+    if (!src) {
+        onResult(SockException(Err_eof, "the media source was released"));
+        return;
+    }
+    // 异步查找直播流
+    std::weak_ptr<SrtPusher> weak_self = static_pointer_cast<SrtPusher>(shared_from_this());
+    _ts_reader = src->getRing()->attach(getPoller());
+    _ts_reader->setDetachCB([weak_self]() {
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            // 本对象已经销毁
+            return;
+        }
+        strong_self->onShutdown(SockException(Err_shutdown));
+    });
+    _ts_reader->setReadCB([weak_self](const TSMediaSource::RingDataType &ts_list) {
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            // 本对象已经销毁
+            return;
+        }
+        size_t i = 0;
+        auto size = ts_list->size();
+        ts_list->for_each([&](const TSPacket::Ptr &ts) { 
+            strong_self->onSendTSData(ts, ++i == size); 
+        });
+    });
+}
+
+size_t SrtPusher::getSendSpeed() {
+    return SrtCaller::getSendSpeed();
+}
+
+size_t SrtPusher::getSendTotalBytes() {
+    return SrtCaller::getSendTotalBytes();
+}
+
+} /* namespace mediakit */
+

srt/SrtPusher.h

@@ -0,0 +1,62 @@
+/*
+ * Copyright (c) 2016-present The ZLMediaKit project authors. All Rights Reserved.
+ *
+ * This file is part of ZLMediaKit(https://github.com/ZLMediaKit/ZLMediaKit).
+ *
+ * Use of this source code is governed by MIT-like 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_SRTPUSHER_H
+#define ZLMEDIAKIT_SRTPUSHER_H
+
+#include "Network/Socket.h"
+#include "Pusher/PusherBase.h"
+#include "Poller/Timer.h"
+#include "Util/TimeTicker.h"
+#include "srt/SrtTransport.hpp"
+#include "Http/HttpRequester.h"
+#include <memory>
+#include <string>
+#include "SrtCaller.h"
+
+namespace mediakit {
+
+// 实现了srt代理推流功能
+class SrtPusher
+    : public PusherBase , public SrtCaller {
+public:
+    using Ptr = std::shared_ptr<SrtPusher>;
+
+    SrtPusher(const toolkit::EventPoller::Ptr &poller,const TSMediaSource::Ptr &src);
+    ~SrtPusher() override;
+
+    //// PusherBase override////
+    void publish(const std::string &url) override;
+    void teardown() override;
+
+    void doPublish();
+protected:
+
+    //// SrtCaller override////
+    void onHandShakeFinished() override;
+    void onResult(const toolkit::SockException &ex) override;
+
+    bool isPlayer() override {return false;}
+    uint16_t getLatency() override;
+    float getTimeOutSec() override;
+    std::string getPassphrase() override;
+
+protected:
+    std::weak_ptr<TSMediaSource> _push_src;
+    TSMediaSource::RingType::RingReader::Ptr _ts_reader;
+
+    size_t getSendSpeed() override;
+    size_t getSendTotalBytes() override;
+};
+
+using SrtPusherImp = PusherImp<SrtPusher, PusherBase>;
+
+} /* namespace mediakit */
+#endif /* ZLMEDIAKIT_SRTPUSHER_H */