Threaded query runner implementation

All database access is now implemented in threads, to avoid
blocking the main thread. The resource communication still resides in
the main thread to keep the coordination simple.

With it comes a test that ensures we don't block the main thread for
too long.

4 files changed, 235 insertions, 76 deletions

diff --git a/common/queryrunner.cpp b/common/queryrunner.cpp
index 25c9d5b..af232c3 100644
--- a/common/queryrunner.cpp
+++ b/common/queryrunner.cpp
@@ -30,57 +30,89 @@
 
 using namespace Akonadi2;
 
-static inline ResultSet fullScan(const Akonadi2::Storage::Transaction &transaction, const QByteArray &bufferType)
+/*
+ * This class wraps the actual query implementation.
+ *
+ * This is a worker object that can be moved to a thread to execute the query.
+ * The only interaction point is the ResultProvider, which handles the threadsafe reporting of the result.
+ */
+template<typename DomainType>
+class QueryWorker : public QObject
 {
-    //TODO use a result set with an iterator, to read values on demand
-    QVector<QByteArray> keys;
-    transaction.openDatabase(bufferType + ".main").scan(QByteArray(), [&](const QByteArray &key, const QByteArray &value) -> bool {
-        //Skip internals
-        if (Akonadi2::Storage::isInternalKey(key)) {
-            return true;
-        }
-        keys << Akonadi2::Storage::uidFromKey(key);
-        return true;
-    },
-    [](const Akonadi2::Storage::Error &error) {
-        qWarning() << "Error during query: " << error.message;
-    });
+public:
+    QueryWorker(const Akonadi2::Query &query, const QByteArray &instanceIdentifier, const DomainTypeAdaptorFactoryInterface::Ptr &, const QByteArray &bufferType);
+    virtual ~QueryWorker();
+
+    qint64 executeIncrementalQuery(const Akonadi2::Query &query, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider);
+    qint64 executeInitialQuery(const Akonadi2::Query &query, const typename DomainType::Ptr &parent, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider);
+
+private:
+    static void replaySet(ResultSet &resultSet, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, const QList<QByteArray> &properties);
+
+    void readEntity(const Akonadi2::Storage::NamedDatabase &db, const QByteArray &key, const std::function<void(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &, Akonadi2::Operation)> &resultCallback);
+
+    ResultSet loadInitialResultSet(const Akonadi2::Query &query, Akonadi2::Storage::Transaction &transaction, QSet<QByteArray> &remainingFilters);
+    ResultSet loadIncrementalResultSet(qint64 baseRevision, const Akonadi2::Query &query, Akonadi2::Storage::Transaction &transaction, QSet<QByteArray> &remainingFilters);
+
+    ResultSet filterSet(const ResultSet &resultSet, const std::function<bool(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &domainObject)> &filter, const Akonadi2::Storage::NamedDatabase &db, bool initialQuery);
+    std::function<bool(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &domainObject)> getFilter(const QSet<QByteArray> remainingFilters, const Akonadi2::Query &query);
+    qint64 load(const Akonadi2::Query &query, const std::function<ResultSet(Akonadi2::Storage::Transaction &, QSet<QByteArray> &)> &baseSetRetriever, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, bool initialQuery);
+
+private:
+    DomainTypeAdaptorFactoryInterface::Ptr mDomainTypeAdaptorFactory;
+    QByteArray mResourceInstanceIdentifier;
+    QByteArray mBufferType;
+    Akonadi2::Query mQuery;
+};
 
-    Trace() << "Full scan on " << bufferType << " found " << keys.size() << " results";
-    return ResultSet(keys);
-}
 
 template<class DomainType>
 QueryRunner<DomainType>::QueryRunner(const Akonadi2::Query &query, const Akonadi2::ResourceAccessInterface::Ptr &resourceAccess, const QByteArray &instanceIdentifier, const DomainTypeAdaptorFactoryInterface::Ptr &factory, const QByteArray &bufferType)
     : QueryRunnerBase(),
-    mResultProvider(new ResultProvider<typename DomainType::Ptr>),
     mResourceAccess(resourceAccess),
-    mDomainTypeAdaptorFactory(factory),
-    mResourceInstanceIdentifier(instanceIdentifier),
-    mBufferType(bufferType),
-    mQuery(query)
+    mResultProvider(new ResultProvider<typename DomainType::Ptr>)
 {
     Trace() << "Starting query";
     //We delegate loading of initial data to the result provider, os it can decide for itself what it needs to load.
-    mResultProvider->setFetcher([this, query](const typename DomainType::Ptr &parent) {
-        Trace() << "Running fetcher";
-        const qint64 newRevision = executeInitialQuery(query, parent, *mResultProvider);
-        //Only send the revision replayed information if we're connected to the resource, there's no need to start the resource otherwise.
-        if (query.liveQuery) {
-            mResourceAccess->sendRevisionReplayedCommand(newRevision);
-        }
+    mResultProvider->setFetcher([this, query, instanceIdentifier, factory, bufferType](const typename DomainType::Ptr &parent) {
+            Trace() << "Running fetcher";
+            auto resultProvider = mResultProvider;
+            auto result = QtConcurrent::run([query, instanceIdentifier, factory, bufferType, parent, resultProvider]() -> qint64 {
+                QueryWorker<DomainType> worker(query, instanceIdentifier, factory, bufferType);
+                const qint64 newRevision = worker.executeInitialQuery(query, parent, *resultProvider);
+                return newRevision;
+            });
+            //Only send the revision replayed information if we're connected to the resource, there's no need to start the resource otherwise.
+            if (query.liveQuery) {
+                auto watcher = new QFutureWatcher<qint64>;
+                watcher->setFuture(result);
+                QObject::connect(watcher, &QFutureWatcher<qint64>::finished, watcher, [this, watcher]() {
+                    const auto newRevision = watcher->future().result();
+                    mResourceAccess->sendRevisionReplayedCommand(newRevision);
+                    delete watcher;
+                });
+            }
     });
 
-
-    //In case of a live query we keep the runner for as long alive as the result provider exists
+    // In case of a live query we keep the runner for as long alive as the result provider exists
     if (query.liveQuery) {
         //Incremental updates are always loaded directly, leaving it up to the result to discard the changes if they are not interesting
-        setQuery([this, query] () -> KAsync::Job<void> {
-            return KAsync::start<void>([this, query](KAsync::Future<void> &future) {
-                //TODO execute in thread
-                const qint64 newRevision = executeIncrementalQuery(query, *mResultProvider);
-                mResourceAccess->sendRevisionReplayedCommand(newRevision);
-                future.setFinished();
+        setQuery([this, query, instanceIdentifier, factory, bufferType] () -> KAsync::Job<void> {
+            return KAsync::start<void>([this, query, instanceIdentifier, factory, bufferType](KAsync::Future<void> &future) {
+                auto resultProvider = mResultProvider;
+                auto result = QtConcurrent::run([query, instanceIdentifier, factory, bufferType, resultProvider]() -> qint64 {
+                    QueryWorker<DomainType> worker(query, instanceIdentifier, factory, bufferType);
+                    const qint64 newRevision = worker.executeIncrementalQuery(query, *resultProvider);
+                    return newRevision;
+                });
+                auto watcher = new QFutureWatcher<qint64>;
+                watcher->setFuture(result);
+                QObject::connect(watcher, &QFutureWatcher<qint64>::finished, watcher, [this, &future, watcher]() {
+                    const auto newRevision = watcher->future().result();
+                    mResourceAccess->sendRevisionReplayedCommand(newRevision);
+                    future.setFinished();
+                    delete watcher;
+                });
             });
         });
         //Ensure the connection is open, if it wasn't already opened
@@ -102,8 +134,48 @@ typename Akonadi2::ResultEmitter<typename DomainType::Ptr>::Ptr QueryRunner<Doma
     return mResultProvider->emitter();
 }
 
+
+
+static inline ResultSet fullScan(const Akonadi2::Storage::Transaction &transaction, const QByteArray &bufferType)
+{
+    //TODO use a result set with an iterator, to read values on demand
+    QVector<QByteArray> keys;
+    transaction.openDatabase(bufferType + ".main").scan(QByteArray(), [&](const QByteArray &key, const QByteArray &value) -> bool {
+        //Skip internals
+        if (Akonadi2::Storage::isInternalKey(key)) {
+            return true;
+        }
+        keys << Akonadi2::Storage::uidFromKey(key);
+        return true;
+    },
+    [](const Akonadi2::Storage::Error &error) {
+        qWarning() << "Error during query: " << error.message;
+    });
+
+    Trace() << "Full scan on " << bufferType << " found " << keys.size() << " results";
+    return ResultSet(keys);
+}
+
+
 template<class DomainType>
-void QueryRunner<DomainType>::replaySet(ResultSet &resultSet, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, const QList<QByteArray> &properties)
+QueryWorker<DomainType>::QueryWorker(const Akonadi2::Query &query, const QByteArray &instanceIdentifier, const DomainTypeAdaptorFactoryInterface::Ptr &factory, const QByteArray &bufferType)
+    : QObject(),
+    mDomainTypeAdaptorFactory(factory),
+    mResourceInstanceIdentifier(instanceIdentifier),
+    mBufferType(bufferType),
+    mQuery(query)
+{
+    Trace() << "Starting query worker";
+}
+
+template<class DomainType>
+QueryWorker<DomainType>::~QueryWorker()
+{
+    Trace() << "Stopped query worker";
+}
+
+template<class DomainType>
+void QueryWorker<DomainType>::replaySet(ResultSet &resultSet, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, const QList<QByteArray> &properties)
 {
     int counter = 0;
     while (resultSet.next([&resultProvider, &counter, &properties](const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &value, Akonadi2::Operation operation) -> bool {
@@ -128,7 +200,7 @@ void QueryRunner<DomainType>::replaySet(ResultSet &resultSet, Akonadi2::ResultPr
 }
 
 template<class DomainType>
-void QueryRunner<DomainType>::readEntity(const Akonadi2::Storage::NamedDatabase &db, const QByteArray &key, const std::function<void(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &, Akonadi2::Operation)> &resultCallback)
+void QueryWorker<DomainType>::readEntity(const Akonadi2::Storage::NamedDatabase &db, const QByteArray &key, const std::function<void(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &, Akonadi2::Operation)> &resultCallback)
 {
     //This only works for a 1:1 mapping of resource to domain types.
     //Not i.e. for tags that are stored as flags in each entity of an imap store.
@@ -150,7 +222,7 @@ void QueryRunner<DomainType>::readEntity(const Akonadi2::Storage::NamedDatabase
 }
 
 template<class DomainType>
-ResultSet QueryRunner<DomainType>::loadInitialResultSet(const Akonadi2::Query &query, Akonadi2::Storage::Transaction &transaction, QSet<QByteArray> &remainingFilters)
+ResultSet QueryWorker<DomainType>::loadInitialResultSet(const Akonadi2::Query &query, Akonadi2::Storage::Transaction &transaction, QSet<QByteArray> &remainingFilters)
 {
     if (!query.ids.isEmpty()) {
         return ResultSet(query.ids.toVector());
@@ -168,7 +240,7 @@ ResultSet QueryRunner<DomainType>::loadInitialResultSet(const Akonadi2::Query &q
 }
 
 template<class DomainType>
-ResultSet QueryRunner<DomainType>::loadIncrementalResultSet(qint64 baseRevision, const Akonadi2::Query &query, Akonadi2::Storage::Transaction &transaction, QSet<QByteArray> &remainingFilters)
+ResultSet QueryWorker<DomainType>::loadIncrementalResultSet(qint64 baseRevision, const Akonadi2::Query &query, Akonadi2::Storage::Transaction &transaction, QSet<QByteArray> &remainingFilters)
 {
     const auto bufferType = mBufferType;
     auto revisionCounter = QSharedPointer<qint64>::create(baseRevision);
@@ -196,7 +268,7 @@ ResultSet QueryRunner<DomainType>::loadIncrementalResultSet(qint64 baseRevision,
 }
 
 template<class DomainType>
-ResultSet QueryRunner<DomainType>::filterSet(const ResultSet &resultSet, const std::function<bool(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &domainObject)> &filter, const Akonadi2::Storage::NamedDatabase &db, bool initialQuery)
+ResultSet QueryWorker<DomainType>::filterSet(const ResultSet &resultSet, const std::function<bool(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &domainObject)> &filter, const Akonadi2::Storage::NamedDatabase &db, bool initialQuery)
 {
     auto resultSetPtr = QSharedPointer<ResultSet>::create(resultSet);
 
@@ -225,9 +297,8 @@ ResultSet QueryRunner<DomainType>::filterSet(const ResultSet &resultSet, const s
     return ResultSet(generator);
 }
 
-
 template<class DomainType>
-std::function<bool(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &domainObject)> QueryRunner<DomainType>::getFilter(const QSet<QByteArray> remainingFilters, const Akonadi2::Query &query)
+std::function<bool(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &domainObject)> QueryWorker<DomainType>::getFilter(const QSet<QByteArray> remainingFilters, const Akonadi2::Query &query)
 {
     return [remainingFilters, query](const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &domainObject) -> bool {
         if (!query.ids.isEmpty()) {
@@ -252,7 +323,7 @@ std::function<bool(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr
 }
 
 template<class DomainType>
-qint64 QueryRunner<DomainType>::load(const Akonadi2::Query &query, const std::function<ResultSet(Akonadi2::Storage::Transaction &, QSet<QByteArray> &)> &baseSetRetriever, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, bool initialQuery)
+qint64 QueryWorker<DomainType>::load(const Akonadi2::Query &query, const std::function<ResultSet(Akonadi2::Storage::Transaction &, QSet<QByteArray> &)> &baseSetRetriever, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, bool initialQuery)
 {
     Akonadi2::Storage storage(Akonadi2::storageLocation(), mResourceInstanceIdentifier);
     storage.setDefaultErrorHandler([](const Akonadi2::Storage::Error &error) {
@@ -269,9 +340,8 @@ qint64 QueryRunner<DomainType>::load(const Akonadi2::Query &query, const std::fu
     return Akonadi2::Storage::maxRevision(transaction);
 }
 
-
 template<class DomainType>
-qint64 QueryRunner<DomainType>::executeIncrementalQuery(const Akonadi2::Query &query, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider)
+qint64 QueryWorker<DomainType>::executeIncrementalQuery(const Akonadi2::Query &query, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider)
 {
     QTime time;
     time.start();
@@ -286,7 +356,7 @@ qint64 QueryRunner<DomainType>::executeIncrementalQuery(const Akonadi2::Query &q
 }
 
 template<class DomainType>
-qint64 QueryRunner<DomainType>::executeInitialQuery(const Akonadi2::Query &query, const typename DomainType::Ptr &parent, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider)
+qint64 QueryWorker<DomainType>::executeInitialQuery(const Akonadi2::Query &query, const typename DomainType::Ptr &parent, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider)
 {
     QTime time;
     time.start();
@@ -312,3 +382,6 @@ qint64 QueryRunner<DomainType>::executeInitialQuery(const Akonadi2::Query &query
 template class QueryRunner<Akonadi2::ApplicationDomain::Folder>;
 template class QueryRunner<Akonadi2::ApplicationDomain::Mail>;
 template class QueryRunner<Akonadi2::ApplicationDomain::Event>;
+template class QueryWorker<Akonadi2::ApplicationDomain::Folder>;
+template class QueryWorker<Akonadi2::ApplicationDomain::Mail>;
+template class QueryWorker<Akonadi2::ApplicationDomain::Event>;
diff --git a/common/queryrunner.h b/common/queryrunner.h
index 8df0ecd..aba7912 100644
--- a/common/queryrunner.h
+++ b/common/queryrunner.h
@@ -20,7 +20,6 @@
 #pragma once
 
 #include <QObject>
-#include "facadeinterface.h"
 #include "resourceaccess.h"
 #include "resultprovider.h"
 #include "domaintypeadaptorfactoryinterface.h"
@@ -28,15 +27,8 @@
 #include "query.h"
 
 /**
- * A QueryRunner runs a query and updates the corresponding result set.
- * 
- * The lifetime of the QueryRunner is defined by the resut set (otherwise it's doing useless work),
- * and by how long a result set must be updated. If the query is one off the runner dies after the execution,
- * otherwise it lives on the react to changes and updates the corresponding result set.
- * 
- * QueryRunner has to keep ResourceAccess alive in order to keep getting updates.
+ * Base clase because you can't have the Q_OBJECT macro in template classes
  */
-
 class QueryRunnerBase : public QObject
 {
     Q_OBJECT
@@ -74,6 +66,15 @@ private:
     QueryFunction queryFunction;
 };
 
+/**
+ * A QueryRunner runs a query and updates the corresponding result set.
+ * 
+ * The lifetime of the QueryRunner is defined by the resut set (otherwise it's doing useless work),
+ * and by how long a result set must be updated. If the query is one off the runner dies after the execution,
+ * otherwise it lives on the react to changes and updates the corresponding result set.
+ * 
+ * QueryRunner has to keep ResourceAccess alive in order to keep getting updates.
+ */
 template<typename DomainType>
 class QueryRunner : public QueryRunnerBase
 {
@@ -84,25 +85,7 @@ public:
     typename Akonadi2::ResultEmitter<typename DomainType::Ptr>::Ptr emitter();
 
 private:
-    static void replaySet(ResultSet &resultSet, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, const QList<QByteArray> &properties);
-
-    void readEntity(const Akonadi2::Storage::NamedDatabase &db, const QByteArray &key, const std::function<void(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &, Akonadi2::Operation)> &resultCallback);
-
-    ResultSet loadInitialResultSet(const Akonadi2::Query &query, Akonadi2::Storage::Transaction &transaction, QSet<QByteArray> &remainingFilters);
-    ResultSet loadIncrementalResultSet(qint64 baseRevision, const Akonadi2::Query &query, Akonadi2::Storage::Transaction &transaction, QSet<QByteArray> &remainingFilters);
-
-    ResultSet filterSet(const ResultSet &resultSet, const std::function<bool(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &domainObject)> &filter, const Akonadi2::Storage::NamedDatabase &db, bool initialQuery);
-    std::function<bool(const Akonadi2::ApplicationDomain::ApplicationDomainType::Ptr &domainObject)> getFilter(const QSet<QByteArray> remainingFilters, const Akonadi2::Query &query);
-    qint64 load(const Akonadi2::Query &query, const std::function<ResultSet(Akonadi2::Storage::Transaction &, QSet<QByteArray> &)> &baseSetRetriever, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider, bool initialQuery);
-    qint64 executeIncrementalQuery(const Akonadi2::Query &query, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider);
-    qint64 executeInitialQuery(const Akonadi2::Query &query, const typename DomainType::Ptr &parent, Akonadi2::ResultProviderInterface<typename DomainType::Ptr> &resultProvider);
-
-private:
-    QSharedPointer<Akonadi2::ResultProvider<typename DomainType::Ptr> > mResultProvider;
     QSharedPointer<Akonadi2::ResourceAccessInterface> mResourceAccess;
-    DomainTypeAdaptorFactoryInterface::Ptr mDomainTypeAdaptorFactory;
-    QByteArray mResourceInstanceIdentifier;
-    QByteArray mBufferType;
-    Akonadi2::Query mQuery;
+    QSharedPointer<Akonadi2::ResultProvider<typename DomainType::Ptr> > mResultProvider;
 };
 
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 5d64511..1e0f6b5 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -50,11 +50,13 @@ auto_tests (
     querytest
     databasepopulationandfacadequerybenchmark
     dummyresourcewritebenchmark
+    modelinteractivitytest
 )
 target_link_libraries(dummyresourcetest akonadi2_resource_dummy)
 target_link_libraries(dummyresourcebenchmark akonadi2_resource_dummy)
 target_link_libraries(dummyresourcewritebenchmark akonadi2_resource_dummy)
 target_link_libraries(querytest akonadi2_resource_dummy)
+target_link_libraries(modelinteractivitytest akonadi2_resource_dummy)
 
 if (BUILD_MAILDIR)
     auto_tests (
diff --git a/tests/modelinteractivitytest.cpp b/tests/modelinteractivitytest.cpp
new file mode 100644
index 0000000..52db932
--- /dev/null
+++ b/tests/modelinteractivitytest.cpp
@@ -0,0 +1,101 @@
+#include <QtTest>
+
+#include <QString>
+#include <iostream>
+
+#include "dummyresource/resourcefactory.h"
+#include "clientapi.h"
+#include "commands.h"
+#include "resourceconfig.h"
+#include "log.h"
+#include "modelresult.h"
+
+static int blockingTime;
+
+class TimeMeasuringApplication : public QCoreApplication
+{
+    QElapsedTimer t;
+public:
+    TimeMeasuringApplication(int& argc, char ** argv) : QCoreApplication(argc, argv) { }
+    virtual ~TimeMeasuringApplication() { }
+
+    virtual bool notify(QObject* receiver, QEvent* event)
+    {
+        t.start();
+        const bool ret = QCoreApplication::notify(receiver, event);
+        if(t.elapsed() > 1)
+            std::cout << QString("processing event type %1 for object %2 took %3ms")
+                .arg((int)event->type())
+                .arg(""/* receiver->objectName().toLocal8Bit().data()*/)
+                .arg((int)t.elapsed())
+                .toStdString() << std::endl;
+        blockingTime += t.elapsed();
+        return ret;
+    }
+};
+
+/**
+ * Ensure that queries don't block the system for an extended period of time.
+ *
+ * This is done by ensuring that the event loop is never blocked.
+ */
+class ModelinteractivityTest : public QObject
+{
+    Q_OBJECT
+private Q_SLOTS:
+    void initTestCase()
+    {
+        Akonadi2::Log::setDebugOutputLevel(Akonadi2::Log::Warning);
+        DummyResource::removeFromDisk("org.kde.dummy.instance1");
+        ResourceConfig::addResource("org.kde.dummy.instance1", "org.kde.dummy");
+    }
+
+    void cleanup()
+    {
+        Akonadi2::Store::shutdown(QByteArray("org.kde.dummy.instance1")).exec().waitForFinished();
+        DummyResource::removeFromDisk("org.kde.dummy.instance1");
+        Akonadi2::Store::start(QByteArray("org.kde.dummy.instance1")).exec().waitForFinished();
+    }
+
+    void init()
+    {
+    }
+
+    void testSingle()
+    {
+        //Setup
+        {
+            Akonadi2::ApplicationDomain::Mail mail("org.kde.dummy.instance1");
+            for (int i = 0; i < 1000; i++) {
+                Akonadi2::Store::create<Akonadi2::ApplicationDomain::Mail>(mail).exec().waitForFinished();
+            }
+        }
+
+        Akonadi2::Query query;
+        query.resources << "org.kde.dummy.instance1";
+        query.syncOnDemand = false;
+        query.processAll = true;
+        query.liveQuery = true;
+
+        Akonadi2::Store::synchronize(query).exec().waitForFinished();
+
+        //Test
+        QTime time;
+        time.start();
+        auto model = Akonadi2::Store::loadModel<Akonadi2::ApplicationDomain::Mail>(query);
+        blockingTime += time.elapsed();
+        QTRY_VERIFY(model->data(QModelIndex(), Akonadi2::Store::ChildrenFetchedRole).toBool());
+        //Never block longer than 10 ms
+        QVERIFY2(blockingTime < 10, QString("Total blocking time: %1").arg(blockingTime).toLatin1().data());
+    }
+};
+
+int main(int argc, char *argv[])
+{
+    blockingTime = 0;
+    TimeMeasuringApplication app(argc, argv);
+    ModelinteractivityTest tc;
+    return QTest::qExec(&tc, argc, argv);
+}
+
+#include "modelinteractivitytest.moc"