qt6-migration-pipeline/Lts/Common/httq/DataStream.cpp

#include <httq/DataStream.h>
#include <httq/HttpRequest.h>

#include <QIODevice>
#include <QTcpSocket>
#include <QTimer>
#include <QDataStream>
#include <QTemporaryFile>
#include <QFile>

#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(); } );
}
}