Merge pull request '添加图马斯特摇杆控制功能框架，按钮未关联吊舱控制键' (#1) from chen into master

Reviewed-on: #1

src/mainwindow.cpp

@@ -505,7 +505,10 @@ void MainWindow::pushStreamSlot(bool bPush) {
         } else {
             pushStreamIP = g_pushStreamInfoStruct.pushStreamIP;
         }
-        ui->videoWidget->pushStream(pushStreamIP);
+        QStringList pushStreamIPs;
+        pushStreamIPs.append(pushStreamIP);
+        // pushStreamIPs.append("rtmp://182.92.130.23/videoclient/fp98C_1_gcs");
+        ui->videoWidget->pushStream(pushStreamIPs);
     } else {
         ui->videoWidget->stopPushStream();
         // ui->videoWidget1->stopPushStream();

src/video/avpacketqueuemanager.cpp

@@ -95,3 +95,63 @@ bool AVPacketQueueManager::isEmptySeekQueue() {
     QMutexLocker locker(&m_seekMutex);
     return m_seekQueue.isEmpty();
 }
+
+void AVPacketQueueManager::waitForStreamDecoderClosed()
+{
+    if(!m_isDecodeEnd){
+        m_waitCloseStreamDecoderMutex.lock();
+        qDebug()<<"waitForStreamDecoderClosed0";
+        m_condition.wait(&m_waitCloseStreamDecoderMutex);
+        m_waitCloseStreamDecoderMutex.unlock();
+        qDebug()<<"waitForStreamDecoderClosed1";
+    }
+}
+
+void AVPacketQueueManager::waitForStreamSaverClosed()
+{
+    if(!m_isStreamSaverEnd){
+        m_waitCloseStreamSaverMutex.lock();
+        qDebug()<<"waitForStreamSaverClosed0";
+        m_condition.wait(&m_waitCloseStreamSaverMutex);
+        m_waitCloseStreamSaverMutex.unlock();
+        qDebug()<<"waitForStreamSaverClosed1";
+    }
+}
+
+void AVPacketQueueManager::waitForStreamPusherClosed()
+{
+    if(!m_isStreamPusherEnd){
+        m_waitCloseStreamPusherMutex.lock();
+        qDebug()<<"waitForStreamPusherClosed0";
+        m_condition.wait(&m_waitCloseStreamPusherMutex);
+        m_waitCloseStreamPusherMutex.unlock();
+        qDebug()<<"waitForStreamPusherClosed1";
+    }
+}
+
+void AVPacketQueueManager::wakeStreamDecoder()
+{
+    m_waitCloseStreamDecoderMutex.lock();
+    qDebug()<<"wakeStreamDecoder0";
+    m_condition.wakeAll();
+    m_waitCloseStreamDecoderMutex.unlock();
+    qDebug()<<"wakeStreamDecoder1";
+}
+
+void AVPacketQueueManager::wakeStreamSaver()
+{
+    m_waitCloseStreamSaverMutex.lock();
+    qDebug()<<"wakeStreamSaver0";
+    m_condition.wakeAll();
+    m_waitCloseStreamSaverMutex.unlock();
+    qDebug()<<"wakeStreamSaver1";
+}
+
+void AVPacketQueueManager::wakeStreamPusher()
+{
+    m_waitCloseStreamPusherMutex.lock();
+    m_condition.wakeAll();
+    qDebug()<<"wakeStreamPusher0";
+    m_waitCloseStreamPusherMutex.unlock();
+    qDebug()<<"wakeStreamPusher1";
+}

src/video/avpacketqueuemanager.h

@@ -3,6 +3,7 @@
 
 #include <QMutex>
 #include <QQueue>
+#include <QWaitCondition>
 
 #include "ffmpeginclude.h"
 
@@ -29,11 +30,22 @@ public:
     bool isEmptySaveQueue();
     bool isEmptySeekQueue();
 
+    void waitForStreamDecoderClosed();
+    void waitForStreamSaverClosed();
+    void waitForStreamPusherClosed();
+
+    void wakeStreamDecoder();
+    void wakeStreamSaver();
+    void wakeStreamPusher();
+
 public:
     std::atomic<bool> m_isSeeking{false};  // 是否正在跳转
     std::atomic<bool> m_isReadEnd{false};
     std::atomic<bool> m_isDecodeEnd{false};
-
+    std::atomic<bool> m_isStreamSaverEnd{false};
+    std::atomic<bool> m_isStreamPusherEnd{false};
+    std::atomic<bool> m_isPullReconnect{false};
+    std::atomic<bool> m_isPushReconnect{false};
 private:
     int QUEUECAPACITY = 100;
     QQueue<AVPacket*> m_decodeQueue;
@@ -44,6 +56,11 @@ private:
     QMutex m_saveMutex;           // 共享的互斥锁
     QMutex m_pushMutex;           // 共享的互斥锁
     QMutex m_seekMutex;           // 共享的互斥锁
+private:
+    QWaitCondition m_condition;
+    QMutex m_waitCloseStreamDecoderMutex;
+    QMutex m_waitCloseStreamSaverMutex;
+    QMutex m_waitCloseStreamPusherMutex;
 };
 
 #endif  // AVPACKETQUEUEMANAGER_H

src/video/decodestream.cpp

@@ -1,6 +1,9 @@
 #include "decodestream.h"
 
-DecodeStream::DecodeStream(QObject *parent) : QObject{parent}, m_playSpeed{1} {}
+DecodeStream::DecodeStream(QObject *parent) : QObject{parent}, m_playSpeed{1} {
+    connect(this,&DecodeStream::initStreamDecoderSignal,this,&DecodeStream::init,Qt::BlockingQueuedConnection);
+    connect(this,&DecodeStream::initUDPStreamDecoderSignal,this,&DecodeStream::initUDPDecodeStream,Qt::BlockingQueuedConnection);
+}
 
 bool DecodeStream::init(AVPacketQueueManager *queueManager,
                         AVFormatContext *formatContext, int videoIndex) {
@@ -22,13 +25,14 @@ bool DecodeStream::init(AVPacketQueueManager *queueManager,
     if (m_fps <= 0) {
         m_fps = 30.0;  // 默认帧率
     }
-
-    return initDecoder(formatContext, videoIndex);
+    initStatus = initDecoder(formatContext, videoIndex);
+    return initStatus;
 }
 
-bool DecodeStream::init(AVPacketQueueManager *queueManager) {
+bool DecodeStream::initUDPDecodeStream(AVPacketQueueManager *queueManager) {
     m_queueManager = queueManager;
-    return initUDPDecoder();
+    initStatus = initUDPDecoder();
+    return initStatus;
 }
 
 // 视频解码线程任务
@@ -142,9 +146,12 @@ void DecodeStream::startDecode() {
     }
 
     free();
+    m_queueManager->m_isDecodeEnd = true;
+    m_queueManager->wakeStreamDecoder();
     qDebug() << "Decoding Thread End!";
     emit decodeEndSignal();
     emit sendErrorMessageSignal("视频解码结束!", 1);
+
 }
 
 void DecodeStream::changePlaySpeedSlot(double speed) {

src/video/decodestream.h

@@ -18,7 +18,7 @@ public:
     DecodeStream(QObject *parent = nullptr);
     bool init(AVPacketQueueManager *queueManager,
               AVFormatContext *formatContext, int videoIndex);
-    bool init(AVPacketQueueManager *queueManager);
+    bool initUDPDecodeStream(AVPacketQueueManager *queueManager);
     void close();
     AVCodecContext *getCodecContext();
     AVCodecParserContext *getParserContext();
@@ -31,7 +31,11 @@ signals:
     void sendErrorMessageSignal(QString message, int type);
     void updateVideoCurrentTime(int currentTime, int duration);
     void decodeEndSignal();
-
+    void initUDPStreamDecoderSignal(AVPacketQueueManager *queueManager);
+    void initStreamDecoderSignal(AVPacketQueueManager *queueManager,
+                                 AVFormatContext *formatContext, int videoIndex);
+public:
+    bool initStatus = false;
 private:
     bool initObject();  // 初始化对象
     bool initDecoder(AVFormatContext *inputFormatContext,
@@ -44,7 +48,7 @@ private:
     bool isValidAVPacket(AVPacket *pkt);
 
 private:
-    bool m_start = true;
+    std::atomic<bool>m_start = true;
     int m_videoIndex = 0;
     AVFormatContext *m_formatContext = nullptr;
     AVCodecContext *m_codecContext = nullptr;      // 解码器上下文

src/video/ffmpeginclude.h

@@ -22,7 +22,6 @@ extern "C" {
 #include <QElapsedTimer>
 #include <QEventLoop>
 
-// #include "global.h"
 
 #define PRINT_LOG 1
 #define ERROR_LEN 1024  // 异常信息数组长度

src/video/pushstream.cpp

@@ -1,17 +1,32 @@
 #include "pushstream.h"
 
-PushStream::PushStream(QObject *parent) : QObject{parent} {}
+PushStream::PushStream(QObject *parent) : QObject{parent} {
+    connect(this,&PushStream::initStreamPusherSignal,this,&PushStream::init,Qt::BlockingQueuedConnection);
+}
 
-void PushStream::setRemoteIP(QString url) {
-    m_pushStreamIP = url;
+void PushStream::setRemoteIP(QStringList& urls) {
+    m_pushStreamIPs = urls;
 }
 
 bool PushStream::init(AVFormatContext *inputFormatCtx,
                       AVPacketQueueManager *queueManager) {
+    qDebug() << "PushStream::init ThreadID:" << QThread::currentThreadId();
     // m_pusherQueue = queue;
     m_queueManager = queueManager;
     m_inputFormatCtx = inputFormatCtx;
-    m_start = openNetworkStream(inputFormatCtx);
+    // m_start = openNetworkStream(inputFormatCtx);
+    bool ss = false;
+    for (int i = 0; i < m_pushStreamIPs.size(); ++i) {
+        AVFormatContext* outputformatCtx = nullptr;
+        ss = openNetworkStream(inputFormatCtx,outputformatCtx,m_pushStreamIPs.at(i));
+        m_outputFormatCtxArray.append(outputformatCtx);
+        m_initStatus.append(ss);
+        if(ss){
+            m_start = ss;
+        }
+    }
+
+    initStatus = m_start;
     return m_start;
 }
 
@@ -22,36 +37,37 @@ void PushStream::close() {
     // qDebug() << "*******m_end0:" << m_end;
 }
 
-bool PushStream::openNetworkStream(AVFormatContext *inputFormatCtx) {
-    if (m_pushStreamIP.isEmpty()) return false;
+bool PushStream::openNetworkStream(AVFormatContext *&inputFormatCtx,AVFormatContext *&outputFormatCtx,QString pushStreamIP) {
+    if (pushStreamIP.isEmpty()) return false;
 
     if (!inputFormatCtx) return false;
 
     // 初始化网络输出流
     // QString m_pushStreamIP = "rtsp://182.92.130.23/app/stream999";
     QString format_name = "flv";
-    if (m_pushStreamIP.left(3) == "udp" || m_pushStreamIP.left(3) == "UDP")
+    if (pushStreamIP.left(3) == "udp" || pushStreamIP.left(3) == "UDP")
         format_name = "mpegts";
     int ret = avformat_alloc_output_context2(
-        &m_outputFormatCtx, NULL, format_name.toStdString().data(),
-        m_pushStreamIP.toUtf8().constData());
+        &outputFormatCtx, NULL, format_name.toStdString().data(),
+        pushStreamIP.toUtf8().constData());
     if (ret < 0) {
         showError(ret);
-        free();
+        free(outputFormatCtx);
         qDebug() << "Could not create output context.";
         return false;
     }
 
+    AVStream *m_ostream = nullptr;
     // 复制流信息
     for (unsigned int i = 0; i < inputFormatCtx->nb_streams; ++i) {
         // AVStream *stream = inputFormatCtx->streams[i];
         if (inputFormatCtx->streams[i]->codecpar->codec_type ==
             AVMEDIA_TYPE_VIDEO) {
             m_istream = inputFormatCtx->streams[i];
-            m_ostream = avformat_new_stream(m_outputFormatCtx, nullptr);
+            m_ostream = avformat_new_stream(outputFormatCtx, nullptr);
             if (!m_ostream) {
                 qDebug() << "Failed allocating output stream.\n";
-                free();
+                free(outputFormatCtx);
                 return false;
             }
             // 复制编解码器参数
@@ -59,7 +75,7 @@ bool PushStream::openNetworkStream(AVFormatContext *inputFormatCtx) {
                                           m_istream->codecpar);
             if (ret < 0) {
                 showError(ret);
-                free();
+                free(outputFormatCtx);
                 qWarning() << "avcodec_parameters_from_context Failed";
                 return false;
             }
@@ -71,25 +87,25 @@ bool PushStream::openNetworkStream(AVFormatContext *inputFormatCtx) {
 
     m_inputTimeBase = m_istream->time_base;
     m_inputFrameRate = m_istream->r_frame_rate;
-    m_outputTimeBase = m_istream->time_base;
+
 
     // 打开输出文件
-    if (!(m_outputFormatCtx->flags & AVFMT_NOFILE)) {
-        if (ret = avio_open(&m_outputFormatCtx->pb,
-                            m_pushStreamIP.toUtf8().constData(),
+    if (!(outputFormatCtx->flags & AVFMT_NOFILE)) {
+        if (ret = avio_open(&outputFormatCtx->pb,
+                            pushStreamIP.toUtf8().constData(),
                             AVIO_FLAG_WRITE) < 0) {
             showError(ret);
-            free();
+            free(outputFormatCtx);
             qDebug() << "Could not open output file.\n";
             return false;
         }
     }
 
     // 写入头文件
-    ret = avformat_write_header(m_outputFormatCtx, NULL);
+    ret = avformat_write_header(outputFormatCtx, NULL);
     if (ret < 0) {
         showError(ret);
-        free();
+        free(outputFormatCtx);
         qDebug() << "Error occurred when write_header into output file.\n";
         return false;
     }
@@ -97,64 +113,89 @@ bool PushStream::openNetworkStream(AVFormatContext *inputFormatCtx) {
     // m_InitStatus = true;
     // startTime = av_gettime_relative();
 
+    m_outputTimeBase = m_ostream->time_base;
+
     m_bwriteHeader = true;
     m_firstPts = AV_NOPTS_VALUE;
     return true;
 }
 
-int PushStream::reconnect(int ret) {
+int PushStream::reconnect(int ret,int id) {
+    if(!m_queueManager->m_isPullReconnect) return 0;
+
     if (ret == -10053 || ret == -10054) {
+        m_queueManager->m_isPushReconnect = true;
         m_end = false;
         // qDebug() << "m_end:" << m_end;
+        AVFormatContext* outputFormatCtx = m_outputFormatCtxArray.at(id);
+        QString message;
         for (int nRetryCount = 0; nRetryCount < MAXCONNECT; ++nRetryCount) {
+            if(m_queueManager->m_isPullReconnect) {
+               break;
+            }
             // 关闭输出
-            if (m_outputFormatCtx &&
-                !(m_outputFormatCtx->flags & AVFMT_NOFILE)) {
-                avio_close(m_outputFormatCtx->pb);
+            if (outputFormatCtx &&
+                !(outputFormatCtx->flags & AVFMT_NOFILE)) {
+                avio_close(outputFormatCtx->pb);
             }
             ret =
-                avio_open(&m_outputFormatCtx->pb,
-                          m_pushStreamIP.toUtf8().constData(), AVIO_FLAG_WRITE);
+                avio_open(&outputFormatCtx->pb,
+                          m_pushStreamIPs[id].toUtf8().constData(), AVIO_FLAG_WRITE);
             if (ret < 0) {
                 showError(ret);
                 qDebug() << "Failed to reconnect"
                          << QString::number(nRetryCount + 1);
-                QString str =
-                    QString("网络中断，尝试重连第%1次!").arg(nRetryCount + 1);
-                emit sendErrorMessageSignal(str, 3);
+                message =
+                    QString("推流地址%1网络中断，尝试重连第%2次!").arg(id+1,nRetryCount + 1);
+                emit sendErrorMessageSignal(message, 3);
                 if (m_end) break;
                 // av_usleep(5 * 1000000);
                 continue;
             }
             // Try to reconnect
-            ret = avformat_write_header(m_outputFormatCtx, nullptr);
+            ret = avformat_write_header(outputFormatCtx, nullptr);
             if (ret < 0) {
+                m_initStatus[id] = false;
                 showError(ret);
                 // free();
                 qDebug() << "Failed to reconnect"
                          << QString::number(nRetryCount + 1);
-                QString str =
-                    QString("网络中断，尝试重连第%1次!").arg(nRetryCount + 1);
-                emit sendErrorMessageSignal(str, 3);
+                message =
+                    QString("推流地址%1网络中断，尝试重连第%2次!").arg(id+1,nRetryCount + 1);
+                emit sendErrorMessageSignal(message, 3);
                 if (m_end) break;
                 // nRetryCount++;
                 // av_usleep(5 * 1000000);
             } else {
+                m_initStatus[id] = true;
                 m_start = true;
                 m_firstPts = AV_NOPTS_VALUE;
                 m_frm_cnt = 0;
                 m_bwriteHeader = true;
-                emit sendErrorMessageSignal("重连成功!", 1);
+                message =
+                    QString("推流地址%1重连成功!").arg(id+1);
+                emit sendErrorMessageSignal(message, 1);
                 m_queueManager->clearPushQueue();
-                qDebug() << "重连成功!";
+                qDebug() << message;
+
+                m_queueManager->m_isPushReconnect = false;
                 return ret;
             }
 
             if (m_end) break;
         }
         m_start = false;
+        for (int i = 0; i < m_initStatus.size(); ++i) {
+            if(m_initStatus.at(i)){
+                m_start = true;
+                break;
+            }
+        }
         m_bwriteHeader = false;
-        emit sendErrorMessageSignal("重连失败，推流停止!", 2);
+        message =
+            QString("推流地址%1重连失败，推流停止!").arg(id+1);
+        emit sendErrorMessageSignal(message, 2);
+        m_queueManager->m_isPushReconnect = false;
         return -1;
     }
     return 0;
@@ -163,6 +204,7 @@ int PushStream::reconnect(int ret) {
 void PushStream::pushStream(int64_t startTime) {
     qDebug() << "PushStreamThreadID:" << QThread::currentThreadId();
     // m_startTime = startTime;
+    m_queueManager->m_isStreamPusherEnd = false;
     while (m_start) {
         AVPacket *inputPacket = m_queueManager->dequeuePushPacket();
         if (inputPacket) {
@@ -213,7 +255,7 @@ void PushStream::pushStream(int64_t startTime) {
                 // sleepMsec(40);
             } else {
                 if (delay < -100000) {
-                    qDebug() << "delay:" << delay;
+                    qDebug() << "push delay:" << delay;
                     // 滞后100ms以上，丢弃非重要帧
                     if (!(inputPacket->flags & AV_PKT_FLAG_KEY)) {
                         av_packet_unref(inputPacket);
@@ -238,42 +280,79 @@ void PushStream::pushStream(int64_t startTime) {
 
             // 向推流服务器推送流数据
             m_frm_cnt++;
-            int ret =
-                av_interleaved_write_frame(m_outputFormatCtx, inputPacket);
-            if (ret < 0) {
-                av_packet_unref(inputPacket);
-                showError(ret);
-                // if (ret == -10053) {
-                //     qDebug() << "网络不稳定";
-                // }
-                if (reconnect(ret) < 0) {
-                    break;
-                };
-                continue;
+            int ret;
+
+            for (int i = 0; i < m_outputFormatCtxArray.size(); ++i) {
+                if(!m_initStatus.at(i)) continue;
+                AVFormatContext* outputFormatCtx = m_outputFormatCtxArray.at(i);
+                if(outputFormatCtx!=nullptr){
+                    AVPacket* clonePacket = nullptr;
+                    if(i!= m_outputFormatCtxArray.size()-1){
+                        clonePacket = av_packet_clone(inputPacket);
+                    }else{
+                        clonePacket = inputPacket;
+                    }
+                    ret =
+                        av_interleaved_write_frame(outputFormatCtx, clonePacket);
+                    if (ret < 0) {
+                        av_packet_unref(clonePacket);
+                        showError(ret);
+                        // if (ret == -10053) {
+                        //     qDebug() << "网络不稳定";
+                        // }
+                        if (reconnect(ret,i) < 0) {
+                            // break;
+                        };
+                        continue;
+                    }
+                    // 数据包写入成功，现在可以释放pkt
+                    av_packet_unref(clonePacket);
+                    av_packet_free(&clonePacket);
+                }
             }
-            // 数据包写入成功，现在可以释放pkt
-            av_packet_unref(inputPacket);
-            av_packet_free(&inputPacket);
         } else {
             // QThread::usleep(1000);
             av_usleep(1000);
         }
     }
 
-    if (m_bwriteHeader) av_write_trailer(m_outputFormatCtx);
+    if (m_bwriteHeader) {
+        for (auto& outputFormatCtx : m_outputFormatCtxArray) {
+            if(outputFormatCtx!=nullptr){
+                av_write_trailer(outputFormatCtx);
+            }
+        }
+    }
     free();
+    m_queueManager->m_isStreamPusherEnd = true;
+    m_queueManager->wakeStreamPusher();
     qDebug() << "Push Stream End!";
     emit sendErrorMessageSignal("推流结束!", 1);
+
 }
 
 void PushStream::free() {
-    m_start = false;
+    for (auto &outputFormatCtx : m_outputFormatCtxArray) {
+        // 关闭输出
+        if (outputFormatCtx && !(outputFormatCtx->flags & AVFMT_NOFILE)) {
+            avio_close(outputFormatCtx->pb);
+        }
+        if (outputFormatCtx) {
+            avformat_free_context(outputFormatCtx);
+            outputFormatCtx = nullptr;
+        }
+    }
+    m_outputFormatCtxArray.clear();
+}
+
+void PushStream::free(AVFormatContext *&outputFormatCtx)
+{
     // 关闭输出
-    if (m_outputFormatCtx && !(m_outputFormatCtx->flags & AVFMT_NOFILE)) {
-        avio_close(m_outputFormatCtx->pb);
+    if (outputFormatCtx && !(outputFormatCtx->flags & AVFMT_NOFILE)) {
+        avio_close(outputFormatCtx->pb);
     }
-    if (m_outputFormatCtx) {
-        avformat_free_context(m_outputFormatCtx);
-        m_outputFormatCtx = nullptr;
+    if (outputFormatCtx) {
+        avformat_free_context(outputFormatCtx);
+        outputFormatCtx = nullptr;
     }
 }

src/video/pushstream.h

@@ -17,30 +17,36 @@ public:
      * @brief 设置推流地址
      * @param url 远端推流地址
      */
-    void setRemoteIP(QString url);
-    bool init(AVFormatContext *inputFormatCtx,
-              AVPacketQueueManager *queueManager);
+    void setRemoteIP(QStringList& urls);
+
     void close();
 public slots:
     void pushStream(int64_t startTime);
+    bool init(AVFormatContext *inputFormatCtx,
+              AVPacketQueueManager *queueManager);
 signals:
     void startPushStreamSignal(int64_t startTime);
     void sendErrorMessageSignal(QString message, int type);
-
+    void initStreamPusherSignal(AVFormatContext *inputFormatCtx,AVPacketQueueManager *queueManager);
 private:
-    bool openNetworkStream(AVFormatContext *inputFormatCtx);
-    int reconnect(int ret);
+    bool openNetworkStream(AVFormatContext *&inputFormatCtx,AVFormatContext *&outputFormatCtx,QString pushStreamIP);
+    int reconnect(int ret,int id=0);
     void free();
+    void free(AVFormatContext *&outputFormatCtx);
+public:
+    bool initStatus = false;
 
 private:
     // QQueue<AVPacket *> *m_pusherQueue = nullptr;
     AVFormatContext *m_inputFormatCtx = nullptr;  //
-    AVFormatContext *m_outputFormatCtx = NULL;    //
+    AVFormatContext *m_outputFormatCtx = nullptr;    //
+    AVFormatContext *m_outputFormatCtx2 = nullptr;    //推流2
+    QVector<AVFormatContext *>m_outputFormatCtxArray;
     AVStream *m_istream = nullptr;
-    AVStream *m_ostream = nullptr;
+    // AVStream *m_ostream = nullptr;
     bool m_bwriteHeader = false;
     int m_videoIndex = -1;
-    QString m_pushStreamIP;  // 推流地址
+    QStringList m_pushStreamIPs;  // 推流地址
     std::atomic<bool> m_start = false;
     std::atomic<bool> m_end = false;
     int64_t m_startTime;
@@ -54,6 +60,8 @@ private:
     AVRational m_inputTimeBase;
     AVRational m_inputFrameRate;
     AVRational m_outputTimeBase;
+
+    QVector<bool> m_initStatus;
 };
 
 #endif  // PUSHSTREAM_H

src/video/readstream.cpp

@@ -105,8 +105,10 @@ bool ReadStream::setStreamDecoder(DecodeStream *decodeStreamer) {
         // QMutexLocker locker(&m_mutex);
         m_streamDecoder = decodeStreamer;
         m_queueManager.clearDecodeQueue();
-        m_decodeStreamFlag = m_streamDecoder->init(
-            &m_queueManager, m_formatContext, m_videoIndex);
+        // m_decodeStreamFlag = m_streamDecoder->init(
+        //     &m_queueManager, m_formatContext, m_videoIndex);
+        m_streamDecoder->initStreamDecoderSignal(&m_queueManager, m_formatContext, m_videoIndex);
+        m_decodeStreamFlag = m_streamDecoder->initStatus;
         if (m_decodeStreamFlag) emit decodeStreamer->startDecodeSignal();
         return m_decodeStreamFlag;
     } else {
@@ -120,9 +122,11 @@ bool ReadStream::setUDPStreamDecoder(DecodeStream *decodeStreamer) {
         // QMutexLocker locker(&m_mutex);
         m_streamDecoder = decodeStreamer;
         m_queueManager.clearDecodeQueue();
-        m_decodeStreamFlag = m_streamDecoder->init(&m_queueManager);
+        // m_decodeStreamFlag = m_streamDecoder->init(&m_queueManager);
         // codec_ctx = m_streamDecoder->getCodecContext();
         // parser = m_streamDecoder->getParserContext();
+        m_streamDecoder->initUDPStreamDecoderSignal(&m_queueManager);
+        m_decodeStreamFlag = m_streamDecoder->initStatus;
         if (m_decodeStreamFlag) emit decodeStreamer->startDecodeSignal();
         return m_decodeStreamFlag;
     } else {
@@ -138,8 +142,10 @@ bool ReadStream::setStreamSaver(SaveStream *streamSaver, bool isUDP) {
         if (isUDP) {
             m_saveStreamFlag = m_streamSaver->initUDP(&m_queueManager);
         } else {
-            m_saveStreamFlag = m_streamSaver->init(
-                m_formatContext, &m_queueManager, m_videoIndex);
+            // m_saveStreamFlag = m_streamSaver->init(
+            //     m_formatContext, &m_queueManager, m_videoIndex);
+            emit m_streamSaver->initStreamSaverSignal(m_formatContext, &m_queueManager, m_videoIndex);
+            m_saveStreamFlag = m_streamSaver->initStatus;
         }
         return m_saveStreamFlag;
     } else {
@@ -149,10 +155,13 @@ bool ReadStream::setStreamSaver(SaveStream *streamSaver, bool isUDP) {
 }
 
 bool ReadStream::setStreamPusher(PushStream *streamPusher) {
+    qDebug() << "setStreamPusher ThreadID:" << QThread::currentThreadId();
     if (streamPusher) {
         m_streamPusher = streamPusher;
         m_queueManager.clearPushQueue();
-        m_pushStreamFlag = streamPusher->init(m_formatContext, &m_queueManager);
+        emit streamPusher->initStreamPusherSignal(m_formatContext, &m_queueManager);
+        // m_pushStreamFlag = streamPusher->init(m_formatContext, &m_queueManager);
+        m_pushStreamFlag = m_streamPusher->initStatus;
         if (m_pushStreamFlag) streamPusher->startPushStreamSignal(0);
         return m_pushStreamFlag;
     } else {
@@ -472,21 +481,25 @@ void ReadStream::free() {
 }
 
 bool ReadStream::reconnect() {
+    m_queueManager.m_isPullReconnect = true;
     m_end = false;
     if (m_streamDecoder) {
         m_streamDecoder->close();
+        m_queueManager.waitForStreamDecoderClosed();
     }
     if (m_streamSaver) {
         m_streamSaver->close();
+        m_queueManager.waitForStreamSaverClosed();
     }
     if (m_streamPusher) {
         m_streamPusher->close();
+        m_queueManager.waitForStreamPusherClosed();
     }
+    qDebug()<<"all closed!";
     free();
     for (int i = 0; i < MAXRECONNECT; ++i) {
         m_start = openFile(m_pullURL);
         if (m_start) {
-            emit sendErrorMessageSignal("重连成功!", 1);
             if (m_streamDecoder) {
                 setStreamDecoder(m_streamDecoder);
             }
@@ -496,6 +509,8 @@ bool ReadStream::reconnect() {
             if (m_streamPusher) {
                 setStreamPusher(m_streamPusher);
             }
+            m_queueManager.m_isPullReconnect = false;
+            emit sendErrorMessageSignal("重连成功!", 1);
             return true;
         } else {
             qDebug() << "reconnect failed:" << QString::number(i + 1);
@@ -507,6 +522,7 @@ bool ReadStream::reconnect() {
         if (m_end) break;
     }
     emit sendErrorMessageSignal("重连失败!", 2);
+    m_queueManager.m_isPullReconnect = false;
     return false;
 }
 

src/video/savestream.cpp

@@ -1,6 +1,8 @@
 #include "savestream.h"
 
-SaveStream::SaveStream(QObject *parent) : QObject{parent} {}
+SaveStream::SaveStream(QObject *parent) : QObject{parent} {
+    connect(this,&SaveStream::initStreamSaverSignal,this,&SaveStream::init,Qt::BlockingQueuedConnection);
+}
 
 bool SaveStream::init(AVFormatContext *formatContext,
                       AVPacketQueueManager *queueManager, int videoIndex) {
@@ -13,6 +15,7 @@ bool SaveStream::init(AVFormatContext *formatContext,
     if (!m_start) {
         free();
     }
+    initStatus = m_start;
     return m_start;
 }
 
@@ -37,8 +40,10 @@ void SaveStream::close() {
 void SaveStream::startSaveStream() {
     qDebug() << "SaveStreamThreadID:" << QThread::currentThreadId();
     if (!m_start) {
+        m_queueManager->m_isStreamSaverEnd = true;
         return;
     }
+    m_queueManager->m_isStreamSaverEnd = false;
     int frameIndex = 0;
     int64_t dts = 0;
     int64_t dts_last = 0;
@@ -101,16 +106,6 @@ void SaveStream::startSaveStream() {
         }
     }
 
-    // AVStream *out_stream = m_formatContextSave->streams[0];
-    // out_stream->start_time = 0;
-    // m_formatContextSave->start_time = 0;
-    // int64_t duration_pts = last_pts - first_pts;
-    // // duration_pts = av_rescale(duration_pts, 1, AV_TIME_BASE);
-    // m_formatContextSave->duration = duration_pts;
-
-    // AVStream *stream = m_formatContextSave->streams[0];
-    // stream->duration = duration_pts;
-
     // 写入文件尾
     if (m_formatContextSave && m_writeHeader) {
         av_write_trailer(m_formatContextSave);
@@ -123,6 +118,8 @@ void SaveStream::startSaveStream() {
     }
 
     free();
+    m_queueManager->m_isStreamSaverEnd = true;
+    m_queueManager->wakeStreamSaver();
     qDebug() << "Save Video End！";
     emit sendErrorMessageSignal("视频保存结束!", 1);
 }
@@ -130,8 +127,11 @@ void SaveStream::startSaveStream() {
 void SaveStream::startSaveUDPStream() {
     qDebug() << "SaveStreamThreadID:" << QThread::currentThreadId();
     if (!m_start) {
+        m_queueManager->m_isStreamSaverEnd = true;
+        qDebug()<<"StreamSaver direct return ";
         return;
     }
+    m_queueManager->m_isStreamSaverEnd = false;
     int frameIndex = 0;
     int64_t pts = 0;
     int64_t dts = 0;
@@ -184,8 +184,11 @@ void SaveStream::startSaveUDPStream() {
     }
 
     free();
+    m_queueManager->m_isStreamSaverEnd = true;
+    m_queueManager->wakeStreamSaver();
     qDebug() << "视频保存结束！";
     emit sendErrorMessageSignal("视频保存结束!", 1);
+
 }
 
 bool SaveStream::openFile(bool isUDP) {

src/video/savestream.h

@@ -30,12 +30,16 @@ public slots:
 signals:
     void startSaveStreamSignal();
     void sendErrorMessageSignal(QString message, int type);
-
+    void initStreamSaverSignal(AVFormatContext *formatContext,
+                         AVPacketQueueManager *queueManager, int videoIndex);
 private:
     bool openFile(bool isUDP = false);
     void free();
     bool isValidAVPacket(AVPacket *pkt);
 
+public:
+    bool initStatus = false;
+
 private:
     AVFormatContext *m_formatContextSave = nullptr;  // 封装上下文
     AVFormatContext *m_inputFormatContext = nullptr;

src/video/videowidget.cpp

@@ -171,8 +171,8 @@ void VideoWidget::stopPlay() {
 }
 
 // 推流
-bool VideoWidget::pushStream(const QString &url) {
-    if (url.isEmpty()) {
+bool VideoWidget::pushStream(QStringList &urls) {
+    if (urls.isEmpty()) {
         return false;
     } else {
         // 先拉流
@@ -192,7 +192,7 @@ bool VideoWidget::pushStream(const QString &url) {
             connect(streamPusher, &PushStream::sendErrorMessageSignal, this,
                     &VideoWidget::receiveErrorMessage, Qt::UniqueConnection);
         }
-        streamPusher->setRemoteIP(url);
+        streamPusher->setRemoteIP(urls);
         streamPusher->moveToThread(&pushStreamThread);
         pushStreamThread.start();
 
@@ -203,7 +203,7 @@ bool VideoWidget::pushStream(const QString &url) {
             return false;
         }
 
-        m_pushURL = url;
+        m_pushURLs = urls;
         m_pushFlag = true;
         return true;
     }
@@ -230,8 +230,8 @@ void VideoWidget::setPullURL(const QString &url) {
     m_pullURL = url;
 }
 
-void VideoWidget::setPushURL(const QString &url) {
-    m_pushURL = url;
+void VideoWidget::setPushURL(QStringList &urls) {
+    m_pushURLs = urls;
 }
 
 void VideoWidget::setVedioSaveFileDirPath(const QString &dirPath) {

src/video/videowidget.h

@@ -46,10 +46,10 @@ public:
     bool play(const QString &url, bool bSave = true);
     bool udpPlay(QString ip, int port);
     void stopPlay();
-    bool pushStream(const QString &url);
+    bool pushStream(QStringList &urls);
     void stopPushStream();
     void setPullURL(const QString &url);
-    void setPushURL(const QString &url);
+    void setPushURL(QStringList &urls);
     void setVedioSaveFileDirPath(const QString &dirPath);
     bool getPlayStatus();
     double getVideoDuration();
@@ -98,7 +98,7 @@ private:
 
 private:
     QString m_pullURL;
-    QString m_pushURL;
+    QStringList m_pushURLs;
     bool m_playFlag = false;
     bool m_pullFlag = false;
     bool m_pushFlag = false;