File indexing completed on 2024-11-10 04:40:30

 /*
     SPDX-FileCopyrightText: 2006 Tobias Koenig <tokoe@kde.org>
                   2006 Marc Mutz <mutz@kde.org>
                   2006 - 2007 Volker Krause <vkrause@kde.org>
 
     SPDX-License-Identifier: LGPL-2.0-or-later
 */
 
 #include "job.h"
 #include "akonadicore_debug.h"
 #include "job_p.h"
 #include "private/instance_p.h"
 #include "private/protocol_p.h"
 #include "session.h"
 #include "session_p.h"
 #include <QDBusConnection>
 #include <QTime>
 
 #include <KLocalizedString>
 
 #include <QDBusConnectionInterface>
 #include <QDBusInterface>
 #include <QElapsedTimer>
 #include <QTimer>
 
 using namespace Akonadi;
 
 static QDBusAbstractInterface *s_jobtracker = nullptr;
 
 /// @cond PRIVATE
 void JobPrivate::handleResponse(qint64 tag, const Protocol::CommandPtr &response)
 {
     Q_Q(Job);
 
     if (mCurrentSubJob) {
         mCurrentSubJob->d_ptr->handleResponse(tag, response);
         return;
     }
 
     if (tag == mTag) {
         if (response->isResponse()) {
             const auto &resp = Protocol::cmdCast<Protocol::Response>(response);
             if (resp.isError()) {
                 q->setError(Job::Unknown);
                 q->setErrorText(resp.errorMessage());
                 q->emitResult();
                 return;
             }
         }
     }
 
     if (mTag != tag) {
         qCWarning(AKONADICORE_LOG) << "Received response with a different tag!";
         qCDebug(AKONADICORE_LOG) << "Response tag:" << tag << ", response type:" << response->type();
         qCDebug(AKONADICORE_LOG) << "Job tag:" << mTag << ", job:" << q;
         return;
     }
 
     if (mStarted) {
         if (mReadingFinished) {
             qCWarning(AKONADICORE_LOG) << "Received response for a job that does not expect any more data, ignoring";
             qCDebug(AKONADICORE_LOG) << "Response tag:" << tag << ", response type:" << response->type();
             qCDebug(AKONADICORE_LOG) << "Job tag:" << mTag << ", job:" << q;
             Q_ASSERT(!mReadingFinished);
             return;
         }
 
         if (q->doHandleResponse(tag, response)) {
             mReadingFinished = true;
             QTimer::singleShot(0, q, [this]() {
                 delayedEmitResult();
             });
         }
     }
 }
 
 void JobPrivate::init(QObject *parent)
 {
     Q_Q(Job);
 
     mParentJob = qobject_cast<Job *>(parent);
     mSession = qobject_cast<Session *>(parent);
 
     if (!mSession) {
         if (!mParentJob) {
             mSession = Session::defaultSession();
         } else {
             mSession = mParentJob->d_ptr->mSession;
         }
     }
 
     if (!mParentJob) {
         mSession->d->addJob(q);
     } else {
         mParentJob->addSubjob(q);
     }
     publishJob();
 }
 
 void JobPrivate::publishJob()
 {
     Q_Q(Job);
     // if there's a job tracker running, tell it about the new job
     if (!s_jobtracker) {
         // Let's only check for the debugging console every 3 seconds, otherwise every single job
         // makes a dbus call to the dbus daemon, doesn't help performance.
         static QElapsedTimer s_lastTime;
         if (!s_lastTime.isValid() || s_lastTime.elapsed() > 3000) {
             if (!s_lastTime.isValid()) {
                 s_lastTime.start();
             }
             const QString suffix = Akonadi::Instance::identifier().isEmpty() ? QString() : QLatin1Char('-') + Akonadi::Instance::identifier();
             if (QDBusConnection::sessionBus().interface()->isServiceRegistered(QStringLiteral("org.kde.akonadiconsole") + suffix)) {
                 s_jobtracker = new QDBusInterface(QLatin1StringView("org.kde.akonadiconsole") + suffix,
                                                   QStringLiteral("/jobtracker"),
                                                   QStringLiteral("org.freedesktop.Akonadi.JobTracker"),
                                                   QDBusConnection::sessionBus(),
                                                   nullptr);
                 mSession->d->publishOtherJobs(q);
             } else {
                 s_lastTime.restart();
             }
         }
         // Note: we never reset s_jobtracker to 0 when a call fails; but if we did
         // then we should restart s_lastTime.
     }
     QMetaObject::invokeMethod(q, "signalCreationToJobTracker", Qt::QueuedConnection);
 }
 
 void JobPrivate::signalCreationToJobTracker()
 {
     Q_Q(Job);
     if (s_jobtracker) {
         // We do these dbus calls manually, so as to avoid having to install (or copy) the console's
         // xml interface document. Since this is purely a debugging aid, that seems preferable to
         // publishing something not intended for public consumption.
         // WARNING: for any signature change here, apply it to resourcescheduler.cpp too
         const QList<QVariant> argumentList = QList<QVariant>() << QLatin1StringView(mSession->sessionId()) << QString::number(reinterpret_cast<quintptr>(q), 16)
                                                                << (mParentJob ? QString::number(reinterpret_cast<quintptr>(mParentJob), 16) : QString())
                                                                << QString::fromLatin1(q->metaObject()->className()) << jobDebuggingString();
         QDBusPendingCall call = s_jobtracker->asyncCallWithArgumentList(QStringLiteral("jobCreated"), argumentList);
 
         auto watcher = new QDBusPendingCallWatcher(call, s_jobtracker);
         QObject::connect(watcher, &QDBusPendingCallWatcher::finished, s_jobtracker, [](QDBusPendingCallWatcher *w) {
             QDBusPendingReply<void> reply = *w;
             if (reply.isError() && s_jobtracker) {
                 qDebug() << reply.error().name() << reply.error().message();
                 s_jobtracker->deleteLater();
                 s_jobtracker = nullptr;
             }
             w->deleteLater();
         });
     }
 }
 
 void JobPrivate::signalStartedToJobTracker()
 {
     Q_Q(Job);
     if (s_jobtracker) {
         // if there's a job tracker running, tell it a job started
         const QList<QVariant> argumentList = {QString::number(reinterpret_cast<quintptr>(q), 16)};
         s_jobtracker->callWithArgumentList(QDBus::NoBlock, QStringLiteral("jobStarted"), argumentList);
     }
 }
 
 void JobPrivate::aboutToFinish()
 {
     // Dummy
 }
 
 void JobPrivate::delayedEmitResult()
 {
     Q_Q(Job);
     if (q->hasSubjobs()) {
         // We still have subjobs, wait for them to finish
         mFinishPending = true;
     } else {
         aboutToFinish();
         q->emitResult();
     }
 }
 
 void JobPrivate::startQueued()
 {
     Q_Q(Job);
     mStarted = true;
 
     Q_EMIT q->aboutToStart(q);
     q->doStart();
     QTimer::singleShot(0, q, [this]() {
         startNext();
     });
     QMetaObject::invokeMethod(q, "signalStartedToJobTracker", Qt::QueuedConnection);
 }
 
 void JobPrivate::lostConnection()
 {
     Q_Q(Job);
 
     if (mCurrentSubJob) {
         mCurrentSubJob->d_ptr->lostConnection();
     } else {
         q->setError(Job::ConnectionFailed);
         q->emitResult();
     }
 }
 
 void JobPrivate::slotSubJobAboutToStart(Job *job)
 {
     Q_ASSERT(mCurrentSubJob == nullptr);
     mCurrentSubJob = job;
 }
 
 void JobPrivate::startNext()
 {
     Q_Q(Job);
 
     if (mStarted && !mCurrentSubJob && q->hasSubjobs()) {
         Job *job = qobject_cast<Akonadi::Job *>(q->subjobs().at(0));
         Q_ASSERT(job);
         job->d_ptr->startQueued();
     } else if (mFinishPending && !q->hasSubjobs()) {
         // The last subjob we've been waiting for has finished, emitResult() finally
         QTimer::singleShot(0, q, [this]() {
             delayedEmitResult();
         });
     }
 }
 
 qint64 JobPrivate::newTag()
 {
     if (mParentJob) {
         mTag = mParentJob->d_ptr->newTag();
     } else {
         mTag = mSession->d->nextTag();
     }
     return mTag;
 }
 
 qint64 JobPrivate::tag() const
 {
     return mTag;
 }
 
 void JobPrivate::sendCommand(qint64 tag, const Protocol::CommandPtr &cmd)
 {
     if (mParentJob) {
         mParentJob->d_ptr->sendCommand(tag, cmd);
     } else {
         mSession->d->sendCommand(tag, cmd);
     }
 }
 
 void JobPrivate::sendCommand(const Protocol::CommandPtr &cmd)
 {
     sendCommand(newTag(), cmd);
 }
 
 void JobPrivate::itemRevisionChanged(Akonadi::Item::Id itemId, int oldRevision, int newRevision)
 {
     mSession->d->itemRevisionChanged(itemId, oldRevision, newRevision);
 }
 
 void JobPrivate::updateItemRevision(Akonadi::Item::Id itemId, int oldRevision, int newRevision)
 {
     Q_Q(Job);
     const auto &subjobs = q->subjobs();
     for (KJob *j : subjobs) {
         auto job = qobject_cast<Akonadi::Job *>(j);
         if (job) {
             job->d_ptr->updateItemRevision(itemId, oldRevision, newRevision);
         }
     }
     doUpdateItemRevision(itemId, oldRevision, newRevision);
 }
 
 void JobPrivate::doUpdateItemRevision(Akonadi::Item::Id itemId, int oldRevision, int newRevision)
 {
     Q_UNUSED(itemId)
     Q_UNUSED(oldRevision)
     Q_UNUSED(newRevision)
 }
 
 int JobPrivate::protocolVersion() const
 {
     return mSession->d->protocolVersion;
 }
 /// @endcond
 
 Job::Job(QObject *parent)
     : KCompositeJob(parent)
     , d_ptr(new JobPrivate(this))
 {
     d_ptr->init(parent);
 }
 
 Job::Job(JobPrivate *dd, QObject *parent)
     : KCompositeJob(parent)
     , d_ptr(dd)
 {
     d_ptr->init(parent);
 }
 
 Job::~Job()
 {
     // if there is a job tracer listening, tell it the job is done now
     if (s_jobtracker) {
         const QList<QVariant> argumentList = {QString::number(reinterpret_cast<quintptr>(this), 16), errorString()};
         s_jobtracker->callWithArgumentList(QDBus::NoBlock, QStringLiteral("jobEnded"), argumentList);
     }
 }
 
 void Job::start()
 {
 }
 
 bool Job::doKill()
 {
     Q_D(Job);
     if (d->mStarted) {
         // the only way to cancel an already started job is reconnecting to the server
         d->mSession->d->forceReconnect();
     }
     d->mStarted = false;
     return true;
 }
 
 QString Job::errorString() const
 {
     QString str;
     switch (error()) {
     case NoError:
         break;
     case ConnectionFailed:
         str = i18n("Cannot connect to the Akonadi service.");
         break;
     case ProtocolVersionMismatch:
         str = i18n("The protocol version of the Akonadi server is incompatible. Make sure you have a compatible version installed.");
         break;
     case UserCanceled:
         str = i18n("User canceled operation.");
         break;
     case Unknown:
         return errorText();
     case UserError:
         str = i18n("Unknown error.");
         break;
     }
     if (!errorText().isEmpty()) {
         str += QStringLiteral(" (%1)").arg(errorText());
     }
     return str;
 }
 
 bool Job::addSubjob(KJob *job)
 {
     bool rv = KCompositeJob::addSubjob(job);
     if (rv) {
         connect(qobject_cast<Job *>(job), &Job::aboutToStart, this, [this](Job *job) {
             d_ptr->slotSubJobAboutToStart(job);
         });
         QTimer::singleShot(0, this, [this]() {
             d_ptr->startNext();
         });
     }
     return rv;
 }
 
 bool Job::removeSubjob(KJob *job)
 {
     bool rv = KCompositeJob::removeSubjob(job);
     if (job == d_ptr->mCurrentSubJob) {
         d_ptr->mCurrentSubJob = nullptr;
         QTimer::singleShot(0, this, [this]() {
             d_ptr->startNext();
         });
     }
     return rv;
 }
 
 bool Akonadi::Job::doHandleResponse(qint64 tag, const Akonadi::Protocol::CommandPtr &response)
 {
     qCDebug(AKONADICORE_LOG) << this << "Unhandled response: " << tag << Protocol::debugString(response);
     setError(Unknown);
     setErrorText(i18n("Unexpected response"));
     emitResult();
     return true;
 }
 
 void Job::slotResult(KJob *job)
 {
     if (d_ptr->mCurrentSubJob == job) {
         // current job finished, start the next one
         d_ptr->mCurrentSubJob = nullptr;
         KCompositeJob::slotResult(job);
         if (!job->error()) {
             QTimer::singleShot(0, this, [this]() {
                 d_ptr->startNext();
             });
         }
     } else {
         // job that was still waiting for execution finished, probably canceled,
         // so just remove it from the queue and move on without caring about
         // its error code
         KCompositeJob::removeSubjob(job);
     }
 }
 
 void Job::emitWriteFinished()
 {
     d_ptr->mWriteFinished = true;
     Q_EMIT writeFinished(this);
 }
 
 #include "moc_job.cpp"