qt-migration-legatus/qt6-after/qt6-migrated-legatus/Lts/Common/httq/DataStream.cpp

#include <QDataStream>
#include <QFile>
#include <QIODevice>
#include <QTcpSocket>
#include <QTemporaryFile>
#include <QTimer>
#include <httq/DataStream.h>
#include <httq/HttpRequest.h>

#ifdef Q_OS_UNIX
#include <unistd.h>
#endif

namespace httq
{
DataStream::DataStream(QIODevice* from, QIODevice* to, const QByteArray& bodyPartial, qint64 fileLength,
                       size_t bufferSize, LoggerFactory* loggerFactory, QObject* parent)
    : QObject(parent), mBodyPartial(bodyPartial), mFrom(from), mTo(to), mBufferSize(bufferSize),
      mFileLength(fileLength), mLogger(loggerFactory->createLogger(this))
{

#ifdef SAVE_PARTIAL_BODY
    {
        QTemporaryFile f;
        f.open();
        f.write(mBodyPartial);
        f.setAutoRemove(false);
        qWarning() << "partial:" << f.fileName();
    }
#endif

    if (mFileLength == 0)
    {
        mLogger->warning(QStringLiteral("file size = 0 - done"));
        QTimer::singleShot(1, this, [this]() {
            emit signalDone();
            deleteLater();
        });
        return;
    }

    mLogger->debug(QStringLiteral("DataStream c_tor"));

    mBuffer.reserve(bufferSize);

    mFromSupportsSignals = (qobject_cast<QFileDevice*>(mFrom) == nullptr);
    mToSupportsSignals = (qobject_cast<QFileDevice*>(mTo) == nullptr);
    mLogger->debug(
        QStringLiteral("signal support - from: %1 - to: %2").arg(mFromSupportsSignals).arg(mToSupportsSignals));

    auto sock = qobject_cast<QTcpSocket*>(mFrom);
    qWarning() << "######################### SOCK" << sock;
    connect(sock, &QTcpSocket::readyRead, this, [this]() {
        mLogger->debug(QStringLiteral(">> DataStream readyRead"));

        QTimer::singleShot(1, this, [this]() { readNext(); });
    });

    if (mToSupportsSignals)
        connect(mTo, &QIODevice::bytesWritten, this, [this]() {
            mLogger->debug(QStringLiteral(">> DataStream write"));

            mReadyWrite = true;
            QTimer::singleShot(1, this, [this]() { readNext(); });
        });

    QTimer::singleShot(1, this, [this]() { readNext(); });
}

DataStream::~DataStream()
{
    mLogger->debug(QStringLiteral("~DataStream"));
}

void DataStream::readNext()
{
#if defined(DEBUG_SLOW) && defined(Q_OS_UNIX)
    mLogger->debug(QStringLiteral("read: slowing down....."));
    usleep(2 * 1000 * 1000);
#endif

    mLogger->debug(QStringLiteral("read: readNext: %1").arg(mBuffer.size()));

    if (mBuffer.size() > 0)
    {
        mLogger->debug(QStringLiteral("read: ignoring read until buffer was written"));
        return;
    }

    if (mToSupportsSignals && !mReadyWrite)
    {
        mLogger->debug(QStringLiteral("read: won't read until write is ready!"));
        return;
    }

    if (!mBodyPartial.isEmpty())
    {
        mBuffer = mBodyPartial;

        alreadyRead += mBuffer.size();
        mLogger->debug(QStringLiteral("read: prebody size: %1").arg(mBuffer.size()));

        mBodyPartial.clear();

        mLogger->debug(QStringLiteral("read: queueing writeNext after prebody read"));
        QTimer::singleShot(1, this, [this]() { writeNext(); });
        return;
    }

    if ((mFrom->bytesAvailable() < 1))
    {

        mLogger->debug(QStringLiteral("read: no bytes available and not ready read"));

        QTimer::singleShot(1000, this, [this]() { readNext(); });

        return;
    }

    mBuffer = mFrom->read(mBufferSize);
    mLogger->debug(QStringLiteral("read: mPos is now %1").arg(mBuffer.size()));

    alreadyRead += mBuffer.size();
    mLogger->debug(QStringLiteral("read: alreadyRead = %1").arg(alreadyRead));

    if (mBuffer.size() < 1)
    {
        mLogger->debug(QStringLiteral("read: cannot read anything"));
        deleteLater();
        return;
    }

    mLogger->debug(QStringLiteral("read: queueing writeNext"));
    QTimer::singleShot(1, this, [this]() { writeNext(); });
}

void DataStream::writeNext()
{
    mLogger->debug(QStringLiteral("write: writeNext: %1").arg(mBuffer.size()));

    if (mBuffer.size() < 1)
    {
        mLogger->debug(QStringLiteral("write: nothing to write"));
        QTimer::singleShot(1, this, [this]() { readNext(); });
        return;
    }

    if (!mTo->isOpen() || !mTo->isWritable())
    {
        mLogger->debug(
            QStringLiteral("write: file error: open? %1 - writable? %2").arg(mTo->isOpen()).arg(mTo->isWritable()));
        deleteLater();
        return;
    }

    if (mToSupportsSignals && !mReadyWrite)
    {
        mLogger->debug(QStringLiteral("write: !mReadyWrite"));

        return;
    }

    if (mToSupportsSignals)
        mReadyWrite = false;

    const qint64 size = mTo->write(mBuffer);
    mLogger->debug(QStringLiteral("write: wrote to buffer %1 bytes").arg(size));

    mLogger->debug(QStringLiteral("write: %1 / %2 bytes done (%3 bytes left until DataStream done)")
                       .arg(alreadyRead)
                       .arg(mFileLength)
                       .arg(mFileLength - alreadyRead));

    if (size < mBuffer.size())
    {
        mLogger->warning(QStringLiteral("write: cannot write all bytes: %1 / %2 bytes").arg(mBuffer.size()).arg(size));
        deleteLater();
        return;
    }

    mLogger->debug(QStringLiteral("write: mPos is now %1 and was set to zero now!!").arg(mBuffer.size()));
    mBuffer.clear();

    if (alreadyRead == mFileLength)
    {
        mLogger->debug(QStringLiteral("write: DONE!"));
        emit signalDone();
        deleteLater();
        return;
    }

    if (alreadyRead > mFileLength)
    {
        mLogger->debug(QStringLiteral("write: oops... we read too much!! data may be corrupt!! already read %1 bytes, "
                                      "but content length is only %2 (%1 / %2)")
                           .arg(alreadyRead)
                           .arg(mFileLength));
        emit signalDone();
        deleteLater();
        return;
    }

    if (mToSupportsSignals)
        mLogger->debug(QStringLiteral("write: waiting for bytes written to be triggered..."));

    mLogger->debug(QStringLiteral("write: queueing writeNext again"));
    QTimer::singleShot(1, this, [this]() { writeNext(); });
}
} // namespace httq