PayloadAPP/Src/VideoGL/readstream.cpp

#include "readstream.h"

ReadStream::ReadStream(QObject *parent) : QObject{parent} {
    // m_packetsQueue.reserve(QUEUECAPACITY);
    // m_saverQueue.reserve(QUEUECAPACITY);
    // m_pusherQueue.reserve(QUEUECAPACITY);
    initFFmpeg();
}

ReadStream::~ReadStream() {}

bool ReadStream::openFile(const QString &url) {
    if (url.isNull()) return false;
    AVDictionary *dict = nullptr;
    // 如果设置失败，则设置UDP传输
    // av_dict_set(&dict, "rtsp_transport", "udp", 0);
    // ////注：设置tcp会导致吊舱拉流中断 设置缓存大小，1080p可将值调大
    av_dict_set(&dict, "buffer_size", "4096000", 0);
    // 设置超时断开连接时间，单位微秒//listen_timeout
    // av_dict_set(&avdic, "listen_timeout", "200000", 0);
    av_dict_set(&dict, "stimeout", "5000000", 0);  // 设置超时5秒
    av_dict_set(&dict, "timeout", "5000000", 0);   // 设置5秒超时
    av_dict_set(&dict, "max_delay", "300000", 0);  // 设置最大时延300ms
    av_dict_set(&dict, "tune", "zerolatency", 0);  // 实时编码
    av_dict_set(&dict, "preset", "faster", 0);     // ultrafast,faster
    av_dict_set(&dict, "threads", "auto", 0);      // 自动开启线程数
    // 设置最大重试时间为1s,解决avformat_open_input打开空流时间过长问题
    av_dict_set(&dict, "max_interleave_delta", "1000000", 0);
    // av_dict_set(&dict, "reconnect", "1", 0); // 开启自动重连
    // av_dict_set(&dict, "reconnect_streamed", "1", 0); // 对于流媒体自动重连
    // av_dict_set(&dict, "reconnect_at_eof", "1", 0);
    av_dict_set(&dict, "reconnect_delay_max", "5",
                0);  // 最大重连延时 5 秒
    // 设置非阻塞标志
    // av_dict_set(&dict, "flags", "+nonblock", 0);
    // 打开输入流之前，设置非阻塞模式
    // m_formatContext = avformat_alloc_context();
    // m_formatContext->flags |= AVFMT_FLAG_NONBLOCK;
    // 打开输入流并返回解封装上下文
    int ret = avformat_open_input(
        &m_formatContext,     // 返回解封装上下文
        url.toUtf8().data(),  // 打开视频地址
        nullptr,  // 如果非null，此参数强制使用特定的输入格式。自动选择解封装器（文件格式）
        &dict);   // 参数设置
    // 释放参数字典
    if (dict) {
        av_dict_free(&dict);
    }
    // 打开视频失败
    if (ret < 0) {
        showError(ret);
        free();
        return false;
    }
    // 读取媒体文件的数据包以获取流信息。
    ret = avformat_find_stream_info(m_formatContext, nullptr);
    if (ret < 0) {
        showError(ret);
        free();
        return false;
    }

    // 通过AVMediaType枚举查询视频流ID（也可以通过遍历查找），最后一个参数无用
    m_videoIndex = av_find_best_stream(m_formatContext, AVMEDIA_TYPE_VIDEO, -1,
                                       -1, nullptr, 0);
    if (m_videoIndex < 0) {
        showError(m_videoIndex);
        free();
        return false;
    }

    m_startTime = -1;
    m_pullURL = url;
    return initObject();
}

bool ReadStream::setStreamDecoder(DecodeStream *decodeStreamer) {
    if (decodeStreamer) {
        // QMutexLocker locker(&m_mutex);
        m_streamDecoder = decodeStreamer;
        m_queueManager.clearDecodeQueue();
        m_decodeStreamFlag = m_streamDecoder->init(
            &m_queueManager, m_formatContext, m_videoIndex);
        if (m_decodeStreamFlag) emit decodeStreamer->startDecodeSignal();
        return m_decodeStreamFlag;
    } else {
        m_decodeStreamFlag = false;
    }
    return false;
}

bool ReadStream::setStreamSaver(SaveStream *streamSaver) {
    if (streamSaver) {
        m_streamSaver = streamSaver;
        m_queueManager.clearSaveQueue();
        m_saveStreamFlag =
            m_streamSaver->init(m_formatContext, &m_queueManager, m_videoIndex);
        return m_saveStreamFlag;
    } else {
        m_saveStreamFlag = false;
    }
    return false;
}

bool ReadStream::setStreamPusher(PushStream *streamPusher) {
    if (streamPusher) {
        m_streamPusher = streamPusher;
        m_queueManager.clearPushQueue();
        m_pushStreamFlag = streamPusher->init(m_formatContext, &m_queueManager);
        if (m_pushStreamFlag) streamPusher->startPushStreamSignal(0);
        return m_pushStreamFlag;
    } else {
        m_pushStreamFlag = false;
    }
    return false;
}

void ReadStream::close() {
    // QMutexLocker locker(&m_mutex);
    m_start = false;
    m_end = true;
    // if (m_streamDecoder) {
    //     m_streamDecoder->close();
    // }
    // if (m_streamSaver) {
    //     m_streamSaver->close();
    // }
    // if (m_streamPusher) {
    //     m_streamPusher->close();
    // }
}

void ReadStream::startPullStream() {
    // 如果没有打开则返回
    if (!m_formatContext) {
        return;
    }
    qDebug() << "readStreamThreadID:" << QThread::currentThreadId();
    emit m_streamSaver->startSaveStreamSignal();
    m_start = true;
    int readRet;
    // 读取数据包
    while (m_start) {
        readRet = av_read_frame(m_formatContext, m_inputPacket);
        if (readRet == AVERROR(EAGAIN)) {  // 暂时没有数据流
            qDebug() << "No Stream Data";
            av_packet_unref(m_inputPacket);
            av_usleep(30000);  // 等待 30 毫秒
            continue;
        } else if (readRet == AVERROR_EOF) {  // 流文件结束
            qDebug() << "Stream End";
            av_packet_unref(m_inputPacket);
            close();
            break;
        } else if (readRet < 0) {  // 发生错误
            qDebug() << "Stream Error";
            if (reconnect()) {
                continue;
            } else {
                if (m_streamDecoder) {
                    m_streamDecoder->close();
                }
                if (m_streamSaver) {
                    m_streamSaver->close();
                }
                if (m_streamPusher) {
                    m_streamPusher->close();
                }
                break;
            }
        }
        if (isValidAVPacket(m_inputPacket)) {
            if (m_inputPacket->stream_index == m_videoIndex) {
                if (m_decodeStreamFlag) {
                    m_queueManager.enqueueDecodePacket(m_inputPacket);
                }

                if (m_saveStreamFlag) {
                    m_queueManager.enqueueSavePacket(m_inputPacket);
                }

                if (m_pushStreamFlag) {
                    m_queueManager.enqueuePushPacket(m_inputPacket);
                }
            }
        }
        av_packet_unref(m_inputPacket);
        QThread::usleep(2000);  // 等待 2 毫秒
    }
    clear();
    free();
    qDebug() << "read Stream End!";
}

void ReadStream::initFFmpeg() {
    avformat_network_init();
    // m_formatContext = avformat_alloc_context();
    // m_formatContext->flags |= AVFMT_FLAG_NONBLOCK;
}

bool ReadStream::initObject() {
    // 分配AVPacket并将其字段设置为默认值。
    m_inputPacket = av_packet_alloc();
    if (!m_inputPacket) {
#if PRINT_LOG
        qWarning() << "av_packet_alloc() Error！";
#endif
        free();
        return false;
    }

    return true;
}

/**
 * @brief 清空读取缓冲
 */
void ReadStream::clear() {
    // 因为avformat_flush不会刷新AVIOContext
    // (s->pb)。如果有必要，在调用此函数之前调用avio_flush(s->pb)。
    if (m_formatContext && m_formatContext->pb) {
        avio_flush(m_formatContext->pb);
    }
    if (m_formatContext) {
        avformat_flush(m_formatContext);  // 清理读取缓冲
    }
}

void ReadStream::free() {
    // 关闭并失败m_formatContext，并将指针置为null
    if (m_formatContext) {
        avformat_close_input(&m_formatContext);
    }
    if (m_inputPacket) {
        av_packet_free(&m_inputPacket);
    }
}

bool ReadStream::reconnect() {
    m_end = false;
    if (m_streamDecoder) {
        m_streamDecoder->close();
    }
    if (m_streamSaver) {
        m_streamSaver->close();
    }
    if (m_streamPusher) {
        m_streamPusher->close();
    }
    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);
            }
            if (m_streamSaver) {
                setStreamSaver(m_streamSaver);
            }
            if (m_streamPusher) {
                setStreamPusher(m_streamPusher);
            }
            return true;
        } else {
            qDebug() << "reconnect failed:" << QString::number(i + 1);
            QString str = QString("流中断，尝试重新连接第%1次").arg(i + 1);
            emit sendErrorMessageSignal(str, 3);
            free();
            // close();
        }
        if (m_end) break;
    }
    emit sendErrorMessageSignal("重连失败!", 2);
    return false;
}

bool ReadStream::isValidAVPacket(AVPacket *pkt) {
    if (pkt == nullptr) {
        qDebug() << "Invalid AVPacket 0: packet pointer is null.";
        return false;
    }
    // 检查数据指针和大小
    if (pkt->data == nullptr || pkt->size <= 0) {
        qDebug()
            << "Invalid AVPacket 0: data is null or size is non-positive.\n";
        return false;
    }

    // 检查时间戳
    if (pkt->pts == AV_NOPTS_VALUE || pkt->dts == AV_NOPTS_VALUE) {
        qDebug() << "Invalid AVPacket 0: pts or dts is AV_NOPTS_VALUE.\n";
        return false;
    }

    // 检查流索引（如果是多流）
    if (pkt->stream_index < 0) {
        qDebug() << "Invalid AVPacket 0: stream_index is invalid.\n";
        return false;
    }

    // 检查是否是有效的关键帧（可选）
    if (pkt->flags & AV_PKT_FLAG_KEY) {
        // 是关键帧
        // qDebug() << "This is a keyframe.\n";
    }

    return true;
}