10 files changed, 393 insertions, 76 deletions

diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt
index b06718f..a97c7f9 100644
--- a/common/CMakeLists.txt
+++ b/common/CMakeLists.txt
@@ -29,6 +29,7 @@ set(command_SRCS
     clientapi.cpp
     commands.cpp
     console.cpp
+    facade.cpp
     pipeline.cpp
     domainadaptor.cpp
     resource.cpp
diff --git a/common/clientapi.h b/common/clientapi.h
index 22448b3..c2b9493 100644
--- a/common/clientapi.h
+++ b/common/clientapi.h
@@ -49,17 +49,44 @@ namespace async {
     */
     template<class T>
     class ResultProvider {
-    public:
-        //Called from worker thread
-        void add(const T &value)
+    private:
+        void callInMainThreadOnEmitter(void (ResultEmitter<T>::*f)())
         {
             //We use the eventloop to call the addHandler directly from the main eventloop.
             //That way the result emitter implementation doesn't have to care about threadsafety at all.
             //The alternative would be to make all handlers of the emitter threadsafe.
-            auto emitter = mResultEmitter;
-            mResultEmitter->mThreadBoundary.callInMainThread([emitter, value]() {
-                if (emitter) {
-                    emitter->addHandler(value);
+            if (auto emitter = mResultEmitter.toStrongRef()) {
+                auto weakEmitter = mResultEmitter;
+                //We don't want to keep the emitter alive here, so we only capture a weak reference
+                emitter->mThreadBoundary.callInMainThread([weakEmitter, f]() {
+                    if (auto strongRef = weakEmitter.toStrongRef()) {
+                        (strongRef.data()->*f)();
+                    }
+                });
+            }
+        }
+
+        void callInMainThreadOnEmitter(const std::function<void()> &f)
+        {
+            //We use the eventloop to call the addHandler directly from the main eventloop.
+            //That way the result emitter implementation doesn't have to care about threadsafety at all.
+            //The alternative would be to make all handlers of the emitter threadsafe.
+            if (auto emitter = mResultEmitter.toStrongRef()) {
+                emitter->mThreadBoundary.callInMainThread([f]() {
+                    f();
+                });
+            }
+        }
+
+    public:
+        //Called from worker thread
+        void add(const T &value)
+        {
+            //Because I don't know how to use bind
+            auto weakEmitter = mResultEmitter;
+            callInMainThreadOnEmitter([weakEmitter, value](){
+                if (auto strongRef = weakEmitter.toStrongRef()) {
+                    strongRef->addHandler(value);
                 }
             });
         }
@@ -67,25 +94,39 @@ namespace async {
         //Called from worker thread
         void complete()
         {
-            auto emitter = mResultEmitter;
-            mResultEmitter->mThreadBoundary.callInMainThread([emitter]() {
-                if (emitter) {
-                    emitter->completeHandler();
-                }
-            });
+            callInMainThreadOnEmitter(&ResultEmitter<T>::complete);
         }
 
+        void clear()
+        {
+            callInMainThreadOnEmitter(&ResultEmitter<T>::clear);
+        }
+
+
         QSharedPointer<ResultEmitter<T> > emitter()
         {
             if (!mResultEmitter) {
-                mResultEmitter = QSharedPointer<ResultEmitter<T> >(new ResultEmitter<T>());
+                //We have to go over a separate var and return that, otherwise we'd delete the emitter immediately again
+                auto sharedPtr = QSharedPointer<ResultEmitter<T> >::create();
+                mResultEmitter = sharedPtr;
+                return sharedPtr;
             }
 
-            return mResultEmitter;
+            return mResultEmitter.toStrongRef();
+        }
+
+        /**
+         * For lifetimemanagement only.
+         * We keep the runner alive as long as the result provider exists.
+         */
+        void setQueryRunner(const QSharedPointer<QObject> &runner)
+        {
+            mQueryRunner = runner;
         }
 
     private:
-        QSharedPointer<ResultEmitter<T> > mResultEmitter;
+        QWeakPointer<ResultEmitter<T> > mResultEmitter;
+        QSharedPointer<QObject> mQueryRunner;
     };
 
     /*
@@ -99,7 +140,6 @@ namespace async {
     * * build async interfaces with signals
     * * build sync interfaces that block when accessing the value
     *
-    * TODO: This should probably be merged with daniels futurebase used in Async
     */
     template<class DomainType>
     class ResultEmitter {
@@ -113,51 +153,36 @@ namespace async {
         {
             completeHandler = handler;
         }
+        void onClear(const std::function<void(void)> &handler)
+        {
+            clearHandler = handler;
+        }
 
-    private:
-        friend class ResultProvider<DomainType>;
-        std::function<void(const DomainType&)> addHandler;
-        // std::function<void(const T&)> removeHandler;
-        std::function<void(void)> completeHandler;
-        ThreadBoundary mThreadBoundary;
-    };
-
+        void add(const DomainType &value)
+        {
+            addHandler(value);
+        }
 
-    /*
-     * A result set specialization that provides a syncronous list
-     */
-    template<class T>
-    class SyncListResult : public QList<T> {
-    public:
-        SyncListResult(const QSharedPointer<ResultEmitter<T> > &emitter)
-            :QList<T>(),
-            mComplete(false),
-            mEmitter(emitter)
+        void complete()
         {
-            emitter->onAdded([this](const T &value) {
-                this->append(value);
-            });
-            emitter->onComplete([this]() {
-                mComplete = true;
-                auto loop = mWaitLoop.toStrongRef();
-                if (loop) {
-                    loop->quit();
-                }
-            });
+            completeHandler();
         }
 
-        void exec()
+        void clear()
         {
-            auto loop = QSharedPointer<QEventLoop>::create();
-            mWaitLoop = loop;
-            loop->exec(QEventLoop::ExcludeUserInputEvents);
+            clearHandler();
         }
 
     private:
-        bool mComplete;
-        QWeakPointer<QEventLoop> mWaitLoop;
-        QSharedPointer<ResultEmitter<T> > mEmitter;
+        friend class ResultProvider<DomainType>;
+
+        std::function<void(const DomainType&)> addHandler;
+        // std::function<void(const T&)> removeHandler;
+        std::function<void(void)> completeHandler;
+        std::function<void(void)> clearHandler;
+        ThreadBoundary mThreadBoundary;
     };
+
 }
 
 namespace Akonadi2 {
@@ -307,7 +332,7 @@ using namespace async;
 class Query
 {
 public:
-    Query() : syncOnDemand(true), processAll(false) {}
+    Query() : syncOnDemand(true), processAll(false), liveQuery(false) {}
     //Could also be a propertyFilter
     QByteArrayList resources;
     //Could also be a propertyFilter
@@ -318,6 +343,8 @@ public:
     QSet<QByteArray> requestedProperties;
     bool syncOnDemand;
     bool processAll;
+    //If live query is false, this query will not continuously be updated
+    bool liveQuery;
 };
 
 
@@ -337,7 +364,9 @@ public:
     virtual Async::Job<void> create(const DomainType &domainObject) = 0;
     virtual Async::Job<void> modify(const DomainType &domainObject) = 0;
     virtual Async::Job<void> remove(const DomainType &domainObject) = 0;
-    virtual Async::Job<void> load(const Query &query, const std::function<void(const typename DomainType::Ptr &)> &resultCallback) = 0;
+    //TODO remove from public API
+    virtual Async::Job<qint64> load(const Query &query, const std::function<void(const typename DomainType::Ptr &)> &resultCallback) = 0;
+    virtual Async::Job<void> load(const Query &query, const QSharedPointer<ResultProvider<typename DomainType::Ptr> > &resultProvider) { return Async::null<void>(); };
 };
 
 
@@ -349,6 +378,8 @@ public:
 
 class FacadeFactory {
 public:
+    typedef std::function<void*(bool &externallyManaged)> FactoryFunction;
+
     //FIXME: proper singleton implementation
     static FacadeFactory &instance()
     {
@@ -365,7 +396,7 @@ public:
     void registerFacade(const QByteArray &resource)
     {
         const QByteArray typeName = ApplicationDomain::getTypeName<DomainType>();
-        mFacadeRegistry.insert(key(resource, typeName), [](){ return new Facade; });
+        mFacadeRegistry.insert(key(resource, typeName), [](bool &externallyManaged){ return new Facade; });
     }
 
     /*
@@ -375,29 +406,51 @@ public:
      * The facade factory takes ovnership of the pointer and typically deletes the instance via shared pointer.
      * Supplied factory functions should therefore always return a new pointer (i.e. via clone())
      *
-     * FIXME the factory function should really be returning QSharedPointer<void>, which doesn't work (std::shared_pointer<void> would though). That way i.e. a test could keep the object alive until it's done.
+     * FIXME the factory function should really be returning QSharedPointer<void>, which doesn't work (std::shared_pointer<void> would though). That way i.e. a test could keep the object alive until it's done. As a workaround the factory function can define wether it manages the lifetime of the facade itself.
      */
     template<class DomainType, class Facade>
-    void registerFacade(const QByteArray &resource, const std::function<void*(void)> &customFactoryFunction)
+    void registerFacade(const QByteArray &resource, const FactoryFunction &customFactoryFunction)
     {
         const QByteArray typeName = ApplicationDomain::getTypeName<DomainType>();
         mFacadeRegistry.insert(key(resource, typeName), customFactoryFunction);
     }
 
+    /*
+     * Can be used to clear the factory.
+     *
+     * Primarily for testing.
+     */
+    void resetFactory()
+    {
+        mFacadeRegistry.clear();
+    }
+
+    static void doNothingDeleter(void *)
+    {
+        qWarning() << "Do nothing";
+    }
+
     template<class DomainType>
     QSharedPointer<StoreFacade<DomainType> > getFacade(const QByteArray &resource)
     {
         const QByteArray typeName = ApplicationDomain::getTypeName<DomainType>();
         auto factoryFunction = mFacadeRegistry.value(key(resource, typeName));
         if (factoryFunction) {
-            return QSharedPointer<StoreFacade<DomainType> >(static_cast<StoreFacade<DomainType>* >(factoryFunction()));
+            bool externallyManaged = false;
+            auto ptr = static_cast<StoreFacade<DomainType>* >(factoryFunction(externallyManaged));
+            if (externallyManaged) {
+                //Allows tests to manage the lifetime of injected facades themselves
+                return QSharedPointer<StoreFacade<DomainType> >(ptr, doNothingDeleter);
+            } else {
+                return QSharedPointer<StoreFacade<DomainType> >(ptr);
+            }
         }
         qWarning() << "Failed to find facade for resource: " << resource << " and type: " << typeName;
         return QSharedPointer<StoreFacade<DomainType> >();
     }
 
 private:
-    QHash<QByteArray, std::function<void*(void)> > mFacadeRegistry;
+    QHash<QByteArray, FactoryFunction> mFacadeRegistry;
 };
 
 /**
@@ -416,7 +469,7 @@ public:
     template <class DomainType>
     static QSharedPointer<ResultEmitter<typename DomainType::Ptr> > load(Query query)
     {
-        QSharedPointer<ResultProvider<typename DomainType::Ptr> > resultSet(new ResultProvider<typename DomainType::Ptr>);
+        auto resultSet = QSharedPointer<ResultProvider<typename DomainType::Ptr> >::create();
 
         //Execute the search in a thread.
         //We must guarantee that the emitter is returned before the first result is emitted.
@@ -428,15 +481,15 @@ public:
             Async::Job<void> job = Async::null<void>();
             for(const QByteArray &resource : query.resources) {
                 auto facade = FacadeFactory::instance().getFacade<DomainType>(resource);
-                //We have to bind an instance to the function callback. Since we use a shared pointer this keeps the result provider instance (and thus also the emitter) alive.
-                std::function<void(const typename DomainType::Ptr &)> addCallback = std::bind(&ResultProvider<typename DomainType::Ptr>::add, resultSet, std::placeholders::_1);
 
                 // TODO The following is a necessary hack to keep the facade alive.
                 // Otherwise this would reduce to:
                 // job = job.then(facade->load(query, addCallback));
                 // We somehow have to guarantee that the facade remains valid for the duration of the job
-                job = job.then<void>([facade, query, addCallback](Async::Future<void> &future) {
-                    Async::Job<void> j = facade->load(query, addCallback);
+                // TODO: Use one result set per facade, and merge the results separately
+                // resultSet->addSubset(facade->query(query));
+                job = job.then<void>([facade, query, resultSet](Async::Future<void> &future) {
+                    Async::Job<void> j = facade->load(query, resultSet);
                     j.then<void>([&future, facade](Async::Future<void> &f) {
                         future.setFinished();
                         f.setFinished();
diff --git a/common/facade.cpp b/common/facade.cpp
new file mode 100644
index 0000000..e51b32a
--- /dev/null
+++ b/common/facade.cpp
@@ -0,0 +1,20 @@
+/*
+ * Copyright (C) 2015 Christian Mollekopf <chrigi_1@fastmail.fm>
+ *
+ *   This program is free software; you can redistribute it and/or modify
+ *   it under the terms of the GNU General Public License as published by
+ *   the Free Software Foundation; either version 2 of the License, or
+ *   (at your option) any later version.
+ *
+ *   This program is distributed in the hope that it will be useful,
+ *   but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *   GNU General Public License for more details.
+ *
+ *   You should have received a copy of the GNU General Public License
+ *   along with this program; if not, write to the
+ *   Free Software Foundation, Inc.,
+ *   51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA.
+ */
+
+#include "facade.h"
diff --git a/common/facade.h b/common/facade.h
index 98bcb38..f9b5a83 100644
--- a/common/facade.h
+++ b/common/facade.h
@@ -30,6 +30,55 @@
 #include "domainadaptor.h"
 #include "entitybuffer.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.
+ */
+class QueryRunner : public QObject
+{
+    Q_OBJECT
+public:
+    typedef std::function<Async::Job<qint64>(qint64 oldRevision, qint64 newRevision)> QueryFunction;
+
+    QueryRunner(const Akonadi2::Query &query) : mLatestRevision(0) {};
+    /**
+     * Starts query
+     */
+    Async::Job<void> run(qint64 newRevision = 0)
+    {
+        //TODO: JOBAPI: that last empty .then should not be necessary
+        return queryFunction(mLatestRevision, newRevision).then<void, qint64>([this](qint64 revision) {
+            mLatestRevision = revision;
+        }).then<void>([](){});
+    }
+
+    /**
+     *
+     */
+    void setQuery(const QueryFunction &query)
+    {
+        queryFunction = query;
+    }
+
+public slots:
+    /**
+     * Rerun query with new revision
+     */
+    void revisionChanged(qint64 newRevision)
+    {
+        run(newRevision).exec();
+    }
+
+private:
+    QueryFunction queryFunction;
+    qint64 mLatestRevision;
+};
+
 namespace Akonadi2 {
     class ResourceAccess;
 /**
@@ -80,7 +129,7 @@ protected:
         return Async::null<void>();
     }
 
-private:
+protected:
     //TODO use one resource access instance per application => make static
     QSharedPointer<Akonadi2::ResourceAccess> mResourceAccess;
 };
diff --git a/common/synclistresult.h b/common/synclistresult.h
new file mode 100644
index 0000000..5fa0efd
--- /dev/null
+++ b/common/synclistresult.h
@@ -0,0 +1,54 @@
+#pragma once
+
+#include <QList>
+#include <functional>
+#include <QSharedPointer>
+#include <clientapi.h>
+
+namespace async {
+
+/*
+* A result set specialization that provides a syncronous list.
+*
+* Only for testing purposes.
+*
+* WARNING: The nested eventloop can cause all sorts of trouble. Use only in testing code.
+*/
+template<class T>
+class SyncListResult : public QList<T> {
+public:
+    SyncListResult(const QSharedPointer<ResultEmitter<T> > &emitter)
+        :QList<T>(),
+        mComplete(false),
+        mEmitter(emitter)
+    {
+        emitter->onAdded([this](const T &value) {
+            this->append(value);
+        });
+        emitter->onComplete([this]() {
+            mComplete = true;
+            if (eventLoopAborter) {
+                eventLoopAborter();
+                //Be safe in case of a second invocation of the complete handler
+                eventLoopAborter = std::function<void()>();
+            }
+        });
+        emitter->onClear([this]() {
+            this->clear();
+        });
+    }
+
+    void exec()
+    {
+        QEventLoop eventLoop;
+        eventLoopAborter = [&eventLoop]() { eventLoop.quit(); };
+        eventLoop.exec();
+    }
+
+private:
+    bool mComplete;
+    QSharedPointer<ResultEmitter<T> > mEmitter;
+    std::function<void()> eventLoopAborter;
+};
+
+}
diff --git a/common/test/clientapitest.cpp b/common/test/clientapitest.cpp
index 24b3fb9..789c656 100644
--- a/common/test/clientapitest.cpp
+++ b/common/test/clientapitest.cpp
@@ -3,6 +3,22 @@
 #include <functional>
 
 #include "../clientapi.h"
+#include "../facade.h"
+#include "../synclistresult.h"
+
+class RevisionNotifier : public QObject
+{
+    Q_OBJECT
+public:
+    RevisionNotifier() : QObject() {};
+    void notify(qint64 revision)
+    {
+        emit revisionChanged(revision);
+    }
+
+Q_SIGNALS:
+    void revisionChanged(qint64);
+};
 
 class DummyResourceFacade : public Akonadi2::StoreFacade<Akonadi2::ApplicationDomain::Event>
 {
@@ -11,18 +27,63 @@ public:
     virtual Async::Job<void> create(const Akonadi2::ApplicationDomain::Event &domainObject){ return Async::null<void>(); };
     virtual Async::Job<void> modify(const Akonadi2::ApplicationDomain::Event &domainObject){ return Async::null<void>(); };
     virtual Async::Job<void> remove(const Akonadi2::ApplicationDomain::Event &domainObject){ return Async::null<void>(); };
-    virtual Async::Job<void> load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback)
+    virtual Async::Job<qint64> load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback)
     {
-        return Async::start<void>([this, resultCallback](Async::Future<void> &future) {
+        return Async::start<qint64>([this, resultCallback](Async::Future<qint64> &future) {
             qDebug() << "load called";
             for(const auto &result : results) {
                 resultCallback(result);
             }
+            future.setValue(0);
             future.setFinished();
         });
     }
 
+    Async::Job<void> load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider)
+    {
+        auto runner = QSharedPointer<QueryRunner>::create(query);
+        //The runner only lives as long as the resultProvider
+        resultProvider->setQueryRunner(runner);
+        runner->setQuery([this, resultProvider, query](qint64 oldRevision, qint64 newRevision) -> Async::Job<qint64> {
+            qDebug() << "Creating query for revisions: " << oldRevision << newRevision;
+            return Async::start<qint64>([this, resultProvider, query](Async::Future<qint64> &future) {
+                //TODO only emit changes and don't replace everything
+                resultProvider->clear();
+                //rerun query
+                std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> addCallback = std::bind(&Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr>::add, resultProvider, std::placeholders::_1);
+                load(query, addCallback).then<void, qint64>([resultProvider, &future](qint64 queriedRevision) {
+                    //TODO set revision in result provider?
+                    //TODO update all existing results with new revision
+                    resultProvider->complete();
+                    future.setValue(queriedRevision);
+                    future.setFinished();
+                }).exec();
+            });
+        });
+
+        //Ensure the notification is emitted in the right thread
+        //Otherwise we get crashes as we call revisionChanged from the test.
+        if (!notifier) {
+            notifier.reset(new RevisionNotifier);
+        }
+
+        //TODO somehow disconnect as resultNotifier is destroyed. Otherwise we keep the runner alive forever.
+        if (query.liveQuery) {
+            QObject::connect(notifier.data(), &RevisionNotifier::revisionChanged, [runner](qint64 newRevision) {
+                runner->revisionChanged(newRevision);
+            });
+        }
+
+        return Async::start<void>([runner](Async::Future<void> &future) {
+            runner->run().then<void>([&future]() {
+                //TODO if not live query, destroy runner.
+                future.setFinished();
+            }).exec();
+        });
+    }
+
     QList<Akonadi2::ApplicationDomain::Event::Ptr> results;
+    QSharedPointer<RevisionNotifier> notifier;
 };
 
 class ClientAPITest : public QObject
@@ -30,21 +91,60 @@ class ClientAPITest : public QObject
     Q_OBJECT
 private Q_SLOTS:
 
+    void initTestCase()
+    {
+        Akonadi2::FacadeFactory::instance().resetFactory();
+    }
+
     void testLoad()
     {
         DummyResourceFacade facade;
         facade.results << QSharedPointer<Akonadi2::ApplicationDomain::Event>::create("resource", "id", 0, QSharedPointer<Akonadi2::ApplicationDomain::BufferAdaptor>());
 
-        Akonadi2::FacadeFactory::instance().registerFacade<Akonadi2::ApplicationDomain::Event, DummyResourceFacade>("dummyresource", [facade](){ return new DummyResourceFacade(facade); });
+        Akonadi2::FacadeFactory::instance().registerFacade<Akonadi2::ApplicationDomain::Event, DummyResourceFacade>("dummyresource",
+            [&facade](bool &externallyManaged) {
+                externallyManaged = true;
+                return &facade;
+            }
+        );
 
         Akonadi2::Query query;
         query.resources << "dummyresource";
+        query.liveQuery = false;
 
         async::SyncListResult<Akonadi2::ApplicationDomain::Event::Ptr> result(Akonadi2::Store::load<Akonadi2::ApplicationDomain::Event>(query));
         result.exec();
         QCOMPARE(result.size(), 1);
     }
 
+    void testLiveQuery()
+    {
+        DummyResourceFacade facade;
+        facade.results << QSharedPointer<Akonadi2::ApplicationDomain::Event>::create("resource", "id", 0, QSharedPointer<Akonadi2::ApplicationDomain::BufferAdaptor>());
+
+        Akonadi2::FacadeFactory::instance().registerFacade<Akonadi2::ApplicationDomain::Event, DummyResourceFacade>("dummyresource",
+            [&facade](bool &externallManage){
+                externallManage = true;
+                return &facade;
+            }
+        );
+
+        Akonadi2::Query query;
+        query.resources << "dummyresource";
+        query.liveQuery = true;
+
+        async::SyncListResult<Akonadi2::ApplicationDomain::Event::Ptr> result(Akonadi2::Store::load<Akonadi2::ApplicationDomain::Event>(query));
+        result.exec();
+        QCOMPARE(result.size(), 1);
+
+        //Enter a second result
+        facade.results << QSharedPointer<Akonadi2::ApplicationDomain::Event>::create("resource", "id2", 0, QSharedPointer<Akonadi2::ApplicationDomain::BufferAdaptor>());
+        qWarning() << &facade;
+        QVERIFY(facade.notifier);
+        facade.notifier->revisionChanged(2);
+        QTRY_COMPARE(result.size(), 2);
+    }
+
 };
 
 QTEST_MAIN(ClientAPITest)
diff --git a/dummyresource/facade.cpp b/dummyresource/facade.cpp
index 1477fcf..f603c56 100644
--- a/dummyresource/facade.cpp
+++ b/dummyresource/facade.cpp
@@ -147,15 +147,51 @@ void DummyResourceFacade::readValue(QSharedPointer<Akonadi2::Storage> storage, c
     });
 }
 
-Async::Job<void> DummyResourceFacade::load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback)
+Async::Job<void> DummyResourceFacade::load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider)
 {
-    return synchronizeResource(query.syncOnDemand, query.processAll).then<void>([=](Async::Future<void> &future) {
+    auto runner = QSharedPointer<QueryRunner>::create(query);
+    //The runner only lives as long as the resultProvider
+    resultProvider->setQueryRunner(runner);
+    runner->setQuery([this, resultProvider, query](qint64 oldRevision, qint64 newRevision) -> Async::Job<qint64> {
+        return Async::start<qint64>([this, resultProvider, query](Async::Future<qint64> &future) {
+            //TODO only emit changes and don't replace everything
+            resultProvider->clear();
+            //rerun query
+            std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> addCallback = std::bind(&Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr>::add, resultProvider, std::placeholders::_1);
+            load(query, addCallback).then<void, qint64>([resultProvider, &future](qint64 queriedRevision) {
+                //TODO set revision in result provider?
+                //TODO update all existing results with new revision
+                resultProvider->complete();
+                future.setValue(queriedRevision);
+                future.setFinished();
+            }).exec();
+        });
+    });
+
+    auto resourceAccess = mResourceAccess;
+    //TODO we need to somehow disconnect this
+    QObject::connect(resourceAccess.data(), &Akonadi2::ResourceAccess::revisionChanged, [runner](qint64 newRevision) {
+        runner->revisionChanged(newRevision);
+    });
+
+    return Async::start<void>([runner](Async::Future<void> &future) {
+        runner->run().then<void>([&future]() {
+            //TODO if not live query, destroy runner.
+            future.setFinished();
+        }).exec();
+    });
+}
+
+Async::Job<qint64> DummyResourceFacade::load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback)
+{
+    return synchronizeResource(query.syncOnDemand, query.processAll).then<qint64>([=](Async::Future<qint64> &future) {
         //Now that the sync is complete we can execute the query
         const auto preparedQuery = prepareQuery(query);
 
         auto storage = QSharedPointer<Akonadi2::Storage>::create(Akonadi2::Store::storageLocation(), "org.kde.dummy");
 
-        //TODO use transaction over full query and record store revision. We'll need it to update the query.
+        storage->startTransaction(Akonadi2::Storage::ReadOnly);
+        const qint64 revision = storage->maxRevision();
 
         //Index lookups
         QVector<QByteArray> keys;
@@ -177,6 +213,8 @@ Async::Job<void> DummyResourceFacade::load(const Akonadi2::Query &query, const s
                 readValue(storage, key, resultCallback, preparedQuery);
             }
         }
+        storage->abortTransaction();
+        future.setValue(revision);
         future.setFinished();
     });
 }
diff --git a/dummyresource/facade.h b/dummyresource/facade.h
index 37ed81d..3ddfe15 100644
--- a/dummyresource/facade.h
+++ b/dummyresource/facade.h
@@ -34,10 +34,11 @@ class DummyResourceFacade : public Akonadi2::GenericFacade<Akonadi2::Application
 public:
     DummyResourceFacade();
     virtual ~DummyResourceFacade();
-    virtual Async::Job<void> create(const Akonadi2::ApplicationDomain::Event &domainObject);
-    virtual Async::Job<void> modify(const Akonadi2::ApplicationDomain::Event &domainObject);
-    virtual Async::Job<void> remove(const Akonadi2::ApplicationDomain::Event &domainObject);
-    virtual Async::Job<void> load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback);
+    Async::Job<void> create(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE;
+    Async::Job<void> modify(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE;
+    Async::Job<void> remove(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE;
+    Async::Job<qint64> load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback) Q_DECL_OVERRIDE;
+    Async::Job<void> load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider) Q_DECL_OVERRIDE;
 
 private:
     void readValue(QSharedPointer<Akonadi2::Storage> storage, const QByteArray &key, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback, std::function<bool(const std::string &key, DummyCalendar::DummyEvent const *buffer, Akonadi2::ApplicationDomain::Buffer::Event const *local)>);
diff --git a/tests/dummyresourcebenchmark.cpp b/tests/dummyresourcebenchmark.cpp
index d83cb70..e350747 100644
--- a/tests/dummyresourcebenchmark.cpp
+++ b/tests/dummyresourcebenchmark.cpp
@@ -7,6 +7,7 @@
 #include "clientapi.h"
 #include "commands.h"
 #include "entitybuffer.h"
+#include "synclistresult.h"
 
 #include "event_generated.h"
 #include "entity_generated.h"
diff --git a/tests/dummyresourcetest.cpp b/tests/dummyresourcetest.cpp
index 5fed7cd..9523e5d 100644
--- a/tests/dummyresourcetest.cpp
+++ b/tests/dummyresourcetest.cpp
@@ -2,13 +2,13 @@
 
 #include <QString>
 
-// #include "dummycalendar_generated.h"
 #include "event_generated.h"
 #include "entity_generated.h"
 #include "metadata_generated.h"
 #include "createentity_generated.h"
 #include "dummyresource/resourcefactory.h"
 #include "clientapi.h"
+#include "synclistresult.h"
 #include "commands.h"
 #include "entitybuffer.h"