7 files changed, 169 insertions, 77 deletions

diff --git a/common/datastorequery.cpp b/common/datastorequery.cpp
index 3218d1a..50158c7 100644
--- a/common/datastorequery.cpp
+++ b/common/datastorequery.cpp
@@ -643,6 +643,7 @@ ResultSet DataStoreQuery::execute()
 {
     SinkTraceCtx(mLogCtx) << "Executing query";
 
+    Q_ASSERT(mCollector);
     ResultSet::ValueGenerator generator = [this](const ResultSet::Callback &callback) -> bool {
         if (mCollector->next([this, callback](const ResultSet::Result &result) {
                 if (result.operation != Sink::Operation_Removal) {
diff --git a/common/datastorequery.h b/common/datastorequery.h
index cc501e6..8800644 100644
--- a/common/datastorequery.h
+++ b/common/datastorequery.h
@@ -115,7 +115,7 @@ public:
     virtual void updateComplete() { }
 
     FilterBase::Ptr mSource;
-    DataStoreQuery *mDatastore;
+    DataStoreQuery *mDatastore{nullptr};
     bool mIncremental = false;
 };
 
diff --git a/common/queryrunner.cpp b/common/queryrunner.cpp
index 928e1e0..0ed4cb5 100644
--- a/common/queryrunner.cpp
+++ b/common/queryrunner.cpp
@@ -69,79 +69,14 @@ QueryRunner<DomainType>::QueryRunner(const Sink::Query &query, const Sink::Resou
     if (query.limit() && query.sortProperty().isEmpty()) {
         SinkWarningCtx(mLogCtx) << "A limited query without sorting is typically a bad idea, because there is no telling what you're going to get.";
     }
-    auto guardPtr = QPointer<QObject>(&guard);
-    auto fetcher = [=]() {
-        SinkTraceCtx(mLogCtx) << "Running fetcher. Batchsize: " << mBatchSize;
-        auto resultProvider = mResultProvider;
-        auto resultTransformation = mResultTransformation;
-        auto batchSize = mBatchSize;
-        auto resourceContext = mResourceContext;
-        auto logCtx = mLogCtx;
-        auto state = mQueryState;
-        const bool runAsync = !query.synchronousQuery();
-        //The lambda will be executed in a separate thread, so copy all arguments
-        async::run<ReplayResult>([=]() {
-            QueryWorker<DomainType> worker(query, resourceContext, bufferType, resultTransformation, logCtx);
-            return worker.executeInitialQuery(query, *resultProvider, batchSize, state);
-        }, runAsync)
-            .then([this, query, resultProvider, guardPtr](const ReplayResult &result) {
-                if (!guardPtr) {
-                    //Not an error, the query can vanish at any time.
-                    return;
-                }
-                mInitialQueryComplete = true;
-                mQueryState = result.queryState;
-                // 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(result.newRevision);
-                }
-                resultProvider->setRevision(result.newRevision);
-                resultProvider->initialResultSetComplete(result.replayedAll);
-            })
-            .exec();
-    };
-
     // We delegate loading of initial data to the result provider, so it can decide for itself what it needs to load.
-    mResultProvider->setFetcher(fetcher);
+    mResultProvider->setFetcher([this, query, bufferType] { fetch(query, bufferType); });
 
     // In case of a live query we keep the runner for as long alive as the result provider exists
     if (query.liveQuery()) {
         Q_ASSERT(!query.synchronousQuery());
         // Incremental updates are always loaded directly, leaving it up to the result to discard the changes if they are not interesting
-        setQuery([=]() -> KAsync::Job<void> {
-            auto resultProvider = mResultProvider;
-            auto resourceContext = mResourceContext;
-            auto logCtx = mLogCtx;
-            auto state = mQueryState;
-            auto resultTransformation = mResultTransformation;
-            if (!mInitialQueryComplete) {
-                SinkWarningCtx(mLogCtx) << "Can't start the incremental query before the initial query is complete";
-                fetcher();
-                return KAsync::null();
-            }
-            if (mQueryInProgress) {
-                //Can happen if the revision come in quicker than we process them.
-                return KAsync::null();
-            }
-            Q_ASSERT(!mQueryInProgress);
-            return KAsync::start([&] {
-                    mQueryInProgress = true;
-                })
-                .then(async::run<ReplayResult>([=]() {
-                       QueryWorker<DomainType> worker(query, resourceContext, bufferType, resultTransformation, logCtx);
-                       return worker.executeIncrementalQuery(query, *resultProvider, state);
-                   }))
-                .then([query, this, resultProvider, guardPtr](const ReplayResult &newRevisionAndReplayedEntities) {
-                    if (!guardPtr) {
-                        //Not an error, the query can vanish at any time.
-                        return;
-                    }
-                    mQueryInProgress = false;
-                    // Only send the revision replayed information if we're connected to the resource, there's no need to start the resource otherwise.
-                    mResourceAccess->sendRevisionReplayedCommand(newRevisionAndReplayedEntities.newRevision);
-                    resultProvider->setRevision(newRevisionAndReplayedEntities.newRevision);
-                });
-        });
+        setQuery([=]() { return incrementalFetch(query, bufferType); });
         // Ensure the connection is open, if it wasn't already opened
         // TODO If we are not connected already, we have to check for the latest revision once connected, otherwise we could miss some updates
         mResourceAccess->open();
@@ -158,6 +93,90 @@ QueryRunner<DomainType>::~QueryRunner()
     SinkTraceCtx(mLogCtx) << "Stopped query";
 }
 
+//This function triggers the initial fetch, and then subsequent calls will simply fetch more data of mBatchSize.
+template <class DomainType>
+void QueryRunner<DomainType>::fetch(const Sink::Query &query, const QByteArray &bufferType)
+{
+    auto guardPtr = QPointer<QObject>(&guard);
+    SinkTraceCtx(mLogCtx) << "Running fetcher. Batchsize: " << mBatchSize;
+    if (mQueryInProgress) {
+        SinkTraceCtx(mLogCtx) << "Query is already in progress, postponing: " << mBatchSize;
+        mRequestFetchMore = true;
+        return;
+    }
+    mQueryInProgress = true;
+    auto resultProvider = mResultProvider;
+    auto resultTransformation = mResultTransformation;
+    auto batchSize = mBatchSize;
+    auto resourceContext = mResourceContext;
+    auto logCtx = mLogCtx;
+    auto state = mQueryState;
+    const bool runAsync = !query.synchronousQuery();
+    //The lambda will be executed in a separate thread, so copy all arguments
+    async::run<ReplayResult>([=]() {
+        QueryWorker<DomainType> worker(query, resourceContext, bufferType, resultTransformation, logCtx);
+        return worker.executeInitialQuery(query, *resultProvider, batchSize, state);
+    }, runAsync)
+        .then([=](const ReplayResult &result) {
+            if (!guardPtr) {
+                //Not an error, the query can vanish at any time.
+                return;
+            }
+            mInitialQueryComplete = true;
+            mQueryInProgress = false;
+            mQueryState = result.queryState;
+            // 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(result.newRevision);
+            }
+            resultProvider->setRevision(result.newRevision);
+            resultProvider->initialResultSetComplete(result.replayedAll);
+            if (mRequestFetchMore) {
+                mRequestFetchMore = false;
+                fetch(query, bufferType);
+            }
+        })
+        .exec();
+}
+
+template <class DomainType>
+KAsync::Job<void> QueryRunner<DomainType>::incrementalFetch(const Sink::Query &query, const QByteArray &bufferType)
+{
+    if (!mInitialQueryComplete) {
+        SinkWarningCtx(mLogCtx) << "Can't start the incremental query before the initial query is complete";
+        fetch(query, bufferType);
+        return KAsync::null();
+    }
+    if (mQueryInProgress) {
+        //Can happen if the revision come in quicker than we process them.
+        return KAsync::null();
+    }
+    auto resultProvider = mResultProvider;
+    auto resourceContext = mResourceContext;
+    auto logCtx = mLogCtx;
+    auto state = mQueryState;
+    auto resultTransformation = mResultTransformation;
+    Q_ASSERT(!mQueryInProgress);
+    auto guardPtr = QPointer<QObject>(&guard);
+    return KAsync::start([&] {
+            mQueryInProgress = true;
+        })
+        .then(async::run<ReplayResult>([=]() {
+                QueryWorker<DomainType> worker(query, resourceContext, bufferType, resultTransformation, logCtx);
+                return worker.executeIncrementalQuery(query, *resultProvider, state);
+            }))
+        .then([query, this, resultProvider, guardPtr](const ReplayResult &newRevisionAndReplayedEntities) {
+            if (!guardPtr) {
+                //Not an error, the query can vanish at any time.
+                return;
+            }
+            mQueryInProgress = false;
+            // Only send the revision replayed information if we're connected to the resource, there's no need to start the resource otherwise.
+            mResourceAccess->sendRevisionReplayedCommand(newRevisionAndReplayedEntities.newRevision);
+            resultProvider->setRevision(newRevisionAndReplayedEntities.newRevision);
+        });
+}
+
 template <class DomainType>
 void QueryRunner<DomainType>::setResultTransformation(const ResultTransformation &transformation)
 {
diff --git a/common/queryrunner.h b/common/queryrunner.h
index a567b3c..af54619 100644
--- a/common/queryrunner.h
+++ b/common/queryrunner.h
@@ -92,6 +92,9 @@ public:
     typename Sink::ResultEmitter<typename DomainType::Ptr>::Ptr emitter();
 
 private:
+    void fetch(const Sink::Query &query, const QByteArray &bufferType);
+    KAsync::Job<void> incrementalFetch(const Sink::Query &query, const QByteArray &bufferType);
+
     Sink::ResourceContext mResourceContext;
     QSharedPointer<Sink::ResourceAccessInterface> mResourceAccess;
     QSharedPointer<Sink::ResultProvider<typename DomainType::Ptr>> mResultProvider;
@@ -102,4 +105,5 @@ private:
     Sink::Log::Context mLogCtx;
     bool mInitialQueryComplete = false;
     bool mQueryInProgress = false;
+    bool mRequestFetchMore = false;
 };
diff --git a/common/storage/entitystore.cpp b/common/storage/entitystore.cpp
index 8fbc2ad..020f3fd 100644
--- a/common/storage/entitystore.cpp
+++ b/common/storage/entitystore.cpp
@@ -459,6 +459,7 @@ void EntityStore::indexLookup(const QByteArray &type, const QByteArray &property
 
 void EntityStore::readLatest(const QByteArray &type, const QByteArray &uid, const std::function<void(const QByteArray &uid, const EntityBuffer &entity)> callback)
 {
+    Q_ASSERT(d);
     auto db = DataStore::mainDatabase(d->getTransaction(), type);
     db.findLatest(uid,
         [=](const QByteArray &key, const QByteArray &value) {
diff --git a/common/storage/entitystore.h b/common/storage/entitystore.h
index 00241f2..3eb0b7b 100644
--- a/common/storage/entitystore.h
+++ b/common/storage/entitystore.h
@@ -37,6 +37,7 @@ class SINK_EXPORT EntityStore
 public:
     typedef QSharedPointer<EntityStore> Ptr;
     EntityStore(const ResourceContext &resourceContext, const Sink::Log::Context &);
+    ~EntityStore() = default;
 
     //Only the pipeline may call the following functions outside of tests
     bool add(const QByteArray &type, ApplicationDomain::ApplicationDomainType newEntity, bool replayToSource);
diff --git a/sinksh/syntax_modules/sink_selftest.cpp b/sinksh/syntax_modules/sink_selftest.cpp
index 21dfbff..8ad4f60 100644
--- a/sinksh/syntax_modules/sink_selftest.cpp
+++ b/sinksh/syntax_modules/sink_selftest.cpp
@@ -46,28 +46,94 @@ bool selfTest(const QStringList &args_, State &state)
         return false;
     }
 
+    using namespace Sink::ApplicationDomain;
     auto options = SyntaxTree::parseOptions(args_);
     if (options.positionalArguments.contains("stresstest")) {
         auto resource = SinkshUtils::parseUid(options.options.value("resource").first().toUtf8());
         qWarning() << "Stresstest on resource: " << resource;
-        Sink::Query query;
-        query.resourceFilter(resource);
-        query.limit(100);
-
         auto models = QSharedPointer<QList<QSharedPointer<QAbstractItemModel>>>::create();
-        for (int i = 0; i < 50; i++) {
+
+        //Simulate the maillist, where we scroll down and trigger lots of fetchMore calls
+        {
+            Sink::Query query;
+            query.resourceFilter(resource);
+            query.limit(100);
+            query.request<Mail::Subject>();
+            query.request<Mail::Sender>();
+            query.request<Mail::To>();
+            query.request<Mail::Cc>();
+            query.request<Mail::Bcc>();
+            query.request<Mail::Date>();
+            query.request<Mail::Unread>();
+            query.request<Mail::Important>();
+            query.request<Mail::Draft>();
+            query.request<Mail::Sent>();
+            query.request<Mail::Trash>();
+            query.request<Mail::Folder>();
+            query.sort<Mail::Date>();
+            query.reduce<Mail::ThreadId>(Sink::Query::Reduce::Selector::max<Mail::Date>())
+                .count("count")
+                .collect<Mail::Unread>("unreadCollected")
+                .collect<Mail::Important>("importantCollected");
+
             auto model = Sink::Store::loadModel<Sink::ApplicationDomain::Mail>(query);
-            *models << model;
+            models->append(model);
             QObject::connect(model.data(), &QAbstractItemModel::dataChanged, [models, model, &state](const QModelIndex &start, const QModelIndex &end, const QVector<int> &roles) {
                 if (roles.contains(Sink::Store::ChildrenFetchedRole)) {
-                    models->removeAll(model);
-                    qWarning() << "Model complete: " << models->count();
+                    if (!model->canFetchMore({})) {
+                        qWarning() << "Model complete: " << models->count();
+                        models->removeAll(model);
+                    } else {
+                        qWarning() << "Fetching more";
+                        //Simulate superfluous fetchMore calls
+                        for (int i = 0; i < 10; i++) {
+                            model->fetchMore({});
+                        }
+                        return;
+                    }
                     if (models->isEmpty()) {
                         state.commandFinished();
                     }
                 }
-                    });
+            });
+
         }
+
+        //Simluate lot's of mailviewers doing a bunch of queries
+        {
+            Sink::Query query;
+            query.resourceFilter(resource);
+            query.limit(10);
+            query.request<Mail::Subject>();
+            query.request<Mail::Sender>();
+            query.request<Mail::To>();
+            query.request<Mail::Cc>();
+            query.request<Mail::Bcc>();
+            query.request<Mail::Date>();
+            query.request<Mail::Unread>();
+            query.request<Mail::Important>();
+            query.request<Mail::Draft>();
+            query.request<Mail::Sent>();
+            query.request<Mail::Trash>();
+            query.request<Mail::Folder>();
+            query.sort<Sink::ApplicationDomain::Mail::Date>();
+            query.bloom<Sink::ApplicationDomain::Mail::ThreadId>();
+
+            for (int i = 0; i < 50; i++) {
+                auto model = Sink::Store::loadModel<Sink::ApplicationDomain::Mail>(query);
+                *models << model;
+                QObject::connect(model.data(), &QAbstractItemModel::dataChanged, [models, model, &state](const QModelIndex &start, const QModelIndex &end, const QVector<int> &roles) {
+                    if (roles.contains(Sink::Store::ChildrenFetchedRole)) {
+                        models->removeAll(model);
+                        qWarning() << "Model complete: " << models->count();
+                        if (models->isEmpty()) {
+                            state.commandFinished();
+                        }
+                    }
+                        });
+            }
+        }
+
         return true;
     }
     return false;