File indexing completed on 2025-01-05 04:46:59

 /*
     SPDX-FileCopyrightText: 2009 Volker Krause <vkrause@kde.org>
 
     SPDX-License-Identifier: LGPL-2.0-or-later
 */
 
 #include "itemretrievalmanager.h"
 #include "akonadiserver_debug.h"
 #include "itemretrievaljob.h"
 
 #include "resourceinterface.h"
 
 #include "private/dbus_p.h"
 
 #include <QDBusConnection>
 #include <QDBusConnectionInterface>
 #include <QScopedPointer>
 
 using namespace Akonadi;
 using namespace Akonadi::Server;
 
 Q_DECLARE_METATYPE(Akonadi::Server::ItemRetrievalResult)
 
 class ItemRetrievalJobFactory : public AbstractItemRetrievalJobFactory
 {
     AbstractItemRetrievalJob *retrievalJob(ItemRetrievalRequest request, QObject *parent) override
     {
         return new ItemRetrievalJob(std::move(request), parent);
     }
 };
 
 ItemRetrievalManager::ItemRetrievalManager(QObject *parent)
     : ItemRetrievalManager(std::make_unique<ItemRetrievalJobFactory>(), parent)
 {
 }
 
 ItemRetrievalManager::ItemRetrievalManager(std::unique_ptr<AbstractItemRetrievalJobFactory> factory, QObject *parent)
     : AkThread(QStringLiteral("ItemRetrievalManager"), QThread::HighPriority, parent)
     , mJobFactory(std::move(factory))
 {
     qRegisterMetaType<ItemRetrievalResult>("Akonadi::Server::ItemRetrievalResult");
     qDBusRegisterMetaType<QByteArrayList>();
 }
 
 ItemRetrievalManager::~ItemRetrievalManager()
 {
     quitThread();
 }
 
 void ItemRetrievalManager::init()
 {
     AkThread::init();
 
     QDBusConnection conn = QDBusConnection::sessionBus();
     connect(conn.interface(), &QDBusConnectionInterface::serviceOwnerChanged, this, &ItemRetrievalManager::serviceOwnerChanged);
     connect(this, &ItemRetrievalManager::requestAdded, this, &ItemRetrievalManager::processRequest, Qt::QueuedConnection);
 }
 
 // called within the retrieval thread
 void ItemRetrievalManager::serviceOwnerChanged(const QString &serviceName, const QString &oldOwner, const QString &newOwner)
 {
     Q_UNUSED(newOwner)
     if (oldOwner.isEmpty()) {
         return;
     }
     const auto service = DBus::parseAgentServiceName(serviceName);
     if (!service.has_value() || service->agentType != DBus::Resource) {
         return;
     }
     qCDebug(AKONADISERVER_LOG) << "ItemRetrievalManager lost connection to resource" << serviceName << ", discarding cached interface";
     mResourceInterfaces.erase(service->identifier);
 }
 
 // called within the retrieval thread
 org::freedesktop::Akonadi::Resource *ItemRetrievalManager::resourceInterface(const QString &id)
 {
     if (id.isEmpty()) {
         return nullptr;
     }
 
     auto ifaceIt = mResourceInterfaces.find(id);
     if (ifaceIt != mResourceInterfaces.cend() && ifaceIt->second->isValid()) {
         return ifaceIt->second.get();
     }
 
     auto iface =
         std::make_unique<org::freedesktop::Akonadi::Resource>(DBus::agentServiceName(id, DBus::Resource), QStringLiteral("/"), QDBusConnection::sessionBus());
     if (!iface->isValid()) {
         qCCritical(AKONADISERVER_LOG,
                    "Cannot connect to agent instance with identifier '%s', error message: '%s'",
                    qUtf8Printable(id),
                    qUtf8Printable(iface ? iface->lastError().message() : QString()));
         return nullptr;
     }
     // DBus calls can take some time to reply -- e.g. if a huge local mbox has to be parsed first.
     iface->setTimeout(5 * 60 * 1000); // 5 minutes, rather than 25 seconds
     std::tie(ifaceIt, std::ignore) = mResourceInterfaces.emplace(id, std::move(iface));
     return ifaceIt->second.get();
 }
 
 // called from any thread
 void ItemRetrievalManager::requestItemDelivery(ItemRetrievalRequest req)
 {
     QWriteLocker locker(&mLock);
     qCDebug(AKONADISERVER_LOG) << "ItemRetrievalManager posting retrieval request for items" << req.ids << "to" << req.resourceId << ". There are"
                                << mPendingRequests.size() << "request queues and" << mPendingRequests[req.resourceId].size() << "items mine";
     mPendingRequests[req.resourceId].emplace_back(std::move(req));
     locker.unlock();
 
     Q_EMIT requestAdded();
 }
 
 QList<AbstractItemRetrievalJob *> ItemRetrievalManager::scheduleJobsForIdleResourcesLocked()
 {
     QList<AbstractItemRetrievalJob *> newJobs;
     for (auto it = mPendingRequests.begin(); it != mPendingRequests.end();) {
         if (it->second.empty()) {
             it = mPendingRequests.erase(it);
             continue;
         }
 
         if (!mCurrentJobs.contains(it->first) || mCurrentJobs.value(it->first) == nullptr) {
             // TODO: check if there is another one for the same uid with more parts requested
             auto req = std::move(it->second.front());
             it->second.pop_front();
             Q_ASSERT(req.resourceId == it->first);
             auto job = mJobFactory->retrievalJob(std::move(req), this);
             connect(job, &AbstractItemRetrievalJob::requestCompleted, this, &ItemRetrievalManager::retrievalJobFinished);
             mCurrentJobs.insert(job->request().resourceId, job);
             // delay job execution until after we unlocked the mutex, since the job can emit the finished signal immediately in some cases
             newJobs.append(job);
             qCDebug(AKONADISERVER_LOG) << "ItemRetrievalJob" << job << "started for request" << job->request().id;
         }
         ++it;
     }
 
     return newJobs;
 }
 
 // called within the retrieval thread
 void ItemRetrievalManager::processRequest()
 {
     QWriteLocker locker(&mLock);
     // look for idle resources
     auto newJobs = scheduleJobsForIdleResourcesLocked();
     // someone asked as to process requests although everything is done already, he might still be waiting
     if (mPendingRequests.empty() && mCurrentJobs.isEmpty() && newJobs.isEmpty()) {
         return;
     }
     locker.unlock();
 
     // Start the jobs
     for (auto job : newJobs) {
         if (auto j = qobject_cast<ItemRetrievalJob *>(job)) {
             j->setInterface(resourceInterface(j->request().resourceId));
         }
         job->start();
     }
 }
 
 namespace
 {
 bool isSubsetOf(const QByteArrayList &superset, const QByteArrayList &subset)
 {
     // For very small lists like these, this is faster than copy, sort and std::include
     return std::all_of(subset.cbegin(), subset.cend(), [&superset](const auto &val) {
         return superset.contains(val);
     });
 }
 
 }
 
 void ItemRetrievalManager::retrievalJobFinished(AbstractItemRetrievalJob *job)
 {
     const auto &request = job->request();
     const auto &result = job->result();
 
     if (result.errorMsg.has_value()) {
         qCWarning(AKONADISERVER_LOG) << "ItemRetrievalJob for request" << request.id << "finished with error:" << *result.errorMsg;
     } else {
         qCInfo(AKONADISERVER_LOG) << "ItemRetrievalJob for request" << request.id << "finished";
     }
 
     QWriteLocker locker(&mLock);
     Q_ASSERT(mCurrentJobs.contains(request.resourceId));
     mCurrentJobs.remove(request.resourceId);
     // Check if there are any pending requests that are satisfied by this retrieval job
     auto &requests = mPendingRequests[request.resourceId];
     for (auto it = requests.begin(); it != requests.end();) {
         // TODO: also complete requests that are subset of the completed one
         if (it->ids == request.ids && isSubsetOf(request.parts, it->parts)) {
             qCDebug(AKONADISERVER_LOG) << "Someone else requested items " << request.ids << "as well, marking as processed.";
             ItemRetrievalResult otherResult{std::move(*it)};
             otherResult.errorMsg = result.errorMsg;
             Q_EMIT requestFinished(otherResult);
             it = requests.erase(it);
         } else {
             ++it;
         }
     }
     locker.unlock();
 
     Q_EMIT requestFinished(result);
     Q_EMIT requestAdded(); // trigger processRequest() again, in case there is more in the queues
 }
 
 // Can be called from any thread
 void ItemRetrievalManager::triggerCollectionSync(const QString &resource, qint64 colId)
 {
     QTimer::singleShot(0, this, [this, resource, colId]() {
         if (auto interface = resourceInterface(resource)) {
             interface->synchronizeCollection(colId);
         }
     });
 }
 
 void ItemRetrievalManager::triggerCollectionTreeSync(const QString &resource)
 {
     QTimer::singleShot(0, this, [this, resource]() {
         if (auto interface = resourceInterface(resource)) {
             interface->synchronizeCollectionTree();
         }
     });
 }
 
 #include "moc_itemretrievalmanager.cpp"