Merge branch 'master' of https://github.com/xia-chu/ZLMediaKit into dev

src/Pusher/PusherProxy.cpp

@@ -0,0 +1,99 @@
+/*
+ * Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
+ *
+ * This file is part of ZLMediaKit(https://github.com/xia-chu/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 "PusherProxy.h"
+
+using namespace toolkit;
+
+namespace mediakit {
+
+PusherProxy::PusherProxy(const MediaSource::Ptr &src, int retry_count, const EventPoller::Ptr &poller)
+            : MediaPusher(src, poller){
+    _retry_count = retry_count;
+    _on_close = [](const SockException &) {};
+    _weak_src = src;
+}
+
+PusherProxy::~PusherProxy() {
+    _timer.reset();
+}
+
+void PusherProxy::setPushCallbackOnce(const function<void(const SockException &ex)> &cb) {
+    _on_publish = cb;
+}
+
+void PusherProxy::setOnClose(const function<void(const SockException &ex)> &cb) {
+    _on_close = cb;
+}
+
+void PusherProxy::publish(const string &dst_url) {
+    std::weak_ptr<PusherProxy> weak_self = shared_from_this();
+    std::shared_ptr<int> failed_cnt(new int(0));
+
+    setOnPublished([weak_self, dst_url, failed_cnt](const SockException &err) {
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            return;
+        }
+
+        if (strong_self->_on_publish) {
+            strong_self->_on_publish(err);
+            strong_self->_on_publish = nullptr;
+        }
+
+        auto src = strong_self->_weak_src.lock();
+        if (!err) {
+            // 推流成功
+            *failed_cnt = 0;
+            InfoL << "Publish " << dst_url << " success";
+        } else if (src && (*failed_cnt < strong_self->_retry_count || strong_self->_retry_count < 0)) {
+            // 推流失败，延时重试推送
+            strong_self->rePublish(dst_url, (*failed_cnt)++);
+        } else {
+            //如果媒体源已经注销, 或达到了最大重试次数，回调关闭
+            strong_self->_on_close(err);
+        }
+    });
+
+    setOnShutdown([weak_self, dst_url, failed_cnt](const SockException &err) {
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            return;
+        }
+
+        auto src = strong_self->_weak_src.lock();
+        //推流异常中断，延时重试播放
+        if (src && (*failed_cnt < strong_self->_retry_count || strong_self->_retry_count < 0)) {
+            strong_self->rePublish(dst_url, (*failed_cnt)++);
+        } else {
+            //如果媒体源已经注销, 或达到了最大重试次数，回调关闭
+            strong_self->_on_close(err);
+        }
+    });
+
+    MediaPusher::publish(dst_url);
+}
+
+void PusherProxy::rePublish(const string &dst_url, int failed_cnt) {
+    auto delay = MAX(2 * 1000, MIN(failed_cnt * 3000, 60 * 1000));
+    weak_ptr<PusherProxy> weak_self = shared_from_this();
+    _timer = std::make_shared<Timer>(delay / 1000.0f, [weak_self, dst_url, failed_cnt]() {
+        //推流失败次数越多，则延时越长
+        auto strong_self = weak_self.lock();
+        if (!strong_self) {
+            return false;
+        }
+        WarnL << "推流重试[" << failed_cnt << "]:" << dst_url;
+        strong_self->MediaPusher::publish(dst_url);
+        return false;
+    }, getPoller());
+}
+
+} /* namespace mediakit */

src/Pusher/PusherProxy.h

@@ -0,0 +1,63 @@
+/*
+ * Copyright (c) 2016 The ZLMediaKit project authors. All Rights Reserved.
+ *
+ * This file is part of ZLMediaKit(https://github.com/xia-chu/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 SRC_DEVICE_PUSHERPROXY_H
+#define SRC_DEVICE_PUSHERPROXY_H
+
+#include "Pusher/MediaPusher.h"
+#include "Util/TimeTicker.h"
+
+using namespace std;
+using namespace toolkit;
+
+namespace mediakit {
+
+class PusherProxy : public MediaPusher, public std::enable_shared_from_this<PusherProxy> {
+public:
+    typedef std::shared_ptr<PusherProxy> Ptr;
+
+    // 如果retry_count<0,则一直重试播放；否则重试retry_count次数
+    // 默认一直重试，创建此对象时候，需要外部保证MediaSource存在
+    PusherProxy(const MediaSource::Ptr &src, int retry_count = -1, const EventPoller::Ptr &poller = nullptr);
+    ~PusherProxy() override;
+
+    /**
+     * 设置push结果回调，只触发一次；在publish执行之前有效
+     * @param cb 回调对象
+     */
+    void setPushCallbackOnce(const function<void(const SockException &ex)> &cb);
+
+    /**
+     * 设置主动关闭回调
+     * @param cb 回调对象
+     */
+    void setOnClose(const function<void(const SockException &ex)> &cb);
+
+    /**
+     * 开始拉流播放
+     * @param dstUrl 目标推流地址
+     */
+    void publish(const string& dstUrl) override;
+
+private:
+    // 重推逻辑函数
+    void rePublish(const string &dstUrl, int iFailedCnt);
+
+private:
+    int _retry_count;
+    Timer::Ptr _timer;
+    std::weak_ptr<MediaSource> _weak_src;
+    function<void(const SockException &ex)> _on_close;
+    function<void(const SockException &ex)> _on_publish;
+};
+
+} /* namespace mediakit */
+
+#endif //SRC_DEVICE_PUSHERPROXY_H

src/Rtmp/RtmpPusher.cpp

@@ -47,6 +47,7 @@ void RtmpPusher::teardown() {
 }
 
 void RtmpPusher::onPublishResult(const SockException &ex, bool handshake_done) {
+    DebugL << ex.what();
     if (ex.getErrCode() == Err_shutdown) {
         //主动shutdown的，不触发回调
         return;

src/Rtp/Decoder.cpp

@@ -103,19 +103,16 @@ static const char *getCodecName(int codec_id) {
 void DecoderImp::onStream(int stream, int codecid, const void *extra, size_t bytes, int finish){
     switch (codecid) {
         case PSI_STREAM_H264: {
-            InfoL << "got video track: H264";
             onTrack(std::make_shared<H264Track>());
             break;
         }
 
         case PSI_STREAM_H265: {
-            InfoL << "got video track: H265";
             onTrack(std::make_shared<H265Track>());
             break;
         }
 
         case PSI_STREAM_AAC: {
-            InfoL<< "got audio track: AAC";
             onTrack(std::make_shared<AACTrack>());
             break;
         }
@@ -123,14 +120,12 @@ void DecoderImp::onStream(int stream, int codecid, const void *extra, size_t byt
         case PSI_STREAM_AUDIO_G711A:
         case PSI_STREAM_AUDIO_G711U: {
             auto codec = codecid == PSI_STREAM_AUDIO_G711A ? CodecG711A : CodecG711U;
-            InfoL << "got audio track: G711";
             //G711传统只支持 8000/1/16的规格，FFmpeg貌似做了扩展，但是这里不管它了
             onTrack(std::make_shared<G711Track>(codec, 8000, 1, 16));
             break;
         }
 
         case PSI_STREAM_AUDIO_OPUS: {
-            InfoL << "got audio track: opus";
             onTrack(std::make_shared<OpusTrack>());
             break;
         }
@@ -223,8 +218,11 @@ void DecoderImp::onStream(int stream,int codecid,const void *extra,size_t bytes,
 #endif
 
 void DecoderImp::onTrack(const Track::Ptr &track) {
-    _tracks[track->getTrackType()] = track;
-    _sink->addTrack(track);
+    if (!_tracks[track->getTrackType()]) {
+        _tracks[track->getTrackType()] = track;
+        _sink->addTrack(track);
+        InfoL << "got track: " << track->getCodecName();
+    }
 }
 
 void DecoderImp::onFrame(const Frame::Ptr &frame) {

src/Rtsp/RtspPusher.cpp

@@ -95,6 +95,7 @@ void RtspPusher::publish(const string &url_str) {
 }
 
 void RtspPusher::onPublishResult(const SockException &ex, bool handshake_done) {
+    DebugL << ex.what();
     if (ex.getErrCode() == Err_shutdown) {
         //主动shutdown的，不触发回调
         return;