Merge branch 'feature_bafc' of https://github.com/xia-chu/ZLMediaKit

src/Common/MediaSource.cpp

@@ -157,6 +157,24 @@ bool MediaSource::seekTo(uint32_t stamp) {
     return listener->seekTo(*this, stamp);
 }
 
+bool MediaSource::pause()
+{
+    auto listener = _listener.lock();
+    if (!listener) {
+        return false;
+    }
+    return listener->pause(*this);
+}
+
+bool MediaSource::speed(float speed)
+{
+    auto listener = _listener.lock();
+    if (!listener) {
+        return false;
+    }
+    return listener->speed(*this, speed);
+}
+
 bool MediaSource::close(bool force) {
     auto listener = _listener.lock();
     if(!listener){
@@ -583,6 +601,24 @@ bool MediaSourceEventInterceptor::seekTo(MediaSource &sender, uint32_t stamp) {
     return listener->seekTo(sender, stamp);
 }
 
+bool MediaSourceEventInterceptor::pause(MediaSource& sender)
+{
+    auto listener = _listener.lock();
+    if (!listener) {
+        return false;
+    }
+    return listener->pause(sender);
+}
+
+bool MediaSourceEventInterceptor::speed(MediaSource& sender, float speed)
+{
+    auto listener = _listener.lock();
+    if (!listener) {
+        return false;
+    }
+    return listener->speed(sender, speed);
+}
+
 bool MediaSourceEventInterceptor::close(MediaSource &sender, bool force) {
     auto listener = _listener.lock();
     if (!listener) {

src/Common/MediaSource.h

@@ -67,6 +67,10 @@ public:
 
     // 通知拖动进度条
     virtual bool seekTo(MediaSource &sender, uint32_t stamp) { return false; }
+    // 通知暂停
+    virtual bool pause(MediaSource& sender) { return false; }
+    // 通知倍数
+    virtual bool speed(MediaSource& sender, float speed) { return false; }
     // 通知其停止产生流
     virtual bool close(MediaSource &sender, bool force) { return false; }
     // 获取观看总人数
@@ -106,6 +110,8 @@ public:
     std::shared_ptr<SockInfo> getOriginSock(MediaSource &sender) const override;
 
     bool seekTo(MediaSource &sender, uint32_t stamp) override;
+    bool pause(MediaSource& sender) override;
+    bool speed(MediaSource& sender, float speed) override;
     bool close(MediaSource &sender, bool force) override;
     int totalReaderCount(MediaSource &sender) override;
     void onReaderChanged(MediaSource &sender, int size) override;
@@ -249,6 +255,11 @@ public:
 
     // 拖动进度条
     bool seekTo(uint32_t stamp);
+    //暂停
+    bool pause();
+
+    //倍数播放
+    bool speed(float speed);
     // 关闭该流
     bool close(bool force);
     // 该流观看人数变化

src/Http/HttpDownloader.h

@@ -30,9 +30,12 @@ public:
     void setOnResult(const onDownloadResult &cb){
         _onResult = cb;
     }
+	
+protected:
+    void onResponseBody(const char *buf, size_t size, size_t recvedSize, size_t totalSize) override;
+	
 private:
     ssize_t onResponseHeader(const string &status, const HttpHeader &headers) override;
-    void onResponseBody(const char *buf, size_t size, size_t recvedSize, size_t totalSize) override;
     void onResponseCompleted() override;
     void onDisconnect(const SockException &ex) override;
     void closeFile();

src/Player/MediaPlayer.cpp

@@ -64,6 +64,13 @@ void MediaPlayer::pause(bool pause) {
     }
 }
 
+void MediaPlayer::speed(float speed)
+{
+    if (_delegate) {
+        _delegate->speed(speed);
+    }
+}
+
 void MediaPlayer::teardown() {
     if (_delegate) {
         _delegate->teardown();

src/Player/MediaPlayer.h

@@ -29,6 +29,7 @@ public:
     virtual ~MediaPlayer();
     void play(const string &url) override;
     void pause(bool pause) override;
+    void speed(float speed)override;
     void teardown() override;
     EventPoller::Ptr getPoller();
     void setOnCreateSocket(Socket::onCreateSocket cb);

src/Player/PlayerBase.h

@@ -65,6 +65,12 @@ public:
      */
     virtual void pause(bool bPause) {}
 
+    /**
+     * 倍数播放
+     * @param speed 1.0 2.0 0.5
+     */
+    virtual void speed(float speed) {}
+
     /**
      * 中断播放
      */
@@ -94,12 +100,24 @@ public:
      */
     virtual float getProgress() const { return 0;}
 
+    /**
+     * 获取播放进度pos，取值 相对开始时间增量 单位秒
+     * @return
+     */
+    virtual uint32_t getProgressPos() const { return 0; }
+
     /**
      * 拖动进度条
      * @param fProgress 进度，取值 0.0 ~ 1.0
      */
     virtual void seekTo(float fProgress) {}
 
+    /**
+     * 拖动进度条
+     * @param seekPos 进度，取值 相对于开始时间的增量 单位秒
+     */
+    virtual void seekTo(uint32_t seekPos) {}
+
     /**
      * 设置一个MediaSource，直接生产rtsp/rtmp代理
      * @param src
@@ -175,6 +193,12 @@ public:
         }
         return Parent::getProgress();
     }
+    uint32_t getProgressPos() const override {
+        if (_delegate) {
+            return _delegate->getProgressPos();
+        }
+        return Parent::getProgressPos();
+    }
     void seekTo(float fProgress) override{
         if (_delegate) {
             return _delegate->seekTo(fProgress);
@@ -182,6 +206,13 @@ public:
         return Parent::seekTo(fProgress);
     }
 
+    void seekTo(uint32_t seekPos) override {
+        if (_delegate) {
+            return _delegate->seekTo(seekPos);
+        }
+        return Parent::seekTo(seekPos);
+    }
+
     void setMediaSource(const MediaSource::Ptr & src) override {
         if (_delegate) {
             _delegate->setMediaSource(src);

src/Record/MP4Recorder.cpp

@@ -107,12 +107,17 @@ void MP4Recorder::closeFile() {
 }
 
 void MP4Recorder::inputFrame(const Frame::Ptr &frame) {
-    if(!_muxer || ((_createFileTicker.elapsedTime() > _max_second * 1000) &&
+	if (_baseSec == 0)
+		_baseSec = frame->dts();
+	auto dis = frame->dts() - _baseSec;
+    if(!_muxer || ((dis > _max_second * 1000) &&
                   (!_haveVideo || (_haveVideo && frame->keyFrame()))) ){
         //成立条件
         //1、_muxer为空
         //2、到了切片时间，并且只有音频
         //3、到了切片时间，有视频并且遇到视频的关键帧
+        DebugL << "create file dts:" << frame->dts() << " baseSec:" << _baseSec;
+        _baseSec = 0;
         createFile();
     }
 

src/Record/MP4Recorder.h

@@ -65,6 +65,8 @@ private:
     RecordInfo _info;
     MP4Muxer::Ptr _muxer;
     list<Track::Ptr> _tracks;
+
+	uint64_t _baseSec = 0;
 };
 
 #endif ///ENABLE_MP4

src/Rtmp/RtmpPlayer.cpp

@@ -171,6 +171,11 @@ void RtmpPlayer::pause(bool bPause) {
     send_pause(bPause);
 }
 
+void RtmpPlayer::speed(float speed)
+{
+    //todo
+}
+
 inline void RtmpPlayer::send_connect() {
     AMFValue obj(AMF_OBJECT);
     obj.set("app", _app);

src/Rtmp/RtmpPlayer.h

@@ -38,6 +38,7 @@ public:
 
     void play(const string &strUrl) override;
     void pause(bool bPause) override;
+    void speed(float speed)override;
     void teardown() override;
 
 protected:

src/Rtmp/RtmpPlayerImp.h

@@ -46,6 +46,11 @@ public:
         seekToMilliSecond((uint32_t)(fProgress * getDuration() * 1000));
     }
 
+    void seekTo(uint32_t seekPos) override {
+        uint32_t pos = MAX(float(0), MIN(seekPos, getDuration()))*1000;
+        seekToMilliSecond(pos);
+    }
+    
     void play(const string &strUrl) override {
         PlayerImp<RtmpPlayer, RtmpDemuxer>::play(strUrl);
     }

src/Rtmp/RtmpSession.cpp

@@ -412,6 +412,36 @@ void RtmpSession::onCmd_pause(AMFDecoder &dec) {
     //streamBegin
     sendUserControl(paused ? CONTROL_STREAM_EOF : CONTROL_STREAM_BEGIN, STREAM_MEDIA);
     _paused = paused;
+
+    auto stongSrc = _player_src.lock();
+    if (stongSrc) {
+        if (_paused)
+            stongSrc->pause();
+        else
+            stongSrc->seekTo(-1);
+    }
+}
+
+void RtmpSession::onCmd_playCtrl(AMFDecoder& dec)
+{
+    dec.load<AMFValue>();
+    auto ctrlObj = dec.load<AMFValue>();
+    int ctrlType = ctrlObj["ctrlType"].as_integer();
+    float speed = ctrlObj["speed"].as_number();
+
+
+    AMFValue status(AMF_OBJECT);
+    status.set("level", "status");
+    status.set("code", "NetStream.Speed.Notify");
+    status.set("description", "Speeding");
+    sendReply("onStatus", nullptr, status);
+    //streamBegin
+    sendUserControl(CONTROL_STREAM_EOF, STREAM_MEDIA);
+
+    auto stongSrc = _player_src.lock();
+    if (stongSrc) {
+        stongSrc->speed(speed);
+    }
 }
 
 void RtmpSession::setMetaData(AMFDecoder &dec) {
@@ -434,6 +464,7 @@ void RtmpSession::onProcessCmd(AMFDecoder &dec) {
         s_cmd_functions.emplace("play2", &RtmpSession::onCmd_play2);
         s_cmd_functions.emplace("seek", &RtmpSession::onCmd_seek);
         s_cmd_functions.emplace("pause", &RtmpSession::onCmd_pause);
+        s_cmd_functions.emplace("onPlayCtrl", &RtmpSession::onCmd_playCtrl);
     });
 
     std::string method = dec.load<std::string>();

src/Rtmp/RtmpSession.h

@@ -53,6 +53,7 @@ private:
 
     void onCmd_seek(AMFDecoder &dec);
     void onCmd_pause(AMFDecoder &dec);
+    void onCmd_playCtrl(AMFDecoder& dec);
     void setMetaData(AMFDecoder &dec);
 
     void onSendMedia(const RtmpPacket::Ptr &pkt);

src/Rtsp/RtspPlayer.cpp

@@ -25,7 +25,8 @@ namespace mediakit {
 enum PlayType {
     type_play = 0,
     type_pause,
-    type_seek
+    type_seek,
+    type_speed
 };
 
 RtspPlayer::RtspPlayer(const EventPoller::Ptr &poller) : TcpClient(poller){
@@ -414,8 +415,13 @@ void RtspPlayer::sendPause(int type , uint32_t seekMS){
     }
 }
 
-void RtspPlayer::pause(bool pause_flag) {
-    sendPause(pause_flag ? type_pause : type_seek, getProgressMilliSecond());
+void RtspPlayer::pause(bool bPause) {
+    sendPause(bPause ? type_pause : type_seek, getProgressMilliSecond());
+}
+
+void RtspPlayer::speed(float speed)
+{
+    sendRtspRequest("PLAY", _content_base, { "Scale",StrPrinter << speed });
 }
 
 void RtspPlayer::handleResPAUSE(const Parser& parser,int type) {

src/Rtsp/RtspPlayer.h

@@ -41,7 +41,8 @@ public:
     ~RtspPlayer() override;
 
     void play(const string &strUrl) override;
-    void pause(bool pause_flag) override;
+    void pause(bool bPause) override;
+    void speed(float speed)override;
     void teardown() override;
     float getPacketLossRate(TrackType type) const override;
 

src/Rtsp/RtspPlayerImp.h

@@ -42,11 +42,24 @@ public:
 
     }
 
-    void seekTo(float fProgress) override {
-        fProgress = MAX(float(0), MIN(fProgress, float(1.0)));
+    uint32_t getProgressPos() const override {
+        if (getDuration() > 0) {
+            return getProgressMilliSecond();
+        }
+        return PlayerBase::getProgressPos();
+
+    };
+
+    void seekTo(float fProgress) override{
+        fProgress = MAX(float(0),MIN(fProgress,float(1.0)));
         seekToMilliSecond((uint32_t)(fProgress * getDuration() * 1000));
     }
 
+    void seekTo(uint32_t seekPos) override {
+        uint32_t pos = MAX(float(0), MIN(seekPos, getDuration()))*1000;
+        seekToMilliSecond(pos);
+    }
+
 private:
     //派生类回调函数
     bool onCheckSDP(const string &sdp) override {

src/Rtsp/RtspSession.cpp

@@ -774,21 +774,36 @@ void RtspSession::handleReq_Play(const Parser &parser) {
     }
 
     bool useGOP = true;
-    _enable_send_rtp = false;
+    //_enable_send_rtp = false;
     float iStartTime = 0;
     auto strRange = parser["Range"];
-    if (strRange.size()) {
-        //这个是seek操作
-        auto strStart = FindField(strRange.data(), "npt=", "-");
-        if (strStart == "now") {
-            strStart = "0";
+    auto strScale = parser["Scale"];
+    if (strScale.size() == 0)
+    {
+
+        if (strRange.size()) {
+            //这个是seek操作
+            auto strStart = FindField(strRange.data(), "npt=", "-");
+            if (strStart == "now") {
+                strStart = "0";
+            }
+            iStartTime = 1000 * (float)atof(strStart.data());
+            InfoP(this) << "rtsp seekTo(ms):" << iStartTime;
+            useGOP = !play_src->seekTo((uint32_t)iStartTime);
         }
-        iStartTime = 1000 * (float)atof(strStart.data());
-        InfoP(this) << "rtsp seekTo(ms):" << iStartTime;
-        useGOP = !play_src->seekTo((uint32_t)iStartTime);
-    } else if (play_src->totalReaderCount() == 0) {
-        //第一个消费者
-        play_src->seekTo(0);
+        else if (play_src->totalReaderCount() == 0) {
+            //第一个消费者
+            play_src->seekTo(0);
+        }
+        else if (play_src->totalReaderCount() > 0)
+        {
+            _enable_send_rtp = true;
+            play_src->seekTo(-1);
+        }
+    }
+    else
+    {
+        play_src->speed(atof(strScale.data()));
     }
 
     _StrPrinter rtp_info;
@@ -845,7 +860,17 @@ void RtspSession::handleReq_Pause(const Parser &parser) {
     }
 
     sendRtspResponse("200 OK");
-    _enable_send_rtp = false;
+
+
+    auto play_src = _play_src.lock();
+    if (!play_src) {
+        send_StreamNotFound();
+        shutdown(SockException(Err_shutdown, "rtsp stream released"));
+        return;
+    }
+
+    play_src->pause();
+    //_enable_send_rtp = false;
 }
 
 void RtspSession::handleReq_Teardown(const Parser &parser) {