Explaining code of compositeuploadjob.cpp

compositeuploadjob.cpp

CompositeUploadJob::CompositeUploadJob(const QString &deviceId, bool displayNotification)
    : KCompositeJob()
    , m_server(new Server(this))
    , m_socket(nullptr)
    , m_port(0)
    , m_deviceId(deviceId)
    , m_running(false)
    , m_currentJobNum(1)
    , m_totalJobs(0)
    , m_currentJobSendPayloadSize(0)
    , m_totalSendPayloadSize(0)
    , m_totalPayloadSize(0)
    , m_currentJob(nullptr)
    , m_prevElapsedTime(0)
    , m_updatePacketPending(false)
{
    setCapabilities(Killable);

    if (displayNotification) {
        Daemon::instance()->jobTracker()->registerJob(this);
    }
}

This constructor builds a composite upload job object and registers itself with the job tracker. It allows the caller to specify the device id, if it is set to display notification.

bool CompositeUploadJob::isRunning()
{
    return m_running;
}

This implements checking if the upload job is running.

void CompositeUploadJob::start()
{
    if (m_running) {
        qCWarning(KDECONNECT_CORE) << "CompositeUploadJob::start() - already running";
        return;
    }

    if (!hasSubjobs()) {
        qCWarning(KDECONNECT_CORE) << "CompositeUploadJob::start() - there are no subjobs to start";
        emitResult();
        return;
    }

    if (!startListening()) {
        return;
    }

    connect(m_server, &QTcpServer::newConnection, this, &CompositeUploadJob::newConnection);

    m_running = true;

    // Give SharePlugin some time to add subjobs
    QMetaObject::invokeMethod(this, "startNextSubJob", Qt::QueuedConnection);
}

This starts the upload job if it is not already running, and connects to the server and starts the next subjob. It also connects the socket to the QTcpServer object and sets the m_running flag to true.

bool CompositeUploadJob::startListening()
{
    m_port = MIN_PORT;
    while (!m_server->listen(QHostAddress::Any, m_port)) {
        m_port++;
        if (m_port > MAX_PORT) { // No ports available?
            qCWarning(KDECONNECT_CORE) << "CompositeUploadJob::startListening() - Error opening a port in range" << MIN_PORT << "-" << MAX_PORT;
            m_port = 0;
            setError(NoPortAvailable);
            setErrorText(i18n("Couldn't find an available port"));
            emitResult();
            return false;
        }
    }

    qCDebug(KDECONNECT_CORE) << "CompositeUploadJob::startListening() - listening on port: " << m_port;
    return true;
}

This starts listening for incoming connections. It returns true on the first successful attempt. It returns false if the port cannot be opened.

void CompositeUploadJob::startNextSubJob()
{
    m_currentJob = qobject_cast(subjobs().at(0));
    m_currentJobSendPayloadSize = 0;
    emitDescription(m_currentJob->getNetworkPacket().get(QStringLiteral("filename")));

#ifdef SAILFISHOS
    connect(m_currentJob, SIGNAL(processedAmount(KJob *, KJob::Unit, qulonglong)), this, SLOT(slotProcessedAmount(KJob *, KJob::Unit, qulonglong)));
#else
    connect(m_currentJob, QOverload::of(&UploadJob::processedAmount), this, &CompositeUploadJob::slotProcessedAmount);
#endif
    // Already done by KCompositeJob
    // connect(m_currentJob, &KJob::result, this, &CompositeUploadJob::slotResult);

    // TODO: Create a copy of the networkpacket that can be re-injected if sending via lan fails?
    NetworkPacket np = m_currentJob->getNetworkPacket();
#if QT_VERSION < QT_VERSION_CHECK(5, 8, 0)
    np.setPayload({}, np.payloadSize());
#else
    np.setPayload(nullptr, np.payloadSize());
#endif
    np.setPayloadTransferInfo({{QStringLiteral("port"), m_port}});
    np.set(QStringLiteral("numberOfFiles"), m_totalJobs);
    np.set(QStringLiteral("totalPayloadSize"), m_totalPayloadSize);

    if (Daemon::instance()->getDevice(m_deviceId)->sendPacket(np)) {
        m_server->resumeAccepting();
    } else {
        setError(SendingNetworkPacketFailed);
        setErrorText(i18n("Failed to send packet to %1", Daemon::instance()->getDevice(m_deviceId)->name()));

        emitResult();
    }
}

This sends a network packet of the next subjob of the current group, and sets the payload size for the current job and sets the payload transfer info to the network packet and resumes accepting. It also emits the result signal.

void CompositeUploadJob::newConnection()
{
    m_server->pauseAccepting();

    m_socket = m_server->nextPendingConnection();

    if (!m_socket) {
        qCDebug(KDECONNECT_CORE) << "CompositeUploadJob::newConnection() - m_server->nextPendingConnection() returned a nullptr";
        return;
    }

    m_currentJob->setSocket(m_socket);

    connect(m_socket, &QSslSocket::disconnected, this, &CompositeUploadJob::socketDisconnected);
#if QT_VERSION < QT_VERSION_CHECK(5, 15, 0)
    connect(m_socket, QOverload::of(&QAbstractSocket::error), this, &CompositeUploadJob::socketError);
#else
    connect(m_socket, &QAbstractSocket::errorOccurred, this, &CompositeUploadJob::socketError);
#endif
    connect(m_socket, QOverload &>::of(&QSslSocket::sslErrors), this, &CompositeUploadJob::sslError);
    connect(m_socket, &QSslSocket::encrypted, this, &CompositeUploadJob::encrypted);

    LanLinkProvider::configureSslSocket(m_socket, m_deviceId, true);

    m_socket->startServerEncryption();
}

This creates a new socket object and connects it to the current upload job. It first pauses accepting, and gets the next pending connection, and sets up connections for error handling, as well creates a signal called "errorOccurred" and "sslError" overload.

void CompositeUploadJob::socketDisconnected()
{
    m_socket->close();
}

This closes the socket after the upload has completed.

void CompositeUploadJob::socketError(QAbstractSocket::SocketError error)
{
    Q_UNUSED(error);

    // Do not close the socket because when android closes the socket (share is cancelled) closing the socket leads to a cyclic socketError and eventually a
    // segv
    setError(SocketError);
    emitResult();

    m_running = false;
}

This implements a socket error signal. It is not used by the upload job code because it is called from the main thread. It also sets the error state of the m_running variable to false. It also emits the result to the caller.

void CompositeUploadJob::sslError(const QList &errors)
{
    Q_UNUSED(errors);

    m_socket->close();
    setError(SslError);
    emitResult();

    m_running = false;
}

This closes the socket and emits a ssl error.

void CompositeUploadJob::encrypted()
{
    if (!m_timer.isValid()) {
        m_timer.start();
    }

    m_currentJob->start();
}

This starts the timer if it is not valid. Then it starts the current upload job.

bool CompositeUploadJob::addSubjob(KJob *job)
{
    if (UploadJob *uploadJob = qobject_cast(job)) {
        NetworkPacket np = uploadJob->getNetworkPacket();

        m_totalJobs++;

        if (np.payloadSize() >= 0) {
            m_totalPayloadSize += np.payloadSize();
            setTotalAmount(Bytes, m_totalPayloadSize);
        }

        QString filename;
        QString filenameArg = QStringLiteral("filename");

        if (m_currentJob) {
            filename = m_currentJob->getNetworkPacket().get(filenameArg);
        } else {
            filename = np.get(filenameArg);
        }

        emitDescription(filename);

        if (m_running && m_currentJob && !m_updatePacketPending) {
            m_updatePacketPending = true;
            QMetaObject::invokeMethod(this, "sendUpdatePacket", Qt::QueuedConnection);
        }

        return KCompositeJob::addSubjob(job);
    } else {
        qCDebug(KDECONNECT_CORE) << "CompositeUploadJob::addSubjob() - you can only add UploadJob's, ignoring";
        return false;
    }
}

This adds a subjob to the current upload job list. It emits the description of the current upload job and sets m_totalJobs to the total jobs count, and emits the sendUpdatePacket method if the upload job is running. It returns false if the job is not able to add a subjob.

void CompositeUploadJob::sendUpdatePacket()
{
    NetworkPacket np(PACKET_TYPE_SHARE_REQUEST_UPDATE);
    np.set(QStringLiteral("numberOfFiles"), m_totalJobs);
    np.set(QStringLiteral("totalPayloadSize"), m_totalPayloadSize);

    Daemon::instance()->getDevice(m_deviceId)->sendPacket(np);

    m_updatePacketPending = false;
}

This sends a network packet that tells the device that the upload job needs to be updated.

bool CompositeUploadJob::doKill()
{
    if (m_running) {
        m_running = false;

        return m_currentJob->stop();
    }

    return true;
}

This stops the current upload job if it is running. It returns true on the first successful attempt. If the current job is not running, it returns false.

void CompositeUploadJob::slotProcessedAmount(KJob *job, KJob::Unit unit, qulonglong amount)
{
    Q_UNUSED(job);

    m_currentJobSendPayloadSize = amount;
    quint64 uploaded = m_totalSendPayloadSize + m_currentJobSendPayloadSize;

    if (uploaded == m_totalPayloadSize || m_prevElapsedTime == 0 || m_timer.elapsed() - m_prevElapsedTime >= 100) {
        m_prevElapsedTime = m_timer.elapsed();
        setProcessedAmount(unit, uploaded);

        const auto elapsed = m_timer.elapsed();
        if (elapsed > 0) {
            emitSpeed((1000 * uploaded) / elapsed);
        }
    }
}

This sets the processed amount of the given job based on the amount of bytes sent to the remote host. It first checks if the total payload size of the job has been uploaded, if it has been written to the remote host, and if it has been written to the remote host, it updates the processed amount by the amount. If the total payload size has not been uploaded, the function emits a rate of 1000ms based on the elapsed time.

void CompositeUploadJob::slotResult(KJob *job)
{
    // Copies job error and errorText and emits result if job is in error otherwise removes job from subjob list
    KCompositeJob::slotResult(job);

    if (error() || !m_running) {
        return;
    }

    m_totalSendPayloadSize += m_currentJobSendPayloadSize;

    if (hasSubjobs()) {
        m_currentJobNum++;
        startNextSubJob();
    } else {
        emitResult();
    }
}

This method is responsible for setting the error message and emits the result if the job is in error. It also starts a new subjob if the job is a subjob. It also starts the next job if there are subjobs.

void CompositeUploadJob::emitDescription(const QString ¤tFileName)
{
    Q_EMIT description(this, i18n("Sending to %1", Daemon::instance()->getDevice(this->m_deviceId)->name()), {i18n("File"), currentFileName}, {});

    setProcessedAmount(Files, m_currentJobNum);
    setTotalAmount(Files, m_totalJobs);
}

This emits a description of the upload job by setting the current file name and total amount of files by calling a Q_EMIT function. It also sets the processedAmount and totalAmount members of the CompositeUploadJob object to the values of the currentFileName parameter.