PayloadAPP/Src/VideoGL/pushstream.cpp

#include "pushstream.h"

PushStream::PushStream(QObject *parent) : QObject{parent} {}

void PushStream::setRemoteIP(QString url) {
    m_pushStreamIP = url;
}

bool PushStream::init(AVFormatContext *inputFormatCtx,
                      AVPacketQueueManager *queueManager) {
    // m_pusherQueue = queue;
    m_queueManager = queueManager;
    m_inputFormatCtx = inputFormatCtx;
    m_start = openNetworkStream(inputFormatCtx);
    return m_start;
}

void PushStream::close() {
    // QMutexLocker locker(&m_mutex);
    m_start = false;
    m_end = true;
    // qDebug() << "*******m_end0:" << m_end;
}

bool PushStream::openNetworkStream(AVFormatContext *inputFormatCtx) {
    if (m_pushStreamIP.isEmpty()) return false;

    if (!inputFormatCtx) return false;

    // 初始化网络输出流
    // QString m_pushStreamIP = "rtsp://182.92.130.23/app/stream999";
    int ret = avformat_alloc_output_context2(
        &m_outputFormatCtx, NULL, "flv", m_pushStreamIP.toUtf8().constData());
    if (ret < 0) {
        showError(ret);
        free();
        qDebug() << "Could not create output context.";
        return false;
    }

    // 复制流信息
    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);
            if (!m_ostream) {
                qDebug() << "Failed allocating output stream.\n";
                free();
                return false;
            }
            // 复制编解码器参数
            ret = avcodec_parameters_copy(m_ostream->codecpar,
                                          m_istream->codecpar);
            if (ret < 0) {
                showError(ret);
                free();
                qWarning() << "avcodec_parameters_from_context Failed";
                return false;
            }
            // outputStream->codecpar = inputStream->codecpar;
            m_ostream->codecpar->codec_tag = 0;
            break;
        }
    }

    m_inputTimeBase = m_istream->time_base;
    m_inputFrameRate = m_istream->r_frame_rate;
    m_outputTimeBase = m_ostream->time_base;

    // 打开输出文件
    if (!(m_outputFormatCtx->flags & AVFMT_NOFILE)) {
        if (ret = avio_open(&m_outputFormatCtx->pb,
                            m_pushStreamIP.toUtf8().constData(),
                            AVIO_FLAG_WRITE) < 0) {
            showError(ret);
            free();
            qDebug() << "Could not open output file.\n";
            return false;
        }
    }

    // 写入头文件
    ret = avformat_write_header(m_outputFormatCtx, NULL);
    if (ret < 0) {
        showError(ret);
        free();
        qDebug() << "Error occurred when write_header into output file.\n";
        return false;
    }
    // sendSPS_PPS(inputFormatCtx, outputFormatCtx);
    // m_InitStatus = true;
    // startTime = av_gettime_relative();

    m_bwriteHeader = true;
    m_firstPts = AV_NOPTS_VALUE;
    return true;
}

int PushStream::reconnect(int ret) {
    if (ret == -10053 || ret == -10054) {
        m_end = false;
        // qDebug() << "m_end:" << m_end;
        for (int nRetryCount = 0; nRetryCount < MAXCONNECT; ++nRetryCount) {
            // 关闭输出
            if (m_outputFormatCtx &&
                !(m_outputFormatCtx->flags & AVFMT_NOFILE)) {
                avio_close(m_outputFormatCtx->pb);
            }
            ret =
                avio_open(&m_outputFormatCtx->pb,
                          m_pushStreamIP.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);
                if (m_end) break;
                // av_usleep(5 * 1000000);
                continue;
            }
            // Try to reconnect
            ret = avformat_write_header(m_outputFormatCtx, nullptr);
            if (ret < 0) {
                showError(ret);
                // free();
                qDebug() << "Failed to reconnect"
                         << QString::number(nRetryCount + 1);
                QString str =
                    QString("网络中断，尝试重连第%1次!").arg(nRetryCount + 1);
                emit sendErrorMessageSignal(str, 3);
                if (m_end) break;
                // nRetryCount++;
                // av_usleep(5 * 1000000);
            } else {
                m_start = true;
                m_firstPts = AV_NOPTS_VALUE;
                m_frm_cnt = 0;
                m_bwriteHeader = true;
                emit sendErrorMessageSignal("重连成功!", 1);
                m_queueManager->clearPushQueue();
                qDebug() << "重连成功!";
                return ret;
            }

            if (m_end) break;
        }
        m_start = false;
        m_bwriteHeader = false;
        emit sendErrorMessageSignal("重连失败，推流停止!", 2);
        return -1;
    }
}

void PushStream::pushStream(int64_t startTime) {
    qDebug() << "PushStreamThreadID:" << QThread::currentThreadId();
    // m_startTime = startTime;
    while (m_start) {
        AVPacket *inputPacket = m_queueManager->dequeuePushPacket();
        if (inputPacket) {
            if (inputPacket->data == nullptr || inputPacket->size <= 0 ||
                inputPacket->pts <= 0) {
                // av_packet_unref(inputPacket);
                continue;
            }
            int outputStreamIndex = 0;
            // int inputStreamIndex = inputPacket->stream_index;
            // AVStream *istream = m_inputFormatCtx->streams[inputStreamIndex];
            // AVStream *ostream =
            // m_outputFormatCtx->streams[outputStreamIndex]; 时间基
            // AVRational istream_base = m_istream->time_base;
            // AVRational ostream_base = ostream->time_base;
            // 没有pts的视频数据，如未解码的H.264裸流，需要重新计算其pts。
            if (inputPacket->pts == AV_NOPTS_VALUE) {
                // Duration between 2 frames (us)
                int64_t calc_duration =
                    (double)AV_TIME_BASE / av_q2d(m_inputFrameRate);
                // Reset Parameters
                inputPacket->pts =
                    (double)(m_frm_cnt * calc_duration) /
                    (double)(av_q2d(m_inputTimeBase) * AV_TIME_BASE);
                inputPacket->dts = inputPacket->pts;
                inputPacket->duration =
                    (double)calc_duration /
                    (double)(av_q2d(m_inputTimeBase) * AV_TIME_BASE);
            }
            // 视频帧推送速度
            if (m_firstPts == AV_NOPTS_VALUE) {
                m_startTime = av_gettime();
                m_firstPts = av_rescale_q(inputPacket->dts, m_inputTimeBase,
                                          AVRational{1, AV_TIME_BASE});
            }
            auto pts_time = av_rescale_q(inputPacket->pts, m_inputTimeBase,
                                         AVRational{1, AV_TIME_BASE});
            int64_t pts_time_corrected = pts_time - m_firstPts;
            auto now_time = av_gettime() - m_startTime;
            int64_t delay = pts_time_corrected - now_time;
            // qDebug() << "****************PushStream sleep time:"
            //          << QString::number(delay / 1000);
            // delay = delay < MAXDELAY ? delay : MAXDELAY;
            if (delay > 0) {
                // qDebug() << "****************PushStream sleep time:"
                //          << QString::number(delay / 1000);
                av_usleep(delay);
                // sleepMsec(40);
            } else {
                if (delay < -50000) {
                    // 滞后50ms以上，丢弃非重要帧
                    if (!(inputPacket->flags & AV_PKT_FLAG_KEY)) {
                        av_packet_unref(inputPacket);
                        continue;
                    }
                    // 调整基准时间，减少滞后
                    m_startTime += 50000;
                }
            }
            // 计算延时后，重新指定时间戳
            av_packet_rescale_ts(inputPacket, m_inputTimeBase,
                                 m_outputTimeBase);
            inputPacket->pos = -1;
            inputPacket->stream_index = 0;
            if (inputPacket->pts < inputPacket->dts) {
                // qDebug() << "pts < dts";
                if (!(inputPacket->flags & AV_PKT_FLAG_KEY)) {
                    av_packet_unref(inputPacket);
                    continue;
                }
            }

            // 向推流服务器推送流数据
            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;
            }
            // 数据包写入成功，现在可以释放pkt
            av_packet_unref(inputPacket);
            av_packet_free(&inputPacket);
        } else {
            QThread::usleep(1000);
        }
    }

    if (m_bwriteHeader) av_write_trailer(m_outputFormatCtx);
    free();
    qDebug() << "Push Stream End!";
    emit sendErrorMessageSignal("推流结束!",
                                NotificationType::NOTIFICATION_SUCCESS);
}

void PushStream::free() {
    m_start = false;
    // 关闭输出
    if (m_outputFormatCtx && !(m_outputFormatCtx->flags & AVFMT_NOFILE)) {
        avio_close(m_outputFormatCtx->pb);
    }
    if (m_outputFormatCtx) {
        avformat_free_context(m_outputFormatCtx);
        m_outputFormatCtx = nullptr;
    }
}